Coverage for tests / test_zenodo_api.py: 99%

1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from unittest.mock import MagicMock, call, patch 

6 

7import pytest 

8import requests 

9 

10from changes_metadata_manager.zenodo_api import ( 

11 CACHE_TTL_SECONDS, 

12 MAX_RETRIES, 

13 ZenodoRecordCache, 

14 create_edit_draft, 

15 fetch_latest_published_record, 

16 fetch_published_record, 

17 fetch_record, 

18 publish_draft, 

19 request_with_retry, 

20 update_draft, 

21) 

22from changes_metadata_manager.zenodo_metadata import ZenodoUpdatePayload 

23 

24 

25@patch("changes_metadata_manager.zenodo_api.time.time", return_value=1000) 

26def test_record_cache_stores_expires_and_invalidates(mock_time, tmp_path): 

27 cache_path = tmp_path / "records.sqlite3" 

28 record = {"metadata": {"title": "Record"}} 

29 

30 with ZenodoRecordCache(cache_path) as cache: 

31 assert cache.get("https://zenodo.org/api/", "123") is None 

32 

33 cache.set("https://zenodo.org/api/", "123", record, False) 

34 assert cache.get("https://zenodo.org/api", "123") == (record, False) 

35 assert cache.get("https://sandbox.zenodo.org/api", "123") is None 

36 

37 mock_time.return_value = 1000 + CACHE_TTL_SECONDS 

38 assert cache.get("https://zenodo.org/api", "123") is None 

39 

40 cache.set("https://zenodo.org/api", "123", record, True) 

41 assert cache.get("https://zenodo.org/api", "123") == (record, True) 

42 cache.invalidate("https://zenodo.org/api/", "123") 

43 assert cache.get("https://zenodo.org/api", "123") is None 

44 

45 

46@patch("changes_metadata_manager.zenodo_api.time.sleep") 

47@patch("changes_metadata_manager.zenodo_api.requests.request") 

48def test_request_retries_rate_limits(mock_request, mock_sleep): 

49 rate_limited = MagicMock(status_code=429) 

50 success = MagicMock(status_code=200) 

51 mock_request.side_effect = [rate_limited, rate_limited, success] 

52 

53 result = request_with_retry("GET", "https://example.org") 

54 

55 assert result == success 

56 assert mock_request.call_count == 3 

57 assert mock_sleep.call_args_list == [call(20), call(40)] 

58 

59 

60@pytest.mark.parametrize("status_code", [502, 503, 504]) 

61@patch("changes_metadata_manager.zenodo_api.time.sleep") 

62@patch("changes_metadata_manager.zenodo_api.requests.request") 

63def test_request_retries_gateway_errors(mock_request, mock_sleep, status_code): 

64 gateway_error = MagicMock(status_code=status_code) 

65 success = MagicMock(status_code=200) 

66 mock_request.side_effect = [gateway_error, success] 

67 

68 result = request_with_retry("GET", "https://example.org") 

69 

70 assert result == success 

71 assert mock_request.call_count == 2 

72 assert mock_sleep.call_args_list == [call(20)] 

73 

74 

75@patch("changes_metadata_manager.zenodo_api.time.sleep") 

76@patch("changes_metadata_manager.zenodo_api.requests.request") 

77def test_request_returns_final_rate_limit(mock_request, mock_sleep): 

78 rate_limited = MagicMock(status_code=429) 

79 mock_request.return_value = rate_limited 

80 

81 result = request_with_retry("GET", "https://example.org") 

82 

83 assert result == rate_limited 

84 assert mock_request.call_count == MAX_RETRIES 

85 

86 

87@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

88def test_fetches_existing_draft(mock_request): 

89 response = MagicMock(status_code=200) 

90 response.json.return_value = {"metadata": {"title": "draft"}} 

91 mock_request.return_value = response 

92 

93 result = fetch_record("https://zenodo.org/api", "123", "token", "agent") 

94 

95 assert result == ({"metadata": {"title": "draft"}}, True) 

96 assert mock_request.call_count == 1 

97 assert mock_request.call_args.args[:2] == ( 

98 "GET", 

99 "https://zenodo.org/api/records/123/draft", 

100 ) 

