diff --git a/tests/test_api.py b/tests/test_api.py index 6aa26c2..461803a 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -13,6 +13,27 @@ def run(self: "Any", envelope: "Any") -> str: raise WorkflowAgentServiceError("sensitive provider detail") +def test_lifespan_invokes_configure_telemetry() -> None: + with patch("cas_reference_product.app.configure_telemetry") as configure: + with TestClient(create_app(Settings())): + configure.assert_called_once() + + +def test_workflow_api_returns_503_when_service_is_none(envelope: "Any") -> None: + with patch( + "cas_reference_product.app.build_workflow_agent_service", + return_value=None, + ): + with TestClient(create_app(Settings())) as client: + response = client.post( + "/api/v1/workflows", + json=envelope.model_dump(mode="json"), + ) + + assert response.status_code == 503 + assert response.json() == {"detail": "Workflow backend is not ready"} + + def test_workflow_api_emits_canonical_events(envelope: "Any") -> None: client = TestClient(create_app(Settings())) diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 7f6e3e3..2457cc5 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -1,10 +1,18 @@ import hashlib import json +import sys from pathlib import Path +from unittest.mock import patch import pytest -from cas_reference_product.evidence import DEFAULT_BUNDLE, EvidenceVerificationError, verify_bundle +from cas_reference_product.evidence import ( + DEFAULT_BUNDLE, + EvidenceVerificationError, + _load_json, + main, + verify_bundle, +) def copy_bundle(tmp_path: Path) -> Path: @@ -22,6 +30,26 @@ def write_json(path: Path, payload: dict[str, object]) -> None: path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") +def rebuild_evidence_digests(bundle: Path) -> None: + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + + for section_name in ("sourceProvenance", "contractRegistry", "evaluation", "platformWhatIf"): + section = descriptor[section_name] + section["sha256"] = hashlib.sha256((bundle / section["path"]).read_bytes()).hexdigest() + + artifact_manifest_path = bundle / "artifact-manifest.json" + artifact_manifest = json.loads(artifact_manifest_path.read_text(encoding="utf-8")) + manifest_entries = {item["uri"]: item for item in artifact_manifest["artifacts"]} + for artifact in descriptor["artifacts"]: + digest = hashlib.sha256((bundle / artifact["path"]).read_bytes()).hexdigest() + artifact["sha256"] = digest + manifest_entries[artifact["uri"]]["sha256"] = digest + + write_json(artifact_manifest_path, artifact_manifest) + write_json(descriptor_path, descriptor) + + def test_committed_immutable_evidence_bundle_verifies() -> None: verify_bundle() @@ -124,3 +152,161 @@ def test_evaluation_response_digest_is_mandatory(tmp_path: Path) -> None: with pytest.raises(EvidenceVerificationError, match="evaluation fixture digest mismatch"): verify_bundle(bundle) + + +def test_load_json_rejects_non_object_json(tmp_path: Path) -> None: + path = tmp_path / "array.json" + path.write_text("[1, 2, 3]", encoding="utf-8") + + with pytest.raises(EvidenceVerificationError, match="must contain a JSON object"): + _load_json(path) + + +def test_bundle_artifacts_must_be_a_non_empty_list(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["artifacts"] = [] + write_json(descriptor_path, descriptor) + + with pytest.raises( + EvidenceVerificationError, + match="bundle artifacts must be a non-empty list", + ): + verify_bundle(bundle) + + +def test_bundle_artifact_entry_must_be_an_object(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["artifacts"] = ["not-an-object"] + write_json(descriptor_path, descriptor) + + with pytest.raises(EvidenceVerificationError, match="bundle artifact entries must be objects"): + verify_bundle(bundle) + + +def test_artifact_missing_path_or_sha256_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["artifacts"] = [{"uri": "urn:test", "path": None, "sha256": None}] + write_json(descriptor_path, descriptor) + + with pytest.raises(EvidenceVerificationError, match="artifact path and sha256 must be strings"): + verify_bundle(bundle) + + +def test_artifact_sha256_with_invalid_pattern_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["artifacts"] = [ + {"uri": "urn:test", "path": "artifact-manifest.json", "sha256": "invalid"} + ] + write_json(descriptor_path, descriptor) + + with pytest.raises(EvidenceVerificationError, match="has an invalid SHA-256 digest"): + verify_bundle(bundle) + + +def test_artifact_path_traversal_outside_bundle_root_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["artifacts"] = [ + {"uri": "urn:test", "path": "../../outside.json", "sha256": "a" * 64} + ] + write_json(descriptor_path, descriptor) + + with pytest.raises(EvidenceVerificationError, match="escapes the bundle root"): + verify_bundle(bundle) + + +def test_invalid_git_sha_in_source_provenance_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + provenance_path = bundle / "artifacts" / "source-provenance.json" + provenance = json.loads(provenance_path.read_text(encoding="utf-8")) + provenance["repositories"][0]["sha"] = "not-a-git-sha" + write_json(provenance_path, provenance) + rebuild_evidence_digests(bundle) + + with pytest.raises(EvidenceVerificationError, match="invalid immutable source reference"): + verify_bundle(bundle) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("summary", {"failed": 1, "passed": 0, "total": 1}, "did not pass exactly one case"), + ("suiteId", "wrong-suite", "unexpected golden path evaluation suite"), + ], +) +def test_evaluation_metadata_mismatch_raises( + tmp_path: Path, field: str, value: object, message: str +) -> None: + bundle = copy_bundle(tmp_path) + evaluation_path = bundle / "artifacts" / "eval-evidence.json" + evaluation = json.loads(evaluation_path.read_text(encoding="utf-8")) + evaluation[field] = value + write_json(evaluation_path, evaluation) + rebuild_evidence_digests(bundle) + + with pytest.raises(EvidenceVerificationError, match=message): + verify_bundle(bundle) + + +def test_evaluation_response_digest_mismatch_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + evaluation_path = bundle / "artifacts" / "eval-evidence.json" + evaluation = json.loads(evaluation_path.read_text(encoding="utf-8")) + evaluation["evidence"][0]["execution"]["responseDigest"] = f"sha256:{'0' * 64}" + write_json(evaluation_path, evaluation) + rebuild_evidence_digests(bundle) + + with pytest.raises(EvidenceVerificationError, match="evaluation response digest mismatch"): + verify_bundle(bundle) + + +def test_available_container_with_invalid_digest_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["containerImage"] = {"status": "available", "digest": "invalid"} + write_json(descriptor_path, descriptor) + + with pytest.raises(EvidenceVerificationError, match="requires a valid digest"): + verify_bundle(bundle) + + +def test_verification_result_outcome_not_passed_raises(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + result_path = bundle / "verification-result.json" + result = json.loads(result_path.read_text(encoding="utf-8")) + result["outcome"] = "failed" + write_json(result_path, result) + + with pytest.raises(EvidenceVerificationError, match="canonical VerificationResult must pass"): + verify_bundle(bundle) + + +def test_main_returns_zero_on_valid_bundle() -> None: + with patch.object(sys, "argv", ["evidence"]): + assert main() == 0 + + +def test_main_returns_one_on_invalid_bundle_path() -> None: + with patch.object(sys, "argv", ["evidence", "/nonexistent/path/bundle"]): + assert main() == 1 + + +def test_main_returns_one_on_verification_failure(tmp_path: Path) -> None: + bundle = copy_bundle(tmp_path) + descriptor_path = bundle / "bundle.json" + descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) + descriptor["platformWhatIf"]["deploymentClaim"] = "deployed" + write_json(descriptor_path, descriptor) + + with patch.object(sys, "argv", ["evidence", str(bundle)]): + assert main() == 1 diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 3f0b6cf..450605a 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -4,6 +4,7 @@ import pytest +from cas_reference_product.models import Actor, PromptEnvelope, TraceContext from cas_reference_product.workflow import WorkflowOrchestrator @@ -40,3 +41,31 @@ def test_orchestrator_returns_traceable_events(envelope: "Any") -> None: def test_orchestrator_propagates_failure(envelope: "Any") -> None: with pytest.raises(RuntimeError, match="expected"): WorkflowOrchestrator(FailingService(), envelope.repo).execute(envelope) + + +def test_event_includes_tracestate_when_present(envelope: "Any") -> None: + envelope_with_tracestate = PromptEnvelope( + correlationId=envelope.correlationId, + promptId=envelope.promptId, + runId=envelope.runId, + repo=envelope.repo, + actor=Actor(id="developer", type="human"), + timestamp=envelope.timestamp, + traceContext=TraceContext( + traceparent="00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + tracestate="vendor1=value1", + ), + intent=envelope.intent, + prompt=envelope.prompt, + constraints=envelope.constraints, + ) + + with patch( + "cas_reference_product.workflow.current_traceparent", + side_effect=lambda fallback: fallback, + ): + result = WorkflowOrchestrator( + SuccessfulService(), envelope.repo + ).execute(envelope_with_tracestate) + + assert all(event.traceContext.tracestate == "vendor1=value1" for event in result.events)