Coverage for pytest_jsonschema_snapshot / plugin.py: 26%

92 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-03-29 08:31 +0000

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 parser.addoption( 

47 "--jsss-ci-cd", 

48 action="store_true", 

49 help="CI/CD mode is incompatible with --schema-reset and --schema-update and implies:\n" 

50 "When a schema mismatch occurs, " 

51 "schemas are created in __snapshots__/ci.cd/*.schema.json / *.json " 

52 "(when --save-original is used)", 

53 ) 

54 

55 parser.addoption( 

56 "--without-delete", 

57 action="store_true", 

58 help="Disable deleting unused schemas", 

59 ) 

60 parser.addoption( 

61 "--without-update", 

62 action="store_true", 

63 help="Disable updating schemas", 

64 ) 

65 parser.addoption( 

66 "--without-add", 

67 action="store_true", 

68 help="Disable adding new schemas", 

69 ) 

70 

71 parser.addini( 

72 "jsss_dir", 

73 default="__snapshots__", 

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

75 ) 

76 parser.addini( 

77 "jsss_callable_regex", 

78 default="{class_method=.}", 

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

80 ) 

81 parser.addini( 

82 "jsss_format_mode", 

83 default="on", 

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

85 ) 

86 

87 

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

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

90 """ 

91 Fixture providing a SchemaShot instance and gathering used schemas. 

92 """ 

93 

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

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

96 root_dir = test_path.parent 

97 

98 # Автополучение значения и валидация 

99 update_mode: str | bool = "--schema-update" 

100 reset_mode: str | bool = "--schema-reset" 

101 ci_cd_mode: str | bool = "--jsss-ci-cd" 

102 

103 modes = [update_mode, reset_mode, ci_cd_mode] 

104 states: list[bool] = [] 

105 enabled = [] 

106 

107 for mode in modes: 

108 state = bool(request.config.getoption(str(mode))) 

109 states.append(state) 

110 if state: 

111 enabled.append(str(mode)) 

112 

113 update_mode, reset_mode, ci_cd_mode = states 

114 

115 if len(enabled) > 1: 

116 raise ValueError(f"Options {' and '.join(enabled)} are mutually exclusive.") 

117 

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

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

120 

121 actions = { 

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

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

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

125 } 

126 

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

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

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

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

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

132 

133 differ = JsonSchemaDiff( 

134 ConfigMaker.make(), 

135 HighlighterPipeline( 

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

137 ), 

138 ) 

139 

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

141 if root_dir not in _schema_managers: 

142 _schema_managers[root_dir] = SchemaShot( 

143 root_dir=root_dir, 

144 differ=differ, 

145 callable_regex=callable_regex, 

146 format_mode=format_mode, 

147 # examples_limit, 

148 update_mode=bool(update_mode), 

149 reset_mode=bool(reset_mode), 

150 ci_cd_mode=bool(ci_cd_mode), 

151 update_actions=actions, 

152 save_original=save_original, 

153 debug_mode=debug_mode, 

154 snapshot_dir_name=schema_dir_name, 

155 ) 

156 

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

158 yield _schema_managers[root_dir] 

159 

160 

161@pytest.hookimpl(trylast=True) 

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

163 """ 

164 Hook that runs after all tests have finished. 

165 Clears global variables. 

166 """ 

167 global GLOBAL_STATS 

168 

169 # Clear the dictionary 

170 _schema_managers.clear() 

171 # Reset stats for next run 

172 GLOBAL_STATS = SchemaStats() 

173 

174 

175@pytest.hookimpl(trylast=True) 

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

177 """ 

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

179 """ 

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

181 if _schema_managers: 

182 

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

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

185 

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

187 

188 actions = { 

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

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

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

192 } 

193 

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

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

196 cleanup_unused_schemas(manager, update_mode, actions, GLOBAL_STATS) 

197 

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

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

200 GLOBAL_STATS.print_summary(terminalreporter, update_mode) 

201 

202 

203def cleanup_unused_schemas( 

204 manager: SchemaShot, 

205 update_mode: bool, 

206 actions: dict[str, bool], 

207 stats: Optional[SchemaStats] = None, 

208) -> None: 

209 """ 

210 Deletes unused schemas in update mode and collects statistics. 

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

212 

213 Args: 

214 manager: SchemaShot instance 

215 update_mode: Update mode 

216 stats: Optional object for collecting statistics 

217 """ 

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

219 if not manager.snapshot_dir.exists(): 

220 return 

221 

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

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

224 

225 for schema_file in all_schemas: 

226 if schema_file.name not in manager.used_schemas: 

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

228 try: 

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

230 schema_file.unlink() 

231 if stats: 

232 stats.add_deleted(schema_file.name) 

233 

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

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

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

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

238 if paired_json.exists(): 

239 try: 

240 paired_json.unlink() 

241 if stats: 

242 stats.add_deleted(paired_json.name) 

243 except OSError as e: 

244 manager.logger.warning( 

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

246 ) 

247 except Exception as e: 

248 manager.logger.error( 

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

250 ) 

251 

252 except OSError as e: 

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

254 manager.logger.warning( 

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

256 ) 

257 except Exception as e: 

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

259 manager.logger.error( 

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

261 ) 

262 else: 

263 if stats: 

264 stats.add_unused(schema_file.name)