101 

102 

103@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

104def test_fetches_published_record_when_draft_is_missing(mock_request): 

105 missing = MagicMock(status_code=404) 

106 published = MagicMock(status_code=200) 

107 published.json.return_value = {"metadata": {"title": "published"}} 

108 mock_request.side_effect = [missing, published] 

109 

110 result = fetch_record("https://zenodo.org/api", "123", "token", "agent") 

111 

112 assert result == ({"metadata": {"title": "published"}}, False) 

113 assert [item.args[:2] for item in mock_request.call_args_list] == [ 

114 ("GET", "https://zenodo.org/api/records/123/draft"), 

115 ("GET", "https://zenodo.org/api/records/123"), 

116 ] 

117 

118 

119@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

120def test_fetches_published_record_without_checking_draft(mock_request): 

121 published = MagicMock(status_code=200) 

122 published.json.return_value = {"metadata": {"title": "published"}} 

123 mock_request.return_value = published 

124 

125 result = fetch_published_record("https://zenodo.org/api", "123", "token", "agent") 

126 

127 assert result == {"metadata": {"title": "published"}} 

128 assert [item.args[:2] for item in mock_request.call_args_list] == [ 

129 ("GET", "https://zenodo.org/api/records/123") 

130 ] 

131 assert published.raise_for_status.call_count == 1 

132 

133 

134@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

135def test_fetches_latest_published_version(mock_request): 

136 published = MagicMock(status_code=200) 

137 published.json.return_value = {"id": "456"} 

138 mock_request.return_value = published 

139 

140 result = fetch_latest_published_record( 

141 "https://zenodo.org/api", "123", "token", "agent" 

142 ) 

143 

144 assert result == {"id": "456"} 

145 assert [item.args[:2] for item in mock_request.call_args_list] == [ 

146 ("GET", "https://zenodo.org/api/records/123/versions/latest") 

147 ] 

148 assert published.raise_for_status.call_count == 1 

149 

150 

151@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

152def test_edit_draft_lifecycle_uses_expected_endpoints(mock_request): 

153 created = MagicMock(status_code=201) 

154 created.json.return_value = {"metadata": {"title": "draft"}} 

155 updated = MagicMock(status_code=200) 

156 published = MagicMock(status_code=202) 

157 mock_request.side_effect = [created, updated, published] 

158 payload: ZenodoUpdatePayload = { 

159 "access": {"record": "public", "files": "public"}, 

160 "files": {"enabled": True}, 

161 "metadata": {"title": "draft"}, 

162 } 

163 

164 draft = create_edit_draft("https://zenodo.org/api", "123", "token", "agent") 

165 update_draft("https://zenodo.org/api", "123", "token", "agent", payload) 

166 publish_draft("https://zenodo.org/api", "123", "token", "agent") 

167 

168 assert draft == {"metadata": {"title": "draft"}} 

169 assert [item.args[:2] for item in mock_request.call_args_list] == [ 

170 ("POST", "https://zenodo.org/api/records/123/draft"), 

171 ("PUT", "https://zenodo.org/api/records/123/draft"), 

172 ( 

173 "POST", 

174 "https://zenodo.org/api/records/123/draft/actions/publish", 

175 ), 

176 ] 

177 assert mock_request.call_args_list[1].kwargs["json"] == payload 

178 assert created.raise_for_status.call_count == 1 

179 assert updated.raise_for_status.call_count == 1 

180 assert published.raise_for_status.call_count == 1 

181 

182 

183@patch("changes_metadata_manager.zenodo_api.request_with_retry") 

184def test_create_edit_draft_propagates_http_error(mock_request): 

185 response = MagicMock(status_code=500) 

186 response.raise_for_status.side_effect = requests.HTTPError("500 Server Error") 

187 mock_request.return_value = response 

188 

189 try: 

190 create_edit_draft("https://zenodo.org/api", "123", "token", "agent") 

191 except requests.HTTPError as exc: 

192 assert str(exc) == "500 Server Error" 

193 else: 

194 raise AssertionError("HTTP error was not propagated")