From 6c4c006c0165a1e8e8fa8de16de8970981369d06 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Mon, 17 Aug 2026 19:28:39 +1000 Subject: [PATCH 1/3] fix(datacontracts): to_checks() emits primary_key_unique (item 8j, task 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BaseDataContract.to_checks() never emitted primary_key_unique - only compile_datacontract(TypeSpec) did, so a TypeSpec and an equivalent BaseDataContract subclass validated a resource differently inside dag.validate (spec §4.1, Fact C). Reuses the existing, already-tested primary_key_check(spec) helper; cls.to_typespec() already collapses natural_key into primary_key, so this fires for both declaration styles with no new logic. --- src/mountainash/datacontracts/contract.py | 5 ++++ tests/datacontracts/test_native_contract.py | 26 ++++++++++++++++++++- 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/mountainash/datacontracts/contract.py b/src/mountainash/datacontracts/contract.py index b8916054..cc57a069 100644 --- a/src/mountainash/datacontracts/contract.py +++ b/src/mountainash/datacontracts/contract.py @@ -85,9 +85,14 @@ def to_typespec(cls) -> "TypeSpec": @classmethod def to_checks(cls) -> "list[ValidationCheck]": + from mountainash.datacontracts.compiler import primary_key_check + checks: "list[ValidationCheck]" = [] for name, contract_field in cls._contract_fields.items(): checks.extend(contract_field.to_checks(name)) + pk_check = primary_key_check(cls.to_typespec()) + if pk_check is not None: + checks.append(pk_check) return checks @classmethod diff --git a/tests/datacontracts/test_native_contract.py b/tests/datacontracts/test_native_contract.py index 5fe373ee..bae9ba2f 100644 --- a/tests/datacontracts/test_native_contract.py +++ b/tests/datacontracts/test_native_contract.py @@ -44,10 +44,34 @@ def test_natural_key_survives_as_primary_key(self): def test_to_checks_ids(self): ids = [c.id for c in UserContract.to_checks()] assert "id__not_null" in ids - assert "id__unique" in ids + assert "email__not_null" in ids assert "email__pattern" in ids assert "age__ge" in ids + def test_to_checks_includes_primary_key_unique(self): + ids = [c.id for c in UserContract.to_checks()] + assert "primary_key_unique" in ids # UserContract: Config.natural_key = ["id"] + + def test_to_checks_includes_primary_key_unique_for_primary_key_config(self): + class OrderContract(BaseDataContract): + order_id: int = Field(nullable=False) + + class Config: + name = "orders" + primary_key = ["order_id"] + + ids = [c.id for c in OrderContract.to_checks()] + assert "primary_key_unique" in ids + + +def test_to_checks_matches_compile_datacontract_check_ids(): + from mountainash.datacontracts.compiler import compile_datacontract, contract_from_typespec + + spec = UserContract.to_typespec() + compiled_ids = {c.id for c in compile_datacontract(spec)} + contract_ids = {c.id for c in contract_from_typespec(spec).to_checks()} + assert compiled_ids == contract_ids + class TestValidate: def test_valid_data_passes(self): From 13cdd282b3053292c37b65af8f33c95ef697e0e2 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Mon, 17 Aug 2026 19:29:40 +1000 Subject: [PATCH 2/3] fix(datacontracts): thread allow_imperfect_key through validate_datacontract* (item 8j, task 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BaseDataContract.validate_datacontract/validate_datacontract_quick had no way to reach the already-shipped, already-tested Validator.validate/allow_imperfect_key escape hatch (spec §4.2, Fact A) - a declared primary_key/natural_key always raised IdentityInvalidError on a null/duplicate key before checks ran, with no way to opt into primary_key_unique reporting instead. Also corrects validate_datacontract's docstring, which claimed "never raises" - false for keyed-identity failures. --- src/mountainash/datacontracts/contract.py | 21 +++++++++++--- tests/datacontracts/test_native_contract.py | 31 +++++++++++++++++++++ 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/src/mountainash/datacontracts/contract.py b/src/mountainash/datacontracts/contract.py index cc57a069..f52fbec3 100644 --- a/src/mountainash/datacontracts/contract.py +++ b/src/mountainash/datacontracts/contract.py @@ -105,13 +105,22 @@ def validate_datacontract( tail: int | None = None, sample: int | None = None, random_seed: int | None = None, + allow_imperfect_key: bool = False, ) -> "ValidationResult": - """Validate data against this contract; returns (never raises).""" + """Validate data against this contract; returns a ValidationResult. + + Raises IdentityInvalidError if this contract declares a keyed identity + (Config.primary_key / Config.natural_key) and the data does not honour + it: always for key fields missing from the data (declaration-phase, + spec §7); for null-key rows or duplicate key tuples, unless + allow_imperfect_key=True — which lets the run proceed and reports the + duplicates via the primary_key_unique check instead (spec §9.3). + """ from mountainash.datacontracts.validator import Validator return Validator(name=cls.contract_name(), contract=cls).validate( data, context=context, head=head, tail=tail, sample=sample, - random_seed=random_seed, + random_seed=random_seed, allow_imperfect_key=allow_imperfect_key, ) @classmethod @@ -124,11 +133,15 @@ def validate_datacontract_quick( tail: int | None = None, sample: int | None = None, random_seed: int | None = None, + allow_imperfect_key: bool = False, ) -> "ValidationResult": - """Quick validation — same runner, fail_fast=True (item 18 subsumed).""" + """Quick validation — same runner, fail_fast=True (item 18 subsumed). + + See validate_datacontract for the allow_imperfect_key contract. + """ from mountainash.datacontracts.validator import Validator return Validator(name=cls.contract_name(), contract=cls).validate_quick( data, context=context, head=head, tail=tail, sample=sample, - random_seed=random_seed, + random_seed=random_seed, allow_imperfect_key=allow_imperfect_key, ) diff --git a/tests/datacontracts/test_native_contract.py b/tests/datacontracts/test_native_contract.py index bae9ba2f..65a80d4c 100644 --- a/tests/datacontracts/test_native_contract.py +++ b/tests/datacontracts/test_native_contract.py @@ -1,9 +1,11 @@ """Native BaseDataContract: declaration collection, TypeSpec round trip, validate.""" import polars as pl +import pytest from mountainash.datacontracts.contract import BaseDataContract from mountainash.datacontracts.field import Field from mountainash.typespec.universal_types import UniversalType +from mountainash.validation.errors import IdentityInvalidError class UserContract(BaseDataContract): @@ -116,6 +118,35 @@ def test_quick_is_fail_fast_same_shapes(self): assert list(full.failure_cases.columns) == list(quick.failure_cases.columns) assert quick.check_summaries.height <= full.check_summaries.height + def test_validate_datacontract_raises_by_default_on_duplicate_key(self): + df = pl.DataFrame( + {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]} + ) + with pytest.raises(IdentityInvalidError): + UserContract.validate_datacontract(df) + + def test_validate_datacontract_allow_imperfect_key_reports_primary_key_unique(self): + df = pl.DataFrame( + {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]} + ) + result = UserContract.validate_datacontract(df, allow_imperfect_key=True) + assert result.passes is False + failing = set( + result.check_summaries.filter( + result.check_summaries["status"] != "passed" + )["check_id"].to_list() + ) + assert "primary_key_unique" in failing + assert result.identity_diagnostics["duplicate_key_tuples"] == 1 + + def test_validate_datacontract_quick_allow_imperfect_key_same_shape(self): + df = pl.DataFrame( + {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]} + ) + result = UserContract.validate_datacontract_quick(df, allow_imperfect_key=True) + assert result.passes is False + assert result.identity_diagnostics["duplicate_key_tuples"] == 1 + class NoKeyContract(BaseDataContract): id: int = Field(unique=True) From f98d0db8bb440302a6aca89324c07b8974892752 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Mon, 17 Aug 2026 19:32:52 +1000 Subject: [PATCH 3/3] fix(validation): allow_imperfect_key threaded through dag.validate*, per-resource identity isolation (item 8j, task 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dag.validate/validate_quick had no allow_imperfect_key parameter, so every declared keyed identity always raised IdentityInvalidError deep inside validate_relation - and that raise escaped the per-resource loop uncaught, aborting validation of every resource after the failing one in the batch (spec §4.3, Fact B). RelationDAG.validate/validate_quick -> relations/dag/validation.py's validate/validate_quick/_run -> ValidationRunner.validate_dag now thread allow_imperfect_key end to end. validate_dag's per-resource loop wraps validate_relation in try/except IdentityInvalidError and isolates the failure into that resource's own ValidationResult (check_id="__identity__", status="error") - never a raised exception, never an aborted batch (spec §3.2). Isolation is categorical across every IdentityInvalidError cause, including the missing-key-fields raise that allow_imperfect_key never suppresses (spec §7 round-2 fix). Updates item 8l's test_duplicate_primary_key_raises_identity_error (tests/relations/dag/cross_backend/test_datapackage_validation_loop.py) to test_duplicate_primary_key_isolates_identity_failure, asserting the new isolated-result contract instead of a raise - the old assertion exercised exactly the DAG-tier behavior this task supersedes. --- src/mountainash/relations/dag/dag.py | 11 +- src/mountainash/relations/dag/validation.py | 6 + src/mountainash/validation/runner.py | 49 +++++++-- .../test_datapackage_validation_loop.py | 20 ++-- tests/relations/dag/test_dag_validation.py | 103 ++++++++++++++++++ 5 files changed, 169 insertions(+), 20 deletions(-) diff --git a/src/mountainash/relations/dag/dag.py b/src/mountainash/relations/dag/dag.py index 97f1cda3..8539367b 100644 --- a/src/mountainash/relations/dag/dag.py +++ b/src/mountainash/relations/dag/dag.py @@ -615,17 +615,22 @@ def validate( context: dict[str, Any] | None = None, backend: Optional[str] = None, failure_sample: Optional[int] = None, + allow_imperfect_key: bool = False, ) -> "DAGValidationResult": """Full validation via the backend-agnostic ValidationRunner. Per-resource checks compile from each spec/contract; FK row-integrity checks are generated from constraint_metadata + spec foreign keys by validation.fk.build_fk_checks and compiled as relation anti-joins. + A resource's invalid keyed identity is isolated into that resource's + own failing result (check_id="__identity__") rather than raised out + of this call - every other resource still validates (spec item 8j §3.2). """ from mountainash.relations.dag.validation import validate return validate( - self, specs, context=context, backend=backend, failure_sample=failure_sample + self, specs, context=context, backend=backend, failure_sample=failure_sample, + allow_imperfect_key=allow_imperfect_key, ) def validate_quick( @@ -635,12 +640,14 @@ def validate_quick( context: dict[str, Any] | None = None, backend: Optional[str] = None, failure_sample: Optional[int] = None, + allow_imperfect_key: bool = False, ) -> "DAGValidationResult": """Fast validation via the ValidationRunner (fail_fast=True; identical shapes).""" from mountainash.relations.dag.validation import validate_quick return validate_quick( - self, specs, context=context, backend=backend, failure_sample=failure_sample + self, specs, context=context, backend=backend, failure_sample=failure_sample, + allow_imperfect_key=allow_imperfect_key, ) def _unknown_ref_error(self, missing: str) -> "UnknownRelationRef": diff --git a/src/mountainash/relations/dag/validation.py b/src/mountainash/relations/dag/validation.py index b44b501d..ba92613f 100644 --- a/src/mountainash/relations/dag/validation.py +++ b/src/mountainash/relations/dag/validation.py @@ -26,11 +26,13 @@ def validate( context: "dict[str, Any] | None" = None, backend: str | None = None, failure_sample: int | None = None, + allow_imperfect_key: bool = False, ) -> DAGValidationResult: """Full validation — all per-resource checks, then all FK checks.""" return _run( dag, specs, context=context, backend=backend, fail_fast=False, failure_sample=failure_sample, + allow_imperfect_key=allow_imperfect_key, ) @@ -41,11 +43,13 @@ def validate_quick( context: "dict[str, Any] | None" = None, backend: str | None = None, failure_sample: int | None = None, + allow_imperfect_key: bool = False, ) -> DAGValidationResult: """Fast validation — same runner, fail_fast=True. Identical shapes.""" return _run( dag, specs, context=context, backend=backend, fail_fast=True, failure_sample=failure_sample, + allow_imperfect_key=allow_imperfect_key, ) @@ -57,6 +61,7 @@ def _run( backend: str | None, fail_fast: bool, failure_sample: int | None, + allow_imperfect_key: bool = False, ) -> DAGValidationResult: from mountainash.datacontracts.compiler import compile_datacontract from mountainash.datacontracts.contract import BaseDataContract @@ -104,4 +109,5 @@ def _run( failure_sample=failure_sample, backend=backend, fk_error_summaries=fk_errors, + allow_imperfect_key=allow_imperfect_key, ) diff --git a/src/mountainash/validation/runner.py b/src/mountainash/validation/runner.py index db995d6a..c86d43b6 100644 --- a/src/mountainash/validation/runner.py +++ b/src/mountainash/validation/runner.py @@ -20,7 +20,7 @@ from mountainash.expressions.core.expression_nodes import ScalarFunctionNode from mountainash.validation.checks import VERDICT_PASSING, check_kind -from mountainash.validation.errors import UnknownCheckTypeError +from mountainash.validation.errors import IdentityInvalidError, UnknownCheckTypeError from mountainash.validation.identity import RowIdentity, validate_keyed_identity from mountainash.validation.result import ( CheckSummary, @@ -434,6 +434,7 @@ def validate_dag( failure_sample: int | None = None, backend: str | None = None, fk_error_summaries: "list[CheckSummary] | None" = None, + allow_imperfect_key: bool = False, ) -> "DAGValidationResult": from mountainash.relations import relation as as_relation from mountainash.validation.checks import ForeignKeyRule @@ -452,15 +453,43 @@ def _resolver(name: str) -> Any: for name, checks in checks_by_resource.items(): intra = [c for c in checks if not isinstance(c, ForeignKeyRule)] fk_rules.extend(c for c in checks if isinstance(c, ForeignKeyRule)) - result = self.validate_relation( - _resolver(name), - intra, - identity=identity_by_resource.get(name), - context=context, - fail_fast=fail_fast, - failure_sample=failure_sample, - validator_name=name, - ) + resource_identity = identity_by_resource.get(name) or RowIdentity("none") + try: + result = self.validate_relation( + _resolver(name), + intra, + identity=resource_identity, + allow_imperfect_key=allow_imperfect_key, + context=context, + fail_fast=fail_fast, + failure_sample=failure_sample, + validator_name=name, + ) + except IdentityInvalidError as exc: + # spec item 8j §3.2: a resource's invalid keyed identity never + # aborts the batch — isolate it into that resource's own failing + # result, same as every other exception in this loop already is + # (materialisation failures, runner.py:118-134). "__identity__" + # mirrors the existing "__fk__" synthetic-result naming + # (runner.py:470, the fail_fast early-return branch this snippet + # mirrors; runner.py:490, the fk_result construction). + summary = CheckSummary( + check_id="__identity__", + check_kind=None, + status="error", + severity="blocking", + error=f"{type(exc).__name__}: {exc}", + ) + result = ValidationResult( + passes=False, + validator_name=name, + datacontract_name=None, + context=dict(context or {}), + check_summaries=summaries_frame([summary]), + failure_cases=combine_failure_frames([], resource_identity), + identity=resource_identity, + identity_diagnostics={}, + ) results[name] = result if fail_fast and not result.passes: return DAGValidationResult( diff --git a/tests/relations/dag/cross_backend/test_datapackage_validation_loop.py b/tests/relations/dag/cross_backend/test_datapackage_validation_loop.py index acf17e27..043c774b 100644 --- a/tests/relations/dag/cross_backend/test_datapackage_validation_loop.py +++ b/tests/relations/dag/cross_backend/test_datapackage_validation_loop.py @@ -15,7 +15,6 @@ import pytest from mountainash.typespec.datapackage import DataPackage -from mountainash.validation.errors import IdentityInvalidError if TYPE_CHECKING: from pathlib import Path @@ -119,12 +118,14 @@ def test_conforming_data_passes_cross_backend(tmp_path, backend_name): @pytest.mark.cross_backend @pytest.mark.parametrize("backend_name", _COLLECT_BACKENDS) -def test_duplicate_primary_key_raises_identity_error(tmp_path, backend_name): +def test_duplicate_primary_key_isolates_identity_failure(tmp_path, backend_name): """Item 8j's characterization: a declared primary_key resolving to keyed - identity raises IdentityInvalidError and aborts the batch — it never - returns a DAGValidationResult for that call. This is the first check - that a descriptor-sourced to_typespec() produces a TypeSpec whose - primary_key still triggers that precondition.""" + identity is isolated into that resource's own failing result + (check_id="__identity__") rather than raised out of dag.validate — the + batch still returns a DAGValidationResult (spec item 8j §3.2). This is + the first check that a descriptor-sourced to_typespec() produces a + TypeSpec whose primary_key still triggers that precondition, now + surfaced through the DAG's per-resource isolation instead of a raise.""" pkg = _load_package( tmp_path, parents=[ @@ -138,8 +139,11 @@ def test_duplicate_primary_key_raises_identity_error(tmp_path, backend_name): dag = pkg.to_relation_dag() specs = {r.name: r.to_typespec() for r in pkg.resources} - with pytest.raises(IdentityInvalidError): - dag.validate(specs, backend=backend_name) + result = dag.validate(specs, backend=backend_name) # must not raise + + assert result.passes is False + parents_summaries = result.results["parents"].check_summaries + assert _status(parents_summaries, "__identity__") == "error" @pytest.mark.cross_backend diff --git a/tests/relations/dag/test_dag_validation.py b/tests/relations/dag/test_dag_validation.py index 6f45afdb..87bd37cb 100644 --- a/tests/relations/dag/test_dag_validation.py +++ b/tests/relations/dag/test_dag_validation.py @@ -292,3 +292,106 @@ def test_fast_stops_on_first_fk_violation(self): fk_summary = result.fk_result.check_summaries.row(0, named=True) assert fk_summary["check_id"] == "fk__c__pid__p" assert fk_summary["fail_count"] == 1 + + +class TestIdentityIsolation: + def _dup_pk_dag(self): + return _build_dag({"users": pl.DataFrame({"id": [1, 1], "age": [30, 40]})}) + + @pytest.mark.parametrize("spec_kind", ["typespec", "primary_key_contract", "natural_key_contract"]) + def test_allow_imperfect_key_reports_primary_key_unique(self, spec_kind): + dag = self._dup_pk_dag() + if spec_kind == "typespec": + spec = TypeSpec( + fields=[FieldSpec(name="id", type=UniversalType.INTEGER), + FieldSpec(name="age", type=UniversalType.INTEGER)], + primary_key=["id"], + ) + elif spec_kind == "primary_key_contract": + class C(BaseDataContract): + id: int + age: int + class Config: + primary_key = ["id"] + spec = C + else: + class C(BaseDataContract): + id: int + age: int + class Config: + natural_key = ["id"] + spec = C + + result = dag.validate(specs={"users": spec}, allow_imperfect_key=True) + assert result.passes is False + failing = set( + result.results["users"].check_summaries.filter( + result.results["users"].check_summaries["status"] != "passed" + )["check_id"].to_list() + ) + assert "primary_key_unique" in failing + + def test_default_isolates_identity_failure_no_exception(self): + dag = self._dup_pk_dag() + dag.add("other", ma.relation(pl.DataFrame({"name": ["ok"]}))) + spec_users = TypeSpec( + fields=[FieldSpec(name="id", type=UniversalType.INTEGER), + FieldSpec(name="age", type=UniversalType.INTEGER)], + primary_key=["id"], + ) + spec_other = TypeSpec(fields=[FieldSpec(name="name", type=UniversalType.STRING)]) + + result = dag.validate(specs={"users": spec_users, "other": spec_other}) # no allow_imperfect_key + + assert result.passes is False + users_summary = result.results["users"].check_summaries.row(0, named=True) + assert users_summary["check_id"] == "__identity__" + assert users_summary["status"] == "error" + assert result.results["other"].passes is True + + @pytest.mark.parametrize("order", [("users", "other"), ("other", "users")]) + def test_quick_fail_fast_ordering(self, order): + dag = self._dup_pk_dag() + dag.add("other", ma.relation(pl.DataFrame({"name": ["ok"]}))) + spec_users = TypeSpec( + fields=[FieldSpec(name="id", type=UniversalType.INTEGER), + FieldSpec(name="age", type=UniversalType.INTEGER)], + primary_key=["id"], + ) + spec_other = TypeSpec(fields=[FieldSpec(name="name", type=UniversalType.STRING)]) + specs = {order[0]: (spec_users if order[0] == "users" else spec_other), + order[1]: (spec_users if order[1] == "users" else spec_other)} + + result = dag.validate_quick(specs=specs) + + assert result.passes is False + assert "users" in result.results + if order == ("users", "other"): + assert "other" not in result.results # stopped before the second resource ran + else: + assert result.results["other"].passes is True # ran and reported before the stop + + def test_single_resource_dict_does_not_raise(self): + dag = self._dup_pk_dag() + spec = TypeSpec( + fields=[FieldSpec(name="id", type=UniversalType.INTEGER), + FieldSpec(name="age", type=UniversalType.INTEGER)], + primary_key=["id"], + ) + result = dag.validate(specs={"users": spec}) # must not raise + assert result.passes is False + + def test_missing_key_fields_isolated_not_raised_even_with_allow_imperfect_key(self): + # identity.py:69-73's missing-key-fields raise is unconditional - allow_imperfect_key + # never suppresses it (spec §7 round-2 fix) - but §3.2's DAG-tier isolation is categorical + # regardless of *why* IdentityInvalidError was raised, so it must still not escape here. + dag = _build_dag({"users": pl.DataFrame({"age": [30, 40]})}) # no "id" column at all + spec = TypeSpec( + fields=[FieldSpec(name="age", type=UniversalType.INTEGER)], + primary_key=["id"], # declared key column absent from the actual data + ) + result = dag.validate(specs={"users": spec}, allow_imperfect_key=True) # must not raise + assert result.passes is False + users_summary = result.results["users"].check_summaries.row(0, named=True) + assert users_summary["check_id"] == "__identity__" + assert users_summary["status"] == "error" \ No newline at end of file