Coverage for ckanext/udc/tests/test_package_actions.py: 87%

1import json 

2import types 

3import uuid 

4from pathlib import Path 

5 

6import pytest 

7import ckan.logic as logic 

8import ckan.model as model 

9import ckan.plugins as plugins 

10import ckan.plugins.toolkit as tk 

11from configparser import ConfigParser 

12from sqlalchemy import create_engine 

13from ckan.tests import helpers as ckan_helpers 

14 

15from ckanext.udc import helpers 

16from ckanext.udc import plugin as udc_plugin_module 

17from ckanext.udc.solr import solr as udc_solr 

18 

19 

20CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.example.json" 

21 

22 

23def _read_ini(path): 

24 parser = ConfigParser(defaults={"here": str(path.parent)}) 

25 parser.read(path) 

26 data = dict(parser.defaults()) 

27 if parser.has_section("app:main"): 

28 data.update(dict(parser.items("app:main"))) 

29 return data 

30 

31 

32@pytest.fixture(scope="session", autouse=True) 

33def _bootstrap_test_ckan(): 

34 if tk.config.get("sqlalchemy.url"): 

35 return 

36 

37 repo_root = Path(__file__).resolve().parents[4] 

38 core_conf = _read_ini(repo_root / "ckan/test-core.ini") 

39 ext_conf = _read_ini(Path(__file__).resolve().parents[3] / "test.ini") 

40 conf = {**core_conf, **ext_conf} 

41 conf.pop("use", None) 

42 conf.setdefault("ckan.site_url", "http://test.ckan.local") 

43 conf.setdefault("ckan.locale_default", "en") 

44 conf.setdefault("udc.multilingual.languages", "en fr") 

45 conf.setdefault("ckan.base_public_folder", "public") 

46 conf.setdefault("ckan.base_templates_folder", "templates") 

47 conf.setdefault("ckan.auth.create_unowned_dataset", "true") 

48 conf.setdefault("ckan.storage_path", "/tmp") 

49 

50 cfg = tk.config 

51 for key, value in conf.items(): 

52 cfg[key] = value 

53 

54 engine = create_engine(conf["sqlalchemy.url"]) 

55 model.init_model(engine) 

56 model.repo.init_db() 

57 

58 

59@pytest.fixture 

60def stub_udc_plugin(monkeypatch): 

61 plugin = types.SimpleNamespace(disable_graphdb=False) 

62 monkeypatch.setattr(helpers.plugins, "get_plugin", lambda name: plugin) 

63 return plugin 

64 

65 

66@pytest.fixture 

67def udc_config(): 

68 with CONFIG_PATH.open("r", encoding="utf-8") as fh: 

69 data = json.load(fh) 

70 

71 for level in data.get("maturity_model", []): 

72 for field in level.get("fields", []): 

73 if field.get("type") in {"multiple_select", "single_select"}: 

74 field.setdefault("options", []) 

75 

76 return data 

77 

78 

79@pytest.fixture 

80def udc_plugin_instance(monkeypatch, udc_config): 

81 udc_solr.update_solr_maturity_model_fields(udc_config["maturity_model"]) 

82 monkeypatch.setattr(udc_plugin_module, "update_solr_maturity_model_fields", lambda *_: None) 

83 plugin = plugins.get_plugin("udc") 

84 if plugin is None: 

85 plugins.load("udc") 

86 plugin = plugins.get_plugin("udc") 

87 plugin.disable_graphdb = True 

88 plugin.reload_config(udc_config) 

89 return plugin 

90 

91 

92@pytest.fixture 

93def clean_db(): 

94 ckan_helpers.reset_db() 

95 try: 

96 yield 

97 finally: 

98 model.Session.remove() 

99 

100 

101def test_package_update_runs_preprocessor_and_updates_graph(monkeypatch, stub_udc_plugin): 

102 calls = [] 

103 

104 def fake_before(context, data_dict): 

105 calls.append(("before", data_dict.get("file_format"))) 

106 data_dict["file_format"] = "normalized" 

107 

108 def fake_action(context, data_dict): 

109 calls.append(("action", data_dict.get("file_format"))) 

110 return {"id": "pkg", "file_format": "normalized"} 

111 

112 def fake_graph(context, result_dict): 

113 calls.append(("graph", result_dict.get("file_format"))) 

