Coverage for src/data_api/checks/release.py: 84%
137 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Validate release versions and compare public contracts with a PR base commit."""
3import argparse
4import json
5import re
6import subprocess
7import tomllib
8from pathlib import Path
9from typing import Any
11VERSION_PATTERN = re.compile(r"(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)", re.ASCII)
12OPENAPI_CONTRACT = "openapi_contract.json"
13MCP_CONTRACT = "mcp_contract.json"
14CONTRACTS = (OPENAPI_CONTRACT, MCP_CONTRACT)
15CONSTRAINTS = {
16 "type",
17 "$ref",
18 "enum",
19 "default",
20 "minimum",
21 "maximum",
22 "exclusiveMinimum",
23 "exclusiveMaximum",
24 "minLength",
25 "maxLength",
26 "minItems",
27 "maxItems",
28 "pattern",
29 "format",
30 "additionalProperties",
31 "const",
32 "multipleOf",
33 "minProperties",
34 "maxProperties",
35 "uniqueItems",
36 "patternProperties",
37 "dependentRequired",
38 "dependencies",
39 "not",
40 "allOf",
41 "anyOf",
42 "oneOf",
43 "if",
44 "then",
45 "else",
46 "unevaluatedProperties",
47 "contains",
48 "minContains",
49 "maxContains",
50 "propertyNames",
51 "prefixItems",
52 "items",
53 "additionalItems",
54}
55NAMED_MAPS = {
56 "properties": "schema",
57 "$defs": "schema",
58 "definitions": "schema",
59 "patternProperties": "schema",
60 "dependentSchemas": "schema",
61 "schemas": "schema",
62 "operations": "operation",
63 "parameters": "parameter",
64 "responses": "response",
65 "content": "media",
66 "headers": "header",
67}
68SCHEMA_CHILDREN = {
69 "input",
70 "output",
71 "schema",
72 "items",
73 "additionalProperties",
74 "additionalItems",
75 "unevaluatedProperties",
76 "not",
77 "if",
78 "then",
79 "else",
80 "contains",
81 "propertyNames",
82}
85def version_tuple(value: str) -> tuple[int, ...]:
86 if VERSION_PATTERN.fullmatch(value) is None:
87 raise ValueError("Expected a stable X.Y.Z version without leading zeroes")
88 return tuple(map(int, value.split(".")))
91def validate_version(root: Path, tag: str | None = None) -> str:
92 project = tomllib.loads((root / "pyproject.toml").read_text(encoding="utf-8"))["project"]
93 version = project["version"]
94 version_tuple(version)
95 lock = tomllib.loads((root / "uv.lock").read_text(encoding="utf-8"))
96 packages = [package for package in lock["package"] if package["name"] == project["name"]]
97 if len(packages) != 1 or packages[0]["version"] != version:
98 raise ValueError("pyproject.toml and uv.lock project versions must match")
99 if tag is not None and tag != f"v{version}":
100 raise ValueError("Release tag must be vX.Y.Z and match the project version")
101 return version
104def _resolve_context(path: str, context: str | None) -> str:
105 if context is not None:
106 return context
107 if path.endswith(".parameters"):
108 return "map:parameter"
109 if path == MCP_CONTRACT:
110 return "map:tool"
111 if path == OPENAPI_CONTRACT:
112 return "document"
113 return "schema"
116def _is_new_constraint_or_required_input(
117 context: str, named_map: bool, key: str, value: Any
118) -> bool:
119 if context == "schema" and key in CONSTRAINTS:
120 return True
121 if not named_map and key == "required" and value:
122 return True
123 is_param_or_request_body = context == "map:parameter" or (
124 context == "operation" and key == "requestBody"
125 )
126 return is_param_or_request_body and isinstance(value, dict) and bool(value.get("required"))
129def _child_context(context: str, named_map: bool, key: str) -> str:
130 if named_map:
131 return context.removeprefix("map:")
132 if key in NAMED_MAPS:
133 return f"map:{NAMED_MAPS[key]}"
134 if context == "schema" and key == "anyOf":
135 return "anyOf"
136 if context == "schema" and key in {"default", "const", "enum"}:
137 return "value"
138 if key in SCHEMA_CHILDREN:
139 return "schema"
140 return "structure"
143def _should_skip_shared_key(context: str, key: str, old_val: Any, new_val: Any) -> bool:
144 if key == "deprecated" and context in {"schema", "operation"}:
145 return True
146 return context == "schema" and key == "enum" and all(item in new_val for item in old_val)
149def _dict_breaking_changes(
150 old: dict[str, Any], new: dict[str, Any], path: str, context: str
151) -> list[str]:
152 named_map = context.startswith("map:")
153 changes = [
154 f"{path}.{key}: removed"
155 for key in sorted(old.keys() - new.keys())
156 if named_map or key != "deprecated" or context not in {"schema", "operation"}
157 ]
158 for key in sorted(new.keys() - old.keys()):
159 if _is_new_constraint_or_required_input(context, named_map, key, new[key]):
160 changes.append(f"{path}.{key}: new constraint or required input")
161 for key in sorted(old.keys() & new.keys()):
162 if _should_skip_shared_key(context, key, old[key], new[key]):
163 continue
164 child_ctx = _child_context(context, named_map, key)
165 changes.extend(breaking_changes(old[key], new[key], f"{path}.{key}", _context=child_ctx))
166 return changes
169def breaking_changes(
170 old: Any, new: Any, path: str = "contract", *, _context: str | None = None
171) -> list[str]:
172 """Conservative structural guard, not a proof of behavioral compatibility."""
173 context = _resolve_context(path, _context)
174 if context == "value":
175 return [] if old == new else [f"{path}: changed"]
176 if isinstance(old, dict) and isinstance(new, dict):
177 return _dict_breaking_changes(old, new, path, context)
178 if (
179 context == "anyOf"
180 and isinstance(old, list)
181 and isinstance(new, list)
182 and len(old) == len(new)
183 ):
184 return [
185 change
186 for index, (previous, current) in enumerate(zip(old, new))
187 for change in breaking_changes(previous, current, f"{path}[{index}]", _context="schema")
188 ]
189 return [f"{path}: changed"] if old != new else []
192def git_file(base_sha: str, path: str) -> str:
193 if re.fullmatch(r"[a-fA-F0-9]{40,64}", base_sha) is None:
194 raise ValueError("Base must be a full commit SHA")
195 return subprocess.run(
196 ["git", "show", f"{base_sha}:{path}"],
197 check=True,
198 capture_output=True,
199 text=True,
200 encoding="utf-8",
201 ).stdout
204def initial_version_is_unpublished(version: str) -> bool:
205 """Query the authoritative repository, never infer publication from a shallow clone."""
206 if version != "3.0.0":
207 return False
208 ref = f"refs/tags/v{version}"
209 try:
210 result = subprocess.run(
211 [
212 "git",
213 "ls-remote",
214 "--exit-code",
215 "--refs",
216 "https://github.com/NTHU-SA/NTHU-Data-API.git",
217 ref,
218 ],
219 capture_output=True,
220 text=True,
221 encoding="utf-8",
222 timeout=30,
223 )
224 except (OSError, subprocess.TimeoutExpired) as exc:
225 raise ValueError("Unable to verify release publication") from exc
226 if result.returncode == 2 and not result.stdout.strip():
227 return True
228 if result.returncode == 0 and re.fullmatch(
229 rf"[0-9a-f]{{40}}\s+{re.escape(ref)}\s*", result.stdout
230 ):
231 return False
232 raise ValueError("Unable to verify release publication")
235def compare_base(
236 root: Path, base_sha: str, version: str, *, check_publication: bool = False
237) -> list[str]:
238 previous = tomllib.loads(git_file(base_sha, "pyproject.toml"))["project"]["version"]
239 current_version, previous_version = version_tuple(version), version_tuple(previous)
240 if current_version < previous_version:
241 raise ValueError("Project version must not decrease")
242 changes = []
243 for name in CONTRACTS:
244 path = f"tests/fixtures/{name}"
245 # The MCP snapshot is introduced with 3.0.0; older releases have no baseline.
246 if name == MCP_CONTRACT and previous_version < (3, 0, 0):
247 continue
248 old = json.loads(git_file(base_sha, path))
249 new = json.loads((root / "tests" / "fixtures" / name).read_text(encoding="utf-8"))
250 changes.extend(breaking_changes(old, new, name))
251 for change in changes:
252 print(change)
253 if (
254 changes
255 and current_version[0] <= previous_version[0]
256 and not (
257 check_publication
258 and version == previous == "3.0.0"
259 and initial_version_is_unpublished(version)
260 )
261 ):
262 raise ValueError("Public contract changes require a major version increase")
263 return changes
266def main() -> int:
267 parser = argparse.ArgumentParser(description=__doc__)
268 parser.add_argument("--tag", help="Required vX.Y.Z tag for deployment validation")
269 parser.add_argument("--base-sha", help="Compare checked-in contracts against a PR base SHA")
270 parser.add_argument(
271 "--check-publication",
272 action="store_true",
273 help="Verify whether the initial 3.0.0 contract is still unpublished",
274 )
275 parser.add_argument("--next-version", help="Validate a proposed version before preparing a PR")
276 args = parser.parse_args()
277 try:
278 version = validate_version(Path.cwd(), args.tag)
279 if args.next_version and version_tuple(args.next_version) <= version_tuple(version):
280 raise ValueError("New release version must be greater than the current version")
281 if args.base_sha:
282 compare_base(
283 Path.cwd(), args.base_sha, version, check_publication=args.check_publication
284 )
285 except (ValueError, KeyError, OSError, subprocess.CalledProcessError) as exc:
286 parser.exit(1, f"Release validation failed: {exc}\n")
287 print(f"Release version validated: {version}")
288 return 0
291if __name__ == "__main__":
292 raise SystemExit(main())