from datetime import UTC, datetime, timedelta, timezone import pytest from pydantic import ValidationError from tht.evidence.contracts import ( AcquiredDocument, EvidenceSource, EvidenceSourceError, EvidenceSourceErrorCategory, SourceObject, ) class StubSource: def discover(self): return iter( [ SourceObject( source_id="source:handbook", uri="https://host/handbook.md", fingerprint="sha256:abc", ) ] ) def acquire(self, item: SourceObject) -> AcquiredDocument: return AcquiredDocument( source=item, content=b"# Handbook", acquired_at=datetime(2026, 7, 12, tzinfo=UTC), media_type="text/markdown", ) def test_runtime_checkable_source_protocol(): source = StubSource() assert isinstance(source, EvidenceSource) assert source.acquire(next(source.discover())).content == b"# Handbook" def test_source_objects_are_frozen_and_metadata_defaults_are_independent(): first = SourceObject(source_id="source:a", uri="file:///a", fingerprint="sha256:a") second = SourceObject(source_id="source:b", uri="file:///b", fingerprint="sha256:b") with pytest.raises(ValidationError): first.uri = "file:///changed" # type: ignore[misc] with pytest.raises(TypeError): first.metadata["owner"] = "team-a" assert second.metadata == {} @pytest.mark.parametrize( "key", [ "password", "PassWd", "api_key", "x-api-key", "accessToken", "refresh.token", "client secret", "privateKey", "session_cookie", "Authorization", ], ) def test_source_metadata_rejects_credential_specific_keys(key): with pytest.raises(ValidationError, match="credential-like"): SourceObject( source_id="source:a", uri="https://host/a", fingerprint="etag:abc", metadata={"nested": [{key: "secret"}]}, ) def test_source_metadata_allows_benign_generic_token_and_secret_labels(): source = SourceObject( source_id="source:a", uri="https://host/a", fingerprint="etag:abc", metadata={"token": "word count token", "secret": False}, ) assert source.metadata["token"] == "word count token" def test_nested_metadata_is_recursively_immutable_and_serializes_as_json(): source = SourceObject( source_id="source:a", uri="https://host/a", fingerprint="etag:abc", metadata={"nested": {"items": [1, {"ok": True}]}}, ) with pytest.raises(TypeError): source.metadata["nested"]["items"][1]["ok"] = False assert '"items":[1,{"ok":true}]' in source.model_dump_json() @pytest.mark.parametrize( "uri", [ "https://user:pass@host/a", "https://host/a?api_key=secret", "https://host/a?accessToken=secret", ], ) def test_source_uri_rejects_embedded_credentials(uri): with pytest.raises(ValidationError, match="credentials"): SourceObject(source_id="source:a", uri=uri, fingerprint="etag:abc") def test_source_metadata_must_be_json_safe(): with pytest.raises(ValidationError): SourceObject( source_id="source:a", uri="file:///a", fingerprint="sha256:a", metadata={"path": object()}, ) def test_source_identity_and_fingerprint_must_be_namespaced(): with pytest.raises(ValidationError, match="namespaced"): SourceObject(source_id="plain", uri="file:///a", fingerprint="sha256:a") with pytest.raises(ValidationError, match="namespaced"): SourceObject(source_id="source:a", uri="file:///a", fingerprint="plain") def test_acquired_document_does_not_accept_credentials_as_extra_fields(): item = SourceObject(source_id="source:a", uri="https://host/a", fingerprint="etag:abc") with pytest.raises(ValidationError): AcquiredDocument(source=item, content=b"a", api_key="secret") def test_acquired_binary_content_has_explicit_json_round_trip(): item = SourceObject(source_id="source:a", uri="https://host/a", fingerprint="etag:abc") acquired = AcquiredDocument(source=item, content=b"\x00\xffbinary\x80") payload = acquired.model_dump_json() restored = AcquiredDocument.model_validate_json(payload) assert restored.content == acquired.content assert "binary" not in payload def test_datetimes_must_be_aware_and_are_normalized_to_utc(): with pytest.raises(ValidationError, match="timezone-aware"): SourceObject( source_id="source:a", uri="https://host/a", fingerprint="etag:abc", modified_at=datetime(2026, 7, 12), ) source = SourceObject( source_id="source:a", uri="https://host/a", fingerprint="etag:abc", modified_at=datetime(2026, 7, 12, 4, tzinfo=timezone(timedelta(hours=2))), ) assert source.modified_at.tzinfo is UTC assert source.modified_at.hour == 2 acquired = AcquiredDocument( source=source, content=b"a", acquired_at=datetime(2026, 7, 12, 2, tzinfo=UTC) + timedelta(hours=0), ) assert acquired.acquired_at.utcoffset() == timedelta(0) def test_source_errors_are_typed_retryable_and_safe(): transient = EvidenceSourceError( "password=hunter2 at https://user:secret@host", category=EvidenceSourceErrorCategory.TRANSIENT, details={"status": 503}, ) permanent = EvidenceSourceError( "unsupported media type", category=EvidenceSourceErrorCategory.PERMANENT, ) assert transient.retryable is True assert permanent.retryable is False assert transient.details["status"] == 503 assert str(transient) == "evidence source operation failed" assert transient.args == ("evidence source operation failed",) assert "hunter2" not in repr(transient) with pytest.raises(AttributeError): transient.category = EvidenceSourceErrorCategory.PERMANENT with pytest.raises(AttributeError): transient.args = ("leak",) with pytest.raises(AttributeError): transient.details = {"unsafe": True} assert "hunter2" not in repr(transient.__dict__) with pytest.raises(TypeError): transient.details["status"] = 200 with pytest.raises(ValueError, match="credential-like"): EvidenceSourceError( "bad", category=EvidenceSourceErrorCategory.PERMANENT, details={"apiKey": "must-not-leak"}, ) with pytest.raises(ValidationError): EvidenceSourceError( "bad", category=EvidenceSourceErrorCategory.PERMANENT, details={"not_json": object()}, ) def test_source_error_preserves_original_only_through_exception_chaining(): cause = RuntimeError("transport diagnostic with password=hunter2") error = EvidenceSourceError( "ignored unsafe diagnostic", category=EvidenceSourceErrorCategory.TRANSIENT, ) try: raise error from cause except EvidenceSourceError as caught: assert caught.__cause__ is cause assert "hunter2" not in str(caught) assert "hunter2" not in caught.args def test_model_copy_revalidates_source_and_acquired_records(): source = SourceObject(source_id="source:a", uri="file:///a", fingerprint="sha256:a") acquired = AcquiredDocument(source=source, content=b"a") with pytest.raises(ValidationError, match="namespaced"): source.model_copy(update={"source_id": "invalid"}) with pytest.raises(ValidationError, match="timezone-aware"): acquired.model_copy(update={"acquired_at": datetime(2026, 7, 12)})