"""Tests for the fork-a-repo feature. Covers: POST /api/repos/{repo_id}/fork - Happy path: fork a public repo - 404 when source repo does not exist - 403 when source repo is private - 403 when caller tries to fork their own repo - 409 when caller has already forked the same repo - 401 when unauthenticated - Optional name / description / visibility fields GET /api/repos/{repo_id}/forks - Returns empty list when no forks exist - Returns all direct forks with source attribution - 404 when source repo does not exist - Public endpoint (no auth required) GET /api/repos/{repo_id}/fork-network - Returns root node with children - Total_forks count is correct - Public endpoint (no auth required) GET /api/users/{username}/forks - Returns empty list when user has no forks - Returns forks with source attribution after forking - 404 when username does not exist Service layer - fork_repo raises ValueError for business rule violations - get_user_forks returns real data after forks are created - list_repo_forks_flat returns real data after forks are created All tests use the shared ``client``, ``auth_headers``, ``test_user``, and ``db_session`` fixtures from conftest.py. """ from __future__ import annotations import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.db.musehub_models import MusehubIdentity, MusehubRepo from musehub.types.json_types import StrDict # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _TEST_HANDLE = "testuser" # matches conftest._TEST_HANDLE async def _create_public_repo( client: AsyncClient, auth_headers: StrDict, name: str = "upstream-beats", ) -> str: """Create a public repo via the API and return its repo_id.""" resp = await client.post( "/api/repos", json={"name": name, "owner": _TEST_HANDLE, "visibility": "public", "initialize": False}, headers=auth_headers, ) assert resp.status_code == 201, resp.text return str(resp.json()["repoId"]) async def _create_private_repo( client: AsyncClient, auth_headers: StrDict, name: str = "secret-project", ) -> str: """Create a private repo via the API and return its repo_id.""" resp = await client.post( "/api/repos", json={"name": name, "owner": _TEST_HANDLE, "visibility": "private", "initialize": False}, headers=auth_headers, ) assert resp.status_code == 201, resp.text return str(resp.json()["repoId"]) async def _seed_identity(db: AsyncSession, handle: str) -> MusehubIdentity: """Seed a secondary identity in the DB (simulating a different user).""" identity = MusehubIdentity( handle=handle, display_name=handle.title(), identity_type="human", ) db.add(identity) await db.commit() await db.refresh(identity) return identity # --------------------------------------------------------------------------- # POST /api/repos/{repo_id}/fork — happy path # --------------------------------------------------------------------------- async def test_fork_public_repo_returns_201( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Forking a public repo returns 201 with fork metadata.""" # Seed a public source repo owned by a different identity so the caller # can fork it. owner_identity = await _seed_identity(db_session, "alice") source = MusehubRepo( name="shared-beats", owner="alice", slug="shared-beats", visibility="public", owner_user_id=owner_identity.identity_id, description="Alice's public beats", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp.status_code == 201, resp.text body = resp.json() assert "forkId" in body assert body["sourceOwner"] == "alice" assert body["sourceSlug"] == "shared-beats" assert "forkRepo" in body fork_repo = body["forkRepo"] assert fork_repo["owner"] == _TEST_HANDLE assert fork_repo["visibility"] == "public" assert "forkedAt" in body async def test_fork_sets_description_with_attribution( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork description defaults to 'Fork of {owner}/{slug}: {source description}'.""" await _seed_identity(db_session, "bob") source = MusehubRepo( name="groove-box", owner="bob", slug="groove-box", visibility="public", owner_user_id="bob", description="Bob's groove box", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp.status_code == 201, resp.text description = resp.json()["forkRepo"]["description"] assert "bob" in description assert "groove-box" in description assert "Bob's groove box" in description async def test_fork_with_custom_name( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork accepts an optional custom name for the new repo.""" await _seed_identity(db_session, "carol") source = MusehubRepo( name="jazz-trio", owner="carol", slug="jazz-trio", visibility="public", owner_user_id="carol", description="Carol's jazz trio", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={"name": "my-jazz-experiment"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text fork_repo = resp.json()["forkRepo"] assert "jazz-experiment" in fork_repo["slug"] async def test_fork_with_private_visibility( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork accepts visibility='private' to create a private fork.""" await _seed_identity(db_session, "dave") source = MusehubRepo( name="open-source-beats", owner="dave", slug="open-source-beats", visibility="public", owner_user_id="dave", description="Dave's open source beats", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={"visibility": "private"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text assert resp.json()["forkRepo"]["visibility"] == "private" async def test_fork_with_custom_description( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork accepts a custom description that overrides the default attribution.""" await _seed_identity(db_session, "eve") source = MusehubRepo( name="synth-lab", owner="eve", slug="synth-lab", visibility="public", owner_user_id="eve", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={"description": "My custom synth fork"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text assert resp.json()["forkRepo"]["description"] == "My custom synth fork" # --------------------------------------------------------------------------- # POST /api/repos/{repo_id}/fork — error cases # --------------------------------------------------------------------------- async def test_fork_nonexistent_repo_returns_404( client: AsyncClient, auth_headers: StrDict, ) -> None: """Forking a repo that doesn't exist returns 404.""" resp = await client.post( "/api/repos/00000000-0000-0000-0000-000000000000/fork", json={}, headers=auth_headers, ) assert resp.status_code == 404 async def test_fork_private_repo_returns_403( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Forking a private repo returns 403.""" await _seed_identity(db_session, "frank") source = MusehubRepo( name="private-session", owner="frank", slug="private-session", visibility="private", owner_user_id="frank", description="Frank's private work", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp.status_code == 403 assert "public" in resp.json()["detail"].lower() async def test_fork_own_repo_returns_403( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Caller cannot fork a repository they already own — returns 403.""" source_id = await _create_public_repo(client, auth_headers, name="my-own-beats") resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp.status_code == 403 assert "own" in resp.json()["detail"].lower() async def test_fork_same_repo_twice_returns_409( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Forking the same repo twice returns 409 Conflict.""" await _seed_identity(db_session, "grace") source = MusehubRepo( name="shared-vibes", owner="grace", slug="shared-vibes", visibility="public", owner_user_id="grace", description="Grace's vibes", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # First fork succeeds resp1 = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp1.status_code == 201, resp1.text # Second fork of the same repo → 409 resp2 = await client.post( f"/api/repos/{source_id}/fork", json={"name": "another-fork"}, headers=auth_headers, ) assert resp2.status_code == 409 async def test_fork_requires_auth(client: AsyncClient, db_session: AsyncSession) -> None: """Unauthenticated fork request returns 401.""" await _seed_identity(db_session, "henry") source = MusehubRepo( name="open-beats", owner="henry", slug="open-beats", visibility="public", owner_user_id="henry", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post(f"/api/repos/{source_id}/fork", json={}) assert resp.status_code == 401 # --------------------------------------------------------------------------- # GET /api/repos/{repo_id}/forks # --------------------------------------------------------------------------- async def test_list_forks_empty_when_no_forks( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """A repo with no forks returns an empty list.""" await _seed_identity(db_session, "iris") source = MusehubRepo( name="unfork-able", owner="iris", slug="unfork-able", visibility="public", owner_user_id="iris", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.get(f"/api/repos/{source_id}/forks") assert resp.status_code == 200 body = resp.json() assert body["forks"] == [] assert body["total"] == 0 async def test_list_forks_shows_fork_after_creation( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """A fork appears in the list after being created.""" await _seed_identity(db_session, "jack") source = MusehubRepo( name="popular-track", owner="jack", slug="popular-track", visibility="public", owner_user_id="jack", description="Jack's popular track", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork it fork_resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert fork_resp.status_code == 201, fork_resp.text # List forks resp = await client.get(f"/api/repos/{source_id}/forks") assert resp.status_code == 200 body = resp.json() assert body["total"] == 1 fork = body["forks"][0] assert fork["sourceOwner"] == "jack" assert fork["sourceSlug"] == "popular-track" assert fork["forkRepo"]["owner"] == _TEST_HANDLE async def test_list_forks_no_auth_required( client: AsyncClient, db_session: AsyncSession, ) -> None: """List forks endpoint is publicly accessible without authentication.""" await _seed_identity(db_session, "kate") source = MusehubRepo( name="public-beats", owner="kate", slug="public-beats", visibility="public", owner_user_id="kate", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # No auth_headers passed resp = await client.get(f"/api/repos/{source_id}/forks") assert resp.status_code == 200 async def test_list_forks_returns_404_for_missing_repo( client: AsyncClient, ) -> None: """List forks for a non-existent repo returns 404.""" resp = await client.get("/api/repos/00000000-0000-0000-0000-000000000000/forks") assert resp.status_code == 404 # --------------------------------------------------------------------------- # GET /api/repos/{repo_id}/fork-network # --------------------------------------------------------------------------- async def test_fork_network_has_root_and_children( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork network returns root with forked repo as a child.""" await _seed_identity(db_session, "liam") source = MusehubRepo( name="groove-machine", owner="liam", slug="groove-machine", visibility="public", owner_user_id="liam", description="Liam's groove machine", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork it fork_resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert fork_resp.status_code == 201, fork_resp.text # Get fork network resp = await client.get(f"/api/repos/{source_id}/fork-network") assert resp.status_code == 200 body = resp.json() assert "root" in body assert body["totalForks"] == 1 root = body["root"] assert root["owner"] == "liam" assert root["repoSlug"] == "groove-machine" assert len(root["children"]) == 1 child = root["children"][0] assert child["owner"] == _TEST_HANDLE assert child["forkedBy"] == _TEST_HANDLE async def test_fork_network_empty_children_when_no_forks( client: AsyncClient, db_session: AsyncSession, ) -> None: """Fork network for a repo with no forks has an empty children list.""" await _seed_identity(db_session, "mia") source = MusehubRepo( name="solo-track", owner="mia", slug="solo-track", visibility="public", owner_user_id="mia", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.get(f"/api/repos/{source_id}/fork-network") assert resp.status_code == 200 body = resp.json() assert body["totalForks"] == 0 assert body["root"]["children"] == [] async def test_fork_network_returns_404_for_missing_repo( client: AsyncClient, ) -> None: """Fork network for a non-existent repo returns 404.""" resp = await client.get("/api/repos/00000000-0000-0000-0000-000000000000/fork-network") assert resp.status_code == 404 async def test_fork_network_no_auth_required( client: AsyncClient, db_session: AsyncSession, ) -> None: """Fork network endpoint is publicly accessible without authentication.""" await _seed_identity(db_session, "noah") source = MusehubRepo( name="collab-beats", owner="noah", slug="collab-beats", visibility="public", owner_user_id="noah", description="", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.get(f"/api/repos/{source_id}/fork-network") assert resp.status_code == 200 # --------------------------------------------------------------------------- # GET /api/users/{username}/forks # --------------------------------------------------------------------------- async def test_get_user_forks_empty_for_new_user( client: AsyncClient, test_user: MusehubIdentity, ) -> None: """A user with no forks returns an empty list.""" resp = await client.get(f"/api/users/{_TEST_HANDLE}/forks") assert resp.status_code == 200 body = resp.json() assert body["forks"] == [] assert body["total"] == 0 async def test_get_user_forks_shows_fork_after_creation( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, test_user: MusehubIdentity, ) -> None: """User's forks list is populated after forking a repo.""" await _seed_identity(db_session, "olivia") source = MusehubRepo( name="soul-session", owner="olivia", slug="soul-session", visibility="public", owner_user_id="olivia", description="Olivia's soul session", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork it fork_resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert fork_resp.status_code == 201, fork_resp.text # Get user forks resp = await client.get(f"/api/users/{_TEST_HANDLE}/forks") assert resp.status_code == 200 body = resp.json() assert body["total"] == 1 entry = body["forks"][0] assert entry["sourceOwner"] == "olivia" assert entry["sourceSlug"] == "soul-session" assert entry["forkRepo"]["owner"] == _TEST_HANDLE assert "forkId" in entry assert "forkedAt" in entry async def test_get_user_forks_404_for_unknown_user( client: AsyncClient, ) -> None: """Requesting forks for an unknown user returns 404.""" resp = await client.get("/api/users/nonexistent-user-xyz/forks") assert resp.status_code == 404 async def test_get_user_forks_no_auth_required( client: AsyncClient, test_user: MusehubIdentity, ) -> None: """User forks endpoint is publicly accessible without authentication.""" resp = await client.get(f"/api/users/{_TEST_HANDLE}/forks") assert resp.status_code == 200 # --------------------------------------------------------------------------- # Fork response shape # --------------------------------------------------------------------------- async def test_fork_response_contains_all_required_fields( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork creation response contains all documented fields.""" await _seed_identity(db_session, "peter") source = MusehubRepo( name="field-check-beats", owner="peter", slug="field-check-beats", visibility="public", owner_user_id="peter", description="Peter's beats", tags=["jazz", "soul"], ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert resp.status_code == 201, resp.text body = resp.json() # Top-level fork entry fields for field in ("forkId", "forkRepo", "sourceOwner", "sourceSlug", "forkedAt"): assert field in body, f"Missing field: {field}" # Fork repo fields fork_repo = body["forkRepo"] for field in ("repoId", "name", "owner", "slug", "visibility", "description", "tags", "createdAt"): assert field in fork_repo, f"Missing forkRepo field: {field}" # Tags are copied from source assert "jazz" in fork_repo["tags"] assert "soul" in fork_repo["tags"] # --------------------------------------------------------------------------- # Multiple forks of the same source # --------------------------------------------------------------------------- async def test_multiple_forks_appear_in_list( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Multiple forks by different users all appear in the source repo's fork list.""" await _seed_identity(db_session, "quinn") source = MusehubRepo( name="viral-track", owner="quinn", slug="viral-track", visibility="public", owner_user_id="quinn", description="Quinn's viral track", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # testuser forks it fork_resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert fork_resp.status_code == 201, fork_resp.text # Seed a second forker via direct DB insert (bypasses auth) second_forker = await _seed_identity(db_session, "rachel") fork_repo_2 = MusehubRepo( name="viral-track", owner="rachel", slug="viral-track", visibility="public", owner_user_id="rachel", description="Fork of quinn/viral-track: Quinn's viral track", ) db_session.add(fork_repo_2) await db_session.commit() await db_session.refresh(fork_repo_2) from musehub.db.musehub_models import MusehubFork fork_record = MusehubFork( source_repo_id=source_id, fork_repo_id=fork_repo_2.repo_id, forked_by="rachel", ) db_session.add(fork_record) await db_session.commit() resp = await client.get(f"/api/repos/{source_id}/forks") assert resp.status_code == 200 body = resp.json() assert body["total"] == 2 owners = {f["forkRepo"]["owner"] for f in body["forks"]} assert _TEST_HANDLE in owners assert "rachel" in owners # --------------------------------------------------------------------------- # Private fork visibility — security hardening # --------------------------------------------------------------------------- async def test_private_fork_hidden_from_source_forks_list( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """A private fork must NOT appear in the public GET /repos/{id}/forks list.""" await _seed_identity(db_session, "sam") source = MusehubRepo( name="secret-upstream", owner="sam", slug="secret-upstream", visibility="public", owner_user_id="sam", description="Sam's upstream", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork with private visibility resp = await client.post( f"/api/repos/{source_id}/fork", json={"visibility": "private"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text # Public listing must be empty — the fork is private list_resp = await client.get(f"/api/repos/{source_id}/forks") assert list_resp.status_code == 200 body = list_resp.json() assert body["total"] == 0 assert body["forks"] == [] async def test_private_fork_hidden_from_fork_network( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """A private fork must NOT appear in the public fork-network tree.""" await _seed_identity(db_session, "tara") source = MusehubRepo( name="silent-upstream", owner="tara", slug="silent-upstream", visibility="public", owner_user_id="tara", description="Tara's upstream", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork with private visibility resp = await client.post( f"/api/repos/{source_id}/fork", json={"visibility": "private"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text # Fork network must show 0 forks — private fork is not in tree net_resp = await client.get(f"/api/repos/{source_id}/fork-network") assert net_resp.status_code == 200 body = net_resp.json() assert body["totalForks"] == 0 assert body["root"]["children"] == [] async def test_private_fork_hidden_from_public_user_forks( client: AsyncClient, test_user: MusehubIdentity, db_session: AsyncSession, ) -> None: """A private fork must NOT appear when an unauthenticated caller views a user's forks. Note: this test does NOT request the ``auth_headers`` fixture because that fixture globally overrides ``optional_signed_request`` to return the test context, making every request in the test look authenticated. Instead we seed the fork directly in the DB so we can make a genuinely anonymous call. """ await _seed_identity(db_session, "uma") source = MusehubRepo( name="covert-upstream", owner="uma", slug="covert-upstream", visibility="public", owner_user_id="uma", description="Uma's upstream", ) fork_repo = MusehubRepo( name="covert-upstream", owner=_TEST_HANDLE, slug="covert-upstream", visibility="private", # private fork owner_user_id=_TEST_HANDLE, description="Fork of uma/covert-upstream: Uma's upstream", ) db_session.add(source) db_session.add(fork_repo) await db_session.commit() await db_session.refresh(source) await db_session.refresh(fork_repo) from musehub.db.musehub_models import MusehubFork fork_record = MusehubFork( source_repo_id=str(source.repo_id), fork_repo_id=str(fork_repo.repo_id), forked_by=_TEST_HANDLE, ) db_session.add(fork_record) await db_session.commit() # Genuinely unauthenticated GET — no auth_headers fixture, no dep override anon_resp = await client.get(f"/api/users/{_TEST_HANDLE}/forks") assert anon_resp.status_code == 200 body = anon_resp.json() assert body["total"] == 0 assert body["forks"] == [] async def test_private_fork_visible_to_owner( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """The fork owner can see their own private fork via the authenticated forks endpoint.""" await _seed_identity(db_session, "vera") source = MusehubRepo( name="owner-visible-upstream", owner="vera", slug="owner-visible-upstream", visibility="public", owner_user_id="vera", description="Vera's upstream", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork with private visibility resp = await client.post( f"/api/repos/{source_id}/fork", json={"visibility": "private"}, headers=auth_headers, ) assert resp.status_code == 201, resp.text # Authenticated as owner — private fork IS visible auth_resp = await client.get(f"/api/users/{_TEST_HANDLE}/forks", headers=auth_headers) assert auth_resp.status_code == 200 body = auth_resp.json() assert body["total"] == 1 assert body["forks"][0]["forkRepo"]["visibility"] == "private" async def test_invalid_visibility_returns_422( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Fork request with invalid visibility value returns 422 Unprocessable Entity.""" await _seed_identity(db_session, "walter") source = MusehubRepo( name="valid-upstream", owner="walter", slug="valid-upstream", visibility="public", owner_user_id="walter", description="Walter's upstream", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) resp = await client.post( f"/api/repos/{source_id}/fork", json={"visibility": "superadmin"}, headers=auth_headers, ) assert resp.status_code == 422 async def test_slug_collision_auto_resolved( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Forking when the caller already owns a repo with the same name auto-suffixes the slug.""" # testuser already owns a repo with the same name as the source existing_resp = await client.post( "/api/repos", json={"name": "classic-track", "owner": _TEST_HANDLE, "visibility": "public", "initialize": False}, headers=auth_headers, ) assert existing_resp.status_code == 201, existing_resp.text await _seed_identity(db_session, "xavier") source = MusehubRepo( name="classic-track", # same name as testuser's existing repo owner="xavier", slug="classic-track", visibility="public", owner_user_id="xavier", description="Xavier's classic track", ) db_session.add(source) await db_session.commit() await db_session.refresh(source) source_id = str(source.repo_id) # Fork should succeed despite slug collision — gets auto-suffixed slug fork_resp = await client.post( f"/api/repos/{source_id}/fork", json={}, headers=auth_headers, ) assert fork_resp.status_code == 201, fork_resp.text fork_slug = fork_resp.json()["forkRepo"]["slug"] # Slug must differ from the existing one (auto-suffixed) assert fork_slug == "classic-track-2"