Coverage for ckanext/udc/tests/test_plugin.py: 100%

1""" 

2Tests for plugin.py. 

3 

4Tests are written using the pytest library (https://docs.pytest.org), and you 

5should read the testing guidelines in the CKAN docs: 

6https://docs.ckan.org/en/2.9/contributing/testing.html 

7 

8To write tests for your extension you should install the pytest-ckan package: 

9 

10 pip install pytest-ckan 

11 

12This will allow you to use CKAN specific fixtures on your tests. 

13 

14For instance, if your test involves database access you can use `clean_db` to 

15reset the database: 

16 

17 import pytest 

18 

19 from ckan.tests import factories 

20 

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

22 def test_some_action(): 

23 

24 dataset = factories.Dataset() 

25 

26 # ... 

27 

28For functional tests that involve requests to the application, you can use the 

29`app` fixture: 

30 

31 from ckan.plugins import toolkit 

32 

33 def test_some_endpoint(app): 

34 

35 url = toolkit.url_for('myblueprint.some_endpoint') 

36 

37 response = app.get(url) 

38 

39 assert response.status_code == 200 

40 

41 

42To temporary patch the CKAN configuration for the duration of a test you can use: 

43 

44 import pytest 

45 

46 @pytest.mark.ckan_config("ckanext.myext.some_key", "some_value") 

47 def test_some_action(): 

48 pass 

49""" 

50from pathlib import Path 

51from types import SimpleNamespace 

52 

53import pytest 

54 

55import ckan.plugins.toolkit as tk 

56 

57import ckanext.udc.plugin as plugin 

58from ckanext.udc.solr import index as udc_index 

59from ckanext.udc.solr import solr as udc_solr 

60from ckanext.udc.i18n import ( 

61 udc_json_load, 

62 udc_lang_object, 

63 udc_json_dump, 

64) 

65 

66 

67@pytest.fixture() 

68def udc_plugin(monkeypatch): 

69 monkeypatch.setattr(plugin, "update_solr_maturity_model_fields", lambda *_args, **_kwargs: None) 

70 

71 instance = plugin.UdcPlugin() 

72 # Force a clean slate for each test since the plugin stores state on the instance 

73 instance.disable_graphdb = True 

74 instance.sparql_client = None 

75 instance.all_fields = [] 

76 instance.facet_titles = {} 

77 instance.facet_titles_raw = {} 

78 instance.text_fields = [] 

79 instance.date_fields = [] 

80 instance.multiple_select_fields = [] 

81 instance.dropdown_options = {} 

82 instance.maturity_model = [] 

83 instance.mappings = {} 

84 instance.preload_ontologies = {} 

85 return instance 

86 

87 

88def _sample_config(): 

89 return { 

90 "maturity_model": [ 

91 { 

92 "title": "Level 1", 

93 "name": "lvl1", 

94 "fields": [ 

95 {"name": "text_field", "label": "Text Field", "type": "text"}, 

96 {"name": "date_field", "label": "Date Field", "type": "date"}, 

97 { 

98 "name": "multi_field", 

99 "label": "Multi Field", 

100 "type": "multiple_select", 

101 "options": [ 

102 {"value": "opt-a", "text": "Option A"}, 

103 {"value": "opt-b", "text": "Option B"}, 

104 ], 

105 }, 

106 { 

107 "name": "single_field", 

108 "label": "Single Field", 

109 "type": "single_select", 

110 "options": [ 

111 {"value": "one", "text": "One"}, 

112 ], 

113 }, 

114 ], 

115 } 

116 ], 

117 "mappings": {"foo": "bar"}, 

118 "preload_ontologies": {"example": "value"}, 

119 } 

120 

121 

122def test_reload_config_populates_field_metadata(udc_plugin): 

123 udc_plugin.reload_config(_sample_config()) 

124 

125 assert udc_plugin.all_fields == [ 

126 "text_field", 

127 "date_field", 

128 "multi_field", 

129 "single_field", 

130 ] 

131 assert udc_plugin.text_fields == ["text_field"] 

132 assert udc_plugin.date_fields == ["date_field"] 

133 assert udc_plugin.multiple_select_fields == ["multi_field"] 

134 assert udc_plugin.dropdown_options["multi_field"] == { 

135 "opt-a": "Option A", 

136 "opt-b": "Option B", 

137 } 

138 assert udc_plugin.facet_titles["single_field"] == "Single Field" 

139 assert udc_plugin.mappings == {"foo": "bar"} 

140 assert udc_plugin.preload_ontologies == {"example": "value"} 

141 

142 

143def test_modify_package_schema_applies_expected_validators(udc_plugin): 

144 udc_plugin.reload_config(_sample_config()) 

145 

146 base_schema = {} 

147 schema = udc_plugin._modify_package_schema(base_schema) 

148 

149 ignore_missing = tk.get_validator("ignore_missing") 

