|
| 1 | +""" |
| 2 | +tests/test_inbox_to_schema.py |
| 3 | +
|
| 4 | +Regression tests for the inbox → schema pipeline (scripts/inbox_to_schema.py), |
| 5 | +focused on adding *new* slots that are assigned to a domain (sub-)class via the |
| 6 | +Excel "domain" column. |
| 7 | +
|
| 8 | +Background |
| 9 | +---------- |
| 10 | +Previously, a new slot whose "domain" column named a subclass (e.g. anode / |
| 11 | +cathode → ElectrochemicalReactor) was silently dropped: it was neither a known |
| 12 | +global slot nor present in the class slot_usage, so plan_changes routed it into |
| 13 | +the "structural display row" skip branch and _plan_new_slot rejected any |
| 14 | +non-empty domain outright. These tests pin the corrected behaviour: |
| 15 | +
|
| 16 | + • a genuinely new domain-assigned slot is planned as a slot_add targeting the |
| 17 | + named subclass and the YAML module that defines it; |
| 18 | + • a real structural slot (own slots:/mixin of the domain class) keeps being |
| 19 | + skipped and is never planned as a new slot; |
| 20 | + • the top-level (empty-domain) path is unchanged; |
| 21 | + • an unrecognised domain is reported as an error rather than written. |
| 22 | +
|
| 23 | +The planning-level tests are read-only (plan_changes never writes); the apply |
| 24 | +test copies the schema into a tmp dir and redirects SCHEMA_DIR so nothing in the |
| 25 | +repository is mutated. |
| 26 | +""" |
| 27 | + |
| 28 | +from __future__ import annotations |
| 29 | + |
| 30 | +import shutil |
| 31 | +import sys |
| 32 | +from pathlib import Path |
| 33 | + |
| 34 | +import pytest |
| 35 | + |
| 36 | +# ── path setup ────────────────────────────────────────────────────────────── |
| 37 | +_ROOT = Path(__file__).parent.parent |
| 38 | +_SCRIPTS = _ROOT / "scripts" |
| 39 | +_SCHEMA = _ROOT / "src" / "coremeta4cat" / "schema" |
| 40 | + |
| 41 | +sys.path.insert(0, str(_SCRIPTS)) |
| 42 | + |
| 43 | +import inbox_to_schema as ib # noqa: E402 |
| 44 | +from generate_schema_docs import load_merged_schema # noqa: E402 |
| 45 | + |
| 46 | + |
| 47 | +# ── helpers ───────────────────────────────────────────────────────────────── |
| 48 | + |
| 49 | +def _slot_row(label: str, domain: str = "", mro: str = "O", |
| 50 | + range_: str = "string", **extra) -> dict: |
| 51 | + """Build one normalised slot row, mirroring parse_excel's output shape.""" |
| 52 | + return { |
| 53 | + "label": label, |
| 54 | + "type": "slot", |
| 55 | + "domain": domain, |
| 56 | + "mro": mro, |
| 57 | + "range": range_, |
| 58 | + "multivalued": extra.get("multivalued", ""), |
| 59 | + "inlined_as_list": extra.get("inlined_as_list", ""), |
| 60 | + "unit": extra.get("unit", ""), |
| 61 | + "uri": extra.get("uri", ""), |
| 62 | + "description": extra.get("description", ""), |
| 63 | + } |
| 64 | + |
| 65 | + |
| 66 | +def _plan(indexes, excel_data): |
| 67 | + """Run the read-only planning phase and return (changes, reporter).""" |
| 68 | + schema, slot_origin, class_origin, label_to_slot, label_to_class = indexes |
| 69 | + reporter = ib.Reporter() |
| 70 | + changes = ib.plan_changes( |
| 71 | + schema, excel_data, |
| 72 | + label_to_slot, label_to_class, |
| 73 | + slot_origin, class_origin, |
| 74 | + reporter, |
| 75 | + ) |
| 76 | + return changes, reporter |
| 77 | + |
| 78 | + |
| 79 | +@pytest.fixture(scope="module") |
| 80 | +def indexes(): |
| 81 | + """Load the merged schema and build the name/label indexes once.""" |
| 82 | + schema = load_merged_schema(str(_SCHEMA)) |
| 83 | + slot_origin, class_origin = ib.build_origin_index(_SCHEMA) |
| 84 | + label_to_slot, label_to_class = ib.build_label_index(schema) |
| 85 | + return schema, slot_origin, class_origin, label_to_slot, label_to_class |
| 86 | + |
| 87 | + |
| 88 | +# ── planning-level tests (read-only) ──────────────────────────────────────── |
| 89 | + |
| 90 | +def test_new_slot_with_domain_is_planned_for_subclass(indexes): |
| 91 | + """A new slot assigned to an existing subclass is planned as a slot_add |
| 92 | + targeting that subclass and the YAML module that defines it. |
| 93 | +
|
| 94 | + Uses a synthetic slot name that is never shipped in the schema, so the test |
| 95 | + stays valid even after real domain slots (e.g. anode/cathode) have been |
| 96 | + applied via the pipeline. A hard-coded real name would stop being "new" once |
| 97 | + it lands in the schema and would silently invalidate the assertion. |
| 98 | + """ |
| 99 | + schema = indexes[0] |
| 100 | + label_to_slot = indexes[3] |
| 101 | + |
| 102 | + label = "qa synthetic electrode probe" |
| 103 | + name = "qa_synthetic_electrode_probe" |
| 104 | + |
| 105 | + # Precondition: the slot must genuinely be absent for this test to be |
| 106 | + # meaningful. Asserting it makes the assumption explicit and self-checking. |
| 107 | + assert name not in schema.get("slots", {}) |
| 108 | + assert label not in label_to_slot |
| 109 | + assert name not in ib.get_all_class_slots(schema, "ElectrochemicalReactor") |
| 110 | + |
| 111 | + excel = {"Reaction": [ |
| 112 | + _slot_row(label, domain="ElectrochemicalReactor", mro="M"), |
| 113 | + ]} |
| 114 | + changes, reporter = _plan(indexes, excel) |
| 115 | + |
| 116 | + adds = [c for c in changes if c["type"] == "slot_add" and c["name"] == name] |
| 117 | + assert len(adds) == 1, f"expected exactly one slot_add for '{name}'" |
| 118 | + add = adds[0] |
| 119 | + assert add["schema_class"] == "ElectrochemicalReactor" |
| 120 | + assert add["mro"] == "M" |
| 121 | + assert Path(add["_target"]).name == "coremeta4cat_reaction_ap.yaml" |
| 122 | + assert not reporter.has_errors |
| 123 | + |
| 124 | + |
| 125 | +def test_structural_subclass_slot_is_not_added(indexes): |
| 126 | + """A slot that already belongs to the domain class -- via the class's own |
| 127 | + slots: list or via a mixin -- keeps being skipped and is never planned as a |
| 128 | + new slot.""" |
| 129 | + # 'title'/'description' are in QuantitativeRange.slots:; 'has concentration' |
| 130 | + # reaches CoPrecipitation via PrecipitationMixin. All three are real |
| 131 | + # structural rows emitted by the Excel generator and must stay skips. |
| 132 | + excel = { |
| 133 | + "Reaction": [ |
| 134 | + _slot_row("title", domain="QuantitativeRange"), |
| 135 | + _slot_row("description", domain="QuantitativeRange"), |
| 136 | + ], |
| 137 | + "Synthesis": [ |
| 138 | + _slot_row("has concentration", domain="CoPrecipitation"), |
| 139 | + ], |
| 140 | + } |
| 141 | + changes, reporter = _plan(indexes, excel) |
| 142 | + |
| 143 | + new_names = {c["name"] for c in changes if c["type"] == "slot_add"} |
| 144 | + assert "title" not in new_names |
| 145 | + assert "description" not in new_names |
| 146 | + assert "has_concentration" not in new_names |
| 147 | + assert not reporter.has_errors |
| 148 | + |
| 149 | + |
| 150 | +def test_top_level_new_slot_still_targets_sheet_class(indexes): |
| 151 | + """Regression guard: an empty-domain new slot still attaches to the sheet's |
| 152 | + top-level data class (CatalyticReaction for the Reaction sheet).""" |
| 153 | + excel = {"Reaction": [ |
| 154 | + _slot_row("brand new toplevel field", domain="", mro="R"), |
| 155 | + ]} |
| 156 | + changes, reporter = _plan(indexes, excel) |
| 157 | + |
| 158 | + add = next( |
| 159 | + (c for c in changes |
| 160 | + if c["type"] == "slot_add" and c["name"] == "brand_new_toplevel_field"), |
| 161 | + None, |
| 162 | + ) |
| 163 | + assert add is not None |
| 164 | + assert add["schema_class"] == "CatalyticReaction" |
| 165 | + assert not reporter.has_errors |
| 166 | + |
| 167 | + |
| 168 | +def test_new_slot_with_unknown_domain_is_an_error(indexes): |
| 169 | + """A new slot whose domain is not a known class is reported as an error and |
| 170 | + not planned for application.""" |
| 171 | + excel = {"Reaction": [ |
| 172 | + _slot_row("weird slot", domain="NotARealClass"), |
| 173 | + ]} |
| 174 | + changes, reporter = _plan(indexes, excel) |
| 175 | + |
| 176 | + assert reporter.has_errors |
| 177 | + assert not any( |
| 178 | + c["type"] == "slot_add" and c["name"] == "weird_slot" for c in changes |
| 179 | + ) |
| 180 | + |
| 181 | + |
| 182 | +# ── apply-level test (writes into a tmp copy, never the repo) ──────────────── |
| 183 | + |
| 184 | +def test_apply_writes_new_domain_slot_into_subclass(tmp_path, monkeypatch): |
| 185 | + """End-to-end: applying a domain-assigned slot_add creates the global slot |
| 186 | + definition and references it from the subclass's slots: list.""" |
| 187 | + dst = tmp_path / "schema" |
| 188 | + dst.mkdir() |
| 189 | + for f in _SCHEMA.glob("*.yaml"): |
| 190 | + shutil.copy(f, dst / f.name) |
| 191 | + monkeypatch.setattr(ib, "SCHEMA_DIR", dst) |
| 192 | + |
| 193 | + reporter = ib.Reporter() |
| 194 | + schema = load_merged_schema(str(dst)) |
| 195 | + slot_origin, class_origin = ib.build_origin_index(dst) |
| 196 | + label_to_slot, label_to_class = ib.build_label_index(schema) |
| 197 | + |
| 198 | + excel = {"Reaction": [ |
| 199 | + _slot_row("test electrode", domain="ElectrochemicalReactor", mro="M", |
| 200 | + description="A test electrode slot."), |
| 201 | + ]} |
| 202 | + changes = ib.plan_changes( |
| 203 | + schema, excel, label_to_slot, label_to_class, |
| 204 | + slot_origin, class_origin, reporter, |
| 205 | + ) |
| 206 | + |
| 207 | + # Apply only our slot_add so unrelated deletion-detection changes (the |
| 208 | + # single-row workbook makes every other Reaction slot look "missing") do |
| 209 | + # not mutate the copy. |
| 210 | + adds = [c for c in changes |
| 211 | + if c["type"] == "slot_add" and c["name"] == "test_electrode"] |
| 212 | + assert len(adds) == 1 |
| 213 | + ib.apply_changes(adds, reporter) |
| 214 | + |
| 215 | + doc = ib._load_yaml(dst / "coremeta4cat_reaction_ap.yaml") |
| 216 | + |
| 217 | + # (a) global slot definition was created with the required flag |
| 218 | + assert "test_electrode" in (doc.get("slots") or {}) |
| 219 | + assert doc["slots"]["test_electrode"].get("required") is True |
| 220 | + |
| 221 | + # (b) the subclass references the new slot in its slots: list |
| 222 | + er = (doc.get("classes") or {}).get("ElectrochemicalReactor") or {} |
| 223 | + assert "test_electrode" in (er.get("slots") or []) |
0 commit comments