150 lines
5.0 KiB
Python
150 lines
5.0 KiB
Python
"""Model-level integration tests for Discovery metadata and WorldVisit.
|
|
|
|
docs/planning/discovery.md (task b377353b8167).
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.config import settings
|
|
from app.main import app
|
|
from app.models.user import User
|
|
from app.models.world import World, WorldVisibility
|
|
from app.models.world_visit import WorldVisit
|
|
|
|
|
|
def _register(client: TestClient, username: str) -> int:
|
|
resp = client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"username": username,
|
|
"email": f"{username}@example.com",
|
|
"password": "secret123",
|
|
"date_of_birth": "1990-01-01",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
headers = {"Authorization": f"Bearer {resp.json()['access_token']}"}
|
|
me = client.get("/api/users/me", headers=headers)
|
|
assert me.status_code == 200, me.text
|
|
return me.json()["id"]
|
|
|
|
|
|
def _session_factory():
|
|
engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
|
|
return engine, async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_world_discovery_metadata_and_visits():
|
|
client = TestClient(app)
|
|
user_id = _register(client, "discoverer_1")
|
|
visitor_id = _register(client, "visitor_1")
|
|
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
# 1. Create a world with discovery metadata
|
|
world = World(
|
|
name="Cyber Adventure",
|
|
description="An epic adventure world",
|
|
owner_id=user_id,
|
|
visibility=WorldVisibility.PUBLIC,
|
|
category="Adventure",
|
|
tags=["cyberpunk", "rpg", "quest"],
|
|
)
|
|
session.add(world)
|
|
await session.commit()
|
|
await session.refresh(world)
|
|
|
|
assert world.id is not None
|
|
assert world.category == "Adventure"
|
|
assert world.tags == ["cyberpunk", "rpg", "quest"]
|
|
assert world.play_count == 0
|
|
assert world.last_played_at is None
|
|
|
|
# 2. Add authenticated visit
|
|
visit1 = WorldVisit(
|
|
world_id=world.id,
|
|
user_id=visitor_id,
|
|
)
|
|
# 3. Add anonymous visit
|
|
visit2 = WorldVisit(
|
|
world_id=world.id,
|
|
user_id=None,
|
|
anonymous_token="anon_token_12345",
|
|
)
|
|
session.add_all([visit1, visit2])
|
|
await session.commit()
|
|
|
|
# Query world with visits
|
|
res = await session.execute(
|
|
select(World).where(World.id == world.id).options(selectinload(World.visits))
|
|
)
|
|
w = res.scalar_one()
|
|
assert len(w.visits) == 2
|
|
v_user = next(v for v in w.visits if v.user_id == visitor_id)
|
|
assert v_user.visited_at is not None
|
|
v_anon = next(v for v in w.visits if v.user_id is None)
|
|
assert v_anon.anonymous_token == "anon_token_12345"
|
|
|
|
# 4. Verify cascade delete on world
|
|
async with sf() as session:
|
|
res = await session.execute(select(World).where(World.id == world.id))
|
|
w = res.scalar_one()
|
|
await session.delete(w)
|
|
await session.commit()
|
|
|
|
# Visits should be gone
|
|
res_v = await session.execute(select(WorldVisit).where(WorldVisit.world_id == world.id))
|
|
assert len(res_v.scalars().all()) == 0
|
|
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_delete_sets_null_on_world_visits():
|
|
client = TestClient(app)
|
|
owner_id = _register(client, "world_owner_2")
|
|
visitor_id = _register(client, "temp_visitor_2")
|
|
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
world = World(
|
|
name="Chill Lobby",
|
|
owner_id=owner_id,
|
|
visibility=WorldVisibility.PUBLIC,
|
|
category="Social",
|
|
tags=["hangout"],
|
|
)
|
|
session.add(world)
|
|
await session.commit()
|
|
await session.refresh(world)
|
|
|
|
visit = WorldVisit(world_id=world.id, user_id=visitor_id)
|
|
session.add(visit)
|
|
await session.commit()
|
|
await session.refresh(visit)
|
|
visit_id = visit.id
|
|
|
|
# Delete visitor user
|
|
visitor_user = await session.get(User, visitor_id)
|
|
await session.delete(visitor_user)
|
|
await session.commit()
|
|
|
|
# In a new session, query from database to see ON DELETE SET NULL effect
|
|
async with sf() as session:
|
|
v = await session.get(WorldVisit, visit_id)
|
|
assert v is not None
|
|
assert v.user_id is None
|
|
assert v.world_id == world.id
|
|
|
|
finally:
|
|
await engine.dispose()
|