Coverage for tests / test_source_notices.py: 100%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import json 

6import zipfile 

7from pathlib import Path 

8from unittest.mock import call, patch 

9 

10import requests 

11import yaml 

12from rdflib import Graph, URIRef 

13 

14from changes_metadata_manager.folder_metadata_builder import BASE_URI 

15from changes_metadata_manager.patch.source_notices import ( 

16 NEW_NOTICE_HTML, 

17 OLD_NOTICE_HTML, 

18 prepare_source_notice_patches, 

19) 

20 

21 

22def _draft( 

23 tmp_path: Path, 

24 record_id: int, 

25 entity_id: str, 

26 stage: str, 

27 *, 

28 content_license: bool = False, 

29) -> dict: 

30 archive_path = tmp_path / f"dataset-{entity_id}-{stage}.zip" 

31 with zipfile.ZipFile(archive_path, "w") as archive: 

32 archive.writestr(f"S1-{entity_id}-Test/{stage}/meta.ttl", "") 

33 config_path = tmp_path / f"{record_id}.yaml" 

34 rights = ( 

35 [ 

36 { 

37 "title": {"en": "Creative Commons Zero (Content license)"}, 

38 "link": "https://creativecommons.org/publicdomain/zero/1.0/", 

39 } 

40 ] 

41 if content_license 

42 else [] 

43 ) 

44 config_path.write_text( 

45 yaml.safe_dump({"files": [str(archive_path)], "rights": rights}) 

46 ) 

47 return { 

48 "draft_id": record_id, 

49 "config_file": str(config_path), 

50 "zenodo_url": "https://zenodo.org/api", 

51 "access_token": "secret-token", 

52 "user_agent": "changes-metadata-manager/1.0.0", 

53 "status": "published", 

54 } 

55 

56 

57def _write_drafts(tmp_path: Path, drafts: list[dict]) -> Path: 

58 drafts_path = tmp_path / "drafts.json" 

59 drafts_path.write_text(json.dumps(drafts)) 

60 return drafts_path 

61 

62 

63def _record(entity_id: str, stage: str, notice: str) -> dict: 

64 return { 

65 "access": { 

66 "record": "public", 

67 "files": "public", 

68 "embargo": {"active": False, "reason": None}, 

69 }, 

70 "files": { 

71 "enabled": True, 

72 "order": [], 

73 "default_preview": None, 

74 "entries": {f"remote-dataset-{stage}.zip": {"size": 123}}, 

75 }, 

76 "metadata": { 

77 "title": "Remote title", 

78 "resource_type": {"id": "dataset", "title": {"en": "Dataset"}}, 

79 "additional_descriptions": [ 

80 { 

81 "description": "<p>Remote method.</p>", 

82 "type": {"id": "methods", "title": {"en": "Methods"}}, 

83 }, 

84 { 

85 "description": notice, 

86 "type": {"id": "notes", "title": {"en": "Notes"}}, 

87 }, 

88 ], 

89 "identifiers": [ 

90 { 

91 "identifier": ( 

92 f"https://w3id.org/changes/4/aldrovandi/itm/{entity_id}/ob00/1" 

93 ), 

94 "scheme": "url", 

95 } 

96 ], 

97 "publisher": "Zenodo", 

98 }, 

99 } 

100 

101 

102def _expected_payload(entity_id: str) -> dict: 

103 return { 

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

105 "files": {"enabled": True, "order": []}, 

106 "metadata": { 

107 "title": "Remote title", 

108 "resource_type": {"id": "dataset"}, 

109 "additional_descriptions": [ 

110 { 

111 "description": "<p>Remote method.</p>", 

112 "type": {"id": "methods"}, 

113 }, 

114 { 

115 "description": NEW_NOTICE_HTML, 

116 "type": {"id": "notes"}, 

117 }, 

118 ], 

119 "identifiers": [ 

120 { 

121 "identifier": ( 

122 f"https://w3id.org/changes/4/aldrovandi/itm/{entity_id}/ob00/1" 

123 ), 

124 "scheme": "url", 

125 } 

126 ], 

127 "publisher": "Zenodo", 

128 }, 

129 } 

130 

131 

132@patch("changes_metadata_manager.patch.source_notices.time.sleep") 

133@patch("changes_metadata_manager.patch.source_notices.fetch_record") 

134def test_prepare_selects_from_rdf_and_reuses_cache(mock_fetch, mock_sleep, tmp_path): 

135 kg = Graph() 

136 kg.add( 

137 ( 

138 URIRef(f"{BASE_URI}/act/43/00/1"), 

139 URIRef("https://example.org/predicate"), 

140 URIRef("https://example.org/object"), 

141 ) 

142 ) 

