Fail closed on SPSS compatible names

Author: TonisOrmisson

src/openstatspec/spss/sav.py

@@ -147,6 +147,8 @@ def inspect_sav(source: str | Path) -> dict[str, Any]:
     _require_source(source_path)
     metadata, loss_report = _dictionary(source_path)
     names = list(metadata.get("var_names") or [])
+    variables = _variables(metadata, names)
+    loss_report = _merge_loss_reports(tuple(loss_report.values()), _compat_name_loss_report(variables))
     return {
         "source_format": source_path.suffix[1:].upper(),
         "source_name": source_path.name,
@@ -157,9 +159,9 @@ def inspect_sav(source: str | Path) -> dict[str, Any]:
         "file_attributes": dict(metadata.get("file_attributes") or {}),
         "case_weight_variable": metadata.get("case_weight_var") or None,
         "multiple_response_sets": dict(metadata.get("mrsets") or {}),
-        "loss_report": tuple(loss_report.values()),
+        "loss_report": loss_report,
         "variable_count": len(names),
-        "variables": _variables(metadata, names),
+        "variables": variables,
     }
 
 
@@ -175,6 +177,7 @@ def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str
     # frame read's values where a future pyspssio version exposes more there.
     metadata = {**dictionary, **metadata}
     variables = _variables(metadata, list(frame.columns))
+    loss_report = _merge_loss_reports(tuple(loss_report.values()), _compat_name_loss_report(variables))
     result = create_wide_dataset(
         database_url=database_url,
         dataset_id=dataset_id,
@@ -196,9 +199,9 @@ def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str
             {"spss.variable_sets": metadata["_var_sets"]}
             if metadata.get("_var_sets") else {}
         ),
-        fidelity_events=tuple(loss_report.values()),
+        fidelity_events=loss_report,
     )
-    return {**result, "loss_report": tuple(loss_report.values())}
+    return {**result, "loss_report": loss_report}
 
 
 def export_sav_dataset(
@@ -215,8 +218,8 @@ def export_sav_dataset(
         multiple_response_sets=dataset.get("multiple_response_sets"),
     )
     loss_report = _merge_loss_reports(
+        _export_loss_report(dataset, variables),
         read_fidelity_events(database_url=database_url, dataset_id=dataset_id),
-        _export_loss_report(dataset),
     )
     rejected = [event["code"] for event in loss_report if event["code"] not in allow_loss]
     if rejected:
@@ -365,14 +368,48 @@ def _engine_loss_report(metadata: dict[str, Any]) -> dict[str, dict[str, Any]]:
     return events
 
 
-def _export_loss_report(dataset: dict[str, Any]) -> tuple[dict[str, Any], ...]:
+def _compat_name_loss_report(variables: list[dict[str, Any]]) -> tuple[dict[str, Any], ...]:
+    """Report legacy SPSS compatible names the writer cannot set explicitly.
+
+    A compatible name is meaningful only when it differs from the long source
+    variable name. ``pyspssio`` exposes these names while reading, but its
+    public writer metadata has no ``var_compat_names`` input. It may derive a
+    name today, but that is not a preservation contract: export must therefore
+    require an explicit, auditable acceptance rather than silently rederive or
+    rename the value.
+    """
+    events: list[dict[str, Any]] = []
+    for variable in variables:
+        source_name = str(variable["source_name"])
+        compat_name = variable.get("compat_name")
+        if compat_name is None or str(compat_name).casefold() == source_name.casefold():
+            continue
+        events.append({
+            "code": "compatible-variable-name-not-exportable",
+            "detail": (
+                "pyspssio exposes the source compatible variable name but its public "
+                "writer API cannot set or guarantee preservation of that name."
+            ),
+            "details": {
+                "source_name": source_name,
+                "compatible_name": str(compat_name),
+                "physical_name": str(variable["physical_name"]),
+            },
+        })
+    return tuple(events)
+
+
+def _export_loss_report(
+    dataset: dict[str, Any], variables: list[dict[str, Any]],
+) -> tuple[dict[str, Any], ...]:
     events: list[dict[str, Any]] = []
     if _is_non_utf8_encoding(dataset.get("source_encoding")):
         events.append({
             "code": "source-encoding-not-preserved",
             "detail": "The pyspssio writer has no source-encoding preservation contract for this legacy code page.",
             "details": {"source_encoding": dataset.get("source_encoding")},
         })
+    events.extend(_compat_name_loss_report(variables))
     return tuple(events)
 
 

tests/test_loss_reports.py

@@ -1,3 +1,4 @@
+import json
 import sqlite3
 
 import pandas as pd
@@ -68,4 +69,66 @@ def test_non_utf8_source_encoding_is_explicit_export_loss(tmp_path) -> None:
     exported = openstatspec.export_sav(database_url=database, dataset_id="legacy-encoding", destination=destination, allow_loss=["source-encoding-not-preserved"])
     assert {diagnostic.code for diagnostic in exported.diagnostics} == {"source-encoding-not-preserved"}
     metadata = pyspssio.read_metadata(str(destination))
-    assert metadata["encoding"] == "UTF-8"
+    assert metadata["encoding"] == "UTF-8"
+
+@pytest.mark.parametrize("suffix", [".sav", ".zsav"])
+def test_compatible_variable_name_requires_explicit_audited_export_loss(tmp_path, suffix: str) -> None:
+    """A long SPSS name exposes a legacy compatible name the writer cannot set."""
+    source = tmp_path / f"compat-source{suffix}"
+    database_path = tmp_path / f"compat-{suffix[1:]}.sqlite"
+    database = f"sqlite:///{database_path}"
+    blocked = tmp_path / f"compat-blocked{suffix}"
+    approved = tmp_path / f"compat-approved{suffix}"
+    source_name = "long_variable_name"
+
+    pyspssio.write_sav(str(source), pd.DataFrame({source_name: [1.0]}))
+    assert pyspssio.read_metadata(str(source))["var_compat_names"][source_name] != source_name
+
+    imported = openstatspec.import_sav(source, database_url=database, dataset_id=f"compat-{suffix[1:]}")
+    diagnostic = next(
+        item for item in imported.diagnostics
+        if item.code == "compatible-variable-name-not-exportable"
+    )
+    assert diagnostic.details == {
+        "source_name": source_name,
+        "compatible_name": "LONG_VAR",
+        "physical_name": source_name,
+    }
+
+    # Export must also assess the current normalized SQL catalog. Clear the
+    # import event and alter the catalog value to prove that no persisted
+    # diagnostic can mask an unguarded re-export or silent renaming path.
+    connection = sqlite3.connect(database_path)
+    connection.execute(
+        "delete from fidelity_event_catalog where dataset_id = ? and code = ?",
+        (f"compat-{suffix[1:]}", "compatible-variable-name-not-exportable"),
+    )
+    connection.execute(
+        "update variable_catalog set compat_name = ? where dataset_id = ? and source_name = ?",
+        ("CUSTOM_NAME", f"compat-{suffix[1:]}", source_name),
+    )
+    connection.commit()
+    expected_catalog_detail = {
+        "source_name": source_name,
+        "compatible_name": "CUSTOM_NAME",
+        "physical_name": source_name,
+    }
+
+    with pytest.raises(UnsupportedOperationError, match="compatible-variable-name-not-exportable"):
+        openstatspec.export_sav(
+            database_url=database, dataset_id=f"compat-{suffix[1:]}", destination=blocked,
+            allow_loss=_REQUIRED_ENGINE_LOSS,
+        )
+    assert not blocked.exists()
+
+    result = openstatspec.export_sav(
+        database_url=database, dataset_id=f"compat-{suffix[1:]}", destination=approved,
+        allow_loss=[*_REQUIRED_ENGINE_LOSS, "compatible-variable-name-not-exportable"],
+    )
+    accepted = next(item for item in result.diagnostics if item.code == "compatible-variable-name-not-exportable")
+    assert accepted.details == expected_catalog_detail
+    persisted = connection.execute(
+        "select details from fidelity_event_catalog where operation_id = ? and code = ?",
+        (result["operation_id"], "compatible-variable-name-not-exportable"),
+    ).fetchone()[0]
+    assert json.loads(persisted) == {**expected_catalog_detail, "accepted_by_user": True}

tests/test_sav_sqlite.py

@@ -16,6 +16,8 @@
     "separate-write-format-unobservable",
 ]
 
