Coverage for src/data_api/checks/openapi.py: 67%

1"""Structured compatibility baseline, excluding documentation-only metadata.""" 

2 

3import argparse 

4import json 

5from pathlib import Path 

6from typing import Any 

7 

8HTTP_METHODS = {"get", "post", "put", "patch", "delete", "head", "options", "trace"} 

9DOCUMENTATION_KEYS = {"title", "description", "summary", "examples", "example", "externalDocs"} 

10BASELINE = Path("tests") / "fixtures" / "openapi_contract.json" 

11 

12 

13def contract_projection(schema: dict[str, Any]) -> dict[str, Any]: 

14 def strip_docs(value: Any) -> Any: 

15 if isinstance(value, dict): 

16 return { 

17 key: ( 

18 {name: strip_docs(field) for name, field in sorted(item.items())} 

19 if key in {"properties", "$defs"} 

20 else strip_docs(item) 

21 ) 

22 for key, item in sorted(value.items()) 

23 if key not in DOCUMENTATION_KEYS 

24 } 

25 if isinstance(value, list): 

26 return [strip_docs(item) for item in value] 

27 return value 

28 

29 operations = {} 

30 for path, path_item in sorted(schema["paths"].items()): 

31 for method, operation in sorted(path_item.items()): 

32 if method not in HTTP_METHODS: 

33 continue 

34 parameters = { 

35 f"{parameter['in']}:{parameter['name']}": { 

36 "required": parameter.get("required", False), 

37 "schema": strip_docs(parameter["schema"]), 

38 } 

39 for parameter in operation.get("parameters", []) 

40 } 

41 operations[f"{method.upper()} {path}"] = { 

42 "operationId": operation["operationId"], 

43 "deprecated": operation.get("deprecated", False), 

44 "parameters": parameters, 

45 "requestBody": strip_docs(operation.get("requestBody")), 

46 "responses": strip_docs(operation["responses"]), 

47 } 

48 return { 

49 "operations": operations, 

50 "schemas": strip_docs(schema.get("components", {}).get("schemas", {})), 

51 } 

52 

53 

54def contract_differences(expected: Any, actual: Any, path: str = "contract") -> list[str]: 

55 if isinstance(expected, dict) and isinstance(actual, dict): 

56 changes = [f"{path}.{key}: removed" for key in sorted(expected.keys() - actual.keys())] 

57 changes += [f"{path}.{key}: added" for key in sorted(actual.keys() - expected.keys())] 

58 for key in sorted(expected.keys() & actual.keys()): 

59 changes.extend(contract_differences(expected[key], actual[key], f"{path}.{key}")) 

60 return changes 

61 if expected != actual: 

62 return [f"{path}: changed from {expected!r} to {actual!r}"] 

63 return [] 

64 

65 

66def main() -> int: 

67 from data_api.api.api import app 

68 

69 parser = argparse.ArgumentParser(description=__doc__) 

70 parser.add_argument( 

71 "--write", action="store_true", help="Regenerate only for reviewed API changes" 

72 ) 

73 args = parser.parse_args() 

74 actual = contract_projection(app.openapi()) 

75 if args.write: 

76 BASELINE.write_text( 

77 json.dumps(actual, ensure_ascii=False, indent=2, sort_keys=True) + "\n", 

78 encoding="utf-8", 

79 ) 

80 print(f"Wrote {BASELINE}") 

81 return 0 

82 expected = json.loads(BASELINE.read_text(encoding="utf-8")) 

83 changes = contract_differences(expected, actual) 

84 for change in changes: 

85 print(change) 

86 return int(bool(changes)) 

87 

88 

89if __name__ == "__main__": 

90 raise SystemExit(main())