Validate catalog storage and SQLite URI persistence

Author: TonisOrmisson

src/openstatspec/sql/wide.py

@@ -1364,24 +1364,26 @@ def _catalog_variable_bijection_state(
     legacy_rows = [
         (
             row["dataset_id"], row["ordinal"], row["source_name"],
-            row["physical_name"],
+            row["physical_name"], row["storage_kind"],
         )
         for row in connection.execute(select(
             variables.c.dataset_id, variables.c.ordinal,
             variables.c.source_name, variables.c.physical_name,
+            variables.c.storage_kind,
         )).mappings()
     ]
     normative_rows = [
         (
             row["dataset_name"], row["source_ordinal"], row["source_name"],
-            row["physical_name"],
+            row["physical_name"], row["storage_kind"],
         )
         for row in connection.execute(
             select(
                 normative.dataset.c.dataset_name,
                 normative.variable.c.source_ordinal,
                 normative.variable.c.source_name,
                 normative.variable.c.physical_name,
+                normative.variable.c.storage_kind,
             ).join(
                 normative.variable,
                 normative.variable.c.dataset_id == normative.dataset.c.dataset_id,
@@ -1394,7 +1396,8 @@ def _catalog_variable_bijection_state(
             or not isinstance(ordinal, int) or ordinal < 1
             or not isinstance(source_name, str) or not source_name.strip()
             or not isinstance(physical_name, str) or not physical_name.strip()
-            for dataset_name, ordinal, source_name, physical_name in rows
+            or storage_kind not in {"numeric", "string"}
+            for dataset_name, ordinal, source_name, physical_name, storage_kind in rows
         ):
             return "unverified"
         if len(set(rows)) != len(rows):
@@ -1675,9 +1678,15 @@ def initialize_wide_catalog(
     """Install or explicitly migrate a dedicated catalog after server preflight."""
     validate_connection_url(database_url)
     parsed_url = make_url(database_url)
+    sqlite_database = parsed_url.database or ""
+    sqlite_mode = str(parsed_url.query.get("mode", "")).lower()
     if (
         parsed_url.get_backend_name() == "sqlite"
-        and parsed_url.database in {None, "", ":memory:"}
+        and (
+            sqlite_database in {"", ":memory:"}
+            or sqlite_database.lower() == "file::memory:"
+            or sqlite_mode == "memory"
+        )
     ):
         raise UnsupportedOperationError(
             "Catalog initialization requires a persistent SQLite database URL."

tests/test_catalog_lifecycle.py

@@ -328,6 +328,34 @@ def test_duplicate_import_failure_preserves_existing_dataset(tmp_path):
 
 
 
+def test_variable_bijection_rejects_storage_kind_mismatch(tmp_path):
+    path = tmp_path / "variable-storage-mismatch.sqlite"
+    database = f"sqlite:///{path}"
+    openstatspec.initialize_catalog(database_url=database)
+    variables = _variables()
+    variables[0].update({
+        "storage_kind": "numeric",
+        "string_width": None,
+    })
+    create_wide_dataset(
+        database_url=database,
+        dataset_id="sample",
+        source_name="fixture.sav",
+        source_format="SAV",
+        rows=[{"name": 1.0}],
+        variables=variables,
+    )
+    connection = sqlite3.connect(path)
+    connection.execute(
+        "update variable set storage_kind = 'string' where source_name = 'name'"
+    )
+    connection.commit()
+    connection.close()
+
+    with pytest.raises(UnsupportedOperationError, match="catalog is unverified"):
+        read_wide_dataset(database_url=database, dataset_id="sample")
+
+
 def test_missing_additive_column_requires_explicit_migration_without_audit_mutation(
     tmp_path,
 ):
@@ -433,14 +461,27 @@ def test_reflected_mysql_integer_display_width_is_not_semantic():
         inspector, mysql.VARCHAR(length=255),
     ) == "VARCHAR(255)"
 
-@pytest.mark.parametrize("database_url", ["sqlite://", "sqlite:///:memory:"])
+@pytest.mark.parametrize("database_url", [
+    "sqlite://",
+    "sqlite:///:memory:",
+    "sqlite:///file:memdb1?mode=memory&cache=shared&uri=true",
+    "sqlite:///file::memory:?cache=shared&uri=true",
+])
 def test_catalog_initialization_rejects_ephemeral_sqlite_url(database_url):
     with pytest.raises(
         UnsupportedOperationError,
         match="requires a persistent SQLite database URL",
     ):
         openstatspec.initialize_catalog(database_url=database_url)
 
+def test_catalog_initialization_allows_file_backed_sqlite_uri(tmp_path):
+    path = tmp_path / "uri-catalog.sqlite"
+    database = f"sqlite:///file:{path}?mode=rwc&uri=true"
+
+    assert openstatspec.initialize_catalog(database_url=database)["catalog"] == "verified"
+    assert path.is_file()
+
+
 def test_import_fidelity_reader_excludes_export_operational_events(tmp_path):
     path = tmp_path / "fidelity-directions.sqlite"
     database = f"sqlite:///{path}"