Enforce run ownership and show owner info #74
@@ -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
|
||||||
@@ -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)
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user