"""Tests for the Object Store — Section 3 of test-coverage-checklist.md.

Coverage layers
───────────────
Unit        — LocalBackend._path sanitisation, traversal guard, _write idempotency,
              put/get/exists/delete async wrappers, uri_for absolute-path contract,
              _detect_file_type and _content_type helpers.
Integration — Service layer (musehub_repository): list_objects, get_object_row,
              get_object_by_path with a real in-memory DB.
E2E         — HTTP handlers: list objects, get object content, blob meta,
              public vs private visibility, 404/410/401 error paths.
Stress      — 50-object fan-out push then list; 20-repo isolation fan-out.
Data        — Content fidelity (byte-for-byte round-trip), _write idempotency
              prevents overwrites, disk_path absolute-path regression
              (sha256: colon sanitisation + objects_dir root must be present).
Security    — _path traversal rejection (repo_id and object_id), cross-repo
              object isolation, private-repo auth gate on every endpoint.
Performance — LocalBackend put/get latency budgets, list_objects at 100+ rows.
"""
from __future__ import annotations

import hashlib
import os
import time
import uuid
from pathlib import Path

import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.config import settings
from musehub.db import musehub_models as db
from musehub.storage.backends import LocalBackend
from tests.factories import create_repo
from musehub.muse_contracts.json_types import JSONObject, StrDict


# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────

