2 Commits

Author SHA1 Message Date
596393d5b8 Merge pull request 'Infer genlocke visibility from first leg's run' (#77) from feature/infer-genlocke-visibility-from-first-legs-run into feature/enforce-run-ownership-on-all-mutation-endpoints
All checks were successful
CI / backend-tests (pull_request) Successful in 29s
CI / frontend-tests (pull_request) Successful in 29s
Reviewed-on: #77
2026-03-22 09:15:16 +01:00
a4fa5bf1e4 feat: infer genlocke visibility from first leg's run
Genlockes now inherit visibility from their first leg's run:
- Private runs make genlockes hidden from public listings
- All genlocke read endpoints now accept optional auth
- Returns 404 for private genlockes to non-owners

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 13:47:05 +01:00
3 changed files with 383 additions and 41 deletions

View File

@@ -0,0 +1,54 @@
---
# nuzlocke-tracker-i0rn
title: Infer genlocke visibility from first leg's run
status: completed
type: feature
created_at: 2026-03-21T12:46:56Z
updated_at: 2026-03-21T12:46:56Z
---
## Problem
Genlockes are always public — they have no visibility setting. They should inherit visibility from their first leg's run, so if a user makes their run private, the genlocke is also hidden from public listings.
## Approach
Rather than adding a `visibility` column to the `genlockes` table, infer it from the first leg's run at query time. This avoids sync issues and keeps the first leg's run as the source of truth.
### Backend
- `list_genlockes` endpoint: filter out genlockes whose first leg's run is private (unless the requesting user is the owner)
- `get_genlocke` endpoint: return 404 if the first leg's run is private and the user is not the owner
- Add optional auth (not required) to genlocke read endpoints to check ownership
### Frontend
- No changes needed — private genlockes simply won't appear in listings for non-owners
## Files modified
- `backend/src/app/api/genlockes.py` — add visibility filtering to all read endpoints
## Checklist
- [x] Add `get_current_user` (optional auth) dependency to genlocke read endpoints
- [x] Filter private genlockes from `list_genlockes` for non-owners
- [x] Return 404 for private genlockes in `get_genlocke` for non-owners
- [x] Apply same filtering to graveyard, lineages, survivors, and retired-families endpoints
- [x] Test: private run's genlocke hidden from unauthenticated users
- [x] Test: owner can still see their private genlocke
## Summary of Changes
- Added `_is_genlocke_visible()` helper function to check visibility based on first leg's run
- Added optional auth (`get_current_user`) to all genlocke read endpoints:
- `list_genlockes`: filters out private genlockes for non-owners
- `get_genlocke`: returns 404 for private genlockes to non-owners
- `get_genlocke_graveyard`: returns 404 for private genlockes
- `get_genlocke_lineages`: returns 404 for private genlockes
- `get_leg_survivors`: returns 404 for private genlockes
- `get_retired_families`: returns 404 for private genlockes
- Added 9 new tests in `TestGenlockeVisibility` class covering:
- Private genlockes hidden from unauthenticated list
- Private genlockes visible to owner in list
- 404 for all detail endpoints when accessed by unauthenticated users
- 404 for private genlockes when accessed by different authenticated user
- Owner can still access their private genlocke

View File

