Reject foreign views and prune variable sets

Author: TonisOrmisson

src/openstatspec/sql/inplace_transform.py

@@ -316,6 +316,11 @@ def _delete_variable_metadata(
 ) -> None:
     """Delete one variable and every normative catalog row owned by it."""
     variable_id = str(variable["variable_id"])
+    variable_set_ids = list(connection.execute(
+        select(core.variable_set_member.c.variable_set_id).where(
+            core.variable_set_member.c.variable_id == variable_id
+        )
+    ).scalars())
     response_set_ids = list(connection.execute(
         select(core.multiple_response_member.c.multiple_response_set_id).where(
             core.multiple_response_member.c.variable_id == variable_id
@@ -332,6 +337,16 @@ def _delete_variable_metadata(
     connection.execute(delete(core.variable_set_member).where(
         core.variable_set_member.c.variable_id == variable_id
     ))
+    for variable_set_id in variable_set_ids:
+        remaining_member = connection.execute(
+            select(core.variable_set_member.c.variable_id).where(
+                core.variable_set_member.c.variable_set_id == variable_set_id
+            )
+        ).first()
+        if remaining_member is None:
+            connection.execute(delete(core.variable_set).where(
+                core.variable_set.c.variable_set_id == variable_set_id
+            ))
     connection.execute(delete(core.multiple_response_member).where(
         core.multiple_response_member.c.variable_id == variable_id
     ))

src/openstatspec/sql/wide.py

@@ -813,14 +813,18 @@ def create_wide_dataset(
         engine=engine, profile_name=profile.name, active=active_connection,
         audit_relations=audit_relations, phase="catalog initialization",
     ) as catalog_connection:
-        catalog_existed = inspect(catalog_connection).has_table(
-            normative.catalog_identity.name
-        )
-        create_normative_catalog(catalog_connection, normative)
-        if catalog_existed:
-            require_verified_catalog(catalog_connection)
+        inspector = inspect(catalog_connection)
+        existing_tables = set(inspector.get_table_names())
+        existing_views = set(inspector.get_view_names())
+        if existing_tables or existing_views:
+            if normative.catalog_identity.name not in existing_tables:
+                raise UnsupportedOperationError(
+                    "The selected database catalog is foreign; "
+                    "import is not permitted."
+                )
         else:
-            _verify_normative_catalog(catalog_connection, normative)
+            create_normative_catalog(catalog_connection, normative)
+        require_verified_catalog(catalog_connection)
     operation_id = str(uuid4())
     normative_dataset_id = str(uuid4())
     fidelity_events = tuple(fidelity_events)
@@ -873,12 +877,13 @@ def create_wide_dataset(
     data_table_created = False
     try:
         with engine.begin() as setup:
-            _verify_normative_catalog(setup, normative)
+            require_verified_catalog(setup)
         namespace_owned = True
         with _bound_catalog_transaction(
             engine=engine, profile_name=profile.name, active=active_connection,
             audit_relations={*audit_relations, data_table.name}, phase="import",
         ) as connection:
+            require_verified_catalog(connection)
             if connection.execute(
                 select(normative.dataset.c.dataset_id).where(or_(
                     normative.dataset.c.dataset_name == dataset_id,

tests/test_atomic_import.py

@@ -50,6 +50,9 @@ def inspect_with_concurrent_table(bind):
         actual = real_inspect(bind)
 
         class Inspector:
+            def __getattr__(self, name):
+                return getattr(actual, name)
+
             def has_table(self, table_name):
                 nonlocal inspected_data_table
                 if table_name == "data_race":
@@ -298,6 +301,39 @@ def fail_after_normative_write(*args, **kwargs):
         "select direction, severity, event_code, dataset_id from fidelity_event"
     ).fetchall() == [("import", "error", "import_failed", None)]
 
+def test_occupied_foreign_view_namespace_fails_without_modification(
+    tmp_path,
+) -> None:
+    database_path = tmp_path / "foreign-view.sqlite"
+    database = f"sqlite:///{database_path}"
+    connection = sqlite3.connect(database_path)
+    connection.execute("create view foreign_view as select 1 as value")
+    before_schema = connection.execute(
+        "select name, type, sql from sqlite_master order by name"
+    ).fetchall()
+    connection.close()
+    variables = [{
+        "ordinal": 1, "source_name": "score", "physical_name": "score",
+        "storage_kind": "numeric", "string_width": None, "label": "",
+        "format": "F8.0", "measure": "scale", "alignment": "right",
+        "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
+    }]
+
+    with pytest.raises(wide.UnsupportedOperationError, match="foreign"):
+        create_wide_dataset(
+            database_url=database, dataset_id="foreign-view",
+            source_name="fixture.sav", source_format="SAV",
+            rows=[{"score": 1.0}], variables=variables,
+        )
+
+    connection = sqlite3.connect(database_path)
+    assert connection.execute(
+        "select name, type, sql from sqlite_master order by name"
+    ).fetchall() == before_schema
+    assert connection.execute("select value from foreign_view").fetchall() == [(1,)]
+    connection.close()
+
+
 def test_occupied_foreign_namespace_fails_without_modification(tmp_path) -> None:
     database_path = tmp_path / "foreign.sqlite"
     database = f"sqlite:///{database_path}"
@@ -315,7 +351,7 @@ def test_occupied_foreign_namespace_fails_without_modification(tmp_path) -> None
         "format": "A8", "measure": "nominal", "alignment": "left",
         "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
     }]
-    with pytest.raises(RuntimeError, match="occupied"):
+    with pytest.raises(RuntimeError, match="foreign"):
         create_wide_dataset(
             database_url=database, dataset_id="foreign", source_name="fixture.sav",
             source_format="SAV", rows=[{"name": "ok"}], variables=variables,

tests/test_inplace_transform.py

@@ -16,7 +16,9 @@
 from openstatspec.sql.profiles import (
     DOLT, SQLITE, TargetCapabilityExceededError,
 )
-from openstatspec.sql.wide import create_wide_dataset, validate_wide_dataset
+from openstatspec.sql.wide import (
+    create_wide_dataset, read_wide_dataset, validate_wide_dataset,
+)
 
 
 def _variables() -> list[dict[str, object]]:
@@ -250,6 +252,50 @@ def test_delete_recanonicalizes_surviving_collision_columns(tmp_path) -> None:
     )["valid"] is True
 
 
+def test_delete_prunes_an_empty_spss_variable_set(catalog) -> None:
+    url, path, dataset_id, _table_name = catalog
+    connection = sqlite3.connect(path)
+    variable_id = connection.execute(
+        "select variable_id from variable where dataset_id = ?",
+        (dataset_id,),
+    ).fetchone()[0]
+    variable_set_id = "00000000-0000-0000-0000-000000000002"
+    connection.execute(
+        "insert into variable_set "
+        "(variable_set_id, dataset_id, source_ordinal, set_name) "
+        "values (?, ?, 1, 'scores')",
+        (variable_set_id, dataset_id),
+    )
+    connection.execute(
+        "insert into variable_set_member "
+        "(variable_set_id, variable_id, source_ordinal) values (?, ?, 1)",
+        (variable_set_id, variable_id),
+    )
+    connection.commit()
+    connection.close()
+
+    openstatspec.apply_spss_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        source_text="COMPUTE other = score. DELETE VARIABLES score.",
+        actor="test-agent",
+    )
+
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        "select count(*) from variable_set where dataset_id = ?",
+        (dataset_id,),
+    ).fetchone() == (0,)
+    assert connection.execute(
+        "select count(*) from variable_set_member"
+    ).fetchone() == (0,)
+    connection.close()
+    dataset, _variables, _rows = read_wide_dataset(
+        database_url=url, dataset_id=dataset_id,
+    )
+    assert dataset["source_extensions"] == {}
+
+
 def test_delete_prunes_an_empty_multiple_response_set(catalog) -> None:
     url, path, dataset_id, _table_name = catalog
     connection = sqlite3.connect(path)