def _sha256_id(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


async def _insert_object(
    session: AsyncSession,
    repo_id: str,
    object_id: str,
    path: str,
    disk_path: str,
    size: int = 0,
) -> db.MusehubObject:
    obj = db.MusehubObject(
        object_id=object_id,
        repo_id=repo_id,
        path=path,
        size_bytes=size,
        disk_path=disk_path,
        storage_uri=f"local://{disk_path}",
    )
    session.add(obj)
    await session.commit()
    await session.refresh(obj)
    return obj


# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 — Unit: LocalBackend
# ─────────────────────────────────────────────────────────────────────────────

class TestLocalBackendPath:
    """_path() must produce the correct file location and reject traversals."""

    def test_safe_path_is_under_root(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("myrepo", "sha256:abc")
        assert str(p).startswith(str(tmp_path))

    def test_colon_in_object_id_sanitised(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("myrepo", "sha256:abc123")
        assert ":" not in p.name
        assert p.name == "sha256_abc123"

    def test_slash_in_object_id_sanitised(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("myrepo", "a/b/c")
        assert "/" not in p.name
        assert p.name == "a_b_c"

    def test_traversal_via_repo_id_raises(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        with pytest.raises(ValueError, match="traversal"):
            backend._path("../../etc/passwd", "object")

    def test_traversal_via_dotdot_repo_id_raises(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        with pytest.raises(ValueError, match="traversal"):
            backend._path("../secret", "obj")

    def test_normal_path_does_not_raise(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("repo-123", "sha256:deadbeef")
        assert p.parent.parent == tmp_path

    def test_uri_for_embeds_absolute_path(self, tmp_path: Path) -> None:
        """uri_for must return local://<absolute_path> so disk_path is correct."""
        backend = LocalBackend(objects_dir=str(tmp_path))
        uri = backend.uri_for("myrepo", "sha256:cafe")
        assert uri.startswith("local:///")
        # Stripping the scheme gives the real absolute disk path
        disk_path = uri.replace("local://", "")
        assert Path(disk_path).is_absolute()
        # The path must contain the sanitised object_id, not the raw one
        assert "sha256_cafe" in disk_path
        assert "sha256:cafe" not in disk_path


class TestLocalBackendWrite:
    """_write() is idempotent — second call must not overwrite existing content."""

    def test_write_creates_file(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("repo", "obj1")
        backend._write(p, b"hello")
        assert p.exists()
        assert p.read_bytes() == b"hello"

    def test_write_is_idempotent(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("repo", "obj2")
        backend._write(p, b"original")
        backend._write(p, b"overwrite-attempt")
        # Content must not change
        assert p.read_bytes() == b"original"

    def test_write_creates_parent_dirs(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = tmp_path / "deep" / "nested" / "dir" / "file.bin"
        p.parent.mkdir(parents=True)
        backend._write(p, b"data")
        assert p.read_bytes() == b"data"


class TestLocalBackendAsyncOps:
    """put/get/exists/delete async interface."""

    @pytest.mark.asyncio
    async def test_put_returns_uri_and_file_exists(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        uri = await backend.put("repo-a", "sha256:aaa", b"content-a")
        assert uri.startswith("local://")
        disk = uri.replace("local://", "")
        assert Path(disk).exists()
        assert Path(disk).read_bytes() == b"content-a"

    @pytest.mark.asyncio
    async def test_get_returns_none_for_missing(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        result = await backend.get("repo-a", "sha256:missing")
        assert result is None

    @pytest.mark.asyncio
    async def test_get_returns_correct_bytes(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        data = b"\x00\x01\x02\x03" * 1024
        await backend.put("repo-b", "sha256:bbb", data)
        result = await backend.get("repo-b", "sha256:bbb")
        assert result == data

    @pytest.mark.asyncio
    async def test_exists_false_before_put(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        assert not await backend.exists("repo-c", "sha256:ccc")

    @pytest.mark.asyncio
    async def test_exists_true_after_put(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        await backend.put("repo-c", "sha256:ccc", b"x")
        assert await backend.exists("repo-c", "sha256:ccc")

    @pytest.mark.asyncio
    async def test_delete_removes_file(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        await backend.put("repo-d", "sha256:ddd", b"y")
        await backend.delete("repo-d", "sha256:ddd")
        assert not await backend.exists("repo-d", "sha256:ddd")

    @pytest.mark.asyncio
    async def test_delete_noop_on_missing(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        # Should not raise
        await backend.delete("repo-d", "sha256:nonexistent")

    @pytest.mark.asyncio
    async def test_separate_repos_isolated(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        await backend.put("repo-e", "sha256:fff", b"for-repo-e")
        assert not await backend.exists("repo-f", "sha256:fff")


# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 (cont.) — Unit: file-type detection helpers
# ─────────────────────────────────────────────────────────────────────────────

class TestDetectFileType:
    """_detect_file_type covers image, json, xml, and other."""

    def test_json_is_json(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("meta.json") == "json"

    def test_png_is_image(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("cover.png") == "image"

    def test_webp_is_image(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("thumb.webp") == "image"

    def test_jpeg_is_image(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("photo.jpeg") == "image"

    def test_xml_is_xml(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("score.xml") == "xml"

    def test_unknown_ext_is_other(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("file.xyz") == "other"

    def test_no_ext_is_other(self) -> None:
        from musehub.api.routes.musehub.objects import _detect_file_type
        assert _detect_file_type("Makefile") == "other"


class TestContentType:
    """_content_type returns correct MIME type."""

    def test_png_content_type(self) -> None:
        from musehub.api.routes.musehub.objects import _content_type
        ct = _content_type("cover.png")
        assert "png" in ct.lower() or "image" in ct.lower()

    def test_json_content_type(self) -> None:
        from musehub.api.routes.musehub.objects import _content_type
        assert "json" in _content_type("data.json")

    def test_webp_content_type(self) -> None:
        from musehub.api.routes.musehub.objects import _content_type
        assert "webp" in _content_type("roll.webp")


# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 — Integration: service layer
# ─────────────────────────────────────────────────────────────────────────────

class TestServiceListObjects:
    @pytest.mark.asyncio
    async def test_empty_repo_returns_empty_list(self, db_session: AsyncSession) -> None:
        from musehub.services import musehub_repository
        repo = await create_repo(db_session, visibility="public")
        result = await musehub_repository.list_objects(db_session, repo.repo_id)
        assert result == []

    @pytest.mark.asyncio
    async def test_returns_all_objects_sorted_by_path(
        self, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        from musehub.services import musehub_repository
        repo = await create_repo(db_session, visibility="public")
        for name in ("z.bin", "a.bin", "m.bin"):
            p = tmp_path / name
            p.write_bytes(b"data")
            await _insert_object(db_session, repo.repo_id, f"sha256:{name}", name, str(p))

        result = await musehub_repository.list_objects(db_session, repo.repo_id)
        paths = [r.path for r in result]
        assert paths == sorted(paths)

    @pytest.mark.asyncio
    async def test_isolated_to_repo(self, db_session: AsyncSession, tmp_path: Path) -> None:
        from musehub.services import musehub_repository
        repo_a = await create_repo(db_session, slug="repo-a-svc", visibility="public")
        repo_b = await create_repo(db_session, slug="repo-b-svc", visibility="public")
        p = tmp_path / "obj.bin"
        p.write_bytes(b"x")
        await _insert_object(db_session, repo_a.repo_id, "sha256:svc-a", "a.bin", str(p))

        result = await musehub_repository.list_objects(db_session, repo_b.repo_id)
        assert result == []


class TestServiceGetObjectRow:
    @pytest.mark.asyncio
    async def test_returns_row_when_found(
        self, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        from musehub.services import musehub_repository
        repo = await create_repo(db_session, visibility="public")
        p = tmp_path / "file.bin"
        p.write_bytes(b"bytes")
        await _insert_object(db_session, repo.repo_id, "sha256:rowtest", "file.bin", str(p))

        row = await musehub_repository.get_object_row(db_session, repo.repo_id, "sha256:rowtest")
        assert row is not None
        assert row.object_id == "sha256:rowtest"

    @pytest.mark.asyncio
    async def test_returns_none_when_missing(self, db_session: AsyncSession) -> None:
        from musehub.services import musehub_repository
        repo = await create_repo(db_session, visibility="public")
        row = await musehub_repository.get_object_row(db_session, repo.repo_id, "sha256:ghost")
        assert row is None

    @pytest.mark.asyncio
    async def test_wrong_repo_returns_none(
        self, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        from musehub.services import musehub_repository
        repo_a = await create_repo(db_session, slug="row-a", visibility="public")
        repo_b = await create_repo(db_session, slug="row-b", visibility="public")
        p = tmp_path / "f.bin"
        p.write_bytes(b"x")
        await _insert_object(db_session, repo_a.repo_id, "sha256:xrepo", "f.bin", str(p))

        row = await musehub_repository.get_object_row(db_session, repo_b.repo_id, "sha256:xrepo")
        assert row is None


class TestServiceGetObjectByPath:
    @pytest.mark.asyncio
    async def test_returns_most_recent_for_path(
        self, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        from musehub.services import musehub_repository
        import asyncio
        repo = await create_repo(db_session, visibility="public")

        p1 = tmp_path / "v1.bin"
        p1.write_bytes(b"v1")
        await _insert_object(db_session, repo.repo_id, "sha256:v1", "track.bin", str(p1))
        await asyncio.sleep(0.01)  # ensure distinct created_at

        p2 = tmp_path / "v2.bin"
        p2.write_bytes(b"v2")
        await _insert_object(db_session, repo.repo_id, "sha256:v2", "track.bin", str(p2))

        row = await musehub_repository.get_object_by_path(db_session, repo.repo_id, "track.bin")
        assert row is not None
        assert row.object_id == "sha256:v2"

    @pytest.mark.asyncio
    async def test_returns_none_for_missing_path(self, db_session: AsyncSession) -> None:
        from musehub.services import musehub_repository
        repo = await create_repo(db_session, visibility="public")
        row = await musehub_repository.get_object_by_path(db_session, repo.repo_id, "ghost.bin")
        assert row is None


# ─────────────────────────────────────────────────────────────────────────────
# Layer 3 — E2E: HTTP endpoints
# ─────────────────────────────────────────────────────────────────────────────

class TestListObjectsEndpoint:
    @pytest.mark.asyncio
    async def test_public_repo_empty_list(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-list-empty", visibility="public")
        resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
        assert resp.status_code == 200
        assert resp.json()["objects"] == []

    @pytest.mark.asyncio
    async def test_public_repo_returns_metadata(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-list-meta", visibility="public")
        p = tmp_path / "t.bin"
        p.write_bytes(b"data")
        await _insert_object(db_session, repo.repo_id, "sha256:list1", "t.bin", str(p), size=4)

        resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
        assert resp.status_code == 200
        objects = resp.json()["objects"]
        assert len(objects) == 1
        assert objects[0]["path"] == "t.bin"
        assert objects[0]["sizeBytes"] == 4

    @pytest.mark.asyncio
    async def test_unknown_repo_returns_404(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await client.get(f"/api/repos/{uuid.uuid4()}/objects")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    async def test_private_repo_without_auth_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-list-priv", visibility="private")
        resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
        assert resp.status_code == 401

    @pytest.mark.asyncio
    async def test_private_repo_with_auth_returns_200(
        self,
        client: AsyncClient,
        db_session: AsyncSession,
        auth_headers: StrDict,
        tmp_path: Path,
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-list-priv-auth", visibility="private")
        resp = await client.get(
            f"/api/repos/{repo.repo_id}/objects", headers=auth_headers
        )
        assert resp.status_code == 200


class TestGetObjectContentEndpoint:
    @pytest.mark.asyncio
    async def test_returns_file_bytes(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(settings, "musehub_objects_dir", str(tmp_path))
        repo = await create_repo(db_session, slug="e2e-content-ok", visibility="public")
        data = b"\x00\x01\x02DATA-content"
        p = tmp_path / "song.bin"
        p.write_bytes(data)
        obj_id = "sha256:content-ok"
        await _insert_object(db_session, repo.repo_id, obj_id, "song.bin", str(p), len(data))

        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/{obj_id}/content")
        assert resp.status_code == 200
        assert resp.content == data

    @pytest.mark.asyncio
    async def test_unknown_object_returns_404(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-content-404obj", visibility="public")
        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/sha256:ghost/content")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    async def test_unknown_repo_returns_404(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await client.get(f"/api/repos/{uuid.uuid4()}/objects/sha256:x/content")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    async def test_missing_disk_file_returns_410(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(settings, "musehub_objects_dir", str(tmp_path))
        repo = await create_repo(db_session, slug="e2e-content-410", visibility="public")
        # Point disk_path at a non-existent file inside the storage root
        gone_path = str(tmp_path / "gone.bin")
        await _insert_object(db_session, repo.repo_id, "sha256:gone", "gone.bin", gone_path)

        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/sha256:gone/content")
        assert resp.status_code == 410

    @pytest.mark.asyncio
    async def test_private_repo_without_auth_returns_401(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-content-priv", visibility="private")
        p = tmp_path / "secret.bin"
        p.write_bytes(b"secret")
        await _insert_object(db_session, repo.repo_id, "sha256:priv1", "secret.bin", str(p))

        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/sha256:priv1/content")
        assert resp.status_code == 401


class TestGetBlobMetaEndpoint:
    @pytest.mark.asyncio
    async def test_returns_blob_metadata(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(settings, "musehub_objects_dir", str(tmp_path))
        repo = await create_repo(db_session, slug="e2e-blob-meta", visibility="public")
        p = tmp_path / "score.json"
        p.write_text('{"notes": []}')
        await _insert_object(
            db_session, repo.repo_id, "sha256:blobmeta1", "score.json", str(p), size=13
        )

        resp = await client.get(f"/api/repos/{repo.repo_id}/blob/main/score.json")
        assert resp.status_code == 200
        body = resp.json()
        assert body["path"] == "score.json"
        assert body["sizeBytes"] == 13
        assert body["fileType"] == "json"
        # JSON files under 256 KB get content_text embedded
        assert body["contentText"] is not None

    @pytest.mark.asyncio
    async def test_unknown_path_returns_404(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-blob-404", visibility="public")
        resp = await client.get(f"/api/repos/{repo.repo_id}/blob/main/ghost.bin")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    async def test_unknown_repo_returns_404(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        resp = await client.get(f"/api/repos/{uuid.uuid4()}/blob/main/any.bin")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    async def test_private_repo_without_auth_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="e2e-blob-priv", visibility="private")
        resp = await client.get(f"/api/repos/{repo.repo_id}/blob/main/any.bin")
        assert resp.status_code == 401

    @pytest.mark.asyncio
    async def test_binary_file_has_no_content_text(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        """Non-text files (type 'other') must not get content_text embedded."""
        repo = await create_repo(db_session, slug="e2e-blob-bin", visibility="public")
        p = tmp_path / "artifact.bin"
        p.write_bytes(b"\x00\x01\x02\x03")
        await _insert_object(
            db_session, repo.repo_id, "sha256:blobbin", "artifact.bin", str(p), size=4
        )

        resp = await client.get(f"/api/repos/{repo.repo_id}/blob/main/artifact.bin")
        assert resp.status_code == 200
        assert resp.json()["contentText"] is None


# ─────────────────────────────────────────────────────────────────────────────
# Layer 4 — Stress
# ─────────────────────────────────────────────────────────────────────────────

def _wire_push(content: bytes, oid: str, path: str = "artifact.bin") -> JSONObject:
    """Build a minimal wire push payload dict (raw msgpack-serialisable)."""
    return {
        "bundle": {
            "commits": [],
            "snapshots": [],
            "objects": [{"object_id": oid, "path": path, "content": content}],
        },
        "branch": "main",
        "force": False,
    }


class TestObjectStoreStress:
    @pytest.mark.asyncio
    async def test_50_objects_push_then_list_all_present(
        self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict
    ) -> None:
        """Push 50 objects via the wire protocol; list must return all 50."""
        import msgpack

        repo = await create_repo(db_session, slug="stress-50obj", owner="test-user-wire",
                                  owner_user_id="wire-test-user-id", visibility="public")

        OBJECT_COUNT = 50
        objects = []
        for i in range(OBJECT_COUNT):
            content = f"DATA-{i:04d}".encode() * 128
            oid = _sha256_id(content)
            objects.append({"object_id": oid, "path": f"artifact_{i:04d}.bin", "content": content})

        payload = {"bundle": {"commits": [], "snapshots": [], "objects": objects},
                   "branch": "main", "force": False}
        push_resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=msgpack.packb(payload),
            headers=wire_headers,
        )
        assert push_resp.status_code == 200

        list_resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
        assert list_resp.status_code == 200
        assert len(list_resp.json()["objects"]) == OBJECT_COUNT

    @pytest.mark.asyncio
    async def test_20_repo_fan_out_isolated(
        self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict
    ) -> None:
        """20 repos each pushed one unique object; listing each repo returns exactly 1."""
        import msgpack

        REPO_COUNT = 20
        repos = []
        for i in range(REPO_COUNT):
            repo = await create_repo(db_session, slug=f"fanout-{i:02d}",
                                      owner="test-user-wire",
                                      owner_user_id="wire-test-user-id",
                                      visibility="public")
            repos.append(repo)
            content = f"unique-{i}".encode()
            oid = _sha256_id(content)
            resp = await client.post(
                f"/{repo.owner}/{repo.slug}/push",
                content=msgpack.packb(_wire_push(content, oid)),
                headers=wire_headers,
            )
            assert resp.status_code == 200

        for repo in repos:
            resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
            assert resp.status_code == 200
            assert len(resp.json()["objects"]) == 1


# ─────────────────────────────────────────────────────────────────────────────
# Layer 5 — Data Integrity
# ─────────────────────────────────────────────────────────────────────────────

class TestDataIntegrity:
    @pytest.mark.asyncio
    async def test_disk_path_points_to_real_file_after_wire_push(
        self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict
    ) -> None:
        """Regression: disk_path must be an absolute path with sanitised colon.

        Before the uri_for fix, disk_path stored "repo-id/sha256:abc" — a relative
        path with a literal colon — so get_object_content always returned 410 Gone.
        """
        import msgpack

        content = b"DATA-regression-test" * 10
        oid = _sha256_id(content)  # e.g. "sha256:<hex>"
        repo = await create_repo(db_session, slug="diskpath-regression",
                                  owner="test-user-wire",
                                  owner_user_id="wire-test-user-id",
                                  visibility="public")

        push_resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=msgpack.packb(_wire_push(content, oid, "artifact.bin")),
            headers=wire_headers,
        )
        assert push_resp.status_code == 200

        # Verify the stored disk_path is absolute and has no colon in the filename
        from musehub.services import musehub_repository
        row = await musehub_repository.get_object_row(db_session, repo.repo_id, oid)
        assert row is not None
        assert Path(row.disk_path).is_absolute(), (
            f"disk_path is not absolute: {row.disk_path!r}"
        )
        assert ":" not in Path(row.disk_path).name, (
            f"disk_path filename still has colon: {row.disk_path!r}"
        )
        assert Path(row.disk_path).exists(), (
            f"disk_path does not exist on disk: {row.disk_path!r}"
        )

    @pytest.mark.asyncio
    async def test_content_round_trip_via_http(
        self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict
    ) -> None:
        """Content retrieved via HTTP must match pushed bytes exactly."""
        import msgpack

        content = os.urandom(512)
        oid = _sha256_id(content)
        repo = await create_repo(db_session, slug="roundtrip-test",
                                  owner="test-user-wire",
                                  owner_user_id="wire-test-user-id",
                                  visibility="public")

        push_resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=msgpack.packb(_wire_push(content, oid)),
            headers=wire_headers,
        )
        assert push_resp.status_code == 200

        content_resp = await client.get(
            f"/api/repos/{repo.repo_id}/objects/{oid}/content"
        )
        assert content_resp.status_code == 200
        assert content_resp.content == content

    @pytest.mark.asyncio
    async def test_idempotent_push_stores_object_once(
        self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict
    ) -> None:
        """Pushing the same object twice must not duplicate DB rows."""
        import msgpack
        from sqlalchemy import select

        content = b"idempotent-content"
        oid = _sha256_id(content)
        repo = await create_repo(db_session, slug="idempotent-obj",
                                  owner="test-user-wire",
                                  owner_user_id="wire-test-user-id",
                                  visibility="public")

        payload = msgpack.packb(_wire_push(content, oid))
        for _ in range(2):
            r = await client.post(
                f"/{repo.owner}/{repo.slug}/push",
                content=payload, headers=wire_headers,
            )
            assert r.status_code == 200

        from sqlalchemy import func
        count_stmt = (
            select(func.count())
            .select_from(db.MusehubObject)
            .where(db.MusehubObject.object_id == oid)
        )
        count = (await db_session.execute(count_stmt)).scalar_one()
        assert count == 1

    @pytest.mark.asyncio
    async def test_local_backend_put_idempotent_does_not_overwrite(
        self, tmp_path: Path
    ) -> None:
        """_write: second put with different content must not overwrite the first."""
        backend = LocalBackend(objects_dir=str(tmp_path))
        await backend.put("repo", "sha256:idem", b"original")
        await backend.put("repo", "sha256:idem", b"changed")
        result = await backend.get("repo", "sha256:idem")
        assert result == b"original"


# ─────────────────────────────────────────────────────────────────────────────
# Layer 6 — Security
# ─────────────────────────────────────────────────────────────────────────────

class TestObjectStoreSecurity:
    def test_traversal_via_repo_id_raises_value_error(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        with pytest.raises(ValueError, match="traversal"):
            backend._path("../../etc", "obj")

    def test_traversal_via_object_id_colon_is_sanitised(self, tmp_path: Path) -> None:
        """Colons in object_id are sanitised to underscores, not interpreted as path."""
        backend = LocalBackend(objects_dir=str(tmp_path))
        p = backend._path("repo", "sha256:../../../../etc/passwd")
        # After sanitisation the entire object_id is flattened to one filename component
        assert p.parent == tmp_path / "repo"

    def test_traversal_via_repo_id_double_slash(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        with pytest.raises(ValueError, match="traversal"):
            backend._path("/../../../etc", "obj")

    @pytest.mark.asyncio
    async def test_private_repo_list_objects_requires_auth(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="sec-priv-list", visibility="private")
        resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
        assert resp.status_code == 401

    @pytest.mark.asyncio
    async def test_private_repo_content_requires_auth(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        repo = await create_repo(db_session, slug="sec-priv-content", visibility="private")
        p = tmp_path / "x.bin"
        p.write_bytes(b"x")
        await _insert_object(db_session, repo.repo_id, "sha256:sec1", "x.bin", str(p))

        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/sha256:sec1/content")
        assert resp.status_code == 401

    @pytest.mark.asyncio
    async def test_cross_repo_object_not_visible_in_other_repo(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        repo_a = await create_repo(db_session, slug="sec-cross-a", visibility="public")
        repo_b = await create_repo(db_session, slug="sec-cross-b", visibility="public")
        p = tmp_path / "a.bin"
        p.write_bytes(b"a")
        await _insert_object(db_session, repo_a.repo_id, "sha256:cross1", "a.bin", str(p))

        # Object in repo_a must not appear in repo_b listing
        resp = await client.get(f"/api/repos/{repo_b.repo_id}/objects")
        assert resp.status_code == 200
        assert resp.json()["objects"] == []

        # Direct content fetch via repo_b must return 404
        resp = await client.get(
            f"/api/repos/{repo_b.repo_id}/objects/sha256:cross1/content"
        )
        assert resp.status_code == 404


# ─────────────────────────────────────────────────────────────────────────────
# Layer 7 — Performance
# ─────────────────────────────────────────────────────────────────────────────

class TestObjectStorePerformance:
    @pytest.mark.asyncio
    async def test_local_backend_put_1kb_under_50ms(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        data = b"x" * 1024
        times = []
        for i in range(50):
            oid = f"sha256:perf{i:04d}"
            t0 = time.perf_counter()
            await backend.put(f"repo", oid, data)
            times.append(time.perf_counter() - t0)
        median_ms = sorted(times)[len(times) // 2] * 1000
        assert median_ms < 50, f"put median {median_ms:.1f}ms exceeded 50ms budget"

    @pytest.mark.asyncio
    async def test_local_backend_get_1kb_under_10ms(self, tmp_path: Path) -> None:
        backend = LocalBackend(objects_dir=str(tmp_path))
        data = b"y" * 1024
        await backend.put("repo", "sha256:getperf", data)
        times = []
        for _ in range(100):
            t0 = time.perf_counter()
            _ = await backend.get("repo", "sha256:getperf")
            times.append(time.perf_counter() - t0)
        median_ms = sorted(times)[len(times) // 2] * 1000
        assert median_ms < 10, f"get median {median_ms:.1f}ms exceeded 10ms budget"

    @pytest.mark.asyncio
    async def test_list_objects_100_rows_under_200ms(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path
    ) -> None:
        repo = await create_repo(db_session, slug="perf-list-100", visibility="public")
        for i in range(100):
            p = tmp_path / f"f{i:03d}.bin"
            p.write_bytes(b"m")
            await _insert_object(
                db_session, repo.repo_id, f"sha256:perf{i:03d}", f"f{i:03d}.bin", str(p), 1
            )
        # warm-up
        await client.get(f"/api/repos/{repo.repo_id}/objects")

        times = []
        for _ in range(10):
            t0 = time.perf_counter()
            resp = await client.get(f"/api/repos/{repo.repo_id}/objects")
            times.append(time.perf_counter() - t0)
        assert resp.status_code == 200
        assert len(resp.json()["objects"]) == 100
        median_ms = sorted(times)[len(times) // 2] * 1000
        assert median_ms < 200, f"list_objects median {median_ms:.1f}ms exceeded 200ms"

    @pytest.mark.asyncio
    async def test_get_object_content_1mb_under_500ms(
        self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(settings, "musehub_objects_dir", str(tmp_path))
        repo = await create_repo(db_session, slug="perf-content-1mb", visibility="public")
        data = os.urandom(1024 * 1024)
        p = tmp_path / "big.bin"
        p.write_bytes(data)
        oid = "sha256:perf1mb"
        await _insert_object(db_session, repo.repo_id, oid, "big.bin", str(p), len(data))

        # warm-up
        await client.get(f"/api/repos/{repo.repo_id}/objects/{oid}/content")

        t0 = time.perf_counter()
        resp = await client.get(f"/api/repos/{repo.repo_id}/objects/{oid}/content")
        elapsed_ms = (time.perf_counter() - t0) * 1000
        assert resp.status_code == 200
        assert elapsed_ms < 500, f"1 MB content serve took {elapsed_ms:.1f}ms > 500ms"