"""Versioned factor-definition and factor-set contract conformance.""" from __future__ import annotations import copy import hashlib import json from dataclasses import FrozenInstanceError from pathlib import Path from typing import Any, Callable import pytest from quant_engine.factor_contracts import ( ActorIdentity, AvailabilityMode, ContractErrorCode, Causation, DataFoundationEnvelope, DatasetSnapshotEnvelope, FactorContractError, FactorDefinition, FactorInput, FactorSetRef, HistoricalAvailability, InputBinding, LegacyFactorBinding, OutputArtifactRef, OutputCoverage, OutputQuality, OutputQualityCheck, PayloadValidation, ProducerIdentity, TypedParameter, ViewAvailability, canonical_json_bytes, factor_definition_from_alpha158, factor_input_schema_digest, validate_factor_catalog, ) from quant_engine.governed_pipeline import ( FactorVersion, bind_legacy_factor, project_legacy_factor, ) FIXTURE_PATH = Path(__file__).parent / "fixtures" / "factor-contracts-v1.golden.json" VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c" VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" def _golden() -> dict[str, Any]: loaded = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) assert isinstance(loaded, dict) return loaded def _sha256(value: bytes) -> str: return f"sha256:{hashlib.sha256(value).hexdigest()}" def _reidentify(item: dict[str, Any], field: str, prefix: str) -> None: payload = {key: value for key, value in item.items() if key != field} item[field] = f"{prefix}{hashlib.sha256(canonical_json_bytes(payload)).hexdigest()}" def _snapshot_and_foundation( fixture: dict[str, Any] | None = None, ) -> tuple[DatasetSnapshotEnvelope, DataFoundationEnvelope]: source = _golden() if fixture is None else fixture return ( DatasetSnapshotEnvelope.from_dict(source["dataset_snapshot"]), DataFoundationEnvelope.from_dict(source["data_foundation"]), ) def _definition( *, inputs: tuple[FactorInput, ...] | None = None, **overrides: Any, ) -> FactorDefinition: factor_inputs = inputs or ( FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume")), ) arguments: dict[str, Any] = { "factor_id": "alpha_005", "version": "1.0.0", "formula": "correlation(close, volume, 10)", "parameters": {}, "implementation_digest": "sha256:" + "1" * 64, "input_schema_digest": factor_input_schema_digest(factor_inputs), "inputs": factor_inputs, "valid_from": "2026-01-01T00:00:00.000000Z", "valid_until": "2027-01-01T00:00:00Z", "warmup_sessions": 10, "lag_sessions": 1, "producer": ProducerIdentity("quant_engine", "1.0.0"), "code_revision": "c" * 40, } arguments.update(overrides) return FactorDefinition.create(**arguments) def _golden_definition() -> FactorDefinition: factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume")) return factor_definition_from_alpha158( "alpha_005", version="1.0.0", parameters={}, inputs=(factor_input,), implementation_digest="sha256:" + "1" * 64, input_schema_digest=factor_input_schema_digest((factor_input,)), valid_from="2026-01-01T00:00:00.000000Z", valid_until="2027-01-01T00:00:00Z", warmup_sessions=10, lag_sessions=1, producer=ProducerIdentity("quant_engine", "1.0.0"), code_revision="c" * 40, ) def _factor_set_arguments( *, fixture: dict[str, Any] | None = None, snapshot: DatasetSnapshotEnvelope | None = None, foundation: DataFoundationEnvelope | None = None, definition: FactorDefinition | None = None, ) -> dict[str, Any]: source = _golden() if fixture is None else fixture if snapshot is None or foundation is None: parsed_snapshot, parsed_foundation = _snapshot_and_foundation(source) snapshot = snapshot or parsed_snapshot foundation = foundation or parsed_foundation selected_definition = definition or _golden_definition() output_schema_bytes = canonical_json_bytes(source["output_schema"]) output_content_bytes = canonical_json_bytes(source["output_content"]) artifact = OutputArtifactRef.create( schema_digest=_sha256(output_schema_bytes), content_digest=_sha256(output_content_bytes), ) return { "definitions": (selected_definition,), "dataset_snapshot": snapshot, "foundation": foundation, "selected_view_ref_ids": (VIEW_REF_ID,), "input_bindings": ( InputBinding( selected_definition.definition_id, "market", VIEW_REF_ID, VIEW_SCHEMA_DIGEST, ), ), "view_availability": ( ViewAvailability(VIEW_REF_ID, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64), ), "output_quality": OutputQuality( "passed", (OutputQualityCheck("finite_values", "passed", "sha256:" + "3" * 64),), ), "output_coverage": OutputCoverage( "complete", 1, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64, ), "output_schema_bytes": output_schema_bytes, "output_content_bytes": output_content_bytes, "output_artifact_ref": artifact, "availability_mode": AvailabilityMode.AS_AVAILABLE, "evaluation_at": "2026-01-03T11:00:00Z", "computed_at": "2026-01-03T10:15:00Z", "artifact_available_at": "2026-01-03T10:20:00Z", "producer": ProducerIdentity("quant_engine", "1.0.0"), "code_revision": "c" * 40, "actor": ActorIdentity("service", "factor_worker_v1"), "correlation_id": "research_run_001", "causation": Causation("foundation", foundation.foundation_id), "evidence_scope": "synthetic_fixture", "decision_eligible": False, } def _factor_set(**overrides: Any) -> FactorSetRef: arguments = _factor_set_arguments() arguments.update(overrides) return FactorSetRef.create(**arguments) def _assert_error( error: pytest.ExceptionInfo[FactorContractError], code: ContractErrorCode, path: str, ) -> None: assert error.value.code is code assert error.value.path == path def _mutate_artifact_schema_binding(value: dict[str, Any]) -> None: artifact = value["output_artifact_ref"] artifact["schema_digest"] = "sha256:" + "0" * 64 _reidentify(artifact, "artifact_id", "rhfactoroutputv1:sha256:") def test_golden_contracts_are_content_addressed_round_trippable_and_deeply_immutable() -> None: fixture = _golden() original_snapshot = copy.deepcopy(fixture["dataset_snapshot"]) original_foundation = copy.deepcopy(fixture["data_foundation"]) snapshot, foundation = _snapshot_and_foundation(fixture) definition = _golden_definition() factor_set = FactorSetRef.create(**_factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition)) binding = LegacyFactorBinding.create( definition=definition, legacy_factor_id="factor:demo-momentum", legacy_version="1.0.0", legacy_definition_sha256="b" * 64, legacy_dataset_schema_version="1.0.0", canonical_input_schema_digest=definition.input_schema_digest, correspondence_evidence_digest="sha256:" + "5" * 64, ) assert snapshot.pit_cutoff == "2026-01-02T07:01:00Z" assert foundation.pit_cutoff == factor_set.pit_cutoff == "2026-01-03T00:00:00Z" assert snapshot.pit_cutoff != foundation.pit_cutoff assert definition.definition_id == fixture["expected"]["definition_id"] assert definition.input_schema_digest == fixture["expected"]["input_schema_digest"] assert factor_set.factor_set_id == fixture["expected"]["factor_set_id"] assert factor_set.output_artifact_ref.artifact_id == fixture["expected"]["output_artifact_id"] assert binding.binding_id == fixture["expected"]["legacy_binding_id"] assert not definition.to_json().endswith("\n") assert not factor_set.to_json().endswith("\n") assert FactorDefinition.from_json(definition.to_json()) == definition reparsed = FactorSetRef.from_json( factor_set.to_json(), definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, output_schema_bytes=canonical_json_bytes(fixture["output_schema"]), output_content_bytes=canonical_json_bytes(fixture["output_content"]), ) reference_only = FactorSetRef.from_json( factor_set.to_json(), definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, ) assert reparsed.factor_set_id == factor_set.factor_set_id assert reparsed.payload_validation is PayloadValidation.PAYLOAD_REVALIDATED assert reference_only.payload_validation is PayloadValidation.REFERENCE_ONLY fixture["dataset_snapshot"]["descriptor"]["dataset"]["dimensions"].append("forbidden") fixture["data_foundation"]["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64 assert snapshot.to_dict() == original_snapshot assert foundation.to_dict() == original_foundation returned = snapshot.to_dict() returned["descriptor"]["dataset"]["dimensions"].append("also_forbidden") assert snapshot.to_dict() == original_snapshot with pytest.raises(FrozenInstanceError): snapshot.snapshot_id = "rhdsv1:sha256:" + "0" * 64 # type: ignore[misc] @pytest.mark.parametrize("variant", ["whitespace", "key_order"]) def test_contract_decoders_reject_non_canonical_json(variant: str) -> None: fixture = _golden() snapshot, foundation = _snapshot_and_foundation(fixture) definition = _golden_definition() factor_set = FactorSetRef.create( **_factor_set_arguments( fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition, ) ) binding = LegacyFactorBinding.create( definition=definition, legacy_factor_id="factor:demo-momentum", legacy_version="1.0.0", legacy_definition_sha256="b" * 64, legacy_dataset_schema_version="1.0.0", canonical_input_schema_digest=definition.input_schema_digest, correspondence_evidence_digest="sha256:" + "5" * 64, ) def non_canonical(value: str) -> str: if variant == "whitespace": return value + "\n" loaded = json.loads(value) reversed_items = dict(reversed(tuple(loaded.items()))) return json.dumps(reversed_items, ensure_ascii=False, separators=(",", ":")) decoders = ( lambda value: FactorDefinition.from_json(value), lambda value: FactorSetRef.from_json( value, definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, ), lambda value: LegacyFactorBinding.from_json(value, definition=definition), ) for decoder, encoded in zip( decoders, (definition.to_json(), factor_set.to_json(), binding.to_json()), strict=True, ): with pytest.raises(FactorContractError) as exc_info: decoder(non_canonical(encoded)) assert exc_info.value.code is ContractErrorCode.INVALID_FORMAT assert exc_info.value.path == "$" def test_factor_definition_identity_is_order_independent_where_semantics_are_unordered() -> None: first_input = FactorInput("prices", "sha256:" + "6" * 64, ("close",)) second_input = FactorInput("volumes", "sha256:" + "7" * 64, ("volume",)) inputs = (first_input, second_input) parameters_a = { "window": TypedParameter("integer", 10), "weights": TypedParameter("json", {"fast": [1, 2], "slow": [3, 4]}), } parameters_b = { "weights": TypedParameter("json", {"slow": [3, 4], "fast": [1, 2]}), "window": TypedParameter("integer", 10), } first = _definition( inputs=inputs, parameters=parameters_a, input_schema_digest=factor_input_schema_digest(inputs), ) second = _definition( inputs=tuple(reversed(inputs)), parameters=parameters_b, input_schema_digest=factor_input_schema_digest(tuple(reversed(inputs))), ) assert first.definition_id == second.definition_id assert first.to_json() == second.to_json() semantic_changes = ( _definition(factor_id="alpha_006"), _definition(version="1.0.1"), _definition(formula="correlation(close, volume, 11)"), _definition(parameters={"window": TypedParameter("integer", 10)}), _definition(implementation_digest="sha256:" + "9" * 64), _definition(valid_until="2027-01-02T00:00:00Z"), _definition(warmup_sessions=11), _definition(lag_sessions=2), _definition(producer=ProducerIdentity("quant_engine", "1.0.1")), _definition(code_revision="d" * 40), ) assert all(changed.definition_id != _golden_definition().definition_id for changed in semantic_changes) assert len({changed.definition_id for changed in semantic_changes}) == len(semantic_changes) def test_parameter_types_decimal_profile_and_detached_nested_values_are_strict() -> None: nested = {"ordered": [1, {"flag": True}]} parameter = TypedParameter("json", nested) nested["ordered"].append(2) definition = _definition(parameters={"payload": parameter}) assert definition.to_dict()["parameters"]["payload"]["value"] == { "ordered": [1, {"flag": True}] } integer_definition = _definition(parameters={"value": TypedParameter("integer", 1)}) string_definition = _definition(parameters={"value": TypedParameter("string", "1")}) assert integer_definition.definition_id != string_definition.definition_id for parameter_type, value, code in ( ("decimal", "1.0", ContractErrorCode.INVALID_FORMAT), ("decimal", "1e3", ContractErrorCode.INVALID_FORMAT), ("decimal", "-0", ContractErrorCode.INVALID_FORMAT), ("integer", True, ContractErrorCode.TYPE_ERROR), ("json", 1.5, ContractErrorCode.TYPE_ERROR), ("json", {"é": "bad-key"}, ContractErrorCode.INVALID_FORMAT), ("json", 9_007_199_254_740_992, ContractErrorCode.INVALID_VALUE), ): with pytest.raises(FactorContractError) as error: TypedParameter(parameter_type, value) assert error.value.code is code assert TypedParameter("decimal", "10.25").to_dict()["value"] == "10.25" def test_catalog_rejects_duplicate_and_overlapping_logical_validity_but_allows_adjacency() -> None: base = _golden_definition() adjacent = _definition(valid_from="2027-01-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z") assert len(validate_factor_catalog((adjacent, base))) == 2 with pytest.raises(FactorContractError) as duplicate: validate_factor_catalog((base, base)) _assert_error(duplicate, ContractErrorCode.INVALID_VALUE, "$.definitions") overlapping = _definition(valid_from="2026-06-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z") with pytest.raises(FactorContractError) as overlap: validate_factor_catalog((base, overlapping)) _assert_error(overlap, ContractErrorCode.TIME_ORDER_VIOLATION, "$.definitions") def test_upstream_contracts_reject_unknown_fields_identity_forgery_and_unqualified_input() -> None: unknown = _golden()["dataset_snapshot"] unknown["provider"] = "forbidden" with pytest.raises(FactorContractError) as unknown_error: DatasetSnapshotEnvelope.from_dict(unknown) _assert_error(unknown_error, ContractErrorCode.UNKNOWN_FIELD, "$.provider") forged = _golden()["data_foundation"] forged["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64 with pytest.raises(FactorContractError) as forged_error: DataFoundationEnvelope.from_dict(forged) assert forged_error.value.code is ContractErrorCode.IDENTITY_MISMATCH assert forged_error.value.path.endswith("view_ref_id") rejected_source = _golden() rejected_source["dataset_snapshot"]["descriptor"]["qualification"]["status"] = "rejected" _reidentify(rejected_source["dataset_snapshot"], "snapshot_id", "rhdsv1:sha256:") rejected_snapshot = DatasetSnapshotEnvelope.from_dict(rejected_source["dataset_snapshot"]) _, foundation = _snapshot_and_foundation() with pytest.raises(FactorContractError) as rejected_error: FactorSetRef.create( **_factor_set_arguments(snapshot=rejected_snapshot, foundation=foundation) ) _assert_error( rejected_error, ContractErrorCode.QUALIFICATION_REJECTED, "$.dataset_snapshot.descriptor.qualification", ) def test_foundation_rejects_future_knowledge_and_per_view_calendar_borrowing() -> None: future = _golden()["data_foundation"] action = future["corporate_action_revisions"][0] old_action_id = action["action_revision_id"] action["knowledge_time"] = "2026-01-03T00:00:01Z" _reidentify(action, "action_revision_id", "rhcav1:sha256:") future["standardized_views"][0]["corporate_action_revision_ids"] = [action["action_revision_id"]] lineage = next(item for item in future["revision_lineage"] if item["revision_id"] == old_action_id) lineage["revision_id"] = action["action_revision_id"] lineage["knowledge_time"] = action["knowledge_time"] _reidentify(future["standardized_views"][0], "view_ref_id", "rhviewrefv1:sha256:") _reidentify(future, "foundation_id", "rhdfv1:sha256:") with pytest.raises(FactorContractError) as future_error: DataFoundationEnvelope.from_dict(future) _assert_error( future_error, ContractErrorCode.TIME_ORDER_VIOLATION, "$.revision_lineage.knowledge_time", ) uncovered = _golden()["data_foundation"] original_route_id = uncovered["instrument_routes"][0]["route_revision_id"] second_calendar = copy.deepcopy(uncovered["trading_calendar_revisions"][0]) second_calendar["calendar_id"] = "rhcalendar:99990000111122223333444455556666" _reidentify(second_calendar, "calendar_revision_id", "rhcalv1:sha256:") uncovered["trading_calendar_revisions"].append(second_calendar) route = uncovered["instrument_routes"][0] route["calendar_id"] = second_calendar["calendar_id"] _reidentify(route, "route_revision_id", "rhroutev1:sha256:") route_lineage = next(item for item in uncovered["revision_lineage"] if item["revision_id"] == original_route_id) route_lineage["revision_id"] = route["route_revision_id"] uncovered["revision_lineage"].append( { "revision_kind": "trading_calendar", "revision_id": second_calendar["calendar_revision_id"], "revision_number": 1, "knowledge_time": second_calendar["knowledge_time"], "evidence_digest": second_calendar["evidence_digest"], } ) view = uncovered["standardized_views"][0] view["instrument_route_revision_ids"] = [route["route_revision_id"]] _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:") _reidentify(uncovered, "foundation_id", "rhdfv1:sha256:") with pytest.raises(FactorContractError) as calendar_error: DataFoundationEnvelope.from_dict(uncovered) assert calendar_error.value.code is ContractErrorCode.INPUT_CLOSURE_VIOLATION assert "selected route calendar" in calendar_error.value.detail def _replay_fixture() -> dict[str, Any]: fixture = _golden() snapshot = fixture["dataset_snapshot"] snapshot["descriptor"]["published_at"] = "2026-01-04T00:00:00Z" _reidentify(snapshot, "snapshot_id", "rhdsv1:sha256:") foundation = fixture["data_foundation"] foundation["dataset_snapshot_id"] = snapshot["snapshot_id"] for view in foundation["standardized_views"]: view["dataset_snapshot_id"] = snapshot["snapshot_id"] _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:") _reidentify(foundation, "foundation_id", "rhdfv1:sha256:") return fixture def test_as_available_and_retrospective_replay_keep_distinct_time_claims() -> None: as_available = _factor_set() assert as_available.historical_availability is HistoricalAvailability.DECLARED_AS_AVAILABLE replay_source = _replay_fixture() snapshot, foundation = _snapshot_and_foundation(replay_source) replay_view_id = next(iter(foundation.views)) arguments = _factor_set_arguments( fixture=replay_source, snapshot=snapshot, foundation=foundation, ) arguments.update( selected_view_ref_ids=(replay_view_id,), input_bindings=( InputBinding( arguments["definitions"][0].definition_id, "market", replay_view_id, VIEW_SCHEMA_DIGEST, ), ), view_availability=( ViewAvailability(replay_view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64), ), availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY, computed_at="2026-01-04T00:20:00Z", artifact_available_at="2026-01-04T00:25:00Z", causation=Causation("foundation", foundation.foundation_id), ) replay = FactorSetRef.create(**arguments) assert replay.evaluation_at == "2026-01-03T11:00:00Z" assert replay.computed_at == "2026-01-04T00:20:00Z" assert replay.historical_availability is HistoricalAvailability.NOT_ESTABLISHED replay_source_args = _factor_set_arguments( fixture=replay_source, snapshot=snapshot, foundation=foundation, ) replay_source_args.update( selected_view_ref_ids=(replay_view_id,), input_bindings=( InputBinding( replay_source_args["definitions"][0].definition_id, "market", replay_view_id, VIEW_SCHEMA_DIGEST, ), ), view_availability=( ViewAvailability(replay_view_id, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64), ), causation=Causation("foundation", foundation.foundation_id), ) with pytest.raises(FactorContractError) as late_publication: FactorSetRef.create(**replay_source_args) _assert_error( late_publication, ContractErrorCode.TIME_ORDER_VIOLATION, "$.dataset_snapshot.descriptor.published_at", ) @pytest.mark.parametrize( ("overrides", "path"), [ ({"view_availability": (ViewAvailability(VIEW_REF_ID, "2026-01-03T00:00:01Z", "sha256:" + "2" * 64),)}, "$.view_availability"), ({"computed_at": "2026-01-02T23:40:00Z"}, "$.computed_at"), ({"artifact_available_at": "2026-01-03T10:14:00Z"}, "$.artifact_available_at"), ({"artifact_available_at": "2026-01-03T11:00:01Z"}, "$.artifact_available_at"), ({"evaluation_at": "2026-01-03T11:00:00"}, "$.evaluation_at"), ], ) def test_as_available_time_failures_are_typed(overrides: dict[str, Any], path: str) -> None: with pytest.raises(FactorContractError) as error: _factor_set(**overrides) assert error.value.code in { ContractErrorCode.INVALID_FORMAT, ContractErrorCode.TIME_ORDER_VIOLATION, } assert error.value.path == path def test_replay_rejects_backdating_and_historical_availability_promotion() -> None: source = _replay_fixture() snapshot, foundation = _snapshot_and_foundation(source) view_id = next(iter(foundation.views)) arguments = _factor_set_arguments(fixture=source, snapshot=snapshot, foundation=foundation) definition = arguments["definitions"][0] arguments.update( selected_view_ref_ids=(view_id,), input_bindings=(InputBinding(definition.definition_id, "market", view_id, VIEW_SCHEMA_DIGEST),), view_availability=(ViewAvailability(view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),), availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY, computed_at="2026-01-04T00:20:00Z", artifact_available_at="2026-01-04T00:25:00Z", causation=Causation("foundation", foundation.foundation_id), ) replay = FactorSetRef.create(**arguments) promoted = replay.to_dict() promoted["historical_availability"] = "declared_as_available" _reidentify(promoted, "factor_set_id", "rhfactorsetv1:sha256:") with pytest.raises(FactorContractError) as promotion_error: FactorSetRef.from_dict( promoted, definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error( promotion_error, ContractErrorCode.READINESS_ESCALATION, "$.historical_availability", ) arguments["computed_at"] = "2026-01-03T11:30:00Z" with pytest.raises(FactorContractError) as backdated_error: FactorSetRef.create(**arguments) _assert_error(backdated_error, ContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at") def _multi_view_fixture() -> tuple[dict[str, Any], str]: fixture = _golden() foundation = fixture["data_foundation"] second = copy.deepcopy(foundation["standardized_views"][0]) second["view_id"] = "rhview:11111111222222223333333344444444" second["schema_digest"] = "sha256:" + "6" * 64 second["content_digest"] = "sha256:" + "7" * 64 second["transformation_digest"] = "sha256:" + "8" * 64 _reidentify(second, "view_ref_id", "rhviewrefv1:sha256:") foundation["standardized_views"].append(second) _reidentify(foundation, "foundation_id", "rhdfv1:sha256:") return fixture, second["view_ref_id"] def test_multi_input_mapping_requires_exact_consumption_closure_and_is_order_independent() -> None: fixture, second_view_id = _multi_view_fixture() snapshot, foundation = _snapshot_and_foundation(fixture) inputs = ( FactorInput("prices", VIEW_SCHEMA_DIGEST, ("close",)), FactorInput("volumes", "sha256:" + "6" * 64, ("volume",)), ) definition = _definition( inputs=inputs, formula="correlation(close, volume, 10)", input_schema_digest=factor_input_schema_digest(inputs), ) first_binding = InputBinding(definition.definition_id, "prices", VIEW_REF_ID, VIEW_SCHEMA_DIGEST) second_binding = InputBinding(definition.definition_id, "volumes", second_view_id, "sha256:" + "6" * 64) first_availability = ViewAvailability(VIEW_REF_ID, "2026-01-02T23:40:00Z", "sha256:" + "2" * 64) second_availability = ViewAvailability(second_view_id, "2026-01-02T23:50:00Z", "sha256:" + "6" * 64) base = _factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition) base.update( selected_view_ref_ids=(VIEW_REF_ID, second_view_id), input_bindings=(first_binding, second_binding), view_availability=(first_availability, second_availability), causation=Causation("foundation", foundation.foundation_id), ) first = FactorSetRef.create(**base) reordered = dict(base) reordered.update( selected_view_ref_ids=(second_view_id, VIEW_REF_ID), input_bindings=(second_binding, first_binding), view_availability=(second_availability, first_availability), ) assert FactorSetRef.create(**reordered).factor_set_id == first.factor_set_id for invalid_bindings, invalid_views in ( ((first_binding,), (VIEW_REF_ID, second_view_id)), ((first_binding, second_binding), (VIEW_REF_ID,)), ((first_binding, second_binding), (VIEW_REF_ID, second_view_id, VIEW_REF_ID)), ): invalid = dict(base) invalid.update(input_bindings=invalid_bindings, selected_view_ref_ids=invalid_views) with pytest.raises(FactorContractError) as error: FactorSetRef.create(**invalid) assert error.value.code in { ContractErrorCode.INPUT_CLOSURE_VIOLATION, ContractErrorCode.INVALID_VALUE, } @pytest.mark.parametrize( ("mutate", "code", "path"), [ (lambda value: value["producer"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.producer.id"), (lambda value: value["producer"].pop("version"), ContractErrorCode.MISSING_FIELD, "$.producer.version"), (lambda value: value["producer"].update(id="other_engine"), ContractErrorCode.LINEAGE_VIOLATION, "$.producer.id"), (lambda value: value["producer"].update(version="latest"), ContractErrorCode.INVALID_FORMAT, "$.producer.version"), (lambda value: value.update(code_revision="bad"), ContractErrorCode.INVALID_FORMAT, "$.code_revision"), (lambda value: value["actor"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.actor.kind"), (lambda value: value["actor"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.actor.id"), (lambda value: value["actor"].update(kind="robot"), ContractErrorCode.INVALID_VALUE, "$.actor.kind"), (lambda value: value["actor"].update(id="latest"), ContractErrorCode.INVALID_VALUE, "$.actor.id"), (lambda value: value.pop("correlation_id"), ContractErrorCode.MISSING_FIELD, "$.correlation_id"), (lambda value: value.update(correlation_id="latest"), ContractErrorCode.INVALID_VALUE, "$.correlation_id"), (lambda value: value["causation"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.causation.kind"), (lambda value: value["causation"].update(kind="run"), ContractErrorCode.INVALID_VALUE, "$.causation.kind"), (lambda value: value["causation"].update(id="rhdfv1:sha256:" + "0" * 64), ContractErrorCode.LINEAGE_VIOLATION, "$.causation.id"), (lambda value: value.pop("output_artifact_ref"), ContractErrorCode.MISSING_FIELD, "$.output_artifact_ref"), (lambda value: value["output_artifact_ref"].update(artifact_id="rhfactoroutputv1:sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.output_artifact_ref.artifact_id"), (_mutate_artifact_schema_binding, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref"), (lambda value: value.update(availability_mode="implicit_fallback"), ContractErrorCode.INVALID_VALUE, "$.availability_mode"), (lambda value: value.pop("computed_at"), ContractErrorCode.MISSING_FIELD, "$.computed_at"), (lambda value: value.update(decision_eligible=True), ContractErrorCode.READINESS_ESCALATION, "$.decision_eligible"), (lambda value: value.update(evidence_scope="real_data"), ContractErrorCode.READINESS_ESCALATION, "$.evidence_scope"), (lambda value: value["upstream_evidence"].update(qualification_evidence_digest="sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.upstream_evidence"), ], ) def test_lineage_artifact_and_readiness_fields_have_independent_typed_negatives( mutate: Callable[[dict[str, Any]], Any], code: ContractErrorCode, path: str, ) -> None: factor_set = _factor_set() value = factor_set.to_dict() mutate(value) if "factor_set_id" in value: _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:") snapshot, foundation = _snapshot_and_foundation() with pytest.raises(FactorContractError) as error: FactorSetRef.from_dict( value, definitions=(_golden_definition(),), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error(error, code, path) def test_output_schema_content_bytes_cannot_be_swapped_or_forged() -> None: factor_set = _factor_set() fixture = _golden() snapshot, foundation = _snapshot_and_foundation() schema_bytes = canonical_json_bytes(fixture["output_schema"]) content_bytes = canonical_json_bytes(fixture["output_content"]) with pytest.raises(FactorContractError) as swapped: FactorSetRef.from_dict( factor_set.to_dict(), definitions=(_golden_definition(),), dataset_snapshot=snapshot, foundation=foundation, output_schema_bytes=content_bytes, output_content_bytes=schema_bytes, ) _assert_error(swapped, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref") with pytest.raises(FactorContractError) as noncanonical: FactorSetRef.create( **{ **_factor_set_arguments(), "output_schema_bytes": json.dumps(fixture["output_schema"], indent=2).encode(), } ) _assert_error(noncanonical, ContractErrorCode.INVALID_FORMAT, "$.output_schema_bytes") def test_unsuccessful_output_quality_or_coverage_cannot_form_a_factor_set() -> None: with pytest.raises(FactorContractError) as failed_quality: _factor_set( output_quality=OutputQuality( "failed", (OutputQualityCheck("finite_values", "failed", "sha256:" + "3" * 64),), ) ) _assert_error(failed_quality, ContractErrorCode.INVALID_VALUE, "$.output_quality") for coverage in ( OutputCoverage("incomplete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64), OutputCoverage("complete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64), ): with pytest.raises(FactorContractError) as incomplete: _factor_set(output_coverage=coverage) _assert_error(incomplete, ContractErrorCode.INVALID_VALUE, "$.output_coverage") def test_external_snapshot_definition_and_view_references_cannot_be_substituted() -> None: factor_set = _factor_set() snapshot, foundation = _snapshot_and_foundation() value = factor_set.to_dict() value["dataset_snapshot_id"] = "rhdsv1:sha256:" + "0" * 64 _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:") with pytest.raises(FactorContractError) as snapshot_error: FactorSetRef.from_dict( value, definitions=(_golden_definition(),), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error( snapshot_error, ContractErrorCode.INPUT_CLOSURE_VIOLATION, "$.dataset_snapshot_id", ) value = factor_set.to_dict() value["definition_ids"] = ["rhfactorv1:sha256:" + "0" * 64] _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:") with pytest.raises(FactorContractError) as definition_error: FactorSetRef.from_dict( value, definitions=(_golden_definition(),), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error( definition_error, ContractErrorCode.INPUT_CLOSURE_VIOLATION, "$.definition_ids", ) arguments = _factor_set_arguments() arguments["selected_view_ref_ids"] = ("rhviewrefv1:sha256:" + "0" * 64,) with pytest.raises(FactorContractError) as view_error: FactorSetRef.create(**arguments) _assert_error( view_error, ContractErrorCode.INPUT_CLOSURE_VIOLATION, "$.selected_view_ref_ids", ) @pytest.mark.parametrize( "invalid_definition_id", [ {"unexpected": "object"}, ["array"], 42, True, None, ], ) def test_factor_set_ref_definition_ids_reject_non_string_types( invalid_definition_id: Any, ) -> None: factor_set = _factor_set() definition = _golden_definition() value = factor_set.to_dict() value["definition_ids"] = [definition.definition_id, invalid_definition_id] _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:") snapshot, foundation = _snapshot_and_foundation() with pytest.raises(FactorContractError) as error: FactorSetRef.from_json( canonical_json_bytes(value), definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error(error, ContractErrorCode.TYPE_ERROR, "$.definition_ids[1]") def test_factor_set_ref_definition_ids_still_reject_duplicate_strings() -> None: factor_set = _factor_set() definition = _golden_definition() value = factor_set.to_dict() value["definition_ids"] = [definition.definition_id, definition.definition_id] _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:") snapshot, foundation = _snapshot_and_foundation() with pytest.raises(FactorContractError) as error: FactorSetRef.from_json( canonical_json_bytes(value), definitions=(definition,), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error(error, ContractErrorCode.INVALID_VALUE, "$.definition_ids") def test_factor_set_parent_requires_exact_identity_and_correlation() -> None: parent = _factor_set() child_arguments = _factor_set_arguments() child_arguments.update( output_content_bytes=canonical_json_bytes({"rows": [{"value": "0.250"}]}), causation=Causation("factor_set", parent.factor_set_id), parent=parent, ) child_arguments["output_artifact_ref"] = OutputArtifactRef.create( schema_digest=_sha256(child_arguments["output_schema_bytes"]), content_digest=_sha256(child_arguments["output_content_bytes"]), ) child = FactorSetRef.create(**child_arguments) assert child.causation.id == parent.factor_set_id missing_parent = child.to_dict() snapshot, foundation = _snapshot_and_foundation() with pytest.raises(FactorContractError) as missing_error: FactorSetRef.from_dict( missing_parent, definitions=(_golden_definition(),), dataset_snapshot=snapshot, foundation=foundation, ) _assert_error(missing_error, ContractErrorCode.LINEAGE_VIOLATION, "$.causation") wrong_correlation = dict(child_arguments) wrong_correlation["correlation_id"] = "different_run" with pytest.raises(FactorContractError) as correlation_error: FactorSetRef.create(**wrong_correlation) _assert_error(correlation_error, ContractErrorCode.LINEAGE_VIOLATION, "$.correlation_id") def test_legacy_bridge_is_explicit_lossy_and_preserves_all_four_historical_fields() -> None: definition = _golden_definition() legacy = FactorVersion( factor_id="factor:demo-momentum", version="1.0.0", definition_sha256="b" * 64, dataset_schema_version="1.0.0", ) binding = LegacyFactorBinding.create( definition=definition, legacy_factor_id=legacy.factor_id, legacy_version=legacy.version, legacy_definition_sha256=legacy.definition_sha256, legacy_dataset_schema_version=legacy.dataset_schema_version, canonical_input_schema_digest=definition.input_schema_digest, correspondence_evidence_digest="sha256:" + "5" * 64, ) assert bind_legacy_factor(legacy, definition, binding) is definition assert project_legacy_factor(definition, binding) == legacy assert legacy.version_id == "factor:demo-momentum@1.0.0" assert legacy.definition_sha256 != definition.definition_id.rsplit(":", maxsplit=1)[-1] assert LegacyFactorBinding.from_json(binding.to_json(), definition=definition) == binding mismatched = FactorVersion( factor_id="factor:different", version=legacy.version, definition_sha256=legacy.definition_sha256, dataset_schema_version=legacy.dataset_schema_version, ) with pytest.raises(FactorContractError) as mismatch_error: bind_legacy_factor(mismatched, definition, binding) _assert_error(mismatch_error, ContractErrorCode.LEGACY_BINDING_MISMATCH, "$.binding") def test_bare_legacy_factor_or_id_cannot_enter_factor_set_contract() -> None: legacy = FactorVersion("factor:demo-momentum", "1.0.0", "b" * 64, "1.0.0") arguments = _factor_set_arguments() arguments["definitions"] = (legacy,) with pytest.raises(FactorContractError) as legacy_error: FactorSetRef.create(**arguments) _assert_error(legacy_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]") arguments["definitions"] = (legacy.version_id,) with pytest.raises(FactorContractError) as id_error: FactorSetRef.create(**arguments) _assert_error(id_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")