ThothII/harness/tests/test_s3_evidence_source.py

from datetime import UTC, datetime

import pytest

from tht.ports.evidence import EvidenceSourceError


class Body:
    def __init__(self, data): self.data, self.closed = data, False
    def read(self, amount): return self.data[:amount]
    def close(self): self.closed = True


class Client:
    def __init__(self): self.body, self.list_calls = Body(b"hello"), 0
    def get_paginator(self, name): return self
    def list_objects_v2(self, **kwargs):
        self.list_calls += 1
        return {"Contents": [{"Key": "clinical/a.md", "ETag": '"abc"',
                              "Size": 5, "LastModified": datetime(2026, 1, 1, tzinfo=UTC)}]}
    def paginate(self, **kwargs):
        yield {"Contents": [{"Key": "clinical/a.md", "ETag": '"abc"',
                              "Size": 5, "LastModified": datetime(2026, 1, 1, tzinfo=UTC)}]}
    def get_object(self, **kwargs):
        assert kwargs == {"Bucket": "evidence", "Key": "clinical/a.md"}
        return {"Body": self.body, "ContentLength": 5, "ContentType": "text/markdown",
                "ETag": '"abc"'}


def test_s3_canonical_uri_version_fingerprint_and_closed_body():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    source = S3EvidenceSource(bucket="evidence", prefix="clinical/", client=client)
    item = next(iter(source.discover()))
    assert item.uri == "s3://evidence/clinical/a.md"
    assert item.fingerprint.startswith("etag:")
    assert source.acquire(item).content == b"hello"
    assert client.body.closed


def test_s3_etag_fallback_and_bounds():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    with pytest.raises(ValueError):
        S3EvidenceSource(bucket="evidence", client=client, max_objects=0)


def test_s3_rejects_private_or_insecure_endpoint_without_explicit_opt_in():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    with pytest.raises(ValueError, match="trusted"):
        S3EvidenceSource(bucket="evidence", endpoint_url="https://127.0.0.1:9000", client=Client())
    with pytest.raises(ValueError, match="HTTPS"):
        S3EvidenceSource(bucket="evidence", endpoint_url="http://s3.example.test", client=Client())
    source = S3EvidenceSource(bucket="evidence", endpoint_url="http://127.0.0.1:9000",
                              trusted_endpoint=True, allow_private_endpoint=True,
                              allow_insecure_endpoint=True, client=Client())
    assert source is not None


@pytest.mark.parametrize("bucket", ["UPPER", "bad_bucket", "-start", "end-", "a..b"])
def test_s3_rejects_invalid_bucket_names(bucket):
    from tht.adapters.evidence.s3 import S3EvidenceSource
    with pytest.raises(ValueError, match="bucket"):
        S3EvidenceSource(bucket=bucket, client=Client())


@pytest.mark.parametrize("bucket", ["127.0.0.1", "192.168.1.1"])
def test_s3_rejects_ip_shaped_bucket(bucket):
    from tht.adapters.evidence.s3 import S3EvidenceSource
    with pytest.raises(ValueError, match="bucket"):
        S3EvidenceSource(bucket=bucket, client=Client())


def test_s3_rejects_endpoint_query_path_fragment_and_untrusted_custom_host():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    for endpoint in ("https://s3.example.test/path", "https://s3.example.test/?x=1",
                     "https://s3.example.test/#x"):
        with pytest.raises(ValueError, match="root"):
            S3EvidenceSource(bucket="evidence", endpoint_url=endpoint,
                             trusted_endpoint=True, client=Client())
    with pytest.raises(ValueError, match="trusted"):
        S3EvidenceSource(bucket="evidence", endpoint_url="https://s3.example.test", client=Client())


def test_s3_rejects_out_of_prefix_key_and_missing_validator():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    client.list_objects_v2 = lambda **kwargs: {"Contents": [{"Key": "other/a.md", "ETag": '"x"'}]}
    with pytest.raises(EvidenceSourceError):
        list(S3EvidenceSource(bucket="evidence", prefix="clinical/", client=client).discover())
    client.list_objects_v2 = lambda **kwargs: {"Contents": [{"Key": "clinical/a.md"}]}
    with pytest.raises(EvidenceSourceError):
        list(S3EvidenceSource(bucket="evidence", prefix="clinical/", client=client).discover())


