Coverage for pytest_jsonschema_snapshot/plugin.py: 19%

1from pathlib import Path 

2from typing import Dict, Generator, Optional 

3 

4import pytest 

5from jsonschema_diff import ConfigMaker, JsonSchemaDiff 

6from jsonschema_diff.color import HighlighterPipeline 

7from jsonschema_diff.color.stages import ( 

8 MonoLinesHighlighter, 

9 PathHighlighter, 

10 ReplaceGenericHighlighter, 

11) 

12 

13from .core import SchemaShot 

14from .stats import GLOBAL_STATS, SchemaStats 

15 

16# Global storage of SchemaShot instances for different directories 

17_schema_managers: Dict[Path, SchemaShot] = {} 

18 

19 

20def pytest_addoption(parser: pytest.Parser) -> None: 

21 """Adds --schema-update option to pytest.""" 

22 parser.addoption( 

23 "--schema-update", 

24 action="store_true", 

25 help=( 

26 "Augmenting mode for updating schemas. " 

27 "If something is valid for the old schema, then it is valid " 

28 "for the new one (and vice versa)." 

29 ), 

30 ) 

31 parser.addoption( 

32 "--schema-reset", 

33 action="store_true", 

34 help="New schema does not take into account the old one during update.", 

35 ) 

36 parser.addoption( 

37 "--save-original", 

38 action="store_true", 

39 help="Save original JSON alongside schema (same name, but without `.schema` prefix)", 

40 ) 

41 parser.addoption( 

42 "--jsss-debug", 

43 action="store_true", 

44 help="Show internal exception stack (stops hiding them)", 

45 ) 

46 

47 parser.addoption( 

48 "--without-delete", 

49 action="store_true", 

50 help="Disable deleting unused schemas", 

51 ) 

52 parser.addoption( 

53 "--without-update", 

54 action="store_true", 

55 help="Disable updating schemas", 

56 ) 

57 parser.addoption( 

58 "--without-add", 

59 action="store_true", 

60 help="Disable adding new schemas", 

61 ) 

62 

63 parser.addini( 

64 "jsss_dir", 

65 default="__snapshots__", 

66 help="Directory for storing schemas (default: __snapshots__)", 

67 ) 

68 parser.addini( 

69 "jsss_callable_regex", 

70 default="{class_method=.}", 

71 help="Regex for saving callable part of path", 

72 ) 

73 parser.addini( 

74 "jsss_format_mode", 

75 default="on", 

76 help="Format mode: 'on' (annotate and validate), 'safe' (annotate), 'off' (disable)", 

77 ) 

78 

79 

80@pytest.fixture(scope="function") 

81def schemashot(request: pytest.FixtureRequest) -> Generator[SchemaShot, None, None]: 

82 """ 

83 Fixture providing a SchemaShot instance and gathering used schemas. 

84 """ 

85 

86 # Получаем путь к тестовому файлу 

87 test_path = Path(request.node.path if hasattr(request.node, "path") else request.node.fspath) 

88 root_dir = test_path.parent 

89 

90 update_mode = bool(request.config.getoption("--schema-update")) 

91 reset_mode = bool(request.config.getoption("--schema-reset")) 

92 if update_mode and reset_mode: 

93 raise ValueError("Options --schema-update and --schema-reset are mutually exclusive.") 

94 

95 save_original = bool(request.config.getoption("--save-original")) 

96 debug_mode = bool(request.config.getoption("--jsss-debug")) 

97 

98 actions = { 

99 "delete": not request.config.getoption("--without-delete"), 

100 "update": not request.config.getoption("--without-update"), 

101 "add": not request.config.getoption("--without-add"), 

102 } 

103 

104 # Получаем настраиваемую директорию для схем 

105 schema_dir_name = str(request.config.getini("jsss_dir")) 

106 callable_regex = str(request.config.getini("jsss_callable_regex")) 

107 format_mode = str(request.config.getini("jsss_format_mode")).lower() 

108 # examples_limit = int(request.config.getini("jsss_examples_limit")) 

109 

110 differ = JsonSchemaDiff( 

111 ConfigMaker.make(), 

112 HighlighterPipeline( 

113 [MonoLinesHighlighter(), PathHighlighter(), ReplaceGenericHighlighter()] 

114 ), 

115 ) 

116 

117 # Создаем или получаем экземпляр SchemaShot для этой директории 

118 if root_dir not in _schema_managers: 

