"""Section 33 — Storage Backends: 7-layer test suite.

Covers:
  musehub/storage/backends.py — StorageBackend protocol, LocalBackend, S3Backend, get_backend, decode_b64

Key behaviour under test:
  LocalBackend:
    - put/get/exists/delete round-trip
    - _path sanitises repo_id and object_id (strips colon/slash from object_id)
    - path traversal in repo_id raises ValueError
    - put is idempotent (second write to same path is a no-op)
    - uri_for returns "local://" prefix
    - missing object → get returns None, exists returns False, delete is a no-op

  S3Backend:
    - _key replaces colons with underscores
    - uri_for returns "s3://<bucket>/..." prefix
    - put/get/exists/delete delegated to boto3 client (tested via mock)
    - credential/boto3 errors on get/exists → return None/False (no raise)
    - _s3_put always calls put_object (idempotent — HeadObject check removed; deduplication is upstream)

  get_backend:
    - returns LocalBackend when aws_s3_asset_bucket is unset
    - returns S3Backend when aws_s3_asset_bucket is set

  decode_b64:
    - decodes standard base64
    - handles missing padding (1, 2, 3 missing "=" chars)
"""
from __future__ import annotations

import asyncio
import time
import uuid
from pathlib import Path
from types import SimpleNamespace
from typing import TypedDict
from unittest.mock import MagicMock, patch

import pytest

from musehub.storage.backends import LocalBackend, S3Backend, decode_b64, get_backend


# ── helpers ───────────────────────────────────────────────────────────────────


class _S3GetResponse(TypedDict):
    Body: MagicMock


def _uid() -> str:
    return str(uuid.uuid4())


def _backend(tmp_path: Path) -> LocalBackend:
    return LocalBackend(objects_dir=str(tmp_path / "objects"))


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 1 — Unit (pure logic, no filesystem I/O)
# ═══════════════════════════════════════════════════════════════════════════════


