Coverage for tests/test_rss_contract.py: 100%

1"""Publisher URL regressions must survive candidate installation and REST serialization.""" 

2 

3import json 

4from copy import deepcopy 

5from pathlib import Path 

6 

7import httpx 

8import pytest 

9from pydantic import HttpUrl, TypeAdapter, ValidationError 

10from test_data_manager import Clock, Publisher 

11 

12from data_api.api.api import app 

13from data_api.api.schemas.libraries import LibraryRssImage, LibraryRssItem 

14from data_api.data.manager import nthudata 

15from data_api.data.nthudata import Freshness 

16from data_api.data.validation import validate_dataset 

17 

18FIXTURE = Path(__file__).parent / "fixtures" / "library_rss_published_urls.json" 

19 

20 

21@pytest.fixture 

22def published_rss(): 

23 # URL values from the 2026-09-30 publication; article text is abbreviated. 

24 return json.loads(FIXTURE.read_text(encoding="utf-8")) 

25 

26 

27def test_fixed_cases_reproduce_original_single_uri_failures(published_rss): 

28 adapter = TypeAdapter(HttpUrl) 

29 for item in published_rss["news"][1:]: 

30 link_json = json.dumps(item["link"]) 

31 with pytest.raises(ValidationError): 

32 adapter.validate_json(link_json, strict=True) 

33 image_url_json = json.dumps(published_rss["news"][0]["image"]["url"]) 

34 with pytest.raises(ValidationError): 

35 adapter.validate_json(image_url_json, strict=True) 

36 

37 

38def test_full_rss_candidate_preserves_text_and_unknown_fields(published_rss): 

39 original = deepcopy(published_rss) 

40 normalized = validate_dataset("/libraries/rss.json", published_rss) 

41 assert published_rss == original 

42 assert normalized["news"][0]["link"] == original["news"][0]["link"] 

43 assert normalized["news"][0]["publisher_item_field"] == "retained" 

44 assert normalized["news"][0]["image"]["publisher_image_field"] == "retained" 

45 for feed in ["news", "eresources"]: 

46 assert normalized[feed][0]["image"]["url"].endswith("CNKI%20Trial.jpg") 

47 assert validate_dataset("/libraries/rss.json", normalized) == normalized 

48 for items in normalized.values(): 

49 for item in items: 

50 LibraryRssItem.model_validate(item).model_dump_json() 

51 

52 

53@pytest.mark.parametrize("value", ["https://", "not a url", 42]) 

54def test_invalid_images_still_fail(value): 

55 with pytest.raises(ValidationError): 

56 LibraryRssImage(url=value) 

57 

58 

59@pytest.mark.parametrize("value", [42, True, [], {}]) 

60def test_rss_link_remains_nullable_text(value): 

61 with pytest.raises(ValidationError): 

62 LibraryRssItem(title="News", description="", link=value) 

63 

64 

65async def test_rss_cold_start_refresh_and_last_known_good(monkeypatch, published_rss): 

66 clock = Clock() 

67 publisher = Publisher(published_rss, "libraries/rss.json") 

68 monkeypatch.setattr(nthudata, "clock", clock) 

69 async with nthudata.lifespan(publisher.client()): 

70 async with httpx.AsyncClient( 

71 transport=httpx.ASGITransport(app=app), base_url="http://test" 

72 ) as client: 

73 for version in ["a", "b"]: 

74 publisher.version = version 

75 for feed, expected in published_rss.items(): 

76 response = await client.get(f"/libraries/rss/{feed}") 

77 assert response.status_code == 200 

78 assert response.headers["X-Data-Commit-Hash"] == version 

79 assert len(response.json()) == len(expected) 

80 for actual, original in zip(response.json(), expected, strict=True): 

81 assert actual["link"] == original["link"] 

82 if expected: 

83 assert response.json()[0]["link"] == expected[0]["link"] 

84 assert response.json()[0]["image"]["url"].endswith("CNKI%20Trial.jpg") 

85 state = nthudata.state_for("libraries/rss.json") 

86 assert state.snapshot.raw["news"][0]["image"]["url"].endswith("CNKI%20Trial.jpg") 

87 assert state.snapshot.raw is state.snapshot.data 

88 clock.advance() 

89 snapshot = state.snapshot 

90 publisher.version = "invalid" 

91 publisher.payload = deepcopy(published_rss) 

92 publisher.payload["news"][0]["image"]["url"] = "https://" 

93 response = await client.get("/libraries/rss/news") 

94 assert response.status_code == 200 

95 assert response.headers["X-Data-Commit-Hash"] == "b" 

96 assert state.snapshot is snapshot 

97 assert state.freshness == Freshness.STALE 

98 assert state.last_error.category == "validation" 

99 

100 

101async def test_invalid_rss_without_snapshot_returns_503(published_rss): 

102 published_rss["news"][0]["image"]["url"] = "https://" 

103 publisher = Publisher(published_rss, "libraries/rss.json") 

104 async with nthudata.lifespan(publisher.client()): 

105 async with httpx.AsyncClient( 

106 transport=httpx.ASGITransport(app=app), base_url="http://test" 

107 ) as client: 

108 response = await client.get("/libraries/rss/news") 

109 assert response.status_code == 503 

110 assert response.json() == {"detail": "Service temporarily unavailable"}