"""Tests for GET /{owner}/{repo_slug}/raw/{ref}/{path} endpoint. Covers: raw_file_semantic (ui_tree.py): - 200: file exists in snapshot manifest and object exists in storage - 404: file exists in manifest but object missing from storage - 404: file not in snapshot manifest at ref - 404: ref does not exist - correct Content-Type for text files (.py, .toml, .md) - correct Content-Type for binary files (.png) - Content-Disposition: inline for text, attachment for binary storage.exists() interface: - LocalBackend.exists(object_id) — single argument, no repo_id - S3Backend.exists(object_id) — single argument, no repo_id - Both satisfy the StorageBackend protocol """ from __future__ import annotations import secrets from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock, patch import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import fake_id from musehub.core.genesis import compute_branch_id, compute_identity_id, compute_repo_id from musehub.db.musehub_models import ( MusehubBranch, MusehubCommit, MusehubRepo, MusehubSnapshot, ) from musehub.storage.backends import LocalBackend, S3Backend def _uid() -> str: return secrets.token_hex(16) # ── DB fixtures ─────────────────────────────────────────────────────────────── async def _make_repo( db: AsyncSession, owner: str = "gabriel", slug: str = "muse", ) -> MusehubRepo: created_at = datetime.now(tz=timezone.utc) owner_id = compute_identity_id(owner.encode()) repo_id = compute_repo_id(owner_id, slug, "code", created_at.isoformat()) repo = MusehubRepo( repo_id=repo_id, name=slug, owner=owner, slug=slug, visibility="public", owner_user_id=owner_id, created_at=created_at, updated_at=created_at, ) db.add(repo) await db.flush() return repo async def _make_snapshot( db: AsyncSession, repo_id: str, manifest: dict[str, str], ) -> MusehubSnapshot: snap = MusehubSnapshot( snapshot_id=fake_id(_uid()), repo_id=repo_id, manifest_blob=msgpack.packb(manifest, use_bin_type=True), entry_count=len(manifest), created_at=datetime.now(tz=timezone.utc), ) db.add(snap) await db.flush() return snap async def _make_branch_at_commit( db: AsyncSession, repo_id: str, branch_name: str, manifest: dict[str, str], ) -> tuple[MusehubCommit, MusehubSnapshot]: snap = await _make_snapshot(db, repo_id, manifest) now = datetime.now(tz=timezone.utc) commit = MusehubCommit( commit_id=fake_id(_uid()), repo_id=repo_id, snapshot_id=snap.snapshot_id, message="test commit", author="gabriel", branch=branch_name, parent_ids=[], timestamp=now, created_at=now, ) db.add(commit) await db.flush() branch = MusehubBranch( branch_id=compute_branch_id(repo_id, branch_name), repo_id=repo_id, name=branch_name, head_commit_id=commit.commit_id, ) db.add(branch) await db.flush() return commit, snap # ═══════════════════════════════════════════════════════════════════════════════ # StorageBackend interface — exists() takes exactly one argument (object_id) # ═══════════════════════════════════════════════════════════════════════════════ class TestStorageBackendExistsInterface: """Regression: exists() must accept a single object_id, never (repo_id, object_id).""" async def test_local_backend_exists_single_arg(self, tmp_path: Path) -> None: backend = LocalBackend(repo_root=tmp_path / "objects") oid = fake_id("test-object") result = await backend.exists(oid) assert result is False async def test_local_backend_exists_returns_true_after_put(self, tmp_path: Path) -> None: backend = LocalBackend(repo_root=tmp_path / "objects") oid = fake_id("test-object") await backend.put(oid, b"hello world") assert await backend.exists(oid) is True async def test_local_backend_exists_two_args_raises(self, tmp_path: Path) -> None: """Unbound LocalBackend.exists() must raise ValueError — no repo_root, no data.""" backend = LocalBackend() with pytest.raises(ValueError): await backend.exists(fake_id("test-object")) async def test_s3_backend_exists_single_arg(self) -> None: mock_client = MagicMock() mock_client.head_object.return_value = {} backend = S3Backend(bucket="test-bucket", region="us-east-1") backend._client = mock_client oid = fake_id("test-object") result = await backend.exists(oid) assert result is True mock_client.head_object.assert_called_once_with( Bucket="test-bucket", Key=f"objects/{oid}" ) async def test_s3_backend_exists_two_args_raises(self) -> None: """S3Backend.exists() raises when head_object fails — standard error path.""" mock_client = MagicMock() mock_client.head_object.side_effect = Exception("not found") backend = S3Backend(bucket="test-bucket", region="us-east-1") backend._client = mock_client result = await backend.exists(fake_id("test-object")) assert result is False # ═══════════════════════════════════════════════════════════════════════════════ # GET /{owner}/{repo_slug}/raw/{ref}/{path} — endpoint tests # ═══════════════════════════════════════════════════════════════════════════════ class TestRawEndpoint: async def test_returns_200_for_file_in_manifest_and_storage( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session) file_content = b"[tool.poetry]\nname = 'muse'\n" oid = fake_id("pyproject-oid") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"pyproject.toml": oid} ) await db_session.commit() backend = LocalBackend(repo_root=tmp_path / "objects") await backend.put(oid, file_content) with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/main/pyproject.toml") assert resp.status_code == 200 assert resp.content == file_content async def test_returns_404_when_file_not_in_manifest( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session, slug="muse2") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"README.md": fake_id("readme-oid")} ) await db_session.commit() backend = LocalBackend(repo_root=tmp_path / "objects") with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/main/pyproject.toml") assert resp.status_code == 404 assert "pyproject.toml" in resp.json()["detail"] async def test_returns_404_when_object_missing_from_storage( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session, slug="muse3") oid = fake_id("missing-oid") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"pyproject.toml": oid} ) await db_session.commit() # Backend has no objects written — exists() returns False backend = LocalBackend(repo_root=tmp_path / "objects") with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/main/pyproject.toml") assert resp.status_code == 404 # Must be distinct from the manifest-miss message so we can tell the two # failure cases apart from logs/responses (critical for staging diagnosis). assert "storage" in resp.json()["detail"].lower() async def test_404_manifest_miss_and_storage_miss_have_distinct_messages( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: """Regression: the two 404 paths must produce different detail strings. Without this the staging 404 is undiagnosable — we can't tell whether the snapshot manifest has the file or whether the object is missing from R2. """ backend = LocalBackend(repo_root=tmp_path / "objects") # Case A: file not in manifest at all repo_a = await _make_repo(db_session, slug="muse3b") _, _ = await _make_branch_at_commit( db_session, repo_a.repo_id, "main", {"README.md": fake_id("readme-oid")} ) await db_session.commit() with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp_a = await client.get(f"/{repo_a.owner}/{repo_a.slug}/raw/main/pyproject.toml") # Case B: file in manifest, object missing from storage repo_b = await _make_repo(db_session, slug="muse3c") _, _ = await _make_branch_at_commit( db_session, repo_b.repo_id, "main", {"pyproject.toml": fake_id("missing-oid")} ) await db_session.commit() with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp_b = await client.get(f"/{repo_b.owner}/{repo_b.slug}/raw/main/pyproject.toml") assert resp_a.status_code == 404 assert resp_b.status_code == 404 assert resp_a.json()["detail"] != resp_b.json()["detail"], ( "manifest-miss and storage-miss must produce different detail strings" ) async def test_returns_404_for_unknown_ref( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session, slug="muse4") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"pyproject.toml": fake_id("oid")} ) await db_session.commit() backend = LocalBackend(repo_root=tmp_path / "objects") with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/nonexistent-branch/pyproject.toml") assert resp.status_code == 404 async def test_text_file_served_as_text_plain( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session, slug="muse5") oid = fake_id("py-oid") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"musehub/main.py": oid} ) await db_session.commit() backend = LocalBackend(repo_root=tmp_path / "objects") await backend.put(oid, b"def main(): pass\n") with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/main/musehub/main.py") assert resp.status_code == 200 assert "text/plain" in resp.headers["content-type"] async def test_binary_file_served_as_attachment( self, client: AsyncClient, db_session: AsyncSession, tmp_path: Path ) -> None: repo = await _make_repo(db_session, slug="muse6") oid = fake_id("png-oid") _, _ = await _make_branch_at_commit( db_session, repo.repo_id, "main", {"logo.png": oid} ) await db_session.commit() backend = LocalBackend(repo_root=tmp_path / "objects") await backend.put(oid, b"\x89PNG\r\n\x1a\n") # PNG magic bytes with patch("musehub.api.routes.musehub.ui_tree._get_storage_backend", return_value=backend): resp = await client.get(f"/{repo.owner}/{repo.slug}/raw/main/logo.png") assert resp.status_code == 200 assert resp.headers["content-type"] == "image/png" assert "attachment" in resp.headers["content-disposition"]