165 lines
5.1 KiB
Python
165 lines
5.1 KiB
Python
"""API tests for world visit tracking endpoint.
|
|
|
|
docs/planning/discovery.md (task 42726e60fe10).
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.config import settings
|
|
from app.main import app
|
|
from app.models.world import World, WorldVisibility
|
|
from app.models.world_visit import WorldVisit
|
|
|
|
|
|
def _register(client: TestClient, username: str) -> tuple[int, dict[str, str]]:
|
|
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
|
|
token = resp.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
me = client.get("/api/users/me", headers=headers)
|
|
assert me.status_code == 200, me.text
|
|
return me.json()["id"], headers
|
|
|
|
|
|
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_visit_world_atomic_increment_authenticated():
|
|
client = TestClient(app)
|
|
owner_id, owner_headers = _register(client, "visit_owner_1")
|
|
visitor_id, visitor_headers = _register(client, "visitor_user_1")
|
|
|
|
# Create world
|
|
resp = client.post(
|
|
"/api/worlds/",
|
|
headers=owner_headers,
|
|
json={"name": "Neon City", "visibility": "public", "category": "Roleplay"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
world_id = resp.json()["id"]
|
|
|
|
# Visitor visits the world
|
|
v_resp = client.post(
|
|
f"/api/discovery/worlds/{world_id}/visit",
|
|
headers=visitor_headers,
|
|
)
|
|
assert v_resp.status_code == 200, v_resp.text
|
|
data = v_resp.json()
|
|
assert data["play_count"] == 1
|
|
assert data["last_played_at"] is not None
|
|
|
|
# Verify DB state
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
w = (await session.execute(select(World).where(World.id == world_id))).scalar_one()
|
|
assert w.play_count == 1
|
|
assert w.last_played_at is not None
|
|
|
|
visits = (
|
|
await session.execute(
|
|
select(WorldVisit).where(WorldVisit.world_id == world_id)
|
|
)
|
|
).scalars().all()
|
|
assert len(visits) == 1
|
|
assert visits[0].user_id == visitor_id
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_visit_world_anonymous_and_idempotency():
|
|
client = TestClient(app)
|
|
owner_id, owner_headers = _register(client, "visit_owner_2")
|
|
|
|
resp = client.post(
|
|
"/api/worlds/",
|
|
headers=owner_headers,
|
|
json={"name": "Space Station", "visibility": "public"},
|
|
)
|
|
world_id = resp.json()["id"]
|
|
|
|
# Anonymous visit with token
|
|
anon_token = "guest_token_999"
|
|
v_resp1 = client.post(
|
|
f"/api/discovery/worlds/{world_id}/visit",
|
|
json={"anonymous_token": anon_token},
|
|
)
|
|
assert v_resp1.status_code == 200, v_resp1.text
|
|
assert v_resp1.json()["play_count"] == 1
|
|
|
|
# Idempotent second call with same anonymous token
|
|
v_resp2 = client.post(
|
|
f"/api/discovery/worlds/{world_id}/visit",
|
|
json={"anonymous_token": anon_token},
|
|
)
|
|
assert v_resp2.status_code == 200, v_resp2.text
|
|
assert v_resp2.json()["play_count"] == 1
|
|
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
visits = (
|
|
await session.execute(
|
|
select(WorldVisit).where(WorldVisit.world_id == world_id)
|
|
)
|
|
).scalars().all()
|
|
assert len(visits) == 1
|
|
assert visits[0].user_id is None
|
|
assert visits[0].anonymous_token == anon_token
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_visit_world_visibility_and_404():
|
|
client = TestClient(app)
|
|
owner_id, owner_headers = _register(client, "visit_owner_3")
|
|
other_id, other_headers = _register(client, "other_user_3")
|
|
|
|
# 404 on non-existent world
|
|
res_404 = client.post("/api/discovery/worlds/999999/visit", headers=owner_headers)
|
|
assert res_404.status_code == 404
|
|
|
|
# Private world
|
|
resp = client.post(
|
|
"/api/worlds/",
|
|
headers=owner_headers,
|
|
json={"name": "Secret World", "visibility": "private"},
|
|
)
|
|
priv_world_id = resp.json()["id"]
|
|
|
|
# Other user gets 404
|
|
other_res = client.post(
|
|
f"/api/discovery/worlds/{priv_world_id}/visit",
|
|
headers=other_headers,
|
|
)
|
|
assert other_res.status_code == 404
|
|
|
|
# Anonymous user gets 404
|
|
anon_res = client.post(f"/api/discovery/worlds/{priv_world_id}/visit")
|
|
assert anon_res.status_code == 404
|
|
|
|
# Owner gets 200
|
|
owner_res = client.post(
|
|
f"/api/discovery/worlds/{priv_world_id}/visit",
|
|
headers=owner_headers,
|
|
)
|
|
assert owner_res.status_code == 200
|
|
assert owner_res.json()["play_count"] == 1
|