Coverage for src/data_api/checks/common.py: 100%

1"""Safe diagnostics and HTTP document checks for external contract checks.""" 

2 

3import hashlib 

4import json 

5from dataclasses import dataclass 

6 

7import httpx 

8from pydantic import ValidationError 

9 

10from data_api.data.nthudata import JsonData 

11 

12 

13@dataclass 

14class CheckFailure(Exception): 

15 target: str 

16 stage: str 

17 reason: str 

18 

19 def __str__(self) -> str: 

20 return f"{self.target}: stage={self.stage} reason={self.reason}" 

21 

22 

23async def fetch_json( 

24 client: httpx.AsyncClient, url: str, target: str, checksum: str | None = None 

25) -> tuple[JsonData, httpx.Response]: 

26 try: 

27 response = await client.get(url) 

28 except httpx.TimeoutException as exc: 

29 raise CheckFailure(target, "http", "timeout") from exc 

30 except httpx.RequestError as exc: 

31 raise CheckFailure(target, "http", "connection/request failure") from exc 

32 if response.status_code != 200: 

33 raise CheckFailure(target, "http", f"status {response.status_code}") 

34 content_type = response.headers.get("content-type", "").split(";")[0].strip().lower() 

35 if content_type != "application/json" and not ( 

36 content_type.startswith("application/") and content_type.endswith("+json") 

37 ): 

38 raise CheckFailure(target, "content-type", "expected application/json") 

39 if checksum is not None and hashlib.sha256(response.content).hexdigest() != checksum: 

40 raise CheckFailure(target, "checksum", "SHA-256 mismatch") 

41 try: 

42 payload = response.json() 

43 except (json.JSONDecodeError, UnicodeDecodeError) as exc: 

44 raise CheckFailure(target, "json", "invalid JSON") from exc 

45 if not isinstance(payload, (dict, list)): 

46 raise CheckFailure(target, "json", "expected an object or array") 

47 return payload, response 

48 

49 

50def validation_reason(exc: ValidationError) -> str: 

51 """Report field locations and categories without copying upstream values.""" 

52 return "; ".join( 

53 f"{'.'.join(map(str, error['loc'])) or '<root>'}: {error['type']}" 

54 for error in exc.errors(include_url=False, include_context=False, include_input=False)[:5] 

55 ) 

56 

57 

58def report(failures: list[CheckFailure], label: str) -> int: 

59 if failures: 

60 for failure in failures: 

61 print(f"FAIL {label} {failure}") 

62 print(f"FAIL {label}: {len(failures)} external contract check(s) failed") 

63 return 1 

64 print(f"PASS {label}") 

65 return 0