@@ -8,14 +8,14 @@ from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload from sqlalchemy.orm import selectinload
from app.core.auth import AuthUser, require_auth, require_run_owner from app.core.auth import AuthUser, get_current_user, require_auth, require_run_owner
from app.core.database import get_session from app.core.database import get_session
from app.models.encounter import Encounter from app.models.encounter import Encounter
from app.models.evolution import Evolution from app.models.evolution import Evolution
from app.models.game import Game from app.models.game import Game
from app.models.genlocke import Genlocke, GenlockeLeg from app.models.genlocke import Genlocke, GenlockeLeg
from app.models.genlocke_transfer import GenlockeTransfer from app.models.genlocke_transfer import GenlockeTransfer
from app.models.nuzlocke_run import NuzlockeRun from app.models.nuzlocke_run import NuzlockeRun, RunVisibility
from app.models.pokemon import Pokemon from app.models.pokemon import Pokemon
from app.models.route import Route from app.models.route import Route
from app.models.user import User from app.models.user import User
@@ -77,8 +77,34 @@ async def _check_genlocke_owner(
require_run_owner(first_leg.run, user) require_run_owner(first_leg.run, user)
def _is_genlocke_visible(genlocke: Genlocke, user: AuthUser | None) -> bool:
"""
Check if a genlocke is visible to the given user.
Visibility is inferred from the first leg's run:
- Public runs are visible to everyone
- Private runs are only visible to the owner
"""
first_leg = next((leg for leg in genlocke.legs if leg.leg_order == 1), None)
if not first_leg or not first_leg.run:
# No first leg or run - treat as visible (legacy data)
return True
if first_leg.run.visibility == RunVisibility.PUBLIC:
return True
# Private run - only visible to owner
if user is None:
return False
if first_leg.run.owner_id is None:
return False
return str(first_leg.run.owner_id) == user.id
@router.get("", response_model=list[GenlockeListItem]) @router.get("", response_model=list[GenlockeListItem])
async def list_genlockes(session: AsyncSession = Depends(get_session)): async def list_genlockes(
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
result = await session.execute( result = await session.execute(
select(Genlocke) select(Genlocke)
.options( .options(
@@ -92,6 +118,10 @@ async def list_genlockes(session: AsyncSession = Depends(get_session)):
items = [] items = []
for g in genlockes: for g in genlockes:
# Filter out private genlockes for non-owners
if not _is_genlocke_visible(g, user):
continue
completed_legs = 0 completed_legs = 0
current_leg_order = None current_leg_order = None
owner = None owner = None
@@ -126,7 +156,11 @@ async def list_genlockes(session: AsyncSession = Depends(get_session)):
@router.get("/{genlocke_id}", response_model=GenlockeDetailResponse) @router.get("/{genlocke_id}", response_model=GenlockeDetailResponse)
async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_session)): async def get_genlocke(
genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
result = await session.execute( result = await session.execute(
select(Genlocke) select(Genlocke)
.where(Genlocke.id == genlocke_id) .where(Genlocke.id == genlocke_id)
@@ -139,6 +173,10 @@ async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_ses
if genlocke is None: if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found") raise HTTPException(status_code=404, detail="Genlocke not found")
# Check visibility - return 404 for private genlockes to non-owners
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Collect run IDs for aggregate query # Collect run IDs for aggregate query
run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None] run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None]
@@ -222,20 +260,26 @@ async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_ses
response_model=GenlockeGraveyardResponse, response_model=GenlockeGraveyardResponse,
) )
async def get_genlocke_graveyard( async def get_genlocke_graveyard(
genlocke_id: int, session: AsyncSession = Depends(get_session) genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
): ):
# Load genlocke with legs + game # Load genlocke with legs + game + run (for visibility check)
result = await session.execute( result = await session.execute(
select(Genlocke) select(Genlocke)
.where(Genlocke.id == genlocke_id) .where(Genlocke.id == genlocke_id)
.options( .options(
selectinload(Genlocke.legs).selectinload(GenlockeLeg.game), selectinload(Genlocke.legs).selectinload(GenlockeLeg.game),
selectinload(Genlocke.legs).selectinload(GenlockeLeg.run),
) )
) )
genlocke = result.scalar_one_or_none() genlocke = result.scalar_one_or_none()
if genlocke is None: if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found") raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Build run_id → (leg_order, game_name) lookup # Build run_id → (leg_order, game_name) lookup
run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None] run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None]
run_lookup: dict[int, tuple[int, str]] = {} run_lookup: dict[int, tuple[int, str]] = {}
@@ -323,20 +367,26 @@ async def get_genlocke_graveyard(
response_model=GenlockeLineageResponse, response_model=GenlockeLineageResponse,
) )
async def get_genlocke_lineages( async def get_genlocke_lineages(
genlocke_id: int, session: AsyncSession = Depends(get_session) genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
): ):
# Load genlocke with legs + game # Load genlocke with legs + game + run (for visibility check)
result = await session.execute( result = await session.execute(
select(Genlocke) select(Genlocke)
.where(Genlocke.id == genlocke_id) .where(Genlocke.id == genlocke_id)
.options( .options(
selectinload(Genlocke.legs).selectinload(GenlockeLeg.game), selectinload(Genlocke.legs).selectinload(GenlockeLeg.game),
selectinload(Genlocke.legs).selectinload(GenlockeLeg.run),
) )
) )
genlocke = result.scalar_one_or_none() genlocke = result.scalar_one_or_none()
if genlocke is None: if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found") raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Query all transfers for this genlocke # Query all transfers for this genlocke
transfer_result = await session.execute( transfer_result = await session.execute(
select(GenlockeTransfer).where(GenlockeTransfer.genlocke_id == genlocke_id) select(GenlockeTransfer).where(GenlockeTransfer.genlocke_id == genlocke_id)
@@ -570,15 +620,23 @@ async def get_leg_survivors(
genlocke_id: int, genlocke_id: int,
leg_order: int, leg_order: int,
session: AsyncSession = Depends(get_session), session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
): ):
# Find the leg # Load genlocke with legs + run for visibility check
result = await session.execute( genlocke_result = await session.execute(
select(GenlockeLeg).where( select(Genlocke)
GenlockeLeg.genlocke_id == genlocke_id, .where(Genlocke.id == genlocke_id)
GenlockeLeg.leg_order == leg_order, .options(selectinload(Genlocke.legs).selectinload(GenlockeLeg.run))
)
) )
leg = result.scalar_one_or_none() genlocke = genlocke_result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Find the leg
leg = next((leg for leg in genlocke.legs if leg.leg_order == leg_order), None)
if leg is None: if leg is None:
raise HTTPException(status_code=404, detail="Leg not found") raise HTTPException(status_code=404, detail="Leg not found")
@@ -846,12 +904,21 @@ class RetiredFamiliesResponse(BaseModel):
async def get_retired_families( async def get_retired_families(
genlocke_id: int, genlocke_id: int,
session: AsyncSession = Depends(get_session), session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
): ):
# Verify genlocke exists # Load genlocke with legs + run for visibility check
genlocke = await session.get(Genlocke, genlocke_id) result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(selectinload(Genlocke.legs).selectinload(GenlockeLeg.run))
)
genlocke = result.scalar_one_or_none()
if genlocke is None: if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found") raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Query all legs with retired_pokemon_ids # Query all legs with retired_pokemon_ids
result = await session.execute( result = await session.execute(
select(GenlockeLeg) select(GenlockeLeg)

View File

@@ -1,10 +1,13 @@
"""Integration tests for the Genlockes & Bosses API.""" """Integration tests for the Genlockes & Bosses API."""
import pytest import pytest
from httpx import AsyncClient from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.core.auth import AuthUser, get_current_user
from app.main import app
from app.models.game import Game from app.models.game import Game
from app.models.nuzlocke_run import NuzlockeRun, RunVisibility
from app.models.pokemon import Pokemon from app.models.pokemon import Pokemon
from app.models.route import Route from app.models.route import Route
from app.models.version_group import VersionGroup from app.models.version_group import VersionGroup
@@ -55,7 +58,9 @@ async def games_ctx(db_session: AsyncSession) -> dict:
@pytest.fixture @pytest.fixture
async def ctx(db_session: AsyncSession, admin_client: AsyncClient, games_ctx: dict) -> dict: async def ctx(
db_session: AsyncSession, admin_client: AsyncClient, games_ctx: dict
) -> dict:
"""Full context: routes + pokemon + genlocke + encounter for advance/transfer tests.""" """Full context: routes + pokemon + genlocke + encounter for advance/transfer tests."""
route1 = Route(name="GT Route 1", version_group_id=games_ctx["vg1_id"], order=1) route1 = Route(name="GT Route 1", version_group_id=games_ctx["vg1_id"], order=1)
route2 = Route(name="GT Route 2", version_group_id=games_ctx["vg2_id"], order=1) route2 = Route(name="GT Route 2", version_group_id=games_ctx["vg2_id"], order=1)
@@ -116,6 +121,178 @@ class TestListGenlockes:
assert "Test Genlocke" in names assert "Test Genlocke" in names
# ---------------------------------------------------------------------------
# Genlockes — visibility (inferred from first leg's run)
# ---------------------------------------------------------------------------
class TestGenlockeVisibility:
"""Test that genlocke visibility is inferred from the first leg's run."""
@pytest.fixture
async def private_genlocke_ctx(
self, db_session: AsyncSession, admin_client: AsyncClient, games_ctx: dict
) -> dict:
"""Create a genlocke and make its first leg's run private."""
r = await admin_client.post(
GENLOCKES_BASE,
json={
"name": "Private Genlocke",
"gameIds": [games_ctx["game1_id"], games_ctx["game2_id"]],
},
)
assert r.status_code == 201
genlocke = r.json()
leg1 = next(leg for leg in genlocke["legs"] if leg["legOrder"] == 1)
run_id = leg1["runId"]
# Make the run private
run = await db_session.get(NuzlockeRun, run_id)
assert run is not None
run.visibility = RunVisibility.PRIVATE
await db_session.commit()
return {
**games_ctx,
"genlocke_id": genlocke["id"],
"run_id": run_id,
"owner_id": str(run.owner_id),
}
async def test_private_genlocke_hidden_from_unauthenticated_list(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should not see private genlockes in the list."""
# Temporarily remove auth override to simulate unauthenticated request
app.dependency_overrides.pop(get_current_user, None)
try:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(GENLOCKES_BASE)
assert response.status_code == 200
names = [g["name"] for g in response.json()]
assert "Private Genlocke" not in names
finally:
pass
async def test_private_genlocke_visible_to_owner_in_list(
self, admin_client: AsyncClient, private_genlocke_ctx: dict
):
"""Owner should still see their private genlocke in the list."""
response = await admin_client.get(GENLOCKES_BASE)
assert response.status_code == 200
names = [g["name"] for g in response.json()]
assert "Private Genlocke" in names
async def test_private_genlocke_404_for_unauthenticated_get(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should get 404 for private genlocke details."""
app.dependency_overrides.pop(get_current_user, None)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}"
)
assert response.status_code == 404
async def test_private_genlocke_accessible_to_owner(
self, admin_client: AsyncClient, private_genlocke_ctx: dict
):
"""Owner should still be able to access their private genlocke."""
response = await admin_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}"
)
assert response.status_code == 200
assert response.json()["name"] == "Private Genlocke"
async def test_private_genlocke_graveyard_404_for_unauthenticated(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should get 404 for private genlocke graveyard."""
app.dependency_overrides.pop(get_current_user, None)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}/graveyard"
)
assert response.status_code == 404
async def test_private_genlocke_lineages_404_for_unauthenticated(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should get 404 for private genlocke lineages."""
app.dependency_overrides.pop(get_current_user, None)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}/lineages"
)
assert response.status_code == 404
async def test_private_genlocke_survivors_404_for_unauthenticated(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should get 404 for private genlocke survivors."""
app.dependency_overrides.pop(get_current_user, None)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}/legs/1/survivors"
)
assert response.status_code == 404
async def test_private_genlocke_retired_families_404_for_unauthenticated(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""Unauthenticated users should get 404 for private retired-families."""
app.dependency_overrides.pop(get_current_user, None)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as unauth_client:
response = await unauth_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}/retired-families"
)
assert response.status_code == 404
async def test_private_genlocke_404_for_different_user(
self, db_session: AsyncSession, private_genlocke_ctx: dict
):
"""A different authenticated user should get 404 for private genlockes."""
# Create a different user's auth
different_user = AuthUser(
id="00000000-0000-4000-a000-000000000099",
email="other@example.com",
role="authenticated",
)
def _override():
return different_user
app.dependency_overrides[get_current_user] = _override
try:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as other_client:
response = await other_client.get(
f"{GENLOCKES_BASE}/{private_genlocke_ctx['genlocke_id']}"
)
assert response.status_code == 404
# Also check list
list_response = await other_client.get(GENLOCKES_BASE)
assert list_response.status_code == 200
names = [g["name"] for g in list_response.json()]
assert "Private Genlocke" not in names
finally:
app.dependency_overrides.pop(get_current_user, None)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Genlockes — create # Genlockes — create
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -259,14 +436,18 @@ class TestGenlockeLegs:
class TestAdvanceLeg: class TestAdvanceLeg:
async def test_uncompleted_run_returns_400(self, admin_client: AsyncClient, ctx: dict): async def test_uncompleted_run_returns_400(
self, admin_client: AsyncClient, ctx: dict
):
"""Cannot advance when leg 1's run is still active.""" """Cannot advance when leg 1's run is still active."""
response = await admin_client.post( response = await admin_client.post(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance" f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance"
) )
assert response.status_code == 400 assert response.status_code == 400
async def test_no_next_leg_returns_400(self, admin_client: AsyncClient, games_ctx: dict): async def test_no_next_leg_returns_400(
self, admin_client: AsyncClient, games_ctx: dict
):
"""A single-leg genlocke cannot be advanced.""" """A single-leg genlocke cannot be advanced."""
r = await admin_client.post( r = await admin_client.post(
GENLOCKES_BASE, GENLOCKES_BASE,
@@ -283,7 +464,9 @@ class TestAdvanceLeg:
async def test_advances_to_next_leg(self, admin_client: AsyncClient, ctx: dict): async def test_advances_to_next_leg(self, admin_client: AsyncClient, ctx: dict):
"""Completing the current run allows advancing to the next leg.""" """Completing the current run allows advancing to the next leg."""
await admin_client.patch(f"{RUNS_BASE}/{ctx['run_id']}", json={"status": "completed"}) await admin_client.patch(
f"{RUNS_BASE}/{ctx['run_id']}", json={"status": "completed"}
)
response = await admin_client.post( response = await admin_client.post(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance" f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance"
@@ -295,7 +478,9 @@ class TestAdvanceLeg:
async def test_advances_with_transfers(self, admin_client: AsyncClient, ctx: dict): async def test_advances_with_transfers(self, admin_client: AsyncClient, ctx: dict):
"""Advancing with transfer_encounter_ids creates egg encounters in the next leg.""" """Advancing with transfer_encounter_ids creates egg encounters in the next leg."""
await admin_client.patch(f"{RUNS_BASE}/{ctx['run_id']}", json={"status": "completed"}) await admin_client.patch(
f"{RUNS_BASE}/{ctx['run_id']}", json={"status": "completed"}
)
response = await admin_client.post( response = await admin_client.post(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance", f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/1/advance",
@@ -319,30 +504,40 @@ class TestAdvanceLeg:
class TestGenlockeGraveyard: class TestGenlockeGraveyard:
async def test_returns_empty_graveyard(self, admin_client: AsyncClient, ctx: dict): async def test_returns_empty_graveyard(self, admin_client: AsyncClient, ctx: dict):
response = await admin_client.get(f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/graveyard") response = await admin_client.get(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/graveyard"
)
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["entries"] == [] assert data["entries"] == []
assert data["totalDeaths"] == 0 assert data["totalDeaths"] == 0
async def test_not_found_returns_404(self, admin_client: AsyncClient): async def test_not_found_returns_404(self, admin_client: AsyncClient):
assert (await admin_client.get(f"{GENLOCKES_BASE}/9999/graveyard")).status_code == 404 assert (
await admin_client.get(f"{GENLOCKES_BASE}/9999/graveyard")
).status_code == 404
class TestGenlockeLineages: class TestGenlockeLineages:
async def test_returns_empty_lineages(self, admin_client: AsyncClient, ctx: dict): async def test_returns_empty_lineages(self, admin_client: AsyncClient, ctx: dict):
response = await admin_client.get(f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/lineages") response = await admin_client.get(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/lineages"
)
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["lineages"] == [] assert data["lineages"] == []
assert data["totalLineages"] == 0 assert data["totalLineages"] == 0
async def test_not_found_returns_404(self, admin_client: AsyncClient): async def test_not_found_returns_404(self, admin_client: AsyncClient):
assert (await admin_client.get(f"{GENLOCKES_BASE}/9999/lineages")).status_code == 404 assert (
await admin_client.get(f"{GENLOCKES_BASE}/9999/lineages")
).status_code == 404
class TestGenlockeRetiredFamilies: class TestGenlockeRetiredFamilies:
async def test_returns_empty_retired_families(self, admin_client: AsyncClient, ctx: dict): async def test_returns_empty_retired_families(
self, admin_client: AsyncClient, ctx: dict
):
response = await admin_client.get( response = await admin_client.get(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/retired-families" f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/retired-families"
) )
@@ -365,9 +560,13 @@ class TestLegSurvivors:
assert response.status_code == 200 assert response.status_code == 200
assert len(response.json()) == 1 assert len(response.json()) == 1
async def test_leg_not_found_returns_404(self, admin_client: AsyncClient, ctx: dict): async def test_leg_not_found_returns_404(
self, admin_client: AsyncClient, ctx: dict
):
assert ( assert (
await admin_client.get(f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/99/survivors") await admin_client.get(
f"{GENLOCKES_BASE}/{ctx['genlocke_id']}/legs/99/survivors"
)
).status_code == 404 ).status_code == 404
@@ -386,7 +585,9 @@ BOSS_PAYLOAD = {
class TestBossCRUD: class TestBossCRUD:
async def test_empty_list(self, admin_client: AsyncClient, games_ctx: dict): async def test_empty_list(self, admin_client: AsyncClient, games_ctx: dict):
response = await admin_client.get(f"{GAMES_BASE}/{games_ctx['game1_id']}/bosses") response = await admin_client.get(
f"{GAMES_BASE}/{games_ctx['game1_id']}/bosses"
)
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == [] assert response.json() == []
@@ -441,7 +642,9 @@ class TestBossCRUD:
async def test_invalid_game_returns_404(self, admin_client: AsyncClient): async def test_invalid_game_returns_404(self, admin_client: AsyncClient):
assert (await admin_client.get(f"{GAMES_BASE}/9999/bosses")).status_code == 404 assert (await admin_client.get(f"{GAMES_BASE}/9999/bosses")).status_code == 404
async def test_game_without_version_group_returns_400(self, admin_client: AsyncClient): async def test_game_without_version_group_returns_400(
self, admin_client: AsyncClient
):
game = ( game = (
await admin_client.post( await admin_client.post(
GAMES_BASE, GAMES_BASE,
@@ -480,7 +683,9 @@ class TestBossResults:
return {"boss_id": boss["id"], "run_id": run["id"]} return {"boss_id": boss["id"], "run_id": run["id"]}
async def test_empty_list(self, admin_client: AsyncClient, boss_ctx: dict): async def test_empty_list(self, admin_client: AsyncClient, boss_ctx: dict):
response = await admin_client.get(f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results") response = await admin_client.get(
f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results"
)
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == [] assert response.json() == []
@@ -495,7 +700,9 @@ class TestBossResults:
assert data["attempts"] == 1 assert data["attempts"] == 1
assert data["completedAt"] is not None assert data["completedAt"] is not None
async def test_upserts_existing_result(self, admin_client: AsyncClient, boss_ctx: dict): async def test_upserts_existing_result(
self, admin_client: AsyncClient, boss_ctx: dict
):
"""POSTing the same boss twice updates the result (upsert).""" """POSTing the same boss twice updates the result (upsert)."""
await admin_client.post( await admin_client.post(
f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results", f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results",
@@ -530,10 +737,16 @@ class TestBossResults:
await admin_client.get(f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results") await admin_client.get(f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results")
).json() == [] ).json() == []
async def test_invalid_run_returns_404(self, admin_client: AsyncClient, boss_ctx: dict): async def test_invalid_run_returns_404(
assert (await admin_client.get(f"{RUNS_BASE}/9999/boss-results")).status_code == 404 self, admin_client: AsyncClient, boss_ctx: dict
):
assert (
await admin_client.get(f"{RUNS_BASE}/9999/boss-results")
).status_code == 404
async def test_invalid_boss_returns_404(self, admin_client: AsyncClient, boss_ctx: dict): async def test_invalid_boss_returns_404(
self, admin_client: AsyncClient, boss_ctx: dict
):
response = await admin_client.post( response = await admin_client.post(
f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results", f"{RUNS_BASE}/{boss_ctx['run_id']}/boss-results",
json={"bossBattleId": 9999, "result": "won"}, json={"bossBattleId": 9999, "result": "won"},
@@ -587,8 +800,16 @@ class TestExport:
assert response.status_code == 200 assert response.status_code == 200
assert isinstance(response.json(), list) assert isinstance(response.json(), list)
async def test_export_game_routes_not_found_returns_404(self, admin_client: AsyncClient): async def test_export_game_routes_not_found_returns_404(
assert (await admin_client.get(f"{EXPORT_BASE}/games/9999/routes")).status_code == 404 self, admin_client: AsyncClient
):
assert (
await admin_client.get(f"{EXPORT_BASE}/games/9999/routes")
).status_code == 404
async def test_export_game_bosses_not_found_returns_404(self, admin_client: AsyncClient): async def test_export_game_bosses_not_found_returns_404(
assert (await admin_client.get(f"{EXPORT_BASE}/games/9999/bosses")).status_code == 404 self, admin_client: AsyncClient
):
assert (
await admin_client.get(f"{EXPORT_BASE}/games/9999/bosses")
).status_code == 404