Coverage for pytest_jsonschema_snapshot/stats.py: 63%

1""" 

2Module for collecting and displaying statistics about schemas. 

3""" 

4 

5from typing import Dict, Generator, List, Optional 

6 

7import pytest 

8 

9 

10class SchemaStats: 

11 """Class for collecting and displaying statistics about schemas""" 

12 

13 def __init__(self) -> None: 

14 self.created: List[str] = [] 

15 self.updated: List[str] = [] 

16 self.updated_diffs: Dict[str, str] = {} # schema_name -> diff 

17 self.uncommitted: List[str] = [] # New category for uncommitted changes 

18 self.uncommitted_diffs: Dict[str, str] = {} # schema_name -> diff 

19 self.deleted: List[str] = [] 

20 self.unused: List[str] = [] 

21 

22 def add_created(self, schema_name: str) -> None: 

23 """Adds created schema""" 

24 self.created.append(schema_name) 

25 

26 def add_updated(self, schema_name: str, diff: Optional[str] = None) -> None: 

27 """Adds updated schema""" 

28 # Generate diff if both schemas are provided 

29 if diff and diff.strip(): 

30 self.updated.append(schema_name) 

31 self.updated_diffs[schema_name] = diff 

32 else: 

33 # If schemas are not provided, assume it was an update 

34 self.updated.append(schema_name) 

35 

36 def add_uncommitted(self, schema_name: str, diff: Optional[str] = None) -> None: 

37 """Adds schema with uncommitted changes""" 

38 # Add only if there are real changes 

39 if diff and diff.strip(): 

40 self.uncommitted.append(schema_name) 

41 self.uncommitted_diffs[schema_name] = diff 

42 

43 def add_deleted(self, schema_name: str) -> None: 

44 """Adds deleted schema""" 

45 self.deleted.append(schema_name) 

46 

47 def add_unused(self, schema_name: str) -> None: 

48 """Adds unused schema""" 

49 self.unused.append(schema_name) 

50 

51 def has_changes(self) -> bool: 

52 """Returns True if any schema has changes""" 

53 return bool(self.created or self.updated or self.deleted) 

54 

55 def has_any_info(self) -> bool: 

56 """Is there any information about schemas""" 

57 return bool(self.created or self.updated or self.deleted or self.unused or self.uncommitted) 

58 

59 def __str__(self) -> str: 

60 parts = [] 

61 if self.created: 

62 parts.append( 

63 f"Created schemas ({len(self.created)}): " 

64 + ", ".join(f"`{s}`" for s in self.created) 

65 ) 

66 if self.updated: 

67 parts.append( 

68 f"Updated schemas ({len(self.updated)}): " 

69 + ", ".join(f"`{s}`" for s in self.updated) 

70 ) 

71 if self.deleted: 

72 parts.append( 

73 f"Deleted schemas ({len(self.deleted)}): " 

74 + ", ".join(f"`{s}`" for s in self.deleted) 

75 ) 

76 if self.unused: 

77 parts.append( 

78 f"Unused schemas ({len(self.unused)}): " + ", ".join(f"`{s}`" for s in self.unused) 

79 ) 

80 

81 return "\n".join(parts) 

82 

83 def print_summary(self, terminalreporter: pytest.TerminalReporter, update_mode: bool) -> None: 

84 """ 

85 Prints schema summary to pytest terminal output. 

86 Pairs of "<name>.schema.json" + "<name>.json" are merged into one line: 

87 "<name>.schema.json + original" (if original is present). 

88 """ 

89 

90 def _iter_merged(names: List[str]) -> Generator[tuple[str, Optional[str]], None, None]: 

91 """ 

92 Iterates over (display, schema_key): 

93 - display: string to display (may have " + original") 

94 - schema_key: file name of the schema (<name>.schema.json) to find diffs, 

95 or None if it's not a schema. 

96 Preserves the original list order: merging happens at .schema.json 

97 position; single .json outputs are left as is. 

98 """ 

99 names = list(names) # порядок важен 

100 schema_sfx = ".schema.json" 

101 json_sfx = ".json" 

102 

103 # множество баз, где имеются схемы/оригиналы 

104 bases_with_schema = {n[: -len(schema_sfx)] for n in names if n.endswith(schema_sfx)} 

105 bases_with_original = { 

106 n[: -len(json_sfx)] 

107 for n in names 

108 if n.endswith(json_sfx) and not n.endswith(schema_sfx) 

109 } 

110 

111 for n in names: 

112 if n.endswith(schema_sfx): 

113 base = n[: -len(schema_sfx)] 

114 if base in bases_with_original: 

115 yield f"{n} + original", n # display, schema_key 

116 else: 

117 yield n, n 

118 elif n.endswith(json_sfx) and not n.endswith(schema_sfx): 

119 base = n[: -len(json_sfx)] 

120 # если есть парная схема — .json не выводим отдельно 

121 if base in bases_with_schema: 

122 continue 

123 yield n, None 

124 else: 

125 # на всякий случай — прочие имена 

126 yield n, n 

127 

128 if not self.has_any_info(): 

129 return 

130 

131 terminalreporter.write_sep("=", "Schema Summary") 

132 

133 # Created 

134 if self.created: 

135 terminalreporter.write_line(f"Created schemas ({len(self.created)}):", green=True) 

136 for display, _key in _iter_merged(self.created): 

137 terminalreporter.write_line(f" - {display}", green=True) 

138 

139 # Updated 

140 if self.updated: 

141 terminalreporter.write_line(f"Updated schemas ({len(self.updated)}):", yellow=True) 

142 for display, key in _iter_merged(self.updated): 

143 terminalreporter.write_line(f" - {display}", yellow=True) 

144 # Показываем diff, если он есть под ключом схемы (.schema.json) 

145 if key and key in self.updated_diffs: 

146 terminalreporter.write_line(" Changes:", yellow=True) 

147 for line in self.updated_diffs[key].split("\n"): 

148 if line.strip(): 

149 terminalreporter.write_line(f" {line}") 

150 terminalreporter.write_line("") # разделение 

151 elif key: 

152 terminalreporter.write_line( 

153 " (Schema unchanged - no differences detected)", cyan=True 

154 ) 

155 

156 # Uncommitted 

157 if self.uncommitted: 

158 terminalreporter.write_line( 

159 f"Uncommitted minor updates ({len(self.uncommitted)}):", bold=True 

160 ) 

161 for display, key in _iter_merged(self.uncommitted): 

162 terminalreporter.write_line(f" - {display}", cyan=True) 

163 if key and key in self.uncommitted_diffs: 

164 terminalreporter.write_line(" Detected changes:", cyan=True) 

165 for line in self.uncommitted_diffs[key].split("\n"): 

166 if line.strip(): 

167 terminalreporter.write_line(f" {line}") 

168 terminalreporter.write_line("") # разделение 

169 terminalreporter.write_line("Use --schema-update to commit these changes", cyan=True) 

170 

171 # Deleted 

172 if self.deleted: 

173 terminalreporter.write_line(f"Deleted schemas ({len(self.deleted)}):", red=True) 

174 for display, _key in _iter_merged(self.deleted): 

175 terminalreporter.write_line(f" - {display}", red=True) 

176 

177 # Unused (только если не update_mode) 

178 if self.unused and not update_mode: 

179 terminalreporter.write_line(f"Unused schemas ({len(self.unused)}):") 

180 for display, _key in _iter_merged(self.unused): 

181 terminalreporter.write_line(f" - {display}") 

182 terminalreporter.write_line("Use --schema-update to delete unused schemas", yellow=True) 

183 

184 

185GLOBAL_STATS = SchemaStats()