119 _schema_managers[root_dir] = SchemaShot( 

120 root_dir, 

121 differ, 

122 callable_regex, 

123 format_mode, 

124 # examples_limit, 

125 update_mode, 

126 reset_mode, 

127 actions, 

128 save_original, 

129 debug_mode, 

130 schema_dir_name, 

131 ) 

132 

133 # Создаем локальный экземпляр для теста 

134 yield _schema_managers[root_dir] 

135 

136 

137@pytest.hookimpl(trylast=True) 

138def pytest_unconfigure(config: pytest.Config) -> None: 

139 """ 

140 Hook that runs after all tests have finished. 

141 Clears global variables. 

142 """ 

143 global GLOBAL_STATS 

144 

145 # Clear the dictionary 

146 _schema_managers.clear() 

147 # Reset stats for next run 

148 GLOBAL_STATS = SchemaStats() 

149 

150 

151@pytest.hookimpl(trylast=True) 

152def pytest_terminal_summary(terminalreporter: pytest.TerminalReporter, exitstatus: int) -> None: 

153 """ 

154 Adds a summary about schemas to the final pytest report in the terminal. 

155 """ 

156 # Выполняем cleanup перед показом summary 

157 if _schema_managers: 

158 

159 def get_opt(opt: str) -> bool: 

160 return bool(terminalreporter.config.getoption(opt)) 

161 

162 update_mode = get_opt("--schema-update") 

163 

164 actions = { 

165 "delete": not get_opt("--without-delete"), 

166 "update": not get_opt("--without-update"), 

167 "add": not get_opt("--without-add"), 

168 } 

169 

170 # Вызываем метод очистки неиспользованных схем для каждого экземпляра 

171 for _root_dir, manager in _schema_managers.items(): 

172 cleanup_unused_schemas(manager, update_mode, actions, GLOBAL_STATS) 

173 

174 # Используем новую функцию для вывода статистики 

175 update_mode = bool(terminalreporter.config.getoption("--schema-update")) 

176 GLOBAL_STATS.print_summary(terminalreporter, update_mode) 

177 

178 

179def cleanup_unused_schemas( 

180 manager: SchemaShot, 

181 update_mode: bool, 

182 actions: dict[str, bool], 

183 stats: Optional[SchemaStats] = None, 

184) -> None: 

185 """ 

186 Deletes unused schemas in update mode and collects statistics. 

187 Additionally, deletes the pair file `<name>.json` if it exists. 

188 

189 Args: 

190 manager: SchemaShot instance 

191 update_mode: Update mode 

192 stats: Optional object for collecting statistics 

193 """ 

194 # Если директория снимков не существует, ничего не делаем 

195 if not manager.snapshot_dir.exists(): 

196 return 

197 

198 # Перебираем все файлы схем 

199 all_schemas = list(manager.snapshot_dir.glob("*.schema.json")) 

200 

201 for schema_file in all_schemas: 

202 if schema_file.name not in manager.used_schemas: 

203 if update_mode and actions.get("delete"): 

204 try: 

205 # Удаляем саму схему 

206 schema_file.unlink() 

207 if stats: 

208 stats.add_deleted(schema_file.name) 

209 

210 # Пытаемся удалить парный JSON: <name>.json 

211 # Преобразуем "<name>.schema.json" -> "<name>.json" 

212 base_name = schema_file.name[: -len(".schema.json")] 

213 paired_json = schema_file.with_name(f"{base_name}.json") 

214 if paired_json.exists(): 

215 try: 

216 paired_json.unlink() 

217 if stats: 

218 stats.add_deleted(paired_json.name) 

219 except OSError as e: 

220 manager.logger.warning( 

221 f"Failed to delete paired JSON for {schema_file.name}: {e}" 

222 ) 

223 except Exception as e: 

224 manager.logger.error( 

225 f"Unexpected error deleting paired JSON for {schema_file.name}: {e}" 

226 ) 

227 

228 except OSError as e: 

229 # Логируем ошибки удаления, но не прерываем работу 

230 manager.logger.warning( 

231 f"Failed to delete unused schema {schema_file.name}: {e}" 

232 ) 

233 except Exception as e: 

234 # Неожиданные ошибки тоже логируем 

235 manager.logger.error( 

236 f"Unexpected error deleting schema {schema_file.name}: {e}" 

237 ) 

238 else: 

239 if stats: 

240 stats.add_unused(schema_file.name)