+_COMPAT_NAME_LOSS = [*_REQUIRED_ENGINE_LOSS, "compatible-variable-name-not-exportable"]
+
 
 def test_pyspssio_round_trip_uses_one_wide_table_and_catalog(tmp_path) -> None:
     source = tmp_path / "tiny.sav"
@@ -85,7 +87,7 @@ def test_supported_pyspssio_metadata_round_trips_through_sqlite_for_sav_and_zsav
     assert openstatspec.validate(database_url=f"sqlite:///{database_path}", dataset_id=f"supported-{suffix[1:]}")["valid"] is True
     connection = sqlite3.connect(database_path)
     assert connection.execute(f"select comment from data_supported_{suffix[1:]} order by __case_ordinal").fetchone() == (expected["long_text"],)
-    openstatspec.export_sav(database_url=f"sqlite:///{database_path}", dataset_id=f"supported-{suffix[1:]}", destination=destination, allow_loss=_REQUIRED_ENGINE_LOSS)
+    openstatspec.export_sav(database_url=f"sqlite:///{database_path}", dataset_id=f"supported-{suffix[1:]}", destination=destination, allow_loss=_COMPAT_NAME_LOSS)
     assert compare_sav_semantics(source, destination) == {"equivalent": True, "differences": []}
 
 

tests/test_sql_services.py

@@ -13,6 +13,7 @@
 
 pytestmark = pytest.mark.services
 _REQUIRED_ENGINE_LOSS = ["file-label-and-documents-unobservable", "separate-write-format-unobservable"]
+_COMPAT_NAME_LOSS = [*_REQUIRED_ENGINE_LOSS, "compatible-variable-name-not-exportable"]
 
 
 @pytest.fixture
@@ -72,5 +73,5 @@ def test_live_profile_preserves_supported_sav_semantics(environment_name, datase
     imported = openstatspec.import_sav(source, database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}")
     assert imported["case_count"] == 4
     assert openstatspec.validate(database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}")["valid"] is True
-    openstatspec.export_sav(database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}", destination=destination, allow_loss=_REQUIRED_ENGINE_LOSS)
+    openstatspec.export_sav(database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}", destination=destination, allow_loss=_COMPAT_NAME_LOSS)
     assert compare_sav_semantics(source, destination) == {"equivalent": True, "differences": []}