class TestUnitLocalBackend:
    def test_path_basic(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        p = b._path("abc123")
        assert p.name == "abc123"
        assert str(p).startswith(str(tmp_path))

    def test_path_sanitises_colon_in_object_id(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        p = b._path("sha256:deadbeef")
        assert ":" not in p.name
        assert p.name == "sha256_deadbeef"

    def test_path_sanitises_slash_in_object_id(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        p = b._path("some/nested/id")
        assert "/" not in p.name

    def test_path_sanitises_dotdot_in_object_id(self, tmp_path: Path) -> None:
        """Path separators are sanitised so object_id cannot escape storage root."""
        b = _backend(tmp_path)
        p = b._path("../../etc/passwd")
        assert str(p).startswith(str(tmp_path))

    def test_path_dotdot_deep_stays_in_root(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        p = b._path("../../../etc/shadow")
        assert str(p).startswith(str(tmp_path))

    def test_uri_for_has_local_prefix(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        uri = b.uri_for("obj1")
        assert uri.startswith("local://")

    def test_uri_for_contains_obj(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        uri = b.uri_for("myobj")
        assert "myobj" in uri


class TestUnitS3Backend:
    def test_key_basic(self) -> None:
        b = S3Backend(bucket="test-bucket", region="us-east-1")
        key = b._key("abc123")
        assert "abc123" in key

    def test_key_replaces_colon(self) -> None:
        b = S3Backend(bucket="b", region="us-east-1")
        key = b._key("sha256:deadbeef")
        assert ":" not in key
        assert "sha256_deadbeef" in key

    def test_uri_for_s3_prefix(self) -> None:
        b = S3Backend(bucket="my-bucket", region="us-east-1")
        uri = b.uri_for("obj1")
        assert uri.startswith("s3://my-bucket/")

    def test_uri_for_no_colon_in_key(self) -> None:
        b = S3Backend(bucket="b", region="us-east-1")
        uri = b.uri_for("sha256:abc")
        assert "sha256:abc" not in uri
        assert "sha256_abc" in uri


class TestUnitDecodeb64:
    def test_standard_base64(self) -> None:
        import base64
        data = b"hello world"
        encoded = base64.b64encode(data).decode()
        assert decode_b64(encoded) == data

    def test_missing_one_padding(self) -> None:
        import base64
        data = b"hi"
        encoded = base64.b64encode(data).decode().rstrip("=")
        assert decode_b64(encoded) == data

    def test_missing_two_padding(self) -> None:
        import base64
        data = b"h"
        encoded = base64.b64encode(data).decode().rstrip("=")
        assert decode_b64(encoded) == data

    def test_already_padded(self) -> None:
        import base64
        data = b"test"
        encoded = base64.b64encode(data).decode()
        assert decode_b64(encoded) == data

    def test_empty_string(self) -> None:
        assert decode_b64("") == b""


class TestUnitGetBackend:
    def test_get_backend_returns_local_when_no_s3_bucket(self) -> None:
        with patch("musehub.storage.backends.settings") as mock_settings:
            mock_settings.r2_bucket = None
            mock_settings.aws_s3_asset_bucket = None
            mock_settings.musehub_objects_dir = "/tmp/test-objects"
            result = get_backend()
        assert isinstance(result, LocalBackend)

    def test_get_backend_returns_s3_when_bucket_set(self) -> None:
        with patch("musehub.storage.backends.settings") as mock_settings:
            mock_settings.aws_s3_asset_bucket = "my-bucket"
            mock_settings.aws_region = "us-east-1"
            result = get_backend()
        assert isinstance(result, S3Backend)


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 2 — Integration (real filesystem I/O via tmp_path)
# ═══════════════════════════════════════════════════════════════════════════════


class TestIntegrationLocalBackend:
    async def test_put_and_get_round_trip(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"hello storage"
        uri = await b.put("obj1", data)
        assert uri.startswith("local://")
        result = await b.get("obj1")
        assert result == data

    async def test_get_missing_returns_none(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        result = await b.get("nonexistent")
        assert result is None

    async def test_exists_after_put(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        assert await b.exists("obj1") is False
        await b.put("obj1", b"data")
        assert await b.exists("obj1") is True

    async def test_exists_missing_returns_false(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        assert await b.exists("ghost") is False

    async def test_delete_removes_object(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("obj1", b"bye")
        assert await b.exists("obj1") is True
        await b.delete("obj1")
        assert await b.exists("obj1") is False

    async def test_delete_nonexistent_is_noop(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        # Must not raise
        await b.delete("ghost")

    async def test_put_idempotent_same_bytes(self, tmp_path: Path) -> None:
        """Putting the same bytes twice is a no-op — content-addressed dedup."""
        b = _backend(tmp_path)
        await b.put("obj1", b"content")
        await b.put("obj1", b"content")  # identical — _write skips
        result = await b.get("obj1")
        assert result == b"content"

    async def test_put_creates_parent_dirs(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("obj", b"data")
        assert await b.exists("obj") is True

    async def test_put_colon_in_object_id(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"sha content"
        await b.put("sha256:abc", data)
        result = await b.get("sha256:abc")
        assert result == data

    async def test_global_storage_same_id_shares_object(self, tmp_path: Path) -> None:
        """With global content-addressed storage, same object_id is shared across repos."""
        b = _backend(tmp_path)
        await b.put("obj", b"shared data")
        # Same object_id accessible from any repo_id — content-addressed globally
        assert await b.get("obj") == b"shared data"
        assert await b.get("obj") == b"shared data"


class TestIntegrationS3BackendMocked:
    """S3Backend with a mock boto3 client — no real AWS calls."""

    def _mock_s3_backend(self, *, head_raises: bool = False) -> S3Backend:
        b = S3Backend(bucket="test-bucket", region="us-east-1")
        mock_client = MagicMock()
        if head_raises:
            mock_client.head_object.side_effect = Exception("NoSuchKey")
        b._client = mock_client
        return b

    async def test_put_calls_put_object_directly(self) -> None:
        """put() is idempotent — no HeadObject check, just put_object unconditionally."""
        b = self._mock_s3_backend()
        await b.put("obj1", b"data")
        b._client.put_object.assert_called_once()
        b._client.head_object.assert_not_called()

    async def test_put_is_idempotent(self) -> None:
        """Uploading the same object twice both succeed — put_object called twice."""
        b = self._mock_s3_backend()
        await b.put("obj1", b"data")
        await b.put("obj1", b"data")
        assert b._client.put_object.call_count == 2

    async def test_get_returns_body_bytes(self) -> None:
        b = self._mock_s3_backend()
        mock_body = MagicMock()
        mock_body.read.return_value = b"s3 content"
        b._client.get_object.return_value = {"Body": mock_body}
        result = await b.get("obj1")
        assert result == b"s3 content"

    async def test_get_returns_none_on_error(self) -> None:
        b = self._mock_s3_backend()
        b._client.get_object.side_effect = Exception("NoSuchKey")
        result = await b.get("obj1")
        assert result is None

    async def test_exists_returns_true_when_head_succeeds(self) -> None:
        b = self._mock_s3_backend()
        b._client.head_object.return_value = {}
        assert await b.exists("obj1") is True

    async def test_exists_returns_false_on_error(self) -> None:
        b = self._mock_s3_backend(head_raises=True)
        assert await b.exists("obj1") is False

    async def test_delete_calls_delete_object(self) -> None:
        b = self._mock_s3_backend()
        await b.delete("obj1")
        b._client.delete_object.assert_called_once()

    async def test_get_batch_returns_all_found_objects(self) -> None:
        """get_batch maps every object_id to its bytes when S3 finds all of them."""
        b = self._mock_s3_backend()

        def _get_object(Bucket: str, Key: str) -> _S3GetResponse:
            # Return the key bytes as the body so we can assert on content
            mock_body = MagicMock()
            mock_body.read.return_value = Key.encode()
            return {"Body": mock_body}

        b._client.get_object.side_effect = _get_object
        result = await b.get_batch(["oid-a", "oid-b", "oid-c"])
        assert set(result.keys()) == {"oid-a", "oid-b", "oid-c"}

    async def test_get_batch_omits_missing_objects(self) -> None:
        """get_batch omits object_ids where S3 raises (NoSuchKey etc.)."""
        b = self._mock_s3_backend()
        b._client.get_object.side_effect = Exception("NoSuchKey")
        result = await b.get_batch(["oid-1", "oid-2"])
        assert result == {}

    async def test_get_batch_is_parallel_not_sequential(self) -> None:
        """N parallel S3 gets must complete in ~1× per-object latency, not N×.

        The base-class fallback is sequential; S3Backend must override get_batch
        with asyncio.gather so that cloning large repos cannot time out at
        Cloudflare's 100-second origin-response limit.
        """
        import asyncio

        DELAY = 0.04  # 40 ms per object
        N = 6

        b = S3Backend(bucket="test-bucket", region="us-east-1")

        # Patch get() with an async stub that sleeps to simulate network latency.
        async def _slow_get(oid: str) -> bytes | None:
            await asyncio.sleep(DELAY)
            return oid.encode()

        setattr(b, "get", _slow_get)

        obj_ids = [f"oid-{i}" for i in range(N)]
        start = time.perf_counter()
        result = await b.get_batch(obj_ids)
        elapsed = time.perf_counter() - start

        assert len(result) == N
        # Parallel: completes in roughly 1× DELAY.
        # Sequential (base-class fallback): would take N× DELAY ≈ 240 ms.
        assert elapsed < DELAY * 2.5, (
            f"get_batch took {elapsed * 1000:.0f} ms; expected ≈{DELAY * 1000:.0f} ms "
            f"(parallel). Sequential would take ≥{DELAY * N * 1000:.0f} ms. "
            "S3Backend.get_batch must use asyncio.gather, not the base-class loop."
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 3 — End-to-End (LocalBackend as the full stack backend)
# ═══════════════════════════════════════════════════════════════════════════════


class TestE2ELocalBackend:
    async def test_full_lifecycle(self, tmp_path: Path) -> None:
        """put → exists(True) → get → delete → exists(False) → get(None)."""
        b = _backend(tmp_path)
        repo_id = _uid()
        obj_id = _uid()
        data = b"end-to-end content"

        uri = await b.put(obj_id, data)
        assert uri.startswith("local://")

        assert await b.exists(obj_id) is True
        assert await b.get(obj_id) == data

        await b.delete(obj_id)
        assert await b.exists(obj_id) is False
        assert await b.get(obj_id) is None

    async def test_binary_data_preserved(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = bytes(range(256))  # all byte values
        await b.put("bin-obj", data)
        result = await b.get("bin-obj")
        assert result == data

    async def test_large_object(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"x" * (4 * 1024 * 1024)  # 4 MiB
        await b.put("large-obj", data)
        result = await b.get("large-obj")
        assert result == data

    async def test_uri_resolves_to_disk_path(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"uri content"
        uri = await b.put("uri-obj", data)
        disk_path = Path(uri.replace("local://", ""))
        assert disk_path.exists()
        assert disk_path.read_bytes() == data


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 4 — Stress
# ═══════════════════════════════════════════════════════════════════════════════


class TestStressLocalBackend:
    async def test_write_many_objects(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        n = 100
        obj_ids = [f"obj-{i}" for i in range(n)]
        for obj_id in obj_ids:
            await b.put(obj_id, obj_id.encode())

        for obj_id in obj_ids:
            assert await b.exists(obj_id) is True

    async def test_concurrent_writes_different_objects(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        obj_ids = [f"concurrent-{i}" for i in range(20)]
        await asyncio.gather(*[b.put(oid, oid.encode()) for oid in obj_ids])
        for oid in obj_ids:
            assert await b.get(oid) == oid.encode()

    async def test_concurrent_reads(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("shared", b"shared content")
        results = await asyncio.gather(*[b.get("shared") for _ in range(20)])
        assert all(r == b"shared content" for r in results)

    async def test_write_many_objects_different_ids(self, tmp_path: Path) -> None:
        """Write many distinct object_ids (global storage, repo_id is ignored for path)."""
        b = _backend(tmp_path)
        obj_ids = [f"obj-stress-{i}" for i in range(30)]
        for obj_id in obj_ids:
            await b.put(obj_id, obj_id.encode())
        for obj_id in obj_ids:
            assert await b.get(obj_id) == obj_id.encode()


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 5 — Data Integrity
# ═══════════════════════════════════════════════════════════════════════════════


class TestDataIntegrityLocalBackend:
    async def test_put_returns_deterministic_uri(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        uri1 = await b.put("obj1", b"data")
        uri2 = b.uri_for("obj1")
        assert uri1 == uri2

    async def test_data_not_corrupted(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        payload = b"\x00\x01\x02\x03" * 1000
        await b.put("integrity", payload)
        result = await b.get("integrity")
        assert result == payload

    async def test_put_is_idempotent_for_identical_bytes(self, tmp_path: Path) -> None:
        """put() with the same bytes twice is a no-op — content stays unchanged."""
        b = _backend(tmp_path)
        await b.put("obj", b"original")
        await b.put("obj", b"original")
        result = await b.get("obj")
        assert result == b"original"

    async def test_put_repairs_corrupted_file(self, tmp_path: Path) -> None:
        """put() must overwrite an existing file whose bytes differ from the new data.

        The legacy wire_push_objects endpoint stored zlib-compressed wire bytes
        under the SHA-256 of the raw (decompressed) content — a hash/content
        mismatch. When the pack endpoint later pushes the correct raw bytes for
        the same object_id, put() must overwrite the corrupted file so that
        subsequent get() calls return the real content, not the old zlib garbage.
        """
        import zlib
        b = _backend(tmp_path)
        raw = b"# real raw content"
        corrupt = zlib.compress(raw)  # what the legacy endpoint mistakenly stored
        object_id = "obj-sha256-of-raw"

        # Simulate legacy corruption: store compressed bytes directly on disk.
        path = b._path(object_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(corrupt)
        import stat
        path.chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)  # immutable, as _write does

        # Pack endpoint now puts the correct raw bytes.
        await b.put(object_id, raw)

        # get() must return the raw bytes, not the old compressed garbage.
        result = await b.get(object_id)
        assert result == raw, (
            f"expected raw content but got {result[:20]!r} — "
            "LocalBackend.put() failed to repair the corrupted file"
        )

    async def test_delete_only_removes_target(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("keep", b"keep")
        await b.put("remove", b"remove")
        await b.delete("remove")
        assert await b.get("keep") == b"keep"
        assert await b.get("remove") is None

    def test_path_traversal_variations_sanitised(self, tmp_path: Path) -> None:
        """object_id path separators are sanitised — all paths stay inside root."""
        b = _backend(tmp_path)
        traversal_attempts = [
            "../../../etc",
            "repo/../../../etc",
            "repo/../../secret",
        ]
        for obj_id in traversal_attempts:
            p = b._path(obj_id)
            assert str(p).startswith(str(tmp_path))

    async def test_empty_bytes_stored_and_retrieved(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("empty-obj", b"")
        result = await b.get("empty-obj")
        assert result == b""


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 6 — Security
# ═══════════════════════════════════════════════════════════════════════════════


class TestSecurityLocalBackend:
    def test_dotdot_in_object_id_sanitised(self, tmp_path: Path) -> None:
        """Path separators in object_id are sanitised — no traversal possible."""
        b = _backend(tmp_path)
        p = b._path("../secrets")
        assert str(p).startswith(str(tmp_path))

    def test_dotdot_deep_in_object_id_sanitised(self, tmp_path: Path) -> None:
        """Multiple levels of '..' in object_id are sanitised."""
        b = _backend(tmp_path)
        p = b._path("../../../../../../etc/passwd")
        assert str(p).startswith(str(tmp_path))

    def test_object_id_colon_sanitised_prevents_ambiguity(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        p1 = b._path("sha256:abc")
        p2 = b._path("sha256_abc")
        # Both sanitise to the same safe filename
        assert p1 == p2

    async def test_put_with_dotdot_object_id_sanitised(self, tmp_path: Path) -> None:
        """put() sanitises object_id path separators — stays inside root."""
        b = _backend(tmp_path)
        uri = await b.put("../outside", b"data")
        assert uri.startswith("local://")
        disk = uri.replace("local://", "")
        assert str(tmp_path) in disk

    async def test_get_with_dotdot_object_id_sanitised(self, tmp_path: Path) -> None:
        """get() sanitises object_id path separators."""
        b = _backend(tmp_path)
        result = await b.get("../../etc/passwd")
        assert result is None  # file doesn't exist, but no error raised

    async def test_exists_with_dotdot_object_id_sanitised(self, tmp_path: Path) -> None:
        """exists() sanitises object_id path separators."""
        b = _backend(tmp_path)
        result = await b.exists("../outside")
        assert result is False  # sanitised path doesn't exist

    async def test_delete_with_dotdot_object_id_is_noop(self, tmp_path: Path) -> None:
        """delete() sanitises object_id — no error when file doesn't exist."""
        b = _backend(tmp_path)
        await b.delete("../outside")  # must not raise

    def test_s3_key_no_colon_injection(self) -> None:
        b = S3Backend(bucket="b", region="us-east-1")
        key = b._key("sha256:evil:colons")
        assert ":" not in key


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 7 — Performance
# ═══════════════════════════════════════════════════════════════════════════════


class TestPerformanceLocalBackend:
    async def test_put_latency(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"perf-test-payload" * 100
        start = time.perf_counter()
        await b.put("perf-obj", data)
        elapsed = time.perf_counter() - start
        assert elapsed < 0.5

    async def test_get_latency(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"x" * (512 * 1024)  # 512 KiB
        await b.put("big-obj", data)

        start = time.perf_counter()
        result = await b.get("big-obj")
        elapsed = time.perf_counter() - start

        assert result == data
        assert elapsed < 0.5

    async def test_exists_latency(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        await b.put("perf-exists", b"data")

        start = time.perf_counter()
        for _ in range(50):
            await b.exists("perf-exists")
        elapsed = time.perf_counter() - start

        assert elapsed < 1.0

    async def test_50_sequential_puts_under_budget(self, tmp_path: Path) -> None:
        b = _backend(tmp_path)
        data = b"payload" * 100

        start = time.perf_counter()
        for i in range(50):
            await b.put(f"seq-{i}", data)
        elapsed = time.perf_counter() - start

        assert elapsed < 2.0