Coverage for tests / test_folder_metadata_builder.py: 96%
102 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-07-17 16:43 +0000
« prev ^ index » next coverage.py v7.12.0, created at 2026-07-17 16:43 +0000
1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import shutil
6import tempfile
8import pytest
9from pathlib import Path
10from rdflib import Dataset, Graph, Namespace
12from changes_metadata_manager.folder_metadata_builder import (
13 extract_metadata_for_stage,
14 extract_id_from_folder_name,
15 load_kg,
16 merge_provenance_files,
17 process_all_folders,
18 scan_folder_structure,
19)
22DATA_DIR = Path(__file__).parent.parent / "data"
23FIXTURE_DIR = Path(__file__).parent / "fixtures" / "folder_metadata"
24REAL_KG_PATH = DATA_DIR / "kg.ttl"
26TEST_ITEMS = [
27 ("Sala1", "S1-01-CNR_CartaNautica", "1"),
28 ("Sala2", "S2-24-CNR_MappaOrtoBotanicoPADOVA", "24"),
29 ("Sala5", "S5-57-FICLIT_VolumePolpo", "57"),
30]
33def load_fixture(relative_path: str) -> Graph:
34 g = Graph()
35 g.parse(FIXTURE_DIR / relative_path, format="turtle")
36 return g
39def assert_graphs_equal(actual: Graph, expected: Graph):
40 actual_triples = set(actual)
41 expected_triples = set(expected)
43 missing = expected_triples - actual_triples
44 extra = actual_triples - expected_triples
46 if missing:
47 missing_str = "\n".join(
48 f" {s} {p} {o}" for s, p, o in sorted(missing, key=str)
49 )
50 raise AssertionError(f"Missing {len(missing)} triples:\n{missing_str}")
52 if extra:
53 extra_str = "\n".join(f" {s} {p} {o}" for s, p, o in sorted(extra, key=str))
54 raise AssertionError(f"Extra {len(extra)} triples:\n{extra_str}")
57@pytest.fixture(scope="module")
58def real_kg():
59 return load_kg(REAL_KG_PATH)
62STAGES = ["raw", "rawp", "dcho", "dchoo"]
65@pytest.fixture
66def test_folder_structure():
67 tmpdir = tempfile.mkdtemp()
68 root_path = Path(tmpdir) / "root"
70 for sala, folder, _ in TEST_ITEMS:
71 for stage in STAGES:
72 stage_dir = root_path / sala / folder / stage
73 stage_dir.mkdir(parents=True)
75 yield root_path
77 shutil.rmtree(tmpdir)
80class TestExtractMetadataForStageExact:
81 @pytest.mark.parametrize(
82 "nr,stage",
83 [
84 ("1", "raw"),
85 ("1", "rawp"),
86 ("1", "dcho"),
87 ("1", "dchoo"),
88 ("24", "raw"),
89 ("24", "rawp"),
90 ("24", "dcho"),
91 ("24", "dchoo"),
92 ("57", "raw"),
93 ("57", "rawp"),
94 ("57", "dcho"),
95 ("57", "dchoo"),
96 ],
97 )
98 def test_stage_output_matches_fixture(self, real_kg, nr, stage):
99 result = extract_metadata_for_stage(real_kg, nr, stage)
100 expected = load_fixture(f"nr_{nr}/{stage}.ttl")
101 assert_graphs_equal(result, expected)
104class TestExtractIdFromFolderName:
105 @pytest.mark.parametrize(
106 "folder_name,expected",
107 [
108 ("S1-5-nome_oggetto", "5"),
109 ("S2-42-altro_nome", "42"),
110 ("S6-123-oggetto_complesso", "123"),
111 ("S1-7-nome con spazi", "7"),
112 ("S1-01-CNR_CartaNautica", "1"),
113 ("S2-27a-FICLIT_DelphiniumStaphisagria", "27a"),
114 ("S6-74b-ISPC-Orchis_morio_L", "74b"),
115 ],
116 )
117 def test_valid_folder_names(self, folder_name, expected):
118 assert extract_id_from_folder_name(folder_name) == expected
120 @pytest.mark.parametrize(
121 "folder_name",
122 [
123 "1-5-nome",
124 "Sala1-5-nome",
125 "S1_5_nome",
126 ],
127 )
128 def test_invalid_folder_names(self, folder_name):
129 with pytest.raises(ValueError, match="Cannot extract ID"):
130 extract_id_from_folder_name(folder_name)
133class TestProcessAllFolders:
134 def test_creates_files_in_place(self, test_folder_structure):
135 root = test_folder_structure
136 process_all_folders(
137 root=root,
138 kg_path=REAL_KG_PATH,
139 )
141 for sala, folder, _ in TEST_ITEMS:
142 folder_dir = root / sala / folder
143 stage_dirs = [d for d in folder_dir.iterdir() if d.is_dir()]
144 assert len(stage_dirs) == 4, (
145 f"Expected 4 stages for {folder}, got {len(stage_dirs)}"
146 )
148 for stage_dir in stage_dirs:
149 meta_file = stage_dir / "meta.ttl"
150 prov_file = stage_dir / "prov.trig"
151 assert meta_file.exists(), (
152 f"meta.ttl not created for {folder}/{stage_dir.name}"
153 )
154 assert prov_file.exists(), (
155 f"prov.trig not created for {folder}/{stage_dir.name}"
156 )
159class TestMergeProvenanceFiles:
160 def test_merges_all_prov_trig_files(self, test_folder_structure):
161 root = test_folder_structure
162 process_all_folders(
163 root=root,
164 kg_path=REAL_KG_PATH,
165 )
167 output_path = root / "all_provenance.trig"
168 merge_provenance_files(root, output_path)
170 assert output_path.exists()
172 merged = Dataset()
173 merged.parse(str(output_path), format="trig")
175 PROV = Namespace("http://www.w3.org/ns/prov#")
176 prov_entities = set(merged.quads((None, PROV.specializationOf, None, None)))
177 individual_count = sum(1 for _ in root.rglob("prov.trig"))
179 assert individual_count == 12
180 assert len(prov_entities) == 249
183class TestScanFolderStructure:
184 def test_scans_folder_structure(self):
185 with tempfile.TemporaryDirectory() as tmpdir:
186 root = Path(tmpdir)
187 sala_dir = root / "Sala1"
188 folder_dir = sala_dir / "S1-01-TestFolder"
189 raw_dir = folder_dir / "raw"
190 dcho_dir = folder_dir / "dcho"
191 raw_dir.mkdir(parents=True)
192 dcho_dir.mkdir(parents=True)
193 (raw_dir / "file1.jpg").touch()
194 (raw_dir / "file2.jpg").touch()
195 (dcho_dir / "model.obj").touch()
197 result = scan_folder_structure(root)
199 assert "structure" in result
200 assert "Sala1" in result["structure"]
201 assert "S1-01-TestFolder" in result["structure"]["Sala1"]
202 folder_data = result["structure"]["Sala1"]["S1-01-TestFolder"]
203 assert "raw" in folder_data
204 assert "dcho" in folder_data
205 assert set(folder_data["raw"]["_files"]) == {"file1.jpg", "file2.jpg"}
206 assert folder_data["dcho"]["_files"] == ["model.obj"]