from __future__ import annotations from datetime import timedelta from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path from uuid import uuid4 import pytest from sqlalchemy import CheckConstraint from app.core.errors import AppError from app.models import ( Dataset, DatasetLineageEdge, DatasetQuarantine, DatasetVersion, SourceRegistry, SourceSnapshot, ) from app.services.coverage_registry_service import SOURCE_DEFINITIONS from app.services.dataset_consumption_gate_service import DatasetConsumptionGate from app.services.dataset_service import DatasetService from app.services.source_registry_service import ( SERVER_OWNED_SOURCE_DEFINITIONS, SourceRegistryService, ) class _Query: def __init__(self, session: "InMemorySession", model: type) -> None: self.session = session self.model = model self.predicates = [] def filter(self, *predicates): self.predicates.extend(predicates) return self def one_or_none(self): matches = self._matches() if len(matches) > 1: raise AssertionError( f"Expected one {self.model.__name__}, found {len(matches)}" ) return matches[0] if matches else None def all(self): return self._matches() def _matches(self): matches = list(self.session.objects.get(self.model, [])) for predicate in self.predicates: field_name = predicate.left.key expected = predicate.right.value operator_name = getattr(predicate.operator, "__name__", "") if operator_name == "in_op": matches = [ item for item in matches if getattr(item, field_name) in expected ] else: matches = [ item for item in matches if getattr(item, field_name) == expected ] return matches class InMemorySession: def __init__(self, *objects: object) -> None: self.objects: dict[type, list[object]] = {} self.added: list[object] = [] self.flushes = 0 for item in objects: self._store(item) def query(self, model: type) -> _Query: return _Query(self, model) def add(self, item: object) -> None: if getattr(item, "id", None) is None: setattr(item, "id", uuid4()) self._store(item) self.added.append(item) def flush(self) -> None: self.flushes += 1 def _store(self, item: object) -> None: self.objects.setdefault(type(item), []).append(item) def _registry(source_key: str) -> SourceRegistry: definition = SERVER_OWNED_SOURCE_DEFINITIONS[source_key] return SourceRegistry(id=uuid4(), **definition.as_model_values()) def _dataset() -> Dataset: return Dataset( id=uuid4(), project_id=uuid4(), name="candidate.tif", dataset_type="raster", source="governed", status="ready", validation_status="not_validated", provenance_status="incomplete", lineage_status="incomplete", quarantine_status="not_quarantined", ) def test_server_owned_definitions_encode_building_authority_and_non_ground_truth_sources() -> ( None ): grb = SERVER_OWNED_SOURCE_DEFINITIONS["grb"] buildings_register = SERVER_OWNED_SOURCE_DEFINITIONS[ "digitaal_vlaanderen_buildings_addresses_register" ] sentinel = SERVER_OWNED_SOURCE_DEFINITIONS["sentinel_2"] dhmv = SERVER_OWNED_SOURCE_DEFINITIONS["digitaal_vlaanderen_dhmv"] osm = SERVER_OWNED_SOURCE_DEFINITIONS["osm"] assert grb.classification == "authoritative" assert grb.usage_policy["ground_truth_allowed"] is True assert grb.usage_policy["validation_authority"]["building_validation"] == "primary" assert buildings_register.classification == "authoritative" assert buildings_register.usage_policy["ground_truth_allowed"] is False assert ( buildings_register.usage_policy["validation_authority"]["building_validation"] == "corroborative" ) assert ( buildings_register.usage_policy["validation_authority"][ "building_register_validation" ] == "primary" ) assert sentinel.classification == "contextual" assert dhmv.classification == "authoritative" assert dhmv.usage_policy["ground_truth_allowed"] is False assert ( dhmv.usage_policy["validation_authority"]["building_validation"] == "corroborative" ) assert ( dhmv.usage_policy["validation_authority"]["elevation_validation"] == "primary" ) assert osm.classification == "contextual" assert osm.usage_policy["ground_truth_allowed"] is False assert osm.usage_policy["automatic_ground_truth"] is False assert osm.usage_policy["training_allowed"] is False assert { "ngi_adminvector", "rbins_marine_reporting_units", "rbins_msp_2026", "grb", "digitaal_vlaanderen", "vrbg", "digitaal_vlaanderen_buildings_addresses_register", "digitaal_vlaanderen_orthophoto", "spw_orthophoto", "urbis_orthophoto", "digitaal_vlaanderen_dhmv", "spw_terrain", "spw_walous_land_cover", "spw_geoportail", "spw_picc", "urbis", "vmm_flood_hazard", "vmm_vha_bathymetry_profiles", "dov_soil_map", "statbel", "waterinfo", "agentschap_landbouw_zeevisserij_agricultural_parcels", "sentinel_2", "osm", "manual", "fixture", "map_selection", "derived", "training_label", "model", "experimental", "mdk_bcp_bathymetry", }.issubset(SERVER_OWNED_SOURCE_DEFINITIONS) for umbrella_key in ("digitaal_vlaanderen", "spw_geoportail"): definition = SERVER_OWNED_SOURCE_DEFINITIONS[umbrella_key] assert definition.classification == "authoritative" assert definition.usage_policy["ground_truth_allowed"] is False assert definition.usage_policy["automatic_ground_truth"] is False assert ( SERVER_OWNED_SOURCE_DEFINITIONS["mdk_bcp_bathymetry"].ingest_status == "not_configured" ) def test_coverage_and_direct_adapter_source_keys_are_registry_backed() -> None: coverage_source_keys = { definition.contract.source_name for definition in SOURCE_DEFINITIONS } coverage_materialization_keys = { source_key for definition in SOURCE_DEFINITIONS for source_key in definition.materialized_source_names } direct_adapter_source_keys = { "digitaal_vlaanderen", "spw_geoportail", "mdk_bcp_bathymetry", } assert ( coverage_source_keys | coverage_materialization_keys | direct_adapter_source_keys <= set(SERVER_OWNED_SOURCE_DEFINITIONS) ) def test_new_adapter_source_seed_rows_match_server_owned_registry_semantics() -> None: migration_path = ( Path(__file__).resolve().parents[1] / "alembic" / "versions" / "202608010001_source_registry_provenance.py" ) spec = spec_from_file_location("phase2_source_registry_migration", migration_path) assert spec is not None and spec.loader is not None migration = module_from_spec(spec) spec.loader.exec_module(migration) seed_rows = {row["source_key"]: row for row in migration._seed_rows()} for source_key in ("digitaal_vlaanderen", "spw_geoportail", "mdk_bcp_bathymetry"): expected = SERVER_OWNED_SOURCE_DEFINITIONS[source_key].as_model_values() observed = seed_rows[source_key] for field in ( "source_key", "display_name", "classification", "authority_name", "authority_scope_json", "provider_adapter_key", "source_url", "default_crs", "default_units", "geographic_coverage_json", "usage_policy_json", "freshness_status", "ingest_status", "known_limitations_json", ): assert observed[field] == expected[field] def test_ensure_source_is_idempotent_and_rejects_caller_owned_unknown_sources() -> None: grb = _registry("grb") session = InMemorySession(grb) assert SourceRegistryService.ensure_server_owned_source(session, "GRB") is grb assert session.added == [] with pytest.raises(AppError) as exc_info: SourceRegistryService.ensure_server_owned_source(session, "caller_claimed_grb") assert exc_info.value.code == "SOURCE_REGISTRY_ENTRY_NOT_FOUND" def test_snapshot_is_checksum_bound_and_idempotent() -> None: grb = _registry("grb") session = InMemorySession(grb) checksum = "a" * 64 snapshot = SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key="2026-08-01-gbg", checksum_sha256=checksum, source_version="2026-08-01", crs="EPSG:31370", units="metres", ) assert snapshot.source_registry_id == grb.id assert snapshot.checksum_sha256 == checksum assert snapshot.ingest_status == "ingested" assert ( SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key="2026-08-01-gbg", checksum_sha256=checksum, ) is snapshot ) with pytest.raises(AppError) as exc_info: SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key="2026-08-01-gbg", checksum_sha256="b" * 64, ) assert exc_info.value.code == "SOURCE_SNAPSHOT_IMMUTABILITY_CONFLICT" with pytest.raises(AppError) as version_conflict: SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key="2026-08-01-gbg", checksum_sha256=checksum, source_version="2026-08-02", ) assert version_conflict.value.code == "SOURCE_SNAPSHOT_IMMUTABILITY_CONFLICT" with pytest.raises(AppError) as invalid_checksum: SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key="bad-checksum", checksum_sha256="not-a-checksum", ) assert invalid_checksum.value.code == "SOURCE_SNAPSHOT_CHECKSUM_INVALID" def test_governed_import_reuses_an_identical_snapshot_without_rewriting_fetched_at() -> None: """A second project may bind the same immutable source snapshot safely.""" grb = _registry("grb") session = InMemorySession(grb) checksum = "a" * 64 observed_at = None metadata = { "dataset_type": "vector", "bounds_json": {"min_x": 4.0, "min_y": 50.0, "max_x": 4.1, "max_y": 50.1}, } source_metadata = {"source_url": "https://example.test/grb", "units": "metres"} # Dataset ingest keys are project-scoped, while a source snapshot is # globally keyed by immutable source evidence. This represents the same # source file arriving through two independently resumable imports. project_one, project_two = uuid4(), uuid4() assert ( DatasetService._ingest_key( project_id=project_one, source_key="grb", checksum_sha256=checksum, dataset_type="vector", dataset_role="source", area_id=None, reference_layer_name=None, source_version="2026-08-01", ) != DatasetService._ingest_key( project_id=project_two, source_key="grb", checksum_sha256=checksum, dataset_type="vector", dataset_role="source", area_id=None, reference_layer_name=None, source_version="2026-08-01", ) ) first_source, first_snapshot = DatasetService._record_snapshot( session, source_key="grb", checksum_sha256=checksum, source_version="2026-08-01", observed_at=observed_at, valid_from=None, valid_to=None, source_crs="EPSG:31370", source_metadata=source_metadata, metadata=metadata, ) original_fetched_at = first_snapshot.fetched_at replay_source, replay_snapshot = DatasetService._record_snapshot( session, source_key="grb", checksum_sha256=checksum, source_version="2026-08-01", observed_at=observed_at, valid_from=None, valid_to=None, source_crs="EPSG:31370", source_metadata=source_metadata, metadata=metadata, ) assert replay_source is first_source assert replay_snapshot is first_snapshot assert replay_snapshot.fetched_at == original_fetched_at assert session.objects[SourceSnapshot] == [first_snapshot] # Outside the governed replay path, a contradictory acquisition timestamp # remains immutable evidence and is still rejected. with pytest.raises(AppError) as fetched_at_conflict: SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key=first_snapshot.snapshot_key, checksum_sha256=checksum, fetched_at=original_fetched_at + timedelta(seconds=1), ) assert fetched_at_conflict.value.code == "SOURCE_SNAPSHOT_IMMUTABILITY_CONFLICT" # Replay mode is narrow: a changed immutable evidence field still fails. with pytest.raises(AppError) as conflict: SourceRegistryService.record_snapshot( session, source_key="grb", snapshot_key=first_snapshot.snapshot_key, checksum_sha256=checksum, crs="EPSG:4326", reuse_existing_snapshot=True, ) assert conflict.value.code == "SOURCE_SNAPSHOT_IMMUTABILITY_CONFLICT" def test_snapshot_schema_requires_a_canonical_sha256() -> None: constraints = { constraint.name: str(constraint.sqltext) for constraint in SourceSnapshot.__table__.constraints if isinstance(constraint, CheckConstraint) } assert SourceSnapshot.__table__.c.checksum_sha256.nullable is False assert "ck_source_snapshots_checksum_sha256" in constraints assert ( "lower(checksum_sha256)" in constraints["ck_source_snapshots_checksum_sha256"] ) def test_complete_provenance_binding_is_required_before_authoritative_validation() -> ( None ): grb = _registry("grb") snapshot = SourceSnapshot( id=uuid4(), source_registry_id=grb.id, snapshot_key="governed-grb", checksum_sha256="c" * 64, ingest_status="ingested", ) dataset = _dataset() SourceRegistryService.bind_dataset_provenance( dataset, source=grb, snapshot=snapshot, data_contract_key="vector.grb.buildings", data_contract_version="1.0.0", validation_status="passed", provenance_status="complete", lineage_status="complete", ) assert SourceRegistryService.is_dataset_eligible_for_authoritative_validation( dataset, source=grb, snapshot=snapshot, task="building_validation", ) osm = _registry("osm") dataset.source_registry_id = osm.id snapshot.source_registry_id = osm.id assert not SourceRegistryService.is_dataset_eligible_for_authoritative_validation( dataset, source=osm, snapshot=snapshot, task="building_validation", ) def test_lineage_and_quarantine_are_fail_closed_and_observable() -> None: session = InMemorySession() parent_id = uuid4() child_id = uuid4() edge = SourceRegistryService.record_lineage_edge( session, parent_dataset_id=parent_id, child_dataset_id=child_id, relation_type="derived_from", transformation_name="vector_clip", input_checksum_sha256="d" * 64, output_checksum_sha256="e" * 64, ) assert isinstance(edge, DatasetLineageEdge) assert ( SourceRegistryService.record_lineage_edge( session, parent_dataset_id=parent_id, child_dataset_id=child_id, relation_type="derived_from", transformation_name="vector_clip", input_checksum_sha256="d" * 64, output_checksum_sha256="e" * 64, ) is edge ) with pytest.raises(AppError) as self_reference: SourceRegistryService.record_lineage_edge( session, parent_dataset_id=parent_id, child_dataset_id=parent_id, relation_type="derived_from", transformation_name="vector_clip", ) assert self_reference.value.code == "DATASET_LINEAGE_SELF_REFERENCE" grandchild_id = uuid4() SourceRegistryService.record_lineage_edge( session, parent_dataset_id=child_id, child_dataset_id=grandchild_id, relation_type="derived_from", transformation_name="vector_buffer", ) with pytest.raises(AppError) as cycle: SourceRegistryService.record_lineage_edge( session, parent_dataset_id=grandchild_id, child_dataset_id=parent_id, relation_type="derived_from", transformation_name="vector_union", ) assert cycle.value.code == "DATASET_LINEAGE_CYCLE_DETECTED" dataset = _dataset() record = SourceRegistryService.quarantine_dataset( session, dataset=dataset, stage="vector_ingest", reason_code="CRS_UNVERIFIED", details={"observed_crs": None}, ) assert isinstance(record, DatasetQuarantine) assert dataset.status == "quarantined" assert dataset.quarantine_status == "quarantined" assert dataset.validation_status == "failed" version_parent = _dataset() version = DatasetVersion(id=uuid4(), dataset_id=version_parent.id, version=1) version_session = InMemorySession(version_parent, version) version_record = SourceRegistryService.quarantine_dataset( version_session, dataset_version=version, stage="dataset_version_validation", reason_code="CHECKSUM_MISMATCH", ) assert version_record.dataset_id == version_parent.id assert version_record.dataset_version_id == version.id assert version_parent.status == "quarantined" assert version_parent.quarantine_status == "quarantined" assert version_parent.validation_status == "failed" assert version_parent.provenance_status == "incomplete" assert version_parent.lineage_status == "incomplete" assert version.validation_status == "failed" assert version.provenance_status == "incomplete" snapshot = SourceSnapshot( id=uuid4(), source_registry_id=uuid4(), snapshot_key="quarantined-source", checksum_sha256="f" * 64, ingest_status="ingested", ) SourceRegistryService.quarantine_dataset( session, source_snapshot=snapshot, stage="source_snapshot_validation", reason_code="CHECKSUM_MISMATCH", ) assert snapshot.ingest_status == "quarantined" def test_quarantine_propagates_transitively_to_descendant_dataset_and_version_consumption_gates() -> ( None ): """A -> B -> C must fail closed when the governing A artifact is rejected.""" source = _registry("grb") snapshot = SourceSnapshot( id=uuid4(), source_registry_id=source.id, snapshot_key="transitive-quarantine-source", checksum_sha256="a" * 64, freshness_status="current", ingest_status="ingested", ) def governed_dataset(name: str) -> Dataset: dataset = _dataset() dataset.name = name dataset.source = "grb" dataset.source_name = "grb" dataset.dataset_role = "source" dataset.checksum_sha256 = snapshot.checksum_sha256 dataset.source_registry_id = source.id dataset.source_snapshot_id = snapshot.id dataset.data_contract_key = "geointel.raster.geotiff" dataset.data_contract_version = "1.0.0" dataset.validation_status = "passed" dataset.provenance_status = "complete" dataset.lineage_status = "complete" dataset.quarantine_status = "not_quarantined" dataset.status = "ready" dataset.source_registry = source dataset.source_snapshot = snapshot return dataset parent = governed_dataset("parent.tif") child = governed_dataset("child.tif") grandchild = governed_dataset("grandchild.tif") parent_version = DatasetVersion( id=uuid4(), dataset_id=parent.id, version=1, validation_status="passed", provenance_status="complete", lineage_status="complete", ) child_version = DatasetVersion( id=uuid4(), dataset_id=child.id, version=1, validation_status="passed", provenance_status="complete", lineage_status="complete", ) grandchild_version = DatasetVersion( id=uuid4(), dataset_id=grandchild.id, version=1, validation_status="passed", provenance_status="complete", lineage_status="complete", ) session = InMemorySession( parent, child, grandchild, parent_version, child_version, grandchild_version, ) SourceRegistryService.record_lineage_edge( session, parent_dataset_id=parent.id, child_dataset_id=child.id, relation_type="derived_from", transformation_name="clip", ) SourceRegistryService.record_lineage_edge( session, parent_dataset_id=child.id, child_dataset_id=grandchild.id, relation_type="derived_from", transformation_name="buffer", ) assert ( DatasetConsumptionGate.evaluate(child, purpose="production_inference").eligible is True ) assert ( DatasetConsumptionGate.evaluate( grandchild, purpose="production_inference" ).eligible is True ) SourceRegistryService.quarantine_dataset( session, dataset=parent, stage="contract_validation", reason_code="CHECKSUM_MISMATCH", ) for dataset in (parent, child, grandchild): decision = DatasetConsumptionGate.evaluate( dataset, purpose="production_inference" ) assert dataset.status == "quarantined" assert dataset.quarantine_status == "quarantined" assert dataset.validation_status == "failed" assert dataset.provenance_status == "incomplete" assert dataset.lineage_status == "incomplete" assert decision.eligible is False assert "dataset_quarantined" in decision.reasons for dataset_version in (parent_version, child_version, grandchild_version): assert dataset_version.validation_status == "failed" assert dataset_version.provenance_status == "incomplete" assert dataset_version.lineage_status == "incomplete" def test_ingest_keys_are_scoped_and_migration_keeps_unknown_legacy_unbound() -> None: project_id = uuid4() dataset = _dataset() dataset.project_id = project_id dataset.ingest_key = "grb:2026-08-01:gbg:area-sha" version = DatasetVersion( id=uuid4(), dataset_id=dataset.id, ingest_key=dataset.ingest_key, validation_status="not_validated", provenance_status="incomplete", lineage_status="incomplete", ) session = InMemorySession(dataset, version) assert ( SourceRegistryService.find_dataset_by_ingest_key( session, project_id, dataset.ingest_key ) is dataset ) assert ( SourceRegistryService.find_dataset_version_by_ingest_key( session, dataset.id, dataset.ingest_key ) is version ) with pytest.raises(AppError) as invalid_key: SourceRegistryService.find_dataset_by_ingest_key(session, project_id, " ") assert invalid_key.value.code == "INGEST_KEY_INVALID" migration = ( Path(__file__).resolve().parents[1] / "alembic" / "versions" / "202608010001_source_registry_provenance.py" ).read_text(encoding="utf-8") assert "uuid_generate_v5" not in migration assert "__unregistered_legacy_source__" in migration assert "uq_datasets_project_ingest_key" in migration assert "uq_dataset_versions_dataset_ingest_key" in migration def test_migration_contains_database_guards_for_snapshot_pairing_contract_lineage_and_quarantine() -> ( None ): migration = ( Path(__file__).resolve().parents[1] / "alembic" / "versions" / "202608010001_source_registry_provenance.py" ).read_text(encoding="utf-8") assert "trg_datasets_snapshot_registry_guard" in migration assert "trg_dataset_versions_snapshot_registry_guard" in migration assert "trg_source_registry_write_guard" in migration assert "trg_source_snapshots_evidence_immutable" in migration assert "geointel_phase2_contract_report_guard" in migration assert "trg_datasets_contract_report_guard" in migration assert "trg_dataset_versions_contract_report_guard" in migration assert "matching complete validation report" in migration assert "geointel_phase2_lineage_cycle_guard" in migration assert "trg_dataset_lineage_edges_cycle_guard" in migration assert "geointel_phase2_lineage_edge_immutable_guard" in migration assert "trg_dataset_lineage_edges_immutable" in migration assert "WITH RECURSIVE descendants" in migration assert "geointel_phase2_quarantine_lineage_descendants" in migration assert "geointel_phase2_quarantine_state_guard" in migration assert "trg_dataset_quarantines_state_guard" in migration assert "accepted dataset artifact and provenance evidence is immutable" in migration