150 convert_to_extras = tk.get_converter("convert_to_extras") 

151 

152 text_pipeline = schema["text_field"] 

153 assert text_pipeline[0] == ignore_missing 

154 # text fields should round-trip JSON with multilingual helpers 

155 assert udc_json_load in text_pipeline 

156 assert udc_lang_object in text_pipeline 

157 assert udc_json_dump in text_pipeline 

158 assert text_pipeline[-1] == convert_to_extras 

159 

160 number_pipeline = schema["date_field"] 

161 assert number_pipeline == [ignore_missing, convert_to_extras] 

162 

163 

164def test_reload_config_supports_portal_type_ckan_field(udc_plugin): 

165 config = { 

166 "maturity_model": [ 

167 { 

168 "title": "Level 1", 

169 "name": "lvl1", 

170 "fields": [ 

171 { 

172 "ckanField": "portal_type", 

173 "label": {"en": "Portal type", "fr": "Type de portail"}, 

174 "type": "single_select", 

175 "options": [ 

176 {"value": "CKAN", "text": "CKAN"}, 

177 {"value": "ArcGIS", "text": "ArcGIS"}, 

178 ], 

179 } 

180 ], 

181 } 

182 ], 

183 "mappings": {}, 

184 "preload_ontologies": {}, 

185 } 

186 

187 udc_plugin.reload_config(config) 

188 

189 assert udc_plugin.facet_titles["portal_type"] == "Portal type" 

190 assert udc_plugin.dropdown_options["portal_type"] == { 

191 "CKAN": "CKAN", 

192 "ArcGIS": "ArcGIS", 

193 } 

194 

195 

196def test_ckan_fields_template_supports_portal_type_macro(): 

197 template_path = Path(__file__).resolve().parents[1] / "templates" / "package" / "macros" / "ckan_fields.html" 

198 content = template_path.read_text() 

199 

200 assert "{% macro portal_type(data, errors, short_description=\"\", long_description=\"\") %}" in content 

201 assert "'portal_type'" in content 

202 

203 

204def test_update_solr_fields_includes_portal_type(monkeypatch): 

205 added_fields = [] 

206 

207 monkeypatch.setattr(udc_solr, "get_udc_langs", lambda: ["en", "fr"]) 

208 monkeypatch.setattr(udc_solr, "ensure_language_dynamic_fields", lambda langs: None) 

209 monkeypatch.setattr(udc_solr, "get_extras_fields", lambda: {}) 

210 monkeypatch.setattr(udc_solr, "get_fields", lambda: {"tags_ngram": {}}) 

211 monkeypatch.setattr(udc_solr, "add_copy_field", lambda *args, **kwargs: None) 

212 monkeypatch.setattr(udc_solr, "delete_copy_field", lambda *args, **kwargs: None) 

213 monkeypatch.setattr(udc_solr, "delete_field", lambda *args, **kwargs: None) 

214 monkeypatch.setattr(udc_solr, "add_field", lambda *args, **kwargs: added_fields.append((args, kwargs))) 

215 

216 udc_solr.update_solr_maturity_model_fields([ 

217 { 

218 "title": "Level 1", 

219 "name": "lvl1", 

220 "fields": [ 

221 { 

222 "ckanField": "portal_type", 

223 "label": {"en": "Portal type"}, 

224 "type": "single_select", 

225 "options": [ 

226 {"value": "ArcGIS", "text": "ArcGIS"}, 

227 {"value": "CKAN", "text": "CKAN"}, 

228 ], 

229 } 

230 ], 

231 } 

232 ]) 

233 

234 field_names = [call[0][0] for call in added_fields] 

235 assert "extras_portal_type" in field_names 

236 

237 portal_type_call = next(call for call in added_fields if call[0][0] == "extras_portal_type") 

238 assert portal_type_call[0][1] == "string" 

239 assert portal_type_call[0][4] is True 

240 

241 

242def test_before_dataset_index_handles_plain_text_values_for_text_fields(monkeypatch): 

243 mock_plugin = SimpleNamespace( 

244 multiple_select_fields=[], 

245 text_fields=["unique_identifier"], 

246 ) 

247 

248 monkeypatch.setattr(udc_index.plugins, "get_plugin", lambda _name: mock_plugin) 

249 monkeypatch.setattr(udc_index, "get_udc_langs", lambda: ["en", "fr"]) 

250 

251 indexed = udc_index.before_dataset_index( 

252 { 

253 "id": "pkg-1", 

254 "name": "example-dataset", 

255 "title": "Example Dataset", 

256 "notes": "Example notes", 

257 "unique_identifier": "10", 

258 } 

259 ) 

260 

261 assert indexed["unique_identifier_en_txt"] == "10" 

262 assert indexed["unique_identifier_en_f"] == ["10"] 

263 assert "unique_identifier_fr_txt" not in indexed