244 lines
7.8 KiB
Python
244 lines
7.8 KiB
Python
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), # noqa: DTZ001 - verifies rejection
|
|
)
|
|
|
|
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) # noqa: DTZ001 - verifies rejection
|
|
}
|
|
)
|