test: bound transformation validation and plan encoding work

Author: TonisOrmisson

tests/test_inplace_transform.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import sqlite3
 from dataclasses import replace
 from types import SimpleNamespace
+from unittest.mock import patch
 
 import pytest
 from sqlalchemy import create_engine, inspect, text
@@ -632,12 +634,25 @@ def test_temporary_target_type_is_not_taken_from_same_name_recreation(catalog) -
 def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None:
     url, path, dataset_id, table_name = catalog
     plan = _plan("RECODE score (1 = 0).")
-    result = openstatspec.apply_spss_in_place(
-        database_url=url,
-        dataset_id=dataset_id,
-        source_text="RECODE score (1 = 0).",
-        actor="test-agent",
+    expected_json = (
+        '{"contract":"openstatspec-transformation-plan-v0.1","input_alias":"parent",'
+        '"operations":[{"op":"recode","rules":[{"match":{"kind":"values","values":'
+        '[{"bits":"3ff0000000000000","type":"binary64"}]},"result":{"kind":"literal",'
+        '"value":{"bits":"0000000000000000","type":"binary64"}}}],"source":"score",'
+        '"target":"score","target_mode":"replace","unmatched":{"kind":"copy"}}]}'
     )
+    expected_hash = "c1a0025028816c7b424312612af4e0a95d00da6b1bcff25e9b742e4be25dbb42"
+    source_hash = "7e86162055308158fa2cf1781eae7d997c07bb870fc66becdabdfdd00580dc7b"
+    with patch.object(
+        openstatspec.TransformationPlan, "canonical_bytes", autospec=True,
+        side_effect=openstatspec.TransformationPlan.canonical_bytes,
+    ) as encodes, patch("hashlib.sha256", wraps=hashlib.sha256) as hashes:
+        result = openstatspec.apply_spss_in_place(
+            database_url=url,
+            dataset_id=dataset_id,
+            source_text="RECODE score (1 = 0).\r\n",
+            actor="test-agent",
+        )
     assert result["dolt_branch"] is None
     assert result["dolt_commit_performed"] is False
     assert result["plan_hash"] == plan.sha256()
@@ -651,6 +666,18 @@ def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None:
         "spss_syntax",
         "openstatspec-spss-syntax-frontend-v0.2",
     )
+    assert (result["dataset_id"], result["physical_table_name"]) == (dataset_id, table_name)
+    assert (result["plan_hash"], result["source_hash"]) == (expected_hash, source_hash)
+    with sqlite3.connect(path) as connection:
+        audit_json, audit_hash, audit_source = connection.execute(
+            "SELECT canonical_plan_json, plan_hash, source_hash FROM transformation_apply"
+        ).fetchone()
+    assert (audit_json.encode("utf-8"), audit_hash, audit_source) == (
+        expected_json.encode("utf-8"), expected_hash, source_hash,
+    )
+    assert (encodes.call_count, sum(
+        call.args == (expected_json.encode("utf-8"),) for call in hashes.call_args_list
+    )) == (1, 1), "public SPSS apply must encode/hash the canonical plan once"
 
 
 def test_schema_commands_record_the_v03_frontend_contract(catalog) -> None:
