| from __future__ import annotations | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| import pytest | |
| from pydantic import BaseModel, ValidationError | |
| from gcmd_classifier.datasets.cmr import resolve_cmr_collection, validate_cmr_url | |
| from gcmd_classifier.datasets.errors import ( | |
| DatasetCacheError, | |
| DatasetOrchestrationError, | |
| DatasetPersistenceError, | |
| ) | |
| from gcmd_classifier.datasets.models import ( | |
| CMRRetrievedSource, | |
| CoverageMode, | |
| DatasetClassificationOutcome, | |
| DatasetClassificationResult, | |
| DatasetIdentity, | |
| DatasetProcessingMetadata, | |
| DatasetProcessingStatus, | |
| DatasetSemanticValidatorState, | |
| ResolvedCMRCollection, | |
| ) | |
| from gcmd_classifier.datasets.orchestration import ( | |
| DatasetRunController, | |
| DurableInferenceClient, | |
| recover_run, | |
| validate_stage_log, | |
| ) | |
| from gcmd_classifier.datasets.pipeline import BlindDatasetPipeline | |
| from gcmd_classifier.datasets.runtime_models import ( | |
| ArtifactReference, | |
| ArtifactType, | |
| BlindRunState, | |
| CheckpointStatus, | |
| ModelCallAccounting, | |
| PipelineCheckpoint, | |
| PipelineStage, | |
| RunIdentity, | |
| SessionIdentity, | |
| StructuredStageLogEvent, | |
| ) | |
| from gcmd_classifier.datasets.runtime_persistence import ( | |
| DatasetArtifactStore, | |
| persist_expert_source_boundary, | |
| verify_persisted_blind_result, | |
| ) | |
| from gcmd_classifier.datasets.stage_cache import ( | |
| BlindClassificationCacheIdentity, | |
| BlindStageCache, | |
| CMRSourceCacheIdentity, | |
| READMEArtifactCacheIdentity, | |
| build_blind_cache_key, | |
| ) | |
| from gcmd_classifier.llm.base import ModelRequest, ModelResponse, ModelStage | |
| from tests.datasets.fixture_manifest import load_fixture_manifest | |
| SHA_A = "a" * 64 | |
| SHA_B = "b" * 64 | |
| def fixture_root() -> Path: | |
| return Path("tests/fixtures/datasets").resolve() | |
| def run() -> RunIdentity: | |
| return RunIdentity(run_id="run-1", concept_id="C123-TEST") | |
| def frozen_collection(index: int = 0) -> ResolvedCMRCollection: | |
| _, cases = load_fixture_manifest(Path("tests/fixtures/datasets/cases.json")) | |
| case = cases[index] | |
| raw = case.cmr_path.read_bytes() | |
| parsed = json.loads(raw) | |
| source = CMRRetrievedSource( | |
| submitted_url=case.metadata.cmr_url, | |
| final_url=case.metadata.cmr_url, | |
| retrieved_at=case.metadata.fixture_recorded_at, | |
| status_code=200, | |
| response_metadata={"content-type": "application/json"}, | |
| response_bytes=raw, | |
| source_text=raw.decode("utf-8"), | |
| sha256=hashlib.sha256(raw).hexdigest(), | |
| parsed_json=parsed, | |
| ) | |
| return resolve_cmr_collection(validate_cmr_url(case.metadata.cmr_url), source) | |
| def collection_for( | |
| identity: DatasetIdentity, keywords: list[dict] | None = None | |
| ) -> ResolvedCMRCollection: | |
| url = ( | |
| "https://cmr.earthdata.nasa.gov/search/collections.umm_json?native_id=" + identity.native_id | |
| ) | |
| parsed = { | |
| "hits": 1, | |
| "items": [ | |
| { | |
| "meta": { | |
| "concept-id": identity.concept_id, | |
| "native-id": identity.native_id, | |
| "revision-id": identity.cmr_revision_id, | |
| }, | |
| "umm": { | |
| "ShortName": identity.short_name, | |
| "Version": identity.version, | |
| "RelatedUrls": [], | |
| "ScienceKeywords": keywords or [], | |
| }, | |
| } | |
| ], | |
| } | |
| raw = json.dumps(parsed, separators=(",", ":")).encode() | |
| source = CMRRetrievedSource( | |
| submitted_url=url, | |
| final_url=url, | |
| retrieved_at="2026-08-13T00:00:00Z", | |
| status_code=200, | |
| response_bytes=raw, | |
| source_text=raw.decode(), | |
| sha256=hashlib.sha256(raw).hexdigest(), | |
| parsed_json=parsed, | |
| ) | |
| return resolve_cmr_collection(validate_cmr_url(url), source) | |
| def initial(run: RunIdentity) -> BlindRunState: | |
| return BlindRunState( | |
| run=run, | |
| session=SessionIdentity(session_id="session-1"), | |
| checkpoints=( | |
| PipelineCheckpoint(stage=PipelineStage.INITIALIZE, status=CheckpointStatus.COMPLETED), | |
| ), | |
| ) | |
| def test_storage_rejects_fixture_root(fixture_root: Path) -> None: | |
| with pytest.raises(DatasetPersistenceError, match="cannot be inside fixtures"): | |
| DatasetArtifactStore(fixture_root, fixture_root=fixture_root) | |
| def test_artifact_store_is_atomic_immutable_and_run_isolated( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| ref = store.persist_bytes( | |
| run=run, | |
| session_id="session-1", | |
| relative_name="source/raw.json", | |
| data=b"exact", | |
| artifact_type=ArtifactType.CMR_RESPONSE, | |
| ) | |
| assert store.load_verified(run, "session-1", ref) == b"exact" | |
| with pytest.raises(DatasetPersistenceError, match="overwrite"): | |
| store.persist_bytes( | |
| run=run, | |
| session_id="session-1", | |
| relative_name="source/raw.json", | |
| data=b"changed", | |
| artifact_type=ArtifactType.CMR_RESPONSE, | |
| ) | |
| other = RunIdentity(run_id="run-2", concept_id=run.concept_id) | |
| with pytest.raises(DatasetPersistenceError): | |
| store.load_verified(other, "session-1", ref) | |
| def test_atomic_failure_never_publishes_success( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| def fail(point: str) -> None: | |
| if point == "before_publish": | |
| raise RuntimeError("interrupt") | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root, failure_hook=fail) | |
| with pytest.raises(DatasetPersistenceError): | |
| store.persist_bytes( | |
| run=run, | |
| session_id="session-1", | |
| relative_name="blind.json", | |
| data=b"{}", | |
| artifact_type=ArtifactType.BLIND_RESULT, | |
| ) | |
| assert not (store.run_directory(run, "session-1") / "blind.json").exists() | |
| assert list(store.run_directory(run, "session-1").glob("*.tmp")) | |
| def test_illegal_and_terminal_transitions_are_rejected( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| controller = DatasetRunController( | |
| initial(run), DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| ) | |
| with pytest.raises(DatasetOrchestrationError): | |
| controller.complete(PipelineStage.CMR_SOURCE, "not begun") | |
| controller.begin(PipelineStage.CMR_SOURCE, "begin") | |
| controller.complete(PipelineStage.CMR_SOURCE, "done") | |
| with pytest.raises(DatasetOrchestrationError): | |
| controller.begin(PipelineStage.CMR_SOURCE, "stale") | |
| class Reply(BaseModel): | |
| value: str | |
| class FakeClient: | |
| calls = 0 | |
| def generate_structured(self, request: ModelRequest[Reply]) -> ModelResponse[Reply]: | |
| self.calls += 1 | |
| return ModelResponse( | |
| parsed=Reply(value="ok"), provider="fake", model_name="fake", prompt_version="v1" | |
| ) | |
| def request() -> ModelRequest[Reply]: | |
| return ModelRequest( | |
| stage=ModelStage.DATASET_COVERAGE, | |
| prompt="safe", | |
| response_schema=Reply, | |
| provider="fake", | |
| model_name="fake", | |
| prompt_version="v1", | |
| temperature=0, | |
| timeout_seconds=1, | |
| ) | |
| def test_inference_attempt_is_durable_before_transport( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| controller = DatasetRunController( | |
| initial(run), DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| ) | |
| fake = FakeClient() | |
| DurableInferenceClient(fake, controller).generate_structured(request()) | |
| assert controller.state.classification_attempted | |
| assert controller.state.model_calls.current_run_calls == 1 | |
| assert fake.calls == 1 | |
| def test_failed_attempt_checkpoint_prevents_transport( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| def fail(point: str) -> None: | |
| if point == "before_publish": | |
| raise RuntimeError("disk") | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| controller = DatasetRunController(initial(run), store) | |
| controller.store._failure_hook = fail | |
| fake = FakeClient() | |
| with pytest.raises(DatasetPersistenceError): | |
| DurableInferenceClient(fake, controller).generate_structured(request()) | |
| assert fake.calls == 0 | |
| def test_recovery_uses_validated_checkpoint_chain( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| controller = DatasetRunController(initial(run), store) | |
| controller.begin(PipelineStage.CMR_SOURCE, "begin") | |
| controller.complete(PipelineStage.CMR_SOURCE, "done") | |
| recovered, decision = recover_run(store, run, "session-1") | |
| assert recovered == controller.state | |
| assert decision.value == "resume_next_stage" | |
| def test_cache_identity_requires_complete_stage_inputs() -> None: | |
| first = CMRSourceCacheIdentity( | |
| submitted_url="https://cmr.example/a", | |
| final_url="https://cmr.example/a", | |
| response_sha256=SHA_A, | |
| retrieval_policy_version="v1", | |
| ) | |
| changed = first.model_copy(update={"response_sha256": SHA_B}) | |
| assert build_blind_cache_key(first) != build_blind_cache_key(changed) | |
| readme_a = READMEArtifactCacheIdentity( | |
| selected_source_entry_sha256=SHA_A, | |
| submitted_url="https://a.example/readme", | |
| final_url="https://a.example/readme", | |
| pdf_sha256=SHA_B, | |
| retrieval_policy_version="v1", | |
| ) | |
| readme_b = readme_a.model_copy(update={"submitted_url": "https://b.example/readme"}) | |
| assert build_blind_cache_key(readme_a) != build_blind_cache_key(readme_b) | |
| def test_blind_cache_round_trip_and_tamper_detection(tmp_path: Path, fixture_root: Path) -> None: | |
| cache = BlindStageCache(tmp_path / "cache", fixture_root=fixture_root) | |
| identity = BlindClassificationCacheIdentity( | |
| evidence_packet_sha256=SHA_A, | |
| vocabulary_sha256=SHA_B, | |
| model_configuration_sha256="c" * 64, | |
| topic_prompt_version="t", | |
| term_prompt_version="u", | |
| variable_prompt_version="v", | |
| review_policy_version="r", | |
| ) | |
| cache.put(identity, {"classification_attempted": True}) | |
| assert cache.get(identity) == {"classification_attempted": True} | |
| key = build_blind_cache_key(identity) | |
| path = cache._store.root / "blind" / "classification" / key.identity_sha256 / "record.json" | |
| path.write_text("{}") | |
| with pytest.raises(DatasetCacheError): | |
| cache.get(identity) | |
| def test_log_contract_rejects_absolute_reference(run: RunIdentity) -> None: | |
| event = StructuredStageLogEvent( | |
| timestamp="2026-01-01T00:00:00Z", | |
| run_id=run.run_id, | |
| session_id="session-1", | |
| stage=PipelineStage.CMR_SOURCE, | |
| status=CheckpointStatus.COMPLETED, | |
| event_code="CMR_COMPLETE", | |
| classification_attempted=False, | |
| current_run_model_calls=0, | |
| relative_reference="/secret/path", | |
| ) | |
| with pytest.raises(DatasetOrchestrationError): | |
| validate_stage_log(event) | |
| def test_verified_blind_result_gate_reads_persisted_bytes( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| identity = DatasetIdentity( | |
| concept_id=run.concept_id, | |
| native_id="SHORT_1", | |
| short_name="SHORT", | |
| version="1", | |
| cmr_revision_id=1, | |
| ) | |
| result = DatasetClassificationResult( | |
| schema_version="2.0", | |
| dataset_key=run.concept_id, | |
| identity=identity, | |
| evidence_packet_sha256=SHA_A, | |
| coverage_mode=CoverageMode.SINGLE_PRODUCT, | |
| semantic_validator_state=DatasetSemanticValidatorState.NOT_RUN_BY_POLICY, | |
| classification_attempted=True, | |
| processing_status=DatasetProcessingStatus.COMPLETED, | |
| classification_outcome=DatasetClassificationOutcome.NOT_CLASSIFIED, | |
| no_classification_reason="No supported path.", | |
| processing_metadata=DatasetProcessingMetadata(model_calls=1), | |
| ) | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| boundary = persist_expert_source_boundary( | |
| store=store, | |
| run=run, | |
| session_id="session-1", | |
| collection=collection_for(identity, [{"Topic": "SENTINEL"}]), | |
| ) | |
| ref = store.persist_json( | |
| run=run, | |
| session_id="session-1", | |
| relative_name="blind_result.json", | |
| value=result, | |
| artifact_type=ArtifactType.BLIND_RESULT, | |
| ) | |
| verified = verify_persisted_blind_result( | |
| store=store, | |
| run=run, | |
| session_id="session-1", | |
| reference=ref, | |
| expected_packet_sha256=SHA_A, | |
| expected_native_id="SHORT_1", | |
| vocabulary_hash=SHA_B, | |
| source_boundary=boundary, | |
| ) | |
| assert ( | |
| verified.artifact.sha256 | |
| == hashlib.sha256(store.load_verified(run, "session-1", ref)).hexdigest() | |
| ) | |
| def test_runtime_models_reject_unknown_fields(run: RunIdentity) -> None: | |
| with pytest.raises(ValidationError): | |
| ModelCallAccounting(current_run_calls=0, unknown=True) | |
| with pytest.raises(ValidationError): | |
| ArtifactReference( | |
| artifact_type="cmr_response", | |
| relative_path="../escape", | |
| sha256=SHA_A, | |
| byte_count=1, | |
| persistence_state="hash_verified", | |
| ) | |
| class NoReadmeServices: | |
| def __init__(self, collection: ResolvedCMRCollection) -> None: | |
| self.collection = collection | |
| self.stages: list[PipelineStage] = [] | |
| def execute(self, stage: PipelineStage, context: dict, model_client: object) -> object: | |
| self.stages.append(stage) | |
| if stage is PipelineStage.CMR_SOURCE: | |
| return self.collection.source | |
| if stage is PipelineStage.IDENTITY: | |
| return self.collection | |
| if stage is PipelineStage.README_DISCOVERY: | |
| return None | |
| return {"stage": stage.value} | |
| def test_ordered_pipeline_no_readme_is_preinference_skip( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| collection = frozen_collection() | |
| actual_run = RunIdentity(run_id=run.run_id, concept_id=collection.dataset_key) | |
| controller = DatasetRunController(initial(actual_run), store) | |
| services = NoReadmeServices(collection) | |
| terminal = BlindDatasetPipeline( | |
| controller=controller, | |
| store=store, | |
| services=services, | |
| model_client=FakeClient(), | |
| vocabulary_hash=SHA_A, | |
| ).run() | |
| assert terminal.terminal_outcome.value == "completed_skipped" | |
| assert not terminal.state.classification_attempted | |
| assert terminal.state.model_calls.current_run_calls == 0 | |
| assert services.stages == [ | |
| PipelineStage.CMR_SOURCE, | |
| PipelineStage.IDENTITY, | |
| PipelineStage.BLIND_BOUNDARY, | |
| PipelineStage.README_DISCOVERY, | |
| ] | |
| class SuccessfulServices: | |
| def __init__(self, packet: object, collection: ResolvedCMRCollection) -> None: | |
| self.packet = packet | |
| self.collection = collection | |
| def execute(self, stage: PipelineStage, context: dict, model_client: object) -> object: | |
| if stage not in {PipelineStage.CMR_SOURCE, PipelineStage.IDENTITY}: | |
| assert "EXPERT-SENTINEL" not in str(context) | |
| if stage is PipelineStage.CMR_SOURCE: | |
| return self.collection.source | |
| if stage is PipelineStage.IDENTITY: | |
| return self.collection | |
| if stage is PipelineStage.PRODUCT_RESOLUTION: | |
| model_client.generate_structured(request()) | |
| if stage is PipelineStage.README_RETRIEVAL: | |
| return b"%PDF-1.4 frozen" | |
| if stage is PipelineStage.EVIDENCE_PACKET: | |
| return self.packet | |
| if stage is PipelineStage.CLASSIFICATION: | |
| return DatasetClassificationResult( | |
| schema_version="2.0", | |
| dataset_key=self.packet.target.concept_id, | |
| identity=self.packet.target, | |
| evidence_packet_sha256=self.packet.packet_sha256, | |
| coverage_mode=self.packet.coverage_mode, | |
| semantic_validator_state=DatasetSemanticValidatorState.NOT_RUN_BY_POLICY, | |
| classification_attempted=True, | |
| processing_status=DatasetProcessingStatus.COMPLETED, | |
| classification_outcome=DatasetClassificationOutcome.NOT_CLASSIFIED, | |
| no_classification_reason="No defensible path.", | |
| processing_metadata=DatasetProcessingMetadata(model_calls=1), | |
| ) | |
| return {"stage": stage.value} | |
| def test_successful_not_classified_pipeline_is_persisted_and_verified( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| from tests.datasets.test_classifier_adapter import _packet | |
| packet = _packet(tmp_path) | |
| collection = collection_for(packet.target, [{"Topic": "EXPERT-SENTINEL"}]) | |
| actual_run = RunIdentity(run_id=run.run_id, concept_id=packet.target.concept_id) | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| controller = DatasetRunController(initial(actual_run), store) | |
| terminal = BlindDatasetPipeline( | |
| controller=controller, | |
| store=store, | |
| services=SuccessfulServices(packet, collection), | |
| model_client=FakeClient(), | |
| vocabulary_hash=SHA_A, | |
| ).run() | |
| assert terminal.terminal_outcome.value == "completed_not_classified" | |
| assert terminal.verified_blind_result is not None | |
| assert terminal.state.blind_result_hash_verified | |
| assert terminal.state.classification_attempted | |
| assert terminal.state.model_calls.current_run_calls == 1 | |
| def test_sealed_artifact_failure_stops_pipeline_before_inference( | |
| tmp_path: Path, fixture_root: Path, run: RunIdentity | |
| ) -> None: | |
| collection = frozen_collection() | |
| actual_run = RunIdentity(run_id=run.run_id, concept_id=collection.dataset_key) | |
| store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root) | |
| controller = DatasetRunController(initial(actual_run), store) | |
| publishes = 0 | |
| def fail(point: str) -> None: | |
| nonlocal publishes | |
| if point == "before_publish": | |
| publishes += 1 | |
| if publishes == 5: | |
| raise RuntimeError("sealed persistence unavailable") | |
| store._failure_hook = fail | |
| model = FakeClient() | |
| terminal = BlindDatasetPipeline( | |
| controller=controller, | |
| store=store, | |
| services=NoReadmeServices(collection), | |
| model_client=model, | |
| vocabulary_hash=SHA_A, | |
| ).run() | |
| assert terminal.terminal_outcome.value == "pre_inference_failure" | |
| assert not terminal.state.classification_attempted | |
| assert terminal.state.model_calls.current_run_calls == 0 | |
| assert model.calls == 0 | |