Coverage for ckanext/udc/tests/graph/test_template.py: 99%
180 statements
« prev ^ index » next coverage.py v7.7.1, created at 2026-01-19 23:48 +0000
1"""
2Tests for graph/template.py - JSON-LD template compilation logic.
4Tests the compile_template and compile_with_temp_value functions that parse
5the mapping config into JSON-LD data.
6"""
7import pytest
8from unittest.mock import Mock, patch
9from ckanext.udc.graph.template import (
10 compile_template,
11 compile_with_temp_value,
12 is_all_attrs_starts_with_at,
13 filter_out_empty_values
14)
15from ckanext.udc.graph.contants import EMPTY_FIELD
18@pytest.fixture(autouse=True)
19def mock_udc_plugin():
20 """Mock the UDC plugin for all tests."""
21 with patch('ckanext.udc.graph.template.get_plugin') as mock_get_plugin:
22 mock_plugin = Mock()
23 mock_plugin.text_fields = []
24 mock_get_plugin.return_value = mock_plugin
25 yield mock_plugin
28class TestIsAllAttrsStartsWithAt:
29 """Test the is_all_attrs_starts_with_at helper function."""
31 def test_all_attrs_start_with_at(self):
32 """Test when all attributes start with @."""
33 data = {"@id": "test", "@type": "Dataset", "@value": "hello"}
34 assert is_all_attrs_starts_with_at(data) is True
36 def test_some_attrs_dont_start_with_at(self):
37 """Test when some attributes don't start with @."""
38 data = {"@id": "test", "name": "Dataset", "@value": "hello"}
39 assert is_all_attrs_starts_with_at(data) is False
41 def test_empty_dict(self):
42 """Test with empty dictionary."""
43 data = {}
44 assert is_all_attrs_starts_with_at(data) is True
47class TestFilterOutEmptyValues:
48 """Test the filter_out_empty_values function."""
50 def test_filter_empty_dicts(self):
51 """Test filtering out empty dictionaries."""
52 data = [{"name": "test"}, {}, {"@id": "uri"}]
53 result = filter_out_empty_values(data)
54 assert len(result) == 2
55 assert {} not in result
57 def test_filter_only_id_dicts(self):
58 """Test keeping dictionaries with only @id."""
59 data = [{"@id": "uri1"}, {"@id": "uri2", "name": "test"}]
60 result = filter_out_empty_values(data)
61 assert len(result) == 2
63 def test_filter_at_attrs_without_value(self):
64 """Test filtering @ attributes without @value."""
65 data = [
66 {"@type": "xsd:string", "@language": "en"}, # No @value
67 {"@type": "xsd:string", "@value": "hello"},
68 ]
69 result = filter_out_empty_values(data)
70 assert len(result) == 1
71 assert result[0]["@value"] == "hello"
74class TestCompileTemplate:
75 """Test the compile_template function."""
77 def test_simple_string_substitution(self):
78 """Test simple f-string style substitution."""
79 template = ["{name}"]
80 global_vars = {}
81 local_vars = {"name": "TestDataset"}
83 result = compile_template(template, global_vars, local_vars)
84 assert result == "TestDataset"
86 def test_empty_string_becomes_empty_field(self):
87 """Test that empty strings are converted to EMPTY_FIELD."""
88 template = ["{name}"]
89 global_vars = {}
90 local_vars = {"name": ""}
92 result = compile_template(template, global_vars, local_vars)
93 assert result == []
95 def test_nested_dict_compilation(self):
96 """Test compilation of nested dictionary structures."""
97 template = [{
98 "@id": "http://example.com/{id}",
99 "@type": "Dataset",
100 "name": "{title}"
101 }]
102 global_vars = {}
103 local_vars = {"id": "123", "title": "Test Dataset"}
105 result = compile_template(template, global_vars, local_vars)
106 assert result["@id"] == "http://example.com/123"
107 assert result["@type"] == "Dataset"
108 assert result["name"] == "Test Dataset"
110 def test_eval_expression(self):
111 """Test eval() expressions in templates."""
112 template = [{
113 "@value": "eval(title.upper())"
114 }]
115 global_vars = {}
116 local_vars = {"title": "hello"}
118 result = compile_template(template, global_vars, local_vars)
119 assert result == "HELLO"
121 def test_eval_with_helper_function(self):
122 """Test eval() with helper functions."""
123 def to_uppercase(val):
124 return val.upper()
126 template = [{
127 "@value": "eval(to_uppercase(name))"
128 }]
129 global_vars = {"to_uppercase": to_uppercase}
130 local_vars = {"name": "dataset"}
132 result = compile_template(template, global_vars, local_vars)
133 assert result == "DATASET"
135 def test_eval_with_text_field_localization(self, mock_udc_plugin):
136 """Test eval() with localized text fields."""
137 # Configure the mock plugin for this test
138 mock_udc_plugin.text_fields = ["title"]
140 def map_to_multiple_languages(val):
141 if isinstance(val, dict):
142 return [{"@language": lang, "@value": value}
143 for lang, value in val.items()]
144 return [{"@language": "en", "@value": val}]
146 template = [{
147 "title": "eval(title)"
148 }]
149 global_vars = {"map_to_multiple_languages": map_to_multiple_languages}
150 local_vars = {"title": {"en": "English Title", "fr": "Titre français"}}
152 result = compile_template(template, global_vars, local_vars)
153 assert len(result["title"]) == 2
154 assert any(item.get("@language") == "en" and item.get("@value") == "English Title"
155 for item in result["title"])
156 assert any(item.get("@language") == "fr" and item.get("@value") == "Titre français"
157 for item in result["title"])
159 def test_nested_list_compilation(self):
160 """Test compilation of nested lists."""
161 template = [{
162 "@id": "http://example.com/dataset",
163 "keywords": [
164 {"@value": "{keyword1}"},
165 {"@value": "{keyword2}"}
166 ]
167 }]
168 global_vars = {}
169 local_vars = {"keyword1": "housing", "keyword2": "transport"}
171 result = compile_template(template, global_vars, local_vars)
172 assert len(result["keywords"]) == 2
173 assert result["keywords"][0]["@value"] == "housing"
174 assert result["keywords"][1]["@value"] == "transport"
176 def test_remove_empty_nested_attrs(self):
177 """Test that empty nested attributes are removed."""
178 template = [{
179 "@id": "http://example.com/dataset",
180 "title": "{title}",
181 "description": "{description}" # Will be empty
182 }]
183 global_vars = {}
184 local_vars = {"title": "Test", "description": ""}
186 result = compile_template(template, global_vars, local_vars)
187 assert "title" in result
188 assert "description" not in result
190 def test_undefined_variable_handling(self):
191 """Test that undefined variables are handled gracefully."""
192 template = [{
193 "@id": "http://example.com/dataset",
194 "name": "{undefined_var}"
195 }]
196 global_vars = {}
197 local_vars = {}
199 result = compile_template(template, global_vars, local_vars)
200 assert "name" not in result
202 def test_complex_nested_structure(self):
203 """Test complex nested structure compilation."""
204 template = [{
205 "@id": "http://example.com/catalogue/{id}",
206 "@type": "Catalogue",
207 "publisher": [{
208 "@id": "http://example.com/org/{org_id}",
209 "@type": "Organization",
210 "name": "{org_name}",
211 "email": "{org_email}"
212 }]
213 }]
214 global_vars = {}
215 local_vars = {
216 "id": "cat123",
217 "org_id": "org456",
218 "org_name": "Test Org",
219 "org_email": "test@example.com"
220 }
222 result = compile_template(template, global_vars, local_vars)
223 assert result["@id"] == "http://example.com/catalogue/cat123"
224 assert len(result["publisher"]) == 1
225 assert result["publisher"][0]["@id"] == "http://example.com/org/org456"
226 assert result["publisher"][0]["name"] == "Test Org"
229class TestCompileWithTempValue:
230 """Test the compile_with_temp_value function."""
232 def test_simple_substitution(self):
233 """Test simple substitution with available values."""
234 mappings = [{"@id": "http://example.com/{id}", "name": "{name}"}]
235 global_vars = {}
236 local_vars = {"id": "123", "name": "Test"}
238 result = compile_with_temp_value(mappings, global_vars, local_vars)
239 assert result["@id"] == "http://example.com/123"
240 assert result["name"] == "Test"
242 def test_temp_value_for_undefined_vars(self):
243 """Test that undefined variables are replaced with TEMP_VALUE."""
244 mappings = [{"@id": "http://example.com/{id}", "name": "{undefined}"}]
245 global_vars = {}
246 local_vars = {"id": "123"}
248 result = compile_with_temp_value(mappings, global_vars, local_vars)
249 assert result["@id"] == "http://example.com/123"
250 assert result["name"] == "TEMP_VALUE"
252 def test_nested_temp_values(self):
253 """Test temp values in nested structures."""
254 mappings = [{
255 "@id": "http://example.com/cat",
256 "creator": [{
257 "@id": "http://example.com/person/{person_id}",
258 "name": "{person_name}"
259 }]
260 }]
261 global_vars = {}
262 local_vars = {}
264 result = compile_with_temp_value(mappings, global_vars, local_vars)
265 assert "TEMP_VALUE" in result["creator"]["@id"]
266 assert result["creator"]["name"] == "TEMP_VALUE"
268 def test_preserves_structure(self):
269 """Test that the structure is preserved even with missing values."""
270 mappings = [{
271 "@id": "uri",
272 "title": "{title}",
273 "description": "{desc}",
274 "publisher": [{
275 "name": "{pub_name}"
276 }]
277 }]
278 global_vars = {}
279 local_vars = {}
281 result = compile_with_temp_value(mappings, global_vars, local_vars)
282 assert "@id" in result
283 assert "title" in result
284 assert "description" in result
285 assert "publisher" in result
286 assert len(result["publisher"]) == 1
289class TestRealWorldScenarios:
290 """Test with real-world-like data and mapping scenarios."""
292 def test_catalogue_entry_mapping(self):
293 """Test mapping a complete catalogue entry."""
294 def map_to_multiple_languages(val):
295 if isinstance(val, dict):
296 return [{"@language": lang, "@value": value}
297 for lang, value in val.items()]
298 return [{"@language": "en", "@value": val}]
300 template = [{
301 "@id": "http://data.urbandatacentre.ca/catalogue/{id}",
302 "@type": "http://data.urbandatacentre.ca/catalogue",
303 "dct:title": "eval(map_to_multiple_languages(title))",
304 "dct:description": "eval(map_to_multiple_languages(description))",
305 "dct:issued": {
306 "@type": "xsd:date",
307 "@value": "{published_date}"
308 }
309 }]
311 global_vars = {"map_to_multiple_languages": map_to_multiple_languages}
312 local_vars = {
313 "id": "dataset-001",
314 "title": {"en": "Housing Data", "fr": "Données sur le logement"},
315 "description": {"en": "Housing statistics", "fr": "Statistiques de logement"},
316 "published_date": "2025-01-01"
317 }
319 result = compile_template(template, global_vars, local_vars)
321 assert result["@id"] == "http://data.urbandatacentre.ca/catalogue/dataset-001"
322 assert len(result["dct:title"]) == 2
323 assert len(result["dct:description"]) == 2
324 assert result["dct:issued"][0]["@value"] == "2025-01-01"
326 def test_with_optional_fields(self):
327 """Test mapping with optional fields that may be empty."""
328 template = [{
329 "@id": "http://example.com/{id}",
330 "required_field": "{title}",
331 "optional_field": "{optional}",
332 "another_optional": "{also_optional}"
333 }]
335 global_vars = {}
336 local_vars = {
337 "id": "123",
338 "title": "Required Title",
339 "optional": "", # Empty optional
340 # also_optional is not provided
341 }
343 result = compile_template(template, global_vars, local_vars)
345 assert result["@id"] == "http://example.com/123"
346 assert result["required_field"] == "Required Title"
347 assert "optional_field" not in result
348 assert "another_optional" not in result
350 def test_uri_list_generation(self):
351 """Test generating a list of URIs from comma-separated values."""
352 def split_to_uris(val, separator=","):
353 return [{"@id": uri.strip()} for uri in val.split(separator)]
355 template = [{
356 "@id": "http://example.com/dataset",
357 "formats": "eval(split_to_uris(file_format))"
358 }]
360 global_vars = {"split_to_uris": split_to_uris}
361 local_vars = {"file_format": "csv,json,xml"}
363 result = compile_template(template, global_vars, local_vars)
365 assert len(result["formats"]) == 3
366 assert {"@id": "csv"} in result["formats"]
367 assert {"@id": "json"} in result["formats"]
368 assert {"@id": "xml"} in result["formats"]