"""Tests for the MuseHub explore/discover API endpoints. Covers acceptance criteria: - test_explore_page_renders — GET /explore returns 200 HTML - test_list_public_repos_empty — no public repos → empty list - test_explore_only_public_repos — private repos are excluded from results - test_explore_filters_by_genre — genre tag filter works - test_explore_filters_by_instrumentation — instrumentation tag filter works - test_explore_sorts_by_created — created sort returns newest first - test_explore_pagination — page 2 returns different repos """ from __future__ import annotations from datetime import datetime, timezone import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.core.genesis import compute_identity_id, compute_repo_id from musehub.db.musehub_repo_models import MusehubRepo # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _make_public_repo( db_session: AsyncSession, *, name: str = "test-jazz-repo", tags: list[str] | None = None, description: str = "", ) -> str: """Seed a public repo and return its repo_id.""" import re as _re slug = _re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:64].strip("-") or "repo" created_at = datetime.now(tz=timezone.utc) owner_id = compute_identity_id(b"testuser") repo = MusehubRepo( repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()), name=name, owner="testuser", slug=slug, visibility="public", owner_user_id=owner_id, description=description, tags=tags or [], created_at=created_at, updated_at=created_at, ) db_session.add(repo) await db_session.commit() await db_session.refresh(repo) return str(repo.repo_id) async def _make_private_repo(db_session: AsyncSession, name: str = "private-beats") -> str: """Seed a private repo and return its repo_id.""" import re as _re slug = _re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:64].strip("-") or "repo" created_at = datetime.now(tz=timezone.utc) owner_id = compute_identity_id(b"testuser") repo = MusehubRepo( repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()), name=name, owner="testuser", slug=slug, visibility="private", owner_user_id=owner_id, description="", tags=[], created_at=created_at, updated_at=created_at, ) db_session.add(repo) await db_session.commit() await db_session.refresh(repo) return str(repo.repo_id) # --------------------------------------------------------------------------- # UI page tests (no auth required) # --------------------------------------------------------------------------- async def test_explore_page_renders(client: AsyncClient) -> None: """GET /explore returns 200 HTML with filter controls.""" response = await client.get("/explore") assert response.status_code == 200 assert "text/html" in response.headers["content-type"] body = response.text assert "MuseHub" in body assert "Explore" in body # Filter sidebar and sort controls rendered by the Jinja2 template assert "filter-form" in body assert 'name="sort"' in body assert 'name="license"' in body assert "/explore" in body async def test_explore_page_no_auth_required(client: AsyncClient) -> None: """GET /explore must not return 401 — it is a public page.""" response = await client.get("/explore") assert response.status_code == 200 # --------------------------------------------------------------------------- # JSON API tests — public browse endpoint (no auth) # --------------------------------------------------------------------------- async def test_list_public_repos_empty(client: AsyncClient, db_session: AsyncSession) -> None: """GET /api/discover/repos returns empty list when no public repos exist.""" response = await client.get("/api/discover/repos") assert response.status_code == 200 body = response.json() assert body["repos"] == [] assert body["total"] == 0 assert body["nextCursor"] is None async def test_explore_only_public_repos( client: AsyncClient, db_session: AsyncSession ) -> None: """Private repos must not appear in discover results.""" await _make_public_repo(db_session, name="public-one") await _make_private_repo(db_session, name="private-one") response = await client.get("/api/discover/repos") assert response.status_code == 200 body = response.json() assert body["total"] == 1 names = [r["name"] for r in body["repos"]] assert "public-one" in names assert "private-one" not in names async def test_explore_filters_by_genre( client: AsyncClient, db_session: AsyncSession ) -> None: """genre= filter returns only repos whose tags contain the genre string.""" await _make_public_repo(db_session, name="jazz-project", tags=["jazz", "swing"]) await _make_public_repo(db_session, name="lofi-project", tags=["lo-fi", "chill"]) response = await client.get("/api/discover/repos?genre=jazz") assert response.status_code == 200 body = response.json() assert body["total"] == 1 assert body["repos"][0]["name"] == "jazz-project" async def test_explore_filters_by_instrumentation( client: AsyncClient, db_session: AsyncSession ) -> None: """instrumentation= filter matches repos whose tags include the instrument.""" await _make_public_repo(db_session, name="bass-heavy", tags=["jazz", "bass", "drums"]) await _make_public_repo(db_session, name="keys-only", tags=["ambient", "keys"]) response = await client.get("/api/discover/repos?instrumentation=bass") assert response.status_code == 200 body = response.json() assert body["total"] == 1 assert body["repos"][0]["name"] == "bass-heavy" async def test_explore_sorts_by_created( client: AsyncClient, db_session: AsyncSession ) -> None: """sort=created returns newest repos first (default sort).""" await _make_public_repo(db_session, name="first-created") await _make_public_repo(db_session, name="second-created") response = await client.get("/api/discover/repos?sort=created") assert response.status_code == 200 body = response.json() # Newest first — second-created was inserted last names = [r["name"] for r in body["repos"]] assert names.index("second-created") < names.index("first-created") async def test_explore_pagination( client: AsyncClient, db_session: AsyncSession ) -> None: """Cursor pagination: second page returns a different set of repos than page 1.""" for i in range(5): await _make_public_repo(db_session, name=f"repo-{i:02d}") page1 = (await client.get("/api/discover/repos?limit=3")).json() assert page1["total"] == 5 assert page1["nextCursor"] is not None next_cursor = page1["nextCursor"] page2 = (await client.get(f"/api/discover/repos?limit=3&cursor={next_cursor}")).json() assert page2["total"] == 5 page1_ids = {r["repoId"] for r in page1["repos"]} page2_ids = {r["repoId"] for r in page2["repos"]} # Pages must not overlap assert not page1_ids & page2_ids async def test_explore_invalid_sort_returns_422( client: AsyncClient, db_session: AsyncSession ) -> None: """sort= with an invalid value returns 422 Unprocessable Entity.""" response = await client.get("/api/discover/repos?sort=invalid") assert response.status_code == 422