@@ -679,12 +706,24 @@ def test_public_generic_plan_apply_accepts_object_and_mapping(
     url, path, dataset_id, table_name = catalog
     plan = _plan("RECODE score (1 = 7).")
     supplied = plan.as_dict() if as_mapping else plan
-    result = openstatspec.apply_transformation_plan_in_place(
-        database_url=url,
-        dataset_id=dataset_id,
-        plan=supplied,
-        actor="test-agent",
+    expected_json = (
+        '{"contract":"openstatspec-transformation-plan-v0.1","input_alias":"parent",'
+        '"operations":[{"op":"recode","rules":[{"match":{"kind":"values","values":'
+        '[{"bits":"3ff0000000000000","type":"binary64"}]},"result":{"kind":"literal",'
+        '"value":{"bits":"401c000000000000","type":"binary64"}}}],"source":"score",'
+        '"target":"score","target_mode":"replace","unmatched":{"kind":"copy"}}]}'
     )
+    expected_hash = "089de9933b97157e9cb8b595d6f00afc11044907ca69e4ab747f6d0aac0bf60f"
+    with patch.object(
+        openstatspec.TransformationPlan, "canonical_bytes", autospec=True,
+        side_effect=openstatspec.TransformationPlan.canonical_bytes,
+    ) as encodes, patch("hashlib.sha256", wraps=hashlib.sha256) as hashes:
+        result = openstatspec.apply_transformation_plan_in_place(
+            database_url=url,
+            dataset_id=dataset_id,
+            plan=supplied,
+            actor="test-agent",
+        )
     assert result["dataset_id"] == dataset_id
     assert result["physical_table_name"] == table_name
     assert result["source_kind"] == "canonical_plan"
@@ -702,6 +741,15 @@ def test_public_generic_plan_apply_accepts_object_and_mapping(
         None,
         plan.sha256(),
     )
+    audit_json, audit_hash = connection.execute(
+        "SELECT canonical_plan_json, plan_hash FROM transformation_apply"
+    ).fetchone()
+    assert (audit_json.encode("utf-8"), audit_hash, result["plan_hash"]) == (
+        expected_json.encode("utf-8"), expected_hash, expected_hash,
+    )
+    assert (encodes.call_count, sum(
+        call.args == (expected_json.encode("utf-8"),) for call in hashes.call_args_list
+    )) == (1, 1), "public canonical apply must encode/hash the plan once"
 
 
 def test_generic_string_width_is_rejected_before_ddl(

tests/test_transform_frontend.py

@@ -14,7 +14,10 @@
     spss_source_hash,
 )
 from openstatspec.transform import (
+    AssignOperation,
     CreateVariableOperation,
+    ExecuteOperation,
+    Operand,
     DeleteVariableOperation,
     RecodeMatch,
     RecodeOperation,
@@ -680,6 +683,112 @@ def test_recode_copy_preserves_source_declared_width() -> None:
     assert bound.output_schema == schema
 
 
+@pytest.mark.parametrize("workload", ["execute", "labels"])
+def test_generic_binding_10k_has_linear_bookkeeping(workload) -> None:
+    size = 10_000
+    budget = 10 * (size + size)
+    probes = 0
+
+    class Name(str):
+        def casefold(self):
+            nonlocal probes
+            probes += 1
+            assert probes <= budget, f"variable-name casefold probes: {probes} > {budget}"
+            return super().casefold()
+
+    class Operations(tuple):
+        def __getitem__(self, key):
+            nonlocal probes
+            if isinstance(key, slice):
+                probes += len(range(*key.indices(len(self))))
+                assert probes <= budget, f"operation suffix entries: {probes} > {budget}"
+            return super().__getitem__(key)
+
+    schema = _schema(*(VariableDefinition(Name(f"V{i}"), "numeric") for i in range(size)))
+    operations = (
+        Operations((ExecuteOperation(),) * size) if workload == "execute"
+        else tuple(SetVariableLabelOperation(f"v{i}", "L") for i in range(size))
+    )
+    plan = TransformationPlan(operations)
+    probes = 0  # Count only the public bind, not input construction.
+    bound = bind_transformation_plan(plan, schema)
+    assert bound.plan == plan
+    assert bound.output_schema == _schema(*(
+        VariableDefinition(f"V{i}", "numeric", variable_label="L" if workload == "labels" else None)
+        for i in range(size)
+    ))
+
+
+@pytest.mark.parametrize("create", [
+    CreateVariableOperation("replacement", "numeric"),
+    AssignOperation("replacement", "create", Operand("literal", value=TypedValue.binary64(1))),
+], ids=["explicit", "assign"])
+def test_generic_distant_create_preserves_first_error_and_empty_schema_exception(create) -> None:
+    schema = _schema(VariableDefinition("Only", "numeric"))
+    prefix = (DeleteVariableOperation("oNLY"),) + (ExecuteOperation(),) * 100
+    plan = TransformationPlan(prefix + (create,), contract="openstatspec-transformation-plan-v0.3")
+    assert bind_transformation_plan(plan, schema).output_schema == _schema(
+        VariableDefinition("replacement", "numeric"),
+    )
+    plan = TransformationPlan(
+        prefix + (create, SetVariableLabelOperation("missing", "L"), create),
+        contract=plan.contract,
+    )
+    with pytest.raises(TransformationFrontendError) as caught:
+        bind_transformation_plan(plan, schema)
+    assert caught.value.as_dict() == {
+        "code": "unknown_variable",
+        "detail": "Variable 'missing' is not present in the current schema.",
+        "details": {"variable": "missing"},
+    }
+
+
+@pytest.mark.parametrize("later", [(), (RecodeOperation(
+    source="missing", target="replacement", target_mode="create",
+    rules=(RecodeRule(RecodeMatch("values", (TypedValue.binary64(1),)),
+                      RecodeResult("literal", TypedValue.binary64(0))),),
+    unmatched=RecodeResult("copy"),
+),)], ids=["no-create", "recode-create"])
+def test_generic_last_delete_precedes_later_lookup_error(later) -> None:
+    plan = TransformationPlan(
+        (DeleteVariableOperation("oNLY"), SetVariableLabelOperation("missing", "L"))
+        + (ExecuteOperation(),) * 100 + later,
+        contract="openstatspec-transformation-plan-v0.3",
+    )
+    with pytest.raises(TransformationFrontendError) as caught:
+        bind_transformation_plan(plan, _schema(VariableDefinition("Only", "numeric")))
+    assert caught.value.as_dict() == {
+        "code": "cannot_delete_last_variable",
+        "detail": "A dataset must retain at least one variable.",
+        "details": {"variable": "Only"},
+    }
+
+
+def test_generic_delete_recreate_appends_with_new_spelling_and_metadata() -> None:
+    plan = TransformationPlan((
+        DeleteVariableOperation("oNLY"),
+        CreateVariableOperation("ONLY", "string", 3),
+        SetVariableLabelOperation("only", "New"),
+    ), contract="openstatspec-transformation-plan-v0.3")
+    bound = bind_transformation_plan(plan, _schema(
+        VariableDefinition("Only", "numeric", variable_label="Old"),
+        VariableDefinition("Keep", "numeric"),
+    ))
+    assert bound.output_schema == _schema(
+        VariableDefinition("Keep", "numeric"),
+        VariableDefinition("ONLY", "string", variable_label="New", declared_string_width=3),
+    )
+
+
+def test_literal_unicode_canonical_plan_vector() -> None:
+    plan = TransformationPlan((SetVariableLabelOperation("Only", "Täpselt"),), input_alias="survey")
+    assert plan.canonical_bytes() == (
+        '{"contract":"openstatspec-transformation-plan-v0.2","input_alias":"survey",'
+        '"operations":[{"label":"Täpselt","op":"set_variable_label","variable":"Only"}]}'
+    ).encode("utf-8")
+    assert plan.sha256() == "2ab3bf3c6dd2c21ce0fed076d43c49d93b6cc38e3e38529a18b7563286af9e03"
+
+
 def test_generic_plan_binding_validates_sequential_schema_state() -> None:
     plan = TransformationPlan((
         RecodeOperation(