114 

115 monkeypatch.setattr(helpers, "before_package_update_for_file_format", fake_before) 

116 monkeypatch.setattr(helpers, "onUpdateCatalogue", fake_graph) 

117 

118 result = helpers.package_update(fake_action, {"user": "tester"}, {"file_format": "csv"}) 

119 

120 assert result == {"id": "pkg", "file_format": "normalized"} 

121 assert calls == [ 

122 ("before", "csv"), 

123 ("action", "normalized"), 

124 ("graph", "normalized"), 

125 ] 

126 

127 

128def test_package_update_wraps_graph_errors(monkeypatch, stub_udc_plugin): 

129 monkeypatch.setattr(helpers, "before_package_update_for_file_format", lambda *_: None) 

130 

131 def fake_action(context, data_dict): 

132 return {"ok": True} 

133 

134 def boom(context, data_dict): 

135 raise RuntimeError("boom") 

136 

137 monkeypatch.setattr(helpers, "onUpdateCatalogue", boom) 

138 

139 with pytest.raises(logic.ValidationError) as excinfo: 

140 helpers.package_update(fake_action, {}, {}) 

141 

142 messages = excinfo.value.error_dict.get("message") 

143 assert messages 

144 assert "Error occurred in updating the knowledge graph" in messages[0] 

145 

146 

147def test_package_update_skips_graph_when_disabled(monkeypatch, stub_udc_plugin): 

148 stub_udc_plugin.disable_graphdb = True 

149 monkeypatch.setattr(helpers, "before_package_update_for_file_format", lambda *_: None) 

150 

151 def fake_action(context, data_dict): 

152 data_dict["visited"] = True 

153 return {"ok": True} 

154 

155 def fail(*_): 

156 raise AssertionError("graph should be skipped") 

157 

158 monkeypatch.setattr(helpers, "onUpdateCatalogue", fail) 

159 

160 payload = {} 

161 result = helpers.package_update(fake_action, {}, payload) 

162 

163 assert result == {"ok": True} 

164 assert payload["visited"] is True 

165 

166 

167def test_package_delete_invokes_graph(monkeypatch, stub_udc_plugin): 

168 calls = [] 

169 

170 def fake_action(context, data_dict): 

171 calls.append("action") 

172 return {"id": data_dict.get("id")} 

173 

174 def fake_graph(context, data_dict): 

175 calls.append("graph") 

176 

177 monkeypatch.setattr(helpers, "onDeleteCatalogue", fake_graph) 

178 

179 result = helpers.package_delete(fake_action, {}, {"id": "pkg"}) 

180 

181 assert result == {"id": "pkg"} 

182 assert calls == ["action", "graph"] 

183 

184 

185def test_package_delete_wraps_graph_errors(monkeypatch, stub_udc_plugin): 

186 def fake_action(context, data_dict): 

187 return {"id": data_dict.get("id")} 

188 

189 def boom(context, data_dict): 

190 raise RuntimeError("boom") 

191 

192 monkeypatch.setattr(helpers, "onDeleteCatalogue", boom) 

193 

194 with pytest.raises(logic.ValidationError) as excinfo: 

195 helpers.package_delete(fake_action, {}, {"id": "pkg"}) 

196 

197 messages = excinfo.value.error_dict.get("message") 

198 assert messages 

199 assert "Error occurred in updating the knowledge graph" in messages[0] 

200 

201 

202def test_package_delete_skips_graph_when_disabled(monkeypatch, stub_udc_plugin): 

203 stub_udc_plugin.disable_graphdb = True 

204 

205 def fake_action(context, data_dict): 

206 return {"id": data_dict.get("id")} 

207 

208 def fail(*_): 

209 raise AssertionError("graph should be skipped") 

210 

211 monkeypatch.setattr(helpers, "onDeleteCatalogue", fail) 

212 

213 result = helpers.package_delete(fake_action, {}, {"id": "pkg"}) 

214 

215 assert result == {"id": "pkg"} 

216 

217 

218pytestmark = pytest.mark.ckan_config("udc.multilingual.languages", "en fr") 

219 

220 

221def _create_multilingual_dataset(plugin): 

222 context = {"model": model, "session": model.Session, "schema": plugin.create_package_schema()} 