143 drafts_path = _write_drafts( 

144 tmp_path, 

145 [ 

146 _draft(tmp_path, 100, "42", "raw"), 

147 _draft(tmp_path, 101, "43", "raw"), 

148 _draft(tmp_path, 102, "44", "raw", content_license=True), 

149 ], 

150 ) 

151 output_dir = tmp_path / "prepared" 

152 record = _record("42", "raw", OLD_NOTICE_HTML) 

153 mock_fetch.return_value = (record, False) 

154 

155 with patch( 

156 "changes_metadata_manager.patch.source_notices.load_kg", return_value=kg 

157 ): 

158 manifest_path = prepare_source_notice_patches( 

159 drafts_path, tmp_path / "kg.ttl", output_dir 

160 ) 

161 second_manifest_path = prepare_source_notice_patches( 

162 drafts_path, tmp_path / "kg.ttl", output_dir 

163 ) 

164 

165 expected_manifest = [ 

166 { 

167 "record_id": 100, 

168 "entity_ids": ["42"], 

169 "stage": "raw", 

170 "status": "would_patch", 

171 "payload_file": "100.yaml", 

172 } 

173 ] 

174 assert json.loads(manifest_path.read_text()) == expected_manifest 

175 assert json.loads(second_manifest_path.read_text()) == expected_manifest 

176 assert yaml.safe_load((output_dir / "100.yaml").read_text()) == _expected_payload( 

177 "42" 

178 ) 

179 assert mock_fetch.call_count == 1 

180 assert mock_sleep.call_args_list == [call(2)] 

181 assert "secret-token" not in manifest_path.read_text() 

182 

183 

184@patch("changes_metadata_manager.patch.source_notices.time.sleep") 

185@patch("changes_metadata_manager.patch.source_notices.fetch_record") 

186def test_prepare_reports_correct_and_blocked_records(mock_fetch, mock_sleep, tmp_path): 

187 kg = Graph() 

188 drafts_path = _write_drafts( 

189 tmp_path, 

190 [ 

191 _draft(tmp_path, 100, "42", "raw"), 

192 _draft(tmp_path, 101, "43", "raw"), 

193 _draft(tmp_path, 102, "44", "raw"), 

194 ], 

195 ) 

196 output_dir = tmp_path / "prepared" 

197 mock_fetch.side_effect = [ 

198 (_record("42", "raw", NEW_NOTICE_HTML), False), 

199 (_record("43", "raw", OLD_NOTICE_HTML), True), 

200 (_record("44", "raw", "<p>Manual remote note.</p>"), False), 

201 ] 

202 

203 with patch( 

204 "changes_metadata_manager.patch.source_notices.load_kg", return_value=kg 

205 ): 

206 manifest_path = prepare_source_notice_patches( 

207 drafts_path, tmp_path / "kg.ttl", output_dir 

208 ) 

209 

210 assert json.loads(manifest_path.read_text()) == [ 

211 { 

212 "record_id": 100, 

213 "entity_ids": ["42"], 

214 "stage": "raw", 

215 "status": "already_correct", 

216 }, 

217 { 

218 "record_id": 101, 

219 "entity_ids": ["43"], 

220 "stage": "raw", 

221 "status": "blocked", 

222 "reason": "An edit draft already exists", 

223 }, 

224 { 

225 "record_id": 102, 

226 "entity_ids": ["44"], 

227 "stage": "raw", 

228 "status": "blocked", 

229 "reason": "Expected source notice not found", 

230 }, 

231 ] 

232 assert {path.name for path in output_dir.iterdir()} == {"manifest.json"} 

233 assert mock_sleep.call_args_list == [call(2), call(2), call(2)] 

234 

235 

236@patch("changes_metadata_manager.patch.source_notices.time.sleep") 

237@patch("changes_metadata_manager.patch.source_notices.fetch_record") 

238def test_prepare_records_http_error(mock_fetch, mock_sleep, tmp_path): 

239 kg = Graph() 

240 drafts_path = _write_drafts(tmp_path, [_draft(tmp_path, 100, "42", "raw")]) 

241 output_dir = tmp_path / "prepared" 

242 mock_fetch.side_effect = requests.HTTPError("500 Server Error") 

243 

244 with patch( 

245 "changes_metadata_manager.patch.source_notices.load_kg", return_value=kg 

246 ): 

247 manifest_path = prepare_source_notice_patches( 

248 drafts_path, tmp_path / "kg.ttl", output_dir 

249 ) 

250 

251 assert json.loads(manifest_path.read_text()) == [ 

252 { 

253 "record_id": 100, 

254 "entity_ids": ["42"], 

255 "stage": "raw", 

256 "status": "error", 

257 "error": "500 Server Error", 

258 } 

259 ] 

260 assert mock_sleep.call_args_list == [call(2)]