def test_s3_rejects_leading_slash_prefix_empty_and_control_keys():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    with pytest.raises(ValueError, match="prefix"):
        S3EvidenceSource(bucket="evidence", prefix="/clinical", client=Client())
    for key in ("", "clinical/a\x00.md", "clinical/a\x7f.md"):
        client = Client()
        client.list_objects_v2 = lambda **kwargs: {"Contents": [{"Key": key, "ETag": '"x"'}]}
        with pytest.raises(EvidenceSourceError):
            list(S3EvidenceSource(bucket="evidence", prefix="clinical/", client=client).discover())


@pytest.mark.parametrize("prefix", ["/bad", "x" * 1025, "bad\x00prefix", "bad\x7fprefix"])
def test_s3_rejects_invalid_prefix_before_client_request(prefix):
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    with pytest.raises(ValueError, match="prefix"):
        S3EvidenceSource(bucket="evidence", prefix=prefix, client=client)
    assert client.list_calls == 0


def test_s3_hard_page_limit_never_requests_page_max_plus_one():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    def listing(**kwargs):
        client.list_calls += 1
        return {"Contents": [{"Key": f"clinical/{client.list_calls}.md", "ETag": '"x"'}],
                "IsTruncated": True, "NextContinuationToken": str(client.list_calls)}
    client.list_objects_v2 = listing
    with pytest.raises(EvidenceSourceError):
        list(S3EvidenceSource(bucket="evidence", prefix="clinical/", max_pages=2,
                              client=client).discover())
    assert client.list_calls == 2


def test_s3_acquire_rejects_exact_etag_drift_and_closes_body():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    source = S3EvidenceSource(bucket="evidence", client=client)
    item = next(iter(source.discover()))
    client.get_object = lambda **kwargs: {"Body": client.body, "ContentLength": 5,
                                          "ETag": '"changed"'}
    with pytest.raises(EvidenceSourceError):
        source.acquire(item)
    assert client.body.closed


def test_s3_acquire_rejects_forged_reconstructed_item_before_get():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    source = S3EvidenceSource(bucket="evidence", client=client)
    item = next(iter(source.discover()))
    forged = item.model_copy(update={"fingerprint": "etag:" + "0" * 64})
    client.get_object = lambda **kwargs: (_ for _ in ()).throw(AssertionError("called"))
    with pytest.raises(EvidenceSourceError):
        source.acquire(forged)


@pytest.mark.parametrize("host", ["127.0.0.1", "10.0.0.1", "169.254.1.1", "0.0.0.0",
                                  "[::1]", "[fe80::1]", "[::]"])
def test_s3_literal_non_global_endpoint_requires_private_opt_in(host):
    from tht.adapters.evidence.s3 import S3EvidenceSource
    with pytest.raises(ValueError, match="private"):
        S3EvidenceSource(bucket="evidence", endpoint_url=f"https://{host}:9000",
                         trusted_endpoint=True, client=Client())


def test_s3_size_limit_closes_body():
    from tht.adapters.evidence.s3 import S3EvidenceSource
    client = Client()
    source = S3EvidenceSource(bucket="evidence", client=client, max_bytes=4)
    item = next(iter(source.discover()))
    with pytest.raises(EvidenceSourceError):
        source.acquire(item)
    assert client.body.closed


def test_s3_config_serialization_masks_credentials():
    from tht.config import S3EvidenceSourceConfig
    config = S3EvidenceSourceConfig(type="s3", bucket="evidence",
                                    access_key="access-secret", secret_key="write-secret")
    assert "access-secret" not in repr(config)
    assert "write-secret" not in repr(config)


def test_s3_config_loads_credentials_from_secret_files(tmp_path):
    from tht.config import load_config
    access, secret = tmp_path / "access", tmp_path / "secret"
    access.write_text("access-value")
    secret.write_text("secret-value")
    workspace = tmp_path / "workspace.yaml"
    workspace.write_text(f"""
dwh:
  type: postgres_direct
  connection: {{database: d, schema: public, user: u, password: p}}
evidence:
  sources:
    - type: s3
      bucket: evidence
      access_key_file: {access}
      secret_key_file: {secret}
""")
    source = load_config(workspace).evidence.sources[0]
    assert source.access_key.get_secret_value() == "access-value"
    assert source.secret_key.get_secret_value() == "secret-value"