"""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 (per-repo layout)
    - _path requires repo_root — raises ValueError without it
    - _path(oid, repo_root) produces objects/sha256/<2hex>/<62hex> layout
    - 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 preserves colons (colons are valid in S3/R2 keys)
    - 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 secrets
import time
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
from muse.core.types import long_id


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


class _S3GetResponse(TypedDict):
    Body: MagicMock


def _uid() -> str:
    return secrets.token_hex(16)


def _backend() -> LocalBackend:
    return LocalBackend()


def _repo_root(tmp_path: Path) -> Path:
    return tmp_path / "repos" / "gabriel" / "test-repo"


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


class TestUnitLocalBackend:
    def test_path_requires_repo_root(self, tmp_path: Path) -> None:
        b = _backend()
        with pytest.raises(ValueError, match="repo_root"):
            b._path(long_id("a" * 64), repo_root=None)

    def test_path_with_repo_root_uses_algo_shard_layout(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        hex64 = "ab" + "c" * 62
        oid = long_id(hex64)
        p = b._path(oid, repo_root=repo_root)
        expected = repo_root / "objects" / "sha256" / "ab" / ("c" * 62)
        assert p == expected

    def test_uri_for_has_local_prefix(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("a" * 64)
        uri = b.uri_for(oid, repo_root=repo_root)
        assert uri.startswith("local://")

    def test_uri_for_contains_obj(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("a" * 64)
        uri = b.uri_for(oid, repo_root=repo_root)
        assert "sha256" 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_preserves_colon(self) -> None:
        b = S3Backend(bucket="b", region="us-east-1")
        key = b._key("sha256:deadbeef")
        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_preserves_colon_in_key(self) -> None:
        b = S3Backend(bucket="b", region="us-east-1")
        uri = b.uri_for("sha256:abc")
        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
            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"
            mock_settings.r2_bucket = None
            result = get_backend()
        assert isinstance(result, S3Backend)

    def test_get_backend_with_owner_slug_returns_bound_local(self, tmp_path: Path) -> None:
        """get_backend(owner, slug) returns a LocalBackend pre-bound to that repo_root."""
        with patch("musehub.storage.backends.settings") as mock_settings:
            mock_settings.r2_bucket = None
            mock_settings.aws_s3_asset_bucket = None
            mock_settings.musehub_repos_dir = str(tmp_path)
            result = get_backend("gabriel", "myrepo")
        assert isinstance(result, LocalBackend)
        assert result._repo_root is not None

    async def test_get_backend_bound_local_get_needs_no_repo_root_kwarg(self, tmp_path: Path) -> None:
        """A backend from get_backend(owner, slug) satisfies storage.get(oid) with no extra args."""
        with patch("musehub.storage.backends.settings") as mock_settings:
            mock_settings.r2_bucket = None
            mock_settings.aws_s3_asset_bucket = None
            mock_settings.musehub_repos_dir = str(tmp_path)
            backend = get_backend("gabriel", "myrepo")

        oid = long_id("a" * 64)
        await backend.put(oid, b"hello")
        result = await backend.get(oid)
        assert result == b"hello"

    async def test_get_backend_bound_local_exists_needs_no_repo_root_kwarg(self, tmp_path: Path) -> 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_repos_dir = str(tmp_path)
            backend = get_backend("gabriel", "myrepo")

        oid = long_id("b" * 64)
        assert await backend.exists(oid) is False
        await backend.put(oid, b"data")
        assert await backend.exists(oid) is True

    def test_get_backend_no_args_local_returns_unbound(self) -> None:
        """get_backend() with no owner/slug still returns LocalBackend — unbound."""
        with patch("musehub.storage.backends.settings") as mock_settings:
            mock_settings.r2_bucket = None
            mock_settings.aws_s3_asset_bucket = None
            result = get_backend()
        assert isinstance(result, LocalBackend)
        assert result._repo_root is None


class TestUnitLocalBackendBoundInit:
    """LocalBackend(repo_root=path) binds repo_root at construction; methods need no kwarg."""

    def test_bound_backend_stores_repo_root(self, tmp_path: Path) -> None:
        root = tmp_path / "gabriel" / "myrepo"
        b = LocalBackend(repo_root=root)
        assert b._repo_root == root

    def test_unbound_backend_has_none_repo_root(self) -> None:
        b = LocalBackend()
        assert b._repo_root is None

    async def test_bound_put_and_get_no_kwarg(self, tmp_path: Path) -> None:
        root = tmp_path / "gabriel" / "myrepo"
        b = LocalBackend(repo_root=root)
        oid = long_id("c" * 64)
        await b.put(oid, b"bound put")
        result = await b.get(oid)
        assert result == b"bound put"

    async def test_bound_exists_no_kwarg(self, tmp_path: Path) -> None:
        root = tmp_path / "gabriel" / "myrepo"
        b = LocalBackend(repo_root=root)
        oid = long_id("d" * 64)
        assert await b.exists(oid) is False
        await b.put(oid, b"x")
        assert await b.exists(oid) is True

    async def test_bound_delete_no_kwarg(self, tmp_path: Path) -> None:
        root = tmp_path / "gabriel" / "myrepo"
        b = LocalBackend(repo_root=root)
        oid = long_id("e" * 64)
        await b.put(oid, b"y")
        await b.delete(oid)
        assert await b.exists(oid) is False

    async def test_kwarg_overrides_bound_repo_root(self, tmp_path: Path) -> None:
        """Explicit repo_root kwarg at call time takes precedence over bound root."""
        root_a = tmp_path / "a" / "repo"
        root_b = tmp_path / "b" / "repo"
        b = LocalBackend(repo_root=root_a)
        oid = long_id("f" * 64)
        await b.put(oid, b"in-b", repo_root=root_b)
        assert await b.exists(oid, repo_root=root_b) is True
        assert await b.exists(oid, repo_root=root_a) is False
        assert await b.get(oid) is None       # root_a doesn't have it
        assert await b.get(oid, repo_root=root_b) == b"in-b"

    async def test_unbound_get_without_kwarg_raises(self) -> None:
        b = LocalBackend()
        oid = long_id("0" * 64)
        with pytest.raises(ValueError, match="repo_root"):
            await b.get(oid)


# ═══════════════════════════════════════════════════════════════════════════════
# 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()
        repo_root = _repo_root(tmp_path)
        data = b"hello storage"
        oid = long_id("a" * 64)
        uri = await b.put(oid, data, repo_root=repo_root)
        assert uri.startswith("local://")
        result = await b.get(oid, repo_root=repo_root)
        assert result == data

    async def test_get_missing_returns_none(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("b" * 64)
        result = await b.get(oid, repo_root=repo_root)
        assert result is None

    async def test_exists_after_put(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("c" * 64)
        assert await b.exists(oid, repo_root=repo_root) is False
        await b.put(oid, b"data", repo_root=repo_root)
        assert await b.exists(oid, repo_root=repo_root) is True

    async def test_exists_missing_returns_false(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("d" * 64)
        assert await b.exists(oid, repo_root=repo_root) is False

    async def test_delete_removes_object(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("e" * 64)
        await b.put(oid, b"bye", repo_root=repo_root)
        assert await b.exists(oid, repo_root=repo_root) is True
        await b.delete(oid, repo_root=repo_root)
        assert await b.exists(oid, repo_root=repo_root) is False

    async def test_delete_nonexistent_is_noop(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("f" * 64)
        await b.delete(oid, repo_root=repo_root)

    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()
        repo_root = _repo_root(tmp_path)
        oid = long_id("0" * 64)
        await b.put(oid, b"content", repo_root=repo_root)
        await b.put(oid, b"content", repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        assert result == b"content"

    async def test_put_creates_parent_dirs(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("1" * 64)
        await b.put(oid, b"data", repo_root=repo_root)
        assert await b.exists(oid, repo_root=repo_root) is True

    async def test_sha256_oid_round_trip(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("ab" + "c" * 62)
        data = b"sha256 content"
        await b.put(oid, data, repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        assert result == data

    async def test_per_repo_isolation(self, tmp_path: Path) -> None:
        """Objects in different repos are isolated — same oid, different repo_root."""
        b = _backend()
        repo_a = tmp_path / "repos" / "alice" / "repo-a"
        repo_b = tmp_path / "repos" / "bob" / "repo-b"
        oid = long_id("2" * 64)
        await b.put(oid, b"alice data", repo_root=repo_a)
        assert await b.exists(oid, repo_root=repo_b) is False
        assert await b.get(oid, repo_root=repo_a) == b"alice 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:
            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×."""
        import asyncio

        DELAY = 0.04
        N = 6

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

        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
        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()
        repo_root = _repo_root(tmp_path)
        obj_id = long_id("3" * 64)
        data = b"end-to-end content"

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

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

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

    async def test_binary_data_preserved(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("4" * 64)
        data = bytes(range(256))
        await b.put(oid, data, repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        assert result == data

    async def test_large_object(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("5" * 64)
        data = b"x" * (4 * 1024 * 1024)
        await b.put(oid, data, repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        assert result == data

    async def test_uri_resolves_to_disk_path(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("6" * 64)
        data = b"uri content"
        uri = await b.put(oid, data, repo_root=repo_root)
        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()
        repo_root = _repo_root(tmp_path)
        obj_ids = [long_id(str(i).zfill(64)) for i in range(100)]
        for oid in obj_ids:
            await b.put(oid, oid.encode(), repo_root=repo_root)
        for oid in obj_ids:
            assert await b.exists(oid, repo_root=repo_root) is True

    async def test_concurrent_writes_different_objects(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        obj_ids = [long_id(str(i).zfill(64)) for i in range(20)]
        await asyncio.gather(*[b.put(oid, oid.encode(), repo_root=repo_root) for oid in obj_ids])
        for oid in obj_ids:
            assert await b.get(oid, repo_root=repo_root) == oid.encode()

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


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


class TestDataIntegrityLocalBackend:
    async def test_put_returns_deterministic_uri(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("8" * 64)
        uri1 = await b.put(oid, b"data", repo_root=repo_root)
        uri2 = b.uri_for(oid, repo_root=repo_root)
        assert uri1 == uri2

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

    async def test_put_is_idempotent_for_identical_bytes(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("a" * 64)
        await b.put(oid, b"original", repo_root=repo_root)
        await b.put(oid, b"original", repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        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."""
        import zlib
        b = _backend()
        repo_root = _repo_root(tmp_path)
        raw = b"# real raw content"
        corrupt = zlib.compress(raw)
        oid = long_id("b" * 64)

        path = b._path(oid, repo_root=repo_root)
        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)

        await b.put(oid, raw, repo_root=repo_root)

        result = await b.get(oid, repo_root=repo_root)
        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()
        repo_root = _repo_root(tmp_path)
        oid_keep = long_id("c" * 64)
        oid_remove = long_id("d" * 64)
        await b.put(oid_keep, b"keep", repo_root=repo_root)
        await b.put(oid_remove, b"remove", repo_root=repo_root)
        await b.delete(oid_remove, repo_root=repo_root)
        assert await b.get(oid_keep, repo_root=repo_root) == b"keep"
        assert await b.get(oid_remove, repo_root=repo_root) is None

    async def test_empty_bytes_stored_and_retrieved(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("e" * 64)
        await b.put(oid, b"", repo_root=repo_root)
        result = await b.get(oid, repo_root=repo_root)
        assert result == b""


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


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

    async def test_put_without_repo_root_raises(self, tmp_path: Path) -> None:
        b = _backend()
        oid = long_id("f" * 64)
        with pytest.raises(ValueError, match="repo_root"):
            await b.put(oid, b"data", repo_root=None)

    async def test_get_without_repo_root_raises(self, tmp_path: Path) -> None:
        b = _backend()
        oid = long_id("0" * 64)
        with pytest.raises(ValueError, match="repo_root"):
            await b.get(oid, repo_root=None)

    async def test_exists_without_repo_root_raises(self, tmp_path: Path) -> None:
        b = _backend()
        oid = long_id("1" * 64)
        with pytest.raises(ValueError, match="repo_root"):
            await b.exists(oid, repo_root=None)


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


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

    async def test_get_latency(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("3" * 64)
        data = b"x" * (512 * 1024)
        await b.put(oid, data, repo_root=repo_root)

        start = time.perf_counter()
        result = await b.get(oid, repo_root=repo_root)
        elapsed = time.perf_counter() - start

        assert result == data
        assert elapsed < 0.5

    async def test_exists_latency(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("4" * 64)
        await b.put(oid, b"data", repo_root=repo_root)

        start = time.perf_counter()
        for _ in range(50):
            await b.exists(oid, repo_root=repo_root)
        elapsed = time.perf_counter() - start

        assert elapsed < 1.0

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

        start = time.perf_counter()
        for i in range(50):
            oid = long_id(str(i).zfill(64))
            await b.put(oid, data, repo_root=repo_root)
        elapsed = time.perf_counter() - start

        assert elapsed < 2.0


# ═══════════════════════════════════════════════════════════════════════════════
# Layer 8 — Atomic Write Safety (Phase 8 crash-safety fix)
# ═══════════════════════════════════════════════════════════════════════════════


class TestAtomicWrite:
    """_write() uses mkstemp → fsync → os.replace — no partial files visible to readers."""

    async def test_write_leaves_complete_file(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("5" * 64)
        data = b"atomic content"
        await b.put(oid, data, repo_root=repo_root)
        path = b._path(oid, repo_root=repo_root)
        assert path.exists()
        assert path.read_bytes() == data

    async def test_write_no_tmp_file_remains_after_success(self, tmp_path: Path) -> None:
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("6" * 64)
        await b.put(oid, b"data", repo_root=repo_root)
        objects_dir = repo_root / "objects"
        tmp_files = list(objects_dir.rglob("tmp*"))
        assert tmp_files == [], f"stray temp files: {tmp_files}"

    async def test_write_file_is_immutable_after_put(self, tmp_path: Path) -> None:
        import stat as _stat
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("7" * 64)
        await b.put(oid, b"readonly data", repo_root=repo_root)
        path = b._path(oid, repo_root=repo_root)
        mode = _stat.S_IMODE(path.stat().st_mode)
        assert mode == 0o444, f"expected 0o444 but got {oct(mode)}"

    async def test_write_tmp_file_cleaned_up_on_fsync_failure(self, tmp_path: Path) -> None:
        import os as _os
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("8" * 64)

        def _failing_fsync(fd: int) -> None:
            raise OSError("simulated fsync failure")

        import unittest.mock as _mock
        objects_dir = repo_root / "objects"
        objects_dir.mkdir(parents=True, exist_ok=True)

        with _mock.patch("os.fsync", side_effect=_failing_fsync):
            with pytest.raises(OSError, match="simulated fsync failure"):
                await b.put(oid, b"never written", repo_root=repo_root)

        tmp_files = list(objects_dir.rglob("tmp*"))
        assert tmp_files == [], f"orphan temp files after fsync failure: {tmp_files}"

    async def test_write_idempotent_identical_bytes_skips_io(self, tmp_path: Path) -> None:
        import time as _time
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("9" * 64)
        data = b"idempotent"
        await b.put(oid, data, repo_root=repo_root)
        path = b._path(oid, repo_root=repo_root)
        mtime_before = path.stat().st_mtime_ns
        _time.sleep(0.01)
        await b.put(oid, data, repo_root=repo_root)
        mtime_after = path.stat().st_mtime_ns
        assert mtime_before == mtime_after, "second put with identical bytes must not touch the file"

    async def test_write_repairs_corrupted_file_atomically(self, tmp_path: Path) -> None:
        import stat as _stat
        b = _backend()
        repo_root = _repo_root(tmp_path)
        raw = b"correct content"
        corrupt = b"wrong zlib garbage"
        oid = long_id("aa" + "b" * 62)

        path = b._path(oid, repo_root=repo_root)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(corrupt)
        path.chmod(_stat.S_IRUSR | _stat.S_IRGRP | _stat.S_IROTH)

        await b.put(oid, raw, repo_root=repo_root)
        assert await b.get(oid, repo_root=repo_root) == raw

    async def test_write_concurrent_same_object_safe(self, tmp_path: Path) -> None:
        import asyncio as _asyncio
        b = _backend()
        repo_root = _repo_root(tmp_path)
        oid = long_id("bb" + "c" * 62)
        data = b"concurrent data"
        results = await _asyncio.gather(*[b.put(oid, data, repo_root=repo_root) for _ in range(10)])
        assert all(r.startswith("local://") for r in results)
        assert await b.get(oid, repo_root=repo_root) == data


# ═══════════════════════════════════════════════════════════════════════════════
# read_object_bytes — unified adapter
# ═══════════════════════════════════════════════════════════════════════════════


class TestUnitReadObjectBytes:
    """read_object_bytes(obj) must handle every storage case without callers
    knowing which backend is active."""

    def _obj(self, **kwargs):
        """Build a minimal MusehubObject-like namespace."""
        defaults = dict(
            object_id=long_id("aa" * 32),
            content_cache=None,
            disk_path="",
            storage_uri="",
        )
        defaults.update(kwargs)
        return SimpleNamespace(**defaults)

    async def test_returns_content_cache_when_present(self) -> None:
        from musehub.storage.backends import read_object_bytes
        obj = self._obj(content_cache=b"cached")
        assert await read_object_bytes(obj) == b"cached"

    async def test_content_cache_takes_priority_over_disk(self, tmp_path: Path) -> None:
        from musehub.storage.backends import read_object_bytes
        disk_file = tmp_path / "obj"
        disk_file.write_bytes(b"disk content")
        obj = self._obj(content_cache=b"cached", disk_path=str(disk_file))
        assert await read_object_bytes(obj) == b"cached"

    async def test_reads_local_disk_path(self, tmp_path: Path) -> None:
        from musehub.storage.backends import read_object_bytes
        disk_file = tmp_path / "obj"
        disk_file.write_bytes(b"from disk")
        obj = self._obj(disk_path=str(disk_file))
        assert await read_object_bytes(obj) == b"from disk"

    async def test_reads_local_uri_prefix(self, tmp_path: Path) -> None:
        from musehub.storage.backends import read_object_bytes
        disk_file = tmp_path / "obj"
        disk_file.write_bytes(b"from local uri")
        obj = self._obj(disk_path=f"local://{disk_file}")
        assert await read_object_bytes(obj) == b"from local uri"

    async def test_reads_from_s3_when_s3_uri(self) -> None:
        from musehub.storage.backends import read_object_bytes
        obj = self._obj(disk_path="s3://my-bucket/objects/sha256_abc123")
        mock_backend = MagicMock()

        async def _fake_get(oid, **kw):
            return b"s3 bytes"

        mock_backend.get = _fake_get
        with patch("musehub.storage.backends.get_backend", return_value=mock_backend):
            result = await read_object_bytes(obj)
        assert result == b"s3 bytes"

    async def test_missing_disk_file_returns_none(self, tmp_path: Path) -> None:
        from musehub.storage.backends import read_object_bytes
        obj = self._obj(disk_path=str(tmp_path / "does_not_exist"))
        assert await read_object_bytes(obj) is None

    async def test_no_path_no_cache_returns_none(self) -> None:
        from musehub.storage.backends import read_object_bytes
        obj = self._obj()
        assert await read_object_bytes(obj) is None