Coverage for pytest_jsonschema_snapshot / stats.py: 60%

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 _iter_schemas(self, names: List[str]) -> Generator[tuple[str, Optional[str]], None, None]: 

84 """ 

85 Iterates over schema displays: (display, schema_key) 

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

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

88 or None if it's not a schema. 

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

90 position; skips .json if paired with schema. 

91 """ 

92 names = list(names) # order matters 

93 schema_sfx = ".schema.json" 

94 json_sfx = ".json" 

95 

96 # sets of bases 

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

98 bases_with_original = { 

99 n[: -len(json_sfx)] 

100 for n in names 

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

102 } 

103 

104 for n in names: 

105 if n.endswith(schema_sfx): 

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

107 if base in bases_with_original: 

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

109 else: 

110 yield n, n 

111 # if .json, skip if paired 

112 # if other, yield n, n (but assume all are .json or .schema.json) 

113 

114 def _iter_only_originals(self, names: List[str]) -> Generator[str, None, None]: 

115 """ 

116 Iterates over only unpaired .json files, in the order they appear. 

117 """ 

118 names = list(names) # order matters 

119 schema_sfx = ".schema.json" 

120 json_sfx = ".json" 

121 

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

123 

124 for n in names: 

125 if n.endswith(json_sfx) and not n.endswith(schema_sfx): 

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

127 if base not in bases_with_schema: 

128 yield n 

129 

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

131 """ 

132 Prints schema summary to pytest terminal output. 

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

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

135 Unpaired .json are shown in separate "only originals" sections. 

136 """ 

137 

138 if not self.has_any_info(): 

139 return 

140 

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

142 

143 # Created 

144 if self.created: 

145 schemas = list(self._iter_schemas(self.created)) 

146 only_originals = list(self._iter_only_originals(self.created)) 

147 if schemas: 

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

149 for display, _key in schemas: 

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

151 if only_originals: 

152 terminalreporter.write_line( 

153 f"Created only originals ({len(only_originals)}):", green=True 

154 ) 

155 for display in only_originals: 

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

157 

158 # Updated 

159 if self.updated: 

160 schemas = list(self._iter_schemas(self.updated)) 

161 only_originals = list(self._iter_only_originals(self.updated)) 

162 if schemas: 

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

164 for display, key in schemas: 

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

166 # Show diff if available for schema 

167 if key and key in self.updated_diffs: 

168 diff = self.updated_diffs[key] 

169 if diff.strip(): 

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

171 for line in diff.split("\n"): 

172 if line.strip(): 

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

174 terminalreporter.write_line("") # separation 

175 else: 

176 terminalreporter.write_line( 

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

178 ) 

179 if only_originals: 

180 terminalreporter.write_line( 

181 f"Updated only originals ({len(only_originals)}):", yellow=True 

182 ) 

183 for display in only_originals: 

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

185 

186 # Uncommitted 

187 if self.uncommitted: 

188 terminalreporter.write_line( 

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

190 ) 

191 for display, key in self._iter_schemas(self.uncommitted): # assuming mostly schemas 

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

193 # Show diff if available 

194 if key and key in self.uncommitted_diffs: 

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

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

197 if line.strip(): 

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

199 terminalreporter.write_line("") # separation 

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

201 

202 # Deleted 

203 if self.deleted: 

204 schemas = list(self._iter_schemas(self.deleted)) 

205 only_originals = list(self._iter_only_originals(self.deleted)) 

206 if schemas: 

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

208 for display, _key in schemas: 

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

210 if only_originals: 

211 terminalreporter.write_line( 

212 f"Deleted only originals ({len(only_originals)}):", red=True 

213 ) 

214 for display in only_originals: 

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

216 

217 # Unused (only if not update_mode) 

218 if self.unused and not update_mode: 

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

220 for display, _key in self._iter_schemas(self.unused): # assuming schemas 

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

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

223 

224 

225GLOBAL_STATS = SchemaStats()