223 data = { 

224 "name": f"udc-{uuid.uuid4().hex[:8]}", 

225 "title": "Multilingual Dataset", 

226 "title_translated": {"en": "Multilingual Dataset", "fr": "Jeu de données multilingue"}, 

227 "notes_translated": {"en": "English description", "fr": "Description française"}, 

228 "tags_translated": {"en": ["roads", "traffic"], "fr": ["routes"]}, 

229 "theme": {"en": "transport", "fr": "transport-fr"}, 

230 "type": "dataset", 

231 } 

232 return ckan_helpers.call_action("package_create", context=context, **data) 

233 

234 

235@pytest.mark.usefixtures("clean_db") 

236def test_package_create_handles_multilingual_fields(udc_plugin_instance): 

237 created = _create_multilingual_dataset(udc_plugin_instance) 

238 

239 show_context = {"model": model, "session": model.Session, "schema": udc_plugin_instance.show_package_schema()} 

240 shown = ckan_helpers.call_action("package_show", context=show_context, id=created["id"]) 

241 

242 # Verify multilingual fields are stored 

243 assert shown["title_translated"]["en"] == "Multilingual Dataset" 

244 assert shown["title_translated"]["fr"] == "Jeu de données multilingue" 

245 assert shown["notes_translated"]["en"] == "English description" 

246 assert shown["notes_translated"]["fr"] == "Description française" 

247 assert shown["tags_translated"]["en"] == ["roads", "traffic"] 

248 assert shown["tags_translated"]["fr"] == ["routes"] 

249 assert shown["theme"] == {"en": "transport", "fr": "transport-fr"} 

250 

251 

252@pytest.mark.usefixtures("clean_db") 

253def test_package_update_persists_multilingual_changes(udc_plugin_instance): 

254 created = _create_multilingual_dataset(udc_plugin_instance) 

255 

256 show_context = {"model": model, "session": model.Session, "schema": udc_plugin_instance.show_package_schema()} 

257 pkg = ckan_helpers.call_action("package_show", context=show_context, id=created["id"]) 

258 

259 pkg["title_translated"] = {"en": "Updated Title", "fr": "Titre mis à jour"} 

260 pkg["notes_translated"] = {"en": "Updated English", "fr": "Description mise à jour"} 

261 pkg["theme"] = {"en": "economy", "fr": "économie"} 

262 pkg["tags_translated"] = {"en": ["economy"], "fr": ["économie"]} 

263 pkg.pop("tags", None) 

264 

265 update_context = {"model": model, "session": model.Session, "schema": udc_plugin_instance.update_package_schema()} 

266 ckan_helpers.call_action("package_update", context=update_context, **pkg) 

267 

268 refreshed = ckan_helpers.call_action("package_show", context=show_context, id=created["id"]) 

269 

270 # Verify multilingual fields are updated 

271 assert refreshed["title_translated"]["en"] == "Updated Title" 

272 assert refreshed["title_translated"]["fr"] == "Titre mis à jour" 

273 assert refreshed["notes_translated"]["en"] == "Updated English" 

274 assert refreshed["notes_translated"]["fr"] == "Description mise à jour" 

275 assert refreshed["theme"] == {"en": "economy", "fr": "économie"} 

276 assert refreshed["tags_translated"]["en"] == ["economy"] 

277 assert refreshed["tags_translated"]["fr"] == ["économie"] 

278 

279 

280@pytest.mark.usefixtures("clean_db") 

281def test_package_delete_succeeds_with_multilingual_extras(udc_plugin_instance): 

282 created = _create_multilingual_dataset(udc_plugin_instance) 

283 

284 delete_context = {"model": model, "session": model.Session, "ignore_auth": True} 

285 ckan_helpers.call_action("package_delete", context=delete_context, id=created["id"]) 

286 

287 # Verify package is marked as deleted 

288 show_context = { 

289 "model": model, 

290 "session": model.Session, 

291 "schema": udc_plugin_instance.show_package_schema(), 

292 "ignore_auth": True, 

293 } 

294 deleted = ckan_helpers.call_action( 

295 "package_show", 

296 context=show_context, 

297 id=created["id"], 

298 ) 

299 

300 assert deleted["state"] == "deleted" 

301 assert deleted["theme"] == {"en": "transport", "fr": "transport-fr"}