Coverage for src/data_api/data/validation.py: 100%

1"""Validate published payloads before any reader can observe them.""" 

2 

3import json 

4from collections.abc import Sequence 

5from datetime import date, datetime 

6from functools import lru_cache 

7 

8from pydantic import BaseModel, HttpUrl, TypeAdapter 

9 

10from data_api.data.nthudata import FetchFailure, JsonData 

11from data_api.models.announcements import AnnouncementDetail, AnnouncementSource 

12from data_api.models.calendars import Calendar, CalendarEvent 

13from data_api.models.departments import Department 

14from data_api.models.dining import DiningBuilding 

15from data_api.models.libraries import LibraryRssItem 

16from data_api.models.newsletters import NewsletterInfo 

17 

18 

19class Coordinates(BaseModel): 

20 latitude: str 

21 longitude: str 

22 

23 

24class CalendarPayload(Calendar): 

25 events: list[CalendarEvent] 

26 

27 

28class LibraryCalendarPayload(CalendarPayload): 

29 url: HttpUrl 

30 

31 

32class LibraryPayload(BaseModel): 

33 name: str 

34 

35 

36LIBRARIES_RSS_ENDPOINT = "/libraries/rss.json" 

37LIBRARIES_CALENDARS_ENDPOINT = "/libraries/calendars.json" 

38CALENDARS_ENDPOINT = "/calendars.json" 

39 

40 

41@lru_cache 

42def _adapters() -> dict[str, TypeAdapter]: 

43 return { 

44 "/announcements.json": TypeAdapter(list[AnnouncementDetail]), 

45 "/announcements_list.json": TypeAdapter(list[AnnouncementSource]), 

46 "/dining.json": TypeAdapter(list[DiningBuilding]), 

47 "/directory.json": TypeAdapter(list[Department]), 

48 "/newsletters.json": TypeAdapter(list[NewsletterInfo]), 

49 "/maps.json": TypeAdapter(dict[str, dict[str, Coordinates]]), 

50 "/libraries.json": TypeAdapter(list[LibraryPayload]), 

51 LIBRARIES_RSS_ENDPOINT: TypeAdapter(dict[str, list[LibraryRssItem]]), 

52 LIBRARIES_CALENDARS_ENDPOINT: TypeAdapter(list[LibraryCalendarPayload]), 

53 CALENDARS_ENDPOINT: TypeAdapter(list[CalendarPayload]), 

54 } 

55 

56 

57def _validate_calendar_events( 

58 calendars: Sequence[CalendarPayload], 

59) -> None: 

60 for calendar in calendars: 

61 for event in calendar.events: 

62 try: 

63 parse = date.fromisoformat if event.all_day else datetime.fromisoformat 

64 if parse(event.end) < parse(event.start): 

65 raise ValueError("End precedes start") 

66 except (ValueError, TypeError) as exc: 

67 raise FetchFailure("validation") from exc 

68 

69 

70def _normalize_library_rss( 

71 raw: dict[str, list[dict]], parsed: dict[str, list[LibraryRssItem]] 

72) -> dict[str, list[dict]]: 

73 normalized = {} 

74 for feed, items in raw.items(): 

75 normalized[feed] = [] 

76 for item, model in zip(items, parsed[feed], strict=True): 

77 result = dict(item) 

78 if model.image is not None: 

79 result["image"] = { 

80 **item["image"], 

81 "url": str(model.image.url) if model.image.url is not None else None, 

82 "link": str(model.image.link) if model.image.link is not None else None, 

83 } 

84 normalized[feed].append(result) 

85 return normalized 

86 

87 

88def _validate_calendars(endpoint: str, parsed: Sequence[CalendarPayload]) -> None: 

89 ids = [calendar.id for calendar in parsed] 

90 has_reserved_prefix = endpoint == CALENDARS_ENDPOINT and any( 

91 calendar_id.startswith("library-") for calendar_id in ids 

92 ) 

93 if len(ids) != len(set(ids)) or has_reserved_prefix: 

94 raise FetchFailure("validation") 

95 _validate_calendar_events(parsed) 

96 

97 

98def validate_dataset(endpoint: str, raw: JsonData) -> JsonData: 

99 adapter = _adapters().get(endpoint) 

100 if adapter is None: 

101 # Other datasets preserve their upstream fields and representation. 

102 return raw 

103 expected_type = dict if endpoint in {"/maps.json", LIBRARIES_RSS_ENDPOINT} else list 

104 if not isinstance(raw, expected_type): 

105 raise FetchFailure("payload_type") 

106 # JSON strict mode accepts enum/URL strings without coercing booleans or numbers. 

107 parsed = adapter.validate_json(json.dumps(raw), strict=True) 

108 if endpoint == LIBRARIES_RSS_ENDPOINT: 

109 return _normalize_library_rss(raw, parsed) 

110 if endpoint in {LIBRARIES_CALENDARS_ENDPOINT, CALENDARS_ENDPOINT}: 

111 _validate_calendars(endpoint, parsed) 

112 return raw