govoplan-core/tests/test_object_storage.py

from __future__ import annotations

from io import BytesIO
from pathlib import Path
import tempfile
import unittest

from govoplan_core.core.object_storage import (
    LocalFilesystemStorageBackend,
    S3StorageBackend,
    StorageBackendError,
    StorageObjectMissing,
    normalize_storage_key,
)


class _S3Error(RuntimeError):
    def __init__(self, code: str, status: int) -> None:
        super().__init__(code)
        self.response = {
            "Error": {"Code": code},
            "ResponseMetadata": {"HTTPStatusCode": status},
        }


class _FakeS3Client:
    def __init__(self) -> None:
        self.objects: dict[str, bytes] = {}
        self.head_error: Exception | None = None

    def put_object(self, *, Key: str, Body: bytes, **_kwargs) -> None:
        self.objects[Key] = Body

    def get_object(self, *, Key: str, **_kwargs):
        try:
            payload = self.objects[Key]
        except KeyError as exc:
            raise _S3Error("NoSuchKey", 404) from exc
        return {"Body": BytesIO(payload), "ContentLength": len(payload)}

    def head_object(self, *, Key: str, **_kwargs):
        if self.head_error is not None:
            raise self.head_error
        try:
            payload = self.objects[Key]
        except KeyError as exc:
            raise _S3Error("NotFound", 404) from exc
        return {"ContentLength": len(payload)}

    def delete_object(self, *, Key: str, **_kwargs) -> None:
        self.objects.pop(Key, None)


class ObjectStorageTests(unittest.TestCase):
    def test_local_backend_round_trip_pagination_and_safe_keys(self) -> None:
        with tempfile.TemporaryDirectory(prefix="govoplan-object-store-") as directory:
            backend = LocalFilesystemStorageBackend(Path(directory))
            backend.put_bytes("campaign/a.eml", b"a")
            backend.put_bytes("campaign/b.eml", b"bb")

            first = backend.list_objects(prefix="campaign/", limit=1)
            second = backend.list_objects(
                prefix="campaign/",
                after=first.next_cursor,
                limit=1,
            )

            self.assertEqual(b"a", backend.get_bytes("campaign/a.eml"))
            self.assertEqual(1, backend.stat("campaign/a.eml").size_bytes)
            self.assertEqual(
                ("campaign/a.eml",), tuple(item.key for item in first.objects)
            )
            self.assertEqual("campaign/a.eml", first.next_cursor)
            self.assertEqual(
                ("campaign/b.eml",), tuple(item.key for item in second.objects)
            )
            self.assertIsNone(second.next_cursor)
            with self.assertRaises(StorageBackendError):
                normalize_storage_key("../outside")
            with self.assertRaises(StorageObjectMissing):
                backend.get_bytes("campaign/missing.eml")

    def test_local_listing_is_globally_sorted_and_ignores_symlinks(self) -> None:
        with (
            tempfile.TemporaryDirectory(prefix="govoplan-object-store-") as directory,
            tempfile.TemporaryDirectory(prefix="govoplan-object-outside-") as outside,
        ):
            root = Path(directory)
            backend = LocalFilesystemStorageBackend(root)
            backend.put_bytes("a/x.txt", b"nested")
            backend.put_bytes("a.txt", b"sibling")
            backend.put_bytes("b.txt", b"last")
            outside_file = Path(outside) / "secret.txt"
            outside_file.write_bytes(b"outside")
            (root / "linked-file.txt").symlink_to(outside_file)
            (root / "linked-directory").symlink_to(
                Path(outside), target_is_directory=True
            )

            first = backend.list_objects(prefix="", limit=2)
            second = backend.list_objects(
                prefix="",
                after=first.next_cursor,
                limit=2,
            )

            self.assertEqual(
                ("a.txt", "a/x.txt"),
                tuple(item.key for item in first.objects),
            )
            self.assertEqual("a/x.txt", first.next_cursor)
            self.assertEqual(
                ("b.txt",),
                tuple(item.key for item in second.objects),
            )
            self.assertIsNone(second.next_cursor)
            with self.assertRaises(StorageBackendError):
                backend.exists("../outside")

    def test_s3_backend_distinguishes_missing_objects_from_backend_failure(
        self,
    ) -> None:
        backend = S3StorageBackend(
            bucket="files",
            endpoint_url="http://garage:3900",
            region_name="garage",
            access_key_id="key",
            secret_access_key="secret",
            deployment_managed=True,
        )
        client = _FakeS3Client()
        backend._client = client

        backend.put_bytes("campaign/message.eml", b"message/rfc822")
        self.assertTrue(backend.exists("campaign/message.eml"))
        self.assertEqual(b"message/rfc822", backend.get_bytes("campaign/message.eml"))
        self.assertFalse(backend.exists("campaign/missing.eml"))

        client.head_error = _S3Error("AccessDenied", 403)
        with self.assertRaises(StorageBackendError):
            backend.exists("campaign/message.eml")

    def test_external_s3_requires_explicit_https_deployment_trust(self) -> None:
        trusted = S3StorageBackend(
            bucket="files",
            endpoint_url="https://s3.internal.example.test",
            region_name="eu-test-1",
            access_key_id="key",
            secret_access_key="secret",
            endpoint_trusted=True,
        )
        trusted._client = _FakeS3Client()
        trusted.put_bytes("files/document.txt", b"document")
        self.assertEqual(b"document", trusted.get_bytes("files/document.txt"))

        rejected = S3StorageBackend(
            bucket="files",
            endpoint_url="http://s3.internal.example.test",
            region_name="eu-test-1",
            access_key_id="key",
            secret_access_key="secret",
            endpoint_trusted=True,
        )
        with self.assertRaisesRegex(StorageBackendError, "HTTPS origin"):
            rejected.client


if __name__ == "__main__":
    unittest.main()