539 lines
20 KiB
Python
539 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from generate_accuracy_phase4_splits import ( # noqa: E402
|
|
LeakageError,
|
|
assert_training_inputs_safe,
|
|
build_manifests,
|
|
generate,
|
|
)
|
|
|
|
|
|
SOURCE = ROOT / "fixtures/accuracy/p4/split-source-manifest.json"
|
|
|
|
|
|
def load_source() -> dict:
|
|
return json.loads(SOURCE.read_text(encoding="utf-8"))
|
|
|
|
|
|
def build_fixture_manifests(source: dict) -> tuple[dict, dict, dict]:
|
|
return build_manifests(source, trusted_fixture_mode=True)
|
|
|
|
|
|
def assert_fixture_training_inputs_safe(
|
|
input_paths: list[Path], input_records: list[dict], protected: dict
|
|
) -> None:
|
|
assert_training_inputs_safe(
|
|
input_paths, input_records, protected, trusted_fixture_mode=True
|
|
)
|
|
|
|
|
|
def test_normative_roles_hashes_and_source_order_are_enforced() -> None:
|
|
source = load_source()
|
|
development, protected, leakage = build_fixture_manifests(source)
|
|
reversed_source = copy.deepcopy(source)
|
|
reversed_source["samples"].reverse()
|
|
reversed_development, reversed_protected, reversed_leakage = (
|
|
build_fixture_manifests(reversed_source)
|
|
)
|
|
|
|
assert leakage["status"] == "pass"
|
|
assert leakage["finding_count"] == 0
|
|
assert leakage["split_counts"] == {
|
|
"background-test": 2,
|
|
"calibration": 2,
|
|
"challenge": 4,
|
|
"test": 7,
|
|
"train": 3,
|
|
"val": 3,
|
|
}
|
|
assert leakage["crs_validation"] == {
|
|
"status": "pass",
|
|
"crs": "EPSG:31370",
|
|
"distance_units": "m",
|
|
}
|
|
assert development["training_access_allowed_by_split"] == {
|
|
"train": True,
|
|
"val": False,
|
|
"calibration": False,
|
|
}
|
|
assert protected["labels_available_by_split"]["challenge"] == "sealed_external"
|
|
assert reversed_development["manifest_sha256"] == development["manifest_sha256"]
|
|
assert reversed_protected["manifest_sha256"] == protected["manifest_sha256"]
|
|
assert reversed_leakage == leakage
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "expected_code"),
|
|
[
|
|
("group_id", "S-SPATIAL-GROUP"),
|
|
("source_family", "S-SOURCE-FAMILY"),
|
|
("temporal_family", "S-TEMPORAL-FAMILY"),
|
|
("raw_image_sha256", "S-RAW-IMAGE-DUPLICATE"),
|
|
("processed_image_sha256", "S-PROCESSED-IMAGE-DUPLICATE"),
|
|
("label_sha256", "S-LABEL-DUPLICATE"),
|
|
("label_geometry_hash", "S-LABEL-GEOMETRY-DUPLICATE"),
|
|
("parent_raster_id", "S-PARENT-RASTER"),
|
|
("acquisition_id", "S-ACQUISITION"),
|
|
],
|
|
)
|
|
def test_cross_split_lineage_and_content_collisions_fail(
|
|
field: str, expected_code: str
|
|
) -> None:
|
|
source = load_source()
|
|
source["samples"][8][field] = source["samples"][0][field]
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
|
|
assert leakage["status"] == "fail"
|
|
assert expected_code in {item["code"] for item in leakage["findings"]}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "expected_code"),
|
|
[
|
|
("perceptual_image_hash", "S-PERCEPTUAL-IMAGE-NEAR-DUPLICATE"),
|
|
("label_geometry_fingerprint", "S-LABEL-GEOMETRY-NEAR-DUPLICATE"),
|
|
],
|
|
)
|
|
def test_near_duplicate_fingerprints_fail(field: str, expected_code: str) -> None:
|
|
source = load_source()
|
|
source["samples"][8][field] = source["samples"][0][field]
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
|
|
assert leakage["status"] == "fail"
|
|
assert expected_code in {item["code"] for item in leakage["findings"]}
|
|
|
|
|
|
def test_object_native_feature_and_spatial_collisions_fail() -> None:
|
|
source = load_source()
|
|
source["samples"][8]["object_ids"] = source["samples"][0]["object_ids"]
|
|
source["samples"][9]["native_feature_ids"] = source["samples"][1][
|
|
"native_feature_ids"
|
|
]
|
|
source["samples"][10]["bbox"] = source["samples"][2]["bbox"]
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
codes = {item["code"] for item in leakage["findings"]}
|
|
|
|
assert {"S-OBJECT-INSTANCE", "S-NATIVE-FEATURE", "S-SPATIAL-OVERLAP"} <= codes
|
|
|
|
|
|
def test_non_metric_crs_and_missing_normative_role_fail_closed() -> None:
|
|
geographic = load_source()
|
|
geographic["crs"] = "EPSG:4326"
|
|
with pytest.raises(LeakageError, match="projected in metres"):
|
|
build_fixture_manifests(geographic)
|
|
|
|
missing = load_source()
|
|
missing["samples"] = [
|
|
item for item in missing["samples"] if item["split"] != "calibration"
|
|
]
|
|
with pytest.raises(LeakageError, match="Required splits are absent"):
|
|
build_fixture_manifests(missing)
|
|
|
|
|
|
def test_training_firewall_only_allows_train_and_binds_protected_lineage() -> None:
|
|
development, protected, leakage = build_fixture_manifests(load_source())
|
|
assert leakage["status"] == "pass"
|
|
train = [item for item in development["samples"] if item["split"] == "train"]
|
|
validation = next(item for item in development["samples"] if item["split"] == "val")
|
|
protected_item = protected["samples"][0]
|
|
|
|
assert_fixture_training_inputs_safe([], train, protected)
|
|
with pytest.raises(LeakageError, match="non_train_role"):
|
|
assert_fixture_training_inputs_safe([], [validation], protected)
|
|
with pytest.raises(LeakageError, match="protected_identity"):
|
|
disguised = copy.deepcopy(train[0])
|
|
disguised["source_family"] = protected_item["source_family"]
|
|
assert_fixture_training_inputs_safe([], [disguised], protected)
|
|
with pytest.raises(LeakageError, match="protected_path"):
|
|
assert_fixture_training_inputs_safe(
|
|
[Path("vault/protected/test.json")], [], protected
|
|
)
|
|
|
|
|
|
def test_failed_generation_writes_status_but_no_consumable_manifests(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = load_source()
|
|
source["samples"][8]["group_id"] = source["samples"][0]["group_id"]
|
|
source_path = tmp_path / "source.json"
|
|
source_path.write_text(json.dumps(source), encoding="utf-8")
|
|
output = tmp_path / "out"
|
|
|
|
with pytest.raises(LeakageError, match="Leakage gate failed"):
|
|
generate(source_path, output, trusted_fixture_mode=True)
|
|
|
|
status = json.loads((output / "generation-status.json").read_text(encoding="utf-8"))
|
|
assert status["status"] == "fail"
|
|
assert status["consumable_manifests_valid"] is False
|
|
for name in (
|
|
"development-split-manifest.json",
|
|
"protected-split-manifest.json",
|
|
):
|
|
tombstone = json.loads((output / name).read_text(encoding="utf-8"))
|
|
assert tombstone["status"] == "invalidated"
|
|
assert tombstone["consumable"] is False
|
|
|
|
|
|
def test_deterministic_grouped_assignment_is_stable_and_keeps_relatives_together() -> (
|
|
None
|
|
):
|
|
source = load_source()
|
|
source["assignment_mode"] = "deterministic_grouped"
|
|
source["split_assignment"] = {
|
|
"seed": "fixed-phase4-test-seed",
|
|
"roles": [
|
|
"train",
|
|
"val",
|
|
"calibration",
|
|
"test",
|
|
"background-test",
|
|
"challenge",
|
|
],
|
|
"weights": {
|
|
"train": 6,
|
|
"val": 2,
|
|
"calibration": 1,
|
|
"test": 2,
|
|
"background-test": 1,
|
|
"challenge": 1,
|
|
},
|
|
"stratify_by": ["task"],
|
|
}
|
|
for item in source["samples"]:
|
|
item.pop("split")
|
|
source["samples"][1]["group_id"] = source["samples"][0]["group_id"]
|
|
|
|
development, protected, leakage = build_fixture_manifests(source)
|
|
reversed_source = copy.deepcopy(source)
|
|
reversed_source["samples"].reverse()
|
|
reversed_development, reversed_protected, reversed_leakage = (
|
|
build_fixture_manifests(reversed_source)
|
|
)
|
|
|
|
assigned = {
|
|
item["sample_id"]: item["split"]
|
|
for item in development["samples"] + protected["samples"]
|
|
}
|
|
assert assigned["det-train-a"] == assigned["seg-train-a"]
|
|
assert set(leakage["split_counts"]) == {
|
|
"train",
|
|
"val",
|
|
"calibration",
|
|
"test",
|
|
"background-test",
|
|
"challenge",
|
|
}
|
|
assert leakage["status"] == "pass"
|
|
assert reversed_development["manifest_sha256"] == development["manifest_sha256"]
|
|
assert reversed_protected["manifest_sha256"] == protected["manifest_sha256"]
|
|
assert reversed_leakage == leakage
|
|
|
|
|
|
def test_source_cannot_weaken_mandatory_roles_or_policy_floors() -> None:
|
|
source = load_source()
|
|
source["required_splits"] = ["train", "val", "test"]
|
|
with pytest.raises(LeakageError, match="mandatory role order"):
|
|
build_fixture_manifests(source)
|
|
|
|
grouped = load_source()
|
|
grouped["assignment_mode"] = "deterministic_grouped"
|
|
grouped["split_assignment"] = {"roles": ["train", "val"]}
|
|
for item in grouped["samples"]:
|
|
item.pop("split")
|
|
with pytest.raises(LeakageError, match="mandatory role order"):
|
|
build_fixture_manifests(grouped)
|
|
|
|
for field, value in (
|
|
("independence_buffer_m", 1999),
|
|
("perceptual_hamming_threshold", 3),
|
|
("label_geometry_hamming_threshold", 1),
|
|
):
|
|
weakened = load_source()
|
|
weakened[field] = value
|
|
with pytest.raises(LeakageError, match="code-owned minimum"):
|
|
build_fixture_manifests(weakened)
|
|
|
|
|
|
def test_task_coverage_gap_and_even_justified_exemption_fail_honestly() -> None:
|
|
source = load_source()
|
|
source["samples"] = [
|
|
item
|
|
for item in source["samples"]
|
|
if item["sample_id"] != "validation-test-national"
|
|
]
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
assert leakage["status"] == "fail"
|
|
assert "S-PROTECTED-TASK-COVERAGE-MISSING" in {
|
|
finding["code"] for finding in leakage["findings"]
|
|
}
|
|
|
|
source["protected_task_exemptions"] = {
|
|
"geospatial_data_validation": (
|
|
"No evaluator-visible reference exists; challenge data remains sealed."
|
|
)
|
|
}
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
assert leakage["status"] == "fail"
|
|
assert "S-PROTECTED-TASK-COVERAGE-EXEMPTED" in {
|
|
finding["code"] for finding in leakage["findings"]
|
|
}
|
|
|
|
|
|
def test_identifiers_and_acquisition_dates_are_canonical_leakage_keys() -> None:
|
|
source = load_source()
|
|
source["samples"][8]["group_id"] = " G01 "
|
|
_development, _protected, leakage = build_fixture_manifests(source)
|
|
assert "S-SPATIAL-GROUP" in {item["code"] for item in leakage["findings"]}
|
|
|
|
temporal = load_source()
|
|
temporal["samples"][8]["acquisition_date"] = temporal["samples"][0][
|
|
"acquisition_date"
|
|
]
|
|
_development, _protected, leakage = build_fixture_manifests(temporal)
|
|
assert "S-ACQUISITION-DATE" in {item["code"] for item in leakage["findings"]}
|
|
|
|
ambiguous = load_source()
|
|
ambiguous["samples"][8]["sample_id"] = " DET-TRAIN-A "
|
|
with pytest.raises(LeakageError, match="ambiguous canonical sample_id"):
|
|
build_fixture_manifests(ambiguous)
|
|
|
|
|
|
def test_challenge_is_sealed_in_standard_manifest() -> None:
|
|
_development, protected, leakage = build_fixture_manifests(load_source())
|
|
assert leakage["status"] == "pass"
|
|
challenge = [item for item in protected["samples"] if item["split"] == "challenge"]
|
|
forbidden = {
|
|
"label_sha256",
|
|
"label_geometry_hash",
|
|
"label_geometry_fingerprint",
|
|
"object_ids",
|
|
"native_feature_ids",
|
|
"record_sha256",
|
|
"label_path",
|
|
"label_geometry_path",
|
|
}
|
|
assert challenge
|
|
assert all(item["sealed"] is True for item in challenge)
|
|
assert all(not (forbidden & set(item)) for item in challenge)
|
|
|
|
|
|
def test_firewall_rejects_empty_tampered_wrong_and_renamed_manifests(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
development, protected, leakage = build_fixture_manifests(load_source())
|
|
assert leakage["status"] == "pass"
|
|
train = [item for item in development["samples"] if item["split"] == "train"]
|
|
|
|
with pytest.raises(LeakageError, match="empty manifest"):
|
|
assert_fixture_training_inputs_safe([], train, {})
|
|
with pytest.raises(LeakageError, match="missing fields"):
|
|
assert_fixture_training_inputs_safe([], train, development)
|
|
tampered = copy.deepcopy(protected)
|
|
tampered["samples"].pop()
|
|
with pytest.raises(LeakageError, match="checksum mismatch"):
|
|
assert_fixture_training_inputs_safe([], train, tampered)
|
|
|
|
renamed = tmp_path / "ordinary-training-input.json"
|
|
renamed.write_text(json.dumps(protected), encoding="utf-8")
|
|
with pytest.raises(LeakageError, match="protected_manifest_content"):
|
|
assert_fixture_training_inputs_safe([renamed], train, protected)
|
|
|
|
|
|
def _canonical_json_sha256(value: object) -> str:
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def make_governed_source(tmp_path: Path) -> dict:
|
|
source = load_source()
|
|
source["dataset_version"] = "governed-production-v1"
|
|
source["claim_boundary"] = "Governed production split source."
|
|
p3_items: list[dict] = []
|
|
provenance_records: list[dict] = []
|
|
asset_fields = {
|
|
"raw_image": ("raw_image_path", "raw_image_sha256"),
|
|
"processed_image": ("processed_image_path", "processed_image_sha256"),
|
|
"label": ("label_path", "label_sha256"),
|
|
"label_geometry": ("label_geometry_path", "label_geometry_hash"),
|
|
}
|
|
for sample in source["samples"]:
|
|
assets: dict[str, dict] = {}
|
|
p3_ids: dict[str, str] = {}
|
|
for role, (path_field, hash_field) in asset_fields.items():
|
|
relative = Path("assets") / sample["sample_id"] / f"{role}.bin"
|
|
path = tmp_path / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(f"{sample['sample_id']}:{role}:governed".encode())
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
p3_id = hashlib.sha256(
|
|
f"{sample['sample_id']}:{role}".encode()
|
|
).hexdigest()[:20]
|
|
relative_posix = relative.as_posix()
|
|
sample[path_field] = relative_posix
|
|
sample[hash_field] = digest
|
|
p3_ids[role] = p3_id
|
|
assets[role] = {
|
|
"path": relative_posix,
|
|
"sha256": digest,
|
|
"size_bytes": path.stat().st_size,
|
|
"p3_item_id": p3_id,
|
|
}
|
|
p3_items.append(
|
|
{
|
|
"item_id": p3_id,
|
|
"path": relative_posix,
|
|
"sha256": digest,
|
|
"size_bytes": path.stat().st_size,
|
|
"status": "examined",
|
|
"read_status": "readable",
|
|
"recommended_action": "accept",
|
|
"empty_content": False,
|
|
"schema_conformity": "conformant",
|
|
"anomalies": [],
|
|
}
|
|
)
|
|
provenance_id = (
|
|
"prov-" + hashlib.sha256(sample["sample_id"].encode()).hexdigest()[:24]
|
|
)
|
|
sample["governance_binding"] = {
|
|
"source_provenance_record_id": provenance_id,
|
|
"p3_item_ids": p3_ids,
|
|
}
|
|
provenance_records.append(
|
|
{
|
|
"record_id": provenance_id,
|
|
"sample_id": sample["sample_id"],
|
|
"status": "accepted",
|
|
"lineage_status": "complete",
|
|
"training_allowed": True,
|
|
"perceptual_image_hash": sample["perceptual_image_hash"],
|
|
"label_geometry_fingerprint": sample["label_geometry_fingerprint"],
|
|
"assets": assets,
|
|
}
|
|
)
|
|
p3 = {
|
|
"schema_version": 1,
|
|
"scan_id": "p3-test-governed",
|
|
"scanner_version": "3.0.3",
|
|
"completed_at": "2026-08-02T12:00:00+02:00",
|
|
"items": p3_items,
|
|
"reconciliation": {
|
|
"examined": len(p3_items),
|
|
"skipped": 0,
|
|
"unreachable": 0,
|
|
"inventory_total": len(p3_items),
|
|
"reconciles": True,
|
|
},
|
|
}
|
|
provenance = {
|
|
"schema_version": 1,
|
|
"manifest_type": "geointel_phase4_source_provenance",
|
|
"status": "pass",
|
|
"records": provenance_records,
|
|
"records_canonical_json_sha256": _canonical_json_sha256(provenance_records),
|
|
}
|
|
p3_path = tmp_path / "p3.json"
|
|
provenance_path = tmp_path / "provenance.json"
|
|
p3_path.write_text(json.dumps(p3), encoding="utf-8")
|
|
provenance_path.write_text(json.dumps(provenance), encoding="utf-8")
|
|
source["governance_evidence"] = {
|
|
"p3_scan_manifest": {
|
|
"path": p3_path.name,
|
|
"sha256": hashlib.sha256(p3_path.read_bytes()).hexdigest(),
|
|
},
|
|
"source_provenance_manifest": {
|
|
"path": provenance_path.name,
|
|
"sha256": hashlib.sha256(provenance_path.read_bytes()).hexdigest(),
|
|
},
|
|
}
|
|
return source
|
|
|
|
|
|
def test_fixture_mode_is_explicit_and_source_metadata_cannot_enable_it() -> None:
|
|
with pytest.raises(LeakageError, match="explicit trusted_fixture_mode"):
|
|
build_manifests(load_source())
|
|
|
|
source = load_source()
|
|
source["trusted_fixture_mode"] = True
|
|
with pytest.raises(LeakageError, match="cannot be enabled by source metadata"):
|
|
build_manifests(source, trusted_fixture_mode=True)
|
|
|
|
|
|
def test_empty_or_arbitrary_governance_json_is_rejected(tmp_path: Path) -> None:
|
|
source = load_source()
|
|
source["dataset_version"] = "governed-production-v1"
|
|
source["claim_boundary"] = "Governed production split source."
|
|
p3 = tmp_path / "p3.json"
|
|
provenance = tmp_path / "provenance.json"
|
|
p3.write_text("{}", encoding="utf-8")
|
|
provenance.write_text("{}", encoding="utf-8")
|
|
source["governance_evidence"] = {
|
|
"p3_scan_manifest": {
|
|
"path": p3.name,
|
|
"sha256": hashlib.sha256(p3.read_bytes()).hexdigest(),
|
|
},
|
|
"source_provenance_manifest": {
|
|
"path": provenance.name,
|
|
"sha256": hashlib.sha256(provenance.read_bytes()).hexdigest(),
|
|
},
|
|
}
|
|
with pytest.raises(LeakageError, match="non-empty JSON object"):
|
|
build_manifests(source, source_root=tmp_path)
|
|
|
|
|
|
def test_governed_records_require_exact_provenance_paths_and_live_bytes(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = make_governed_source(tmp_path)
|
|
development, protected, leakage = build_manifests(source, source_root=tmp_path)
|
|
assert leakage["status"] == "pass"
|
|
assert protected["source_trust"]["production_accuracy_use_allowed"] is True
|
|
train = [item for item in development["samples"] if item["split"] == "train"]
|
|
assert_training_inputs_safe([], train, protected)
|
|
|
|
no_paths = copy.deepcopy(train[0])
|
|
no_paths.pop("content_path_bindings")
|
|
with pytest.raises(LeakageError, match="missing_accessible_content_paths"):
|
|
assert_training_inputs_safe([], [no_paths], protected)
|
|
|
|
relabeled = copy.deepcopy(
|
|
next(item for item in protected["samples"] if item["split"] == "test")
|
|
)
|
|
relabeled["split"] = "train"
|
|
relabeled.pop("content_path_bindings")
|
|
for field in ("sample_id", "group_id", "source_family", "temporal_family"):
|
|
relabeled[field] = f"spoofed-{field}"
|
|
with pytest.raises(
|
|
LeakageError, match="unavailable_provenance_record|protected_identity"
|
|
):
|
|
assert_training_inputs_safe([], [relabeled], protected)
|
|
|
|
broken_binding = copy.deepcopy(source)
|
|
broken_binding["samples"][0]["governance_binding"]["p3_item_ids"]["raw_image"] = (
|
|
"0" * 20
|
|
)
|
|
with pytest.raises(LeakageError, match="provenance binding mismatch|P3 record"):
|
|
build_manifests(broken_binding, source_root=tmp_path)
|
|
|
|
raw_path = Path(train[0]["content_path_bindings"]["raw_image"]["resolved_path"])
|
|
raw_path.write_bytes(b"mutated after manifest creation")
|
|
with pytest.raises(LeakageError, match="record_path_hash_binding_mismatch"):
|
|
assert_training_inputs_safe([], train, protected)
|