WIP checkpoint: Address review feedback: Implement Godot dedicated server scene + player spawning

This commit is contained in:
2026-08-20 23:31:27 +00:00
parent 1c3c57dbd0
commit 4276a89a6a
94 changed files with 9220 additions and 0 deletions
View File
+38
View File
@@ -0,0 +1,38 @@
"""Shared pytest fixtures.
The auth integration tests hit the *real* Postgres database (per
``DATABASE_URL``), so they persist data between runs. Without cleanup a
second run of the suite fails with 409 conflicts because the users it
registered on the first run already exist. The session-scoped fixture below
truncates all application tables once before the first test, making the suite
re-runnable.
Truncation order doesn't strictly matter because ``CASCADE`` follows foreign
keys, but the tables are listed explicitly for clarity.
"""
import asyncio
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from app.config import settings
TEST_TABLES = ("assets", "worlds", "sessions", "refresh_tokens", "users")
@pytest.fixture(scope="session", autouse=True)
def _reset_database():
"""Truncate all application tables once per test session."""
async def _truncate() -> None:
engine = create_async_engine(settings.DATABASE_URL)
async with engine.begin() as conn:
await conn.execute(
text(f"TRUNCATE TABLE {', '.join(TEST_TABLES)} CASCADE")
)
await engine.dispose()
asyncio.run(_truncate())
yield
+94
View File
@@ -0,0 +1,94 @@
"""Tests for the 18+ age gate on user registration."""
from datetime import date, timedelta
import pytest
from app.routers.auth import calculate_age
class TestCalculateAge:
"""Unit tests for the calculate_age helper."""
def test_exactly_18(self):
"""A person born exactly 18 years ago today is 18."""
today = date(2025, 6, 15)
dob = today.replace(year=today.year - 18)
assert calculate_age(dob, today=today) == 18
def test_just_turned_18(self):
"""A person who turned 18 yesterday is 18."""
today = date(2025, 6, 15)
dob = today.replace(year=today.year - 18) - timedelta(days=1)
assert calculate_age(dob, today=today) == 18
def test_one_day_short_of_18(self):
"""A person born one day after their 18th birthday is 17."""
today = date(2025, 6, 15)
dob = today.replace(year=today.year - 18) + timedelta(days=1)
assert calculate_age(dob, today=today) == 17
def test_17_years_old(self):
"""A person who is 17 years old today is 17."""
today = date(2025, 6, 15)
dob = date(2008, 6, 15)
assert calculate_age(dob, today=today) == 17
def test_65_years_old(self):
"""A person who is 65 is correctly calculated."""
today = date(2025, 6, 15)
dob = today.replace(year=today.year - 65)
assert calculate_age(dob, today=today) == 65
def test_newborn(self):
"""A newborn is 0."""
today = date(2025, 6, 15)
dob = today
assert calculate_age(dob, today=today) == 0
def test_birthday_not_yet_this_year(self):
"""Person born Dec 31, 2000 — on Jan 1, 2025 they are still 24."""
dob = date(2000, 12, 31)
assert calculate_age(dob, today=date(2025, 1, 1)) == 24
def test_birthday_today(self):
"""Person born Dec 31, 2000 — on Dec 31, 2024 they turn 24."""
dob = date(2000, 12, 31)
assert calculate_age(dob, today=date(2024, 12, 31)) == 24
class TestRegisterRequestOver18Field:
"""Tests for RegisterRequest's over_18 field.
Replaces TestRegisterRequestDateValidation -- date_of_birth and its
field_validator were removed from RegisterRequest 2026-08-20 (interim
simplification: a plain client-side over_18 confirmation instead of
collecting/validating an actual date of birth; see calculate_age's
own docstring in app/routers/auth.py for why that function is still
here even though this endpoint no longer calls it). There's no date
format left to validate, so the equivalent coverage is just: the
field exists, is a real bool, and is required."""
def test_over_18_true_accepted(self):
from app.routers.auth import RegisterRequest
req = RegisterRequest(username="u", email="u@example.com", password="p", over_18=True)
assert req.over_18 is True
def test_over_18_false_accepted(self):
"""False is a valid value at the schema level -- the endpoint,
not the schema, is what rejects it (see
TestRegister::test_register_rejects_not_over_18 in
test_auth_endpoints.py)."""
from app.routers.auth import RegisterRequest
req = RegisterRequest(username="u", email="u@example.com", password="p", over_18=False)
assert req.over_18 is False
def test_over_18_missing_rejected(self):
"""over_18 has no default -- omitting it entirely is a validation error."""
from app.routers.auth import RegisterRequest
from pydantic import ValidationError
with pytest.raises(ValidationError):
RegisterRequest(username="u", email="u@example.com", password="p")
+259
View File
@@ -0,0 +1,259 @@
"""Integration tests for the auth endpoints (register + login).
These tests hit the real FastAPI app against the real configured database
(DATABASE_URL -- Postgres via Alembic migrations in CI/dev, not an isolated
in-memory DB), so they exercise the full stack: Pydantic validation → DB →
JWT → response. Requires migrations already applied (alembic upgrade head)
before running.
"""
import pytest
from fastapi.testclient import TestClient
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.database import get_db
from app.routers.auth import pwd_context
# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _override_db():
"""Override get_db so the app uses a fresh session per test.
Deliberately does NOT reuse app.database's shared engine: FastAPI's
(sync) TestClient runs each request through anyio's blocking-portal
machinery, which can hand different requests different event loops.
asyncpg connections are bound to the loop they were created on, so
reusing one from the shared engine's connection pool across requests
raises "cannot perform operation: another operation is in progress" the
moment a test makes more than one request (register-then-login,
register-twice, etc.) -- confirmed this is the actual cause, not a bug
in the endpoints themselves. NullPool sidesteps it: every checkout is a
brand-new connection, never handed to a different loop than the one
that opened it.
"""
test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
async def _get_db():
async_session = test_session_factory()
try:
yield async_session
await async_session.commit()
except Exception:
await async_session.rollback()
raise
finally:
await async_session.close()
app.dependency_overrides[get_db] = _get_db
yield
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Register endpoint
# ---------------------------------------------------------------------------
class TestRegister:
"""Tests for POST /api/auth/register."""
def test_register_success(self):
"""A valid registration returns 201 with a token."""
client = TestClient(app)
resp = client.post(
"/api/auth/register",
json={
"username": "newuser",
"email": "new@example.com",
"password": "secret123",
"over_18": True,
},
)
assert resp.status_code == 201
data = resp.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
def test_register_rejects_not_over_18(self):
"""over_18=false gets a 400 with a detail message.
The interim design (2026-08-20) trusts a client-side boolean
confirmation instead of collecting/validating an actual date of
birth -- this only checks that the flag is respected, not real
age math (there's no date to verify against anymore). Real DOB
collection/verification may come back in a future pass."""
client = TestClient(app)
resp = client.post(
"/api/auth/register",
json={
"username": "kid",
"email": "kid@example.com",
"password": "secret123",
"over_18": False,
},
)
assert resp.status_code == 400
assert "18 or older" in resp.json()["detail"]
def test_register_rejects_duplicate_username(self):
"""Registering the same username twice returns 409."""
client = TestClient(app)
# First registration
client.post(
"/api/auth/register",
json={
"username": "duplicate",
"email": "dup1@example.com",
"password": "secret123",
"over_18": True,
},
)
# Second registration with same username, different email
resp = client.post(
"/api/auth/register",
json={
"username": "duplicate",
"email": "dup2@example.com",
"password": "secret123",
"over_18": True,
},
)
assert resp.status_code == 409
assert "Username already registered" in resp.json()["detail"]
def test_register_rejects_duplicate_email(self):
"""Registering the same email twice returns 409."""
client = TestClient(app)
client.post(
"/api/auth/register",
json={
"username": "user1",
"email": "same@example.com",
"password": "secret123",
"over_18": True,
},
)
resp = client.post(
"/api/auth/register",
json={
"username": "user2",
"email": "same@example.com",
"password": "secret123",
"over_18": True,
},
)
assert resp.status_code == 409
assert "Email already registered" in resp.json()["detail"]
def test_register_missing_over_18(self):
"""Missing the required over_18 field returns 422.
Replaces the old test_register_invalid_date_format -- there's no
date format to get wrong anymore (no date field at all), so the
closest equivalent "malformed request" case is the required
over_18 boolean being absent entirely."""
client = TestClient(app)
resp = client.post(
"/api/auth/register",
json={
"username": "user",
"email": "user@example.com",
"password": "secret123",
},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# Login endpoint
# ---------------------------------------------------------------------------
class TestLogin:
"""Tests for POST /api/auth/login."""
def _create_user(self, username: str, password: str, email: str = None):
"""Helper to create a user via register."""
client = TestClient(app)
if email is None:
email = f"{username}@example.com"
resp = client.post(
"/api/auth/register",
json={
"username": username,
"email": email,
"password": password,
"over_18": True,
},
)
assert resp.status_code == 201
return resp.json()["access_token"]
def test_login_success(self):
"""Correct credentials return a token."""
self._create_user("loginuser", "correctpass")
client = TestClient(app)
resp = client.post(
"/api/auth/login",
data={"username": "loginuser", "password": "correctpass"},
)
assert resp.status_code == 200
data = resp.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
def test_login_wrong_password(self):
"""Wrong password returns 401."""
# Distinct username from test_login_success -- there's no per-test
# DB rollback/isolation here (real Postgres via the app's own
# dependency-overridden session), so reusing "loginuser" collided
# with the user that test already created (409 on register instead
# of the intended 401-on-login assertion this test is actually for).
self._create_user("loginuser_wrongpass", "correctpass")
client = TestClient(app)
resp = client.post(
"/api/auth/login",
data={"username": "loginuser_wrongpass", "password": "wrongpass"},
)
assert resp.status_code == 401
assert "Incorrect username or password" in resp.json()["detail"]
def test_login_nonexistent_user(self):
"""Non-existent user returns 401 (not 404)."""
client = TestClient(app)
resp = client.post(
"/api/auth/login",
data={"username": "nobody", "password": "any"},
)
assert resp.status_code == 401
assert "Incorrect username or password" in resp.json()["detail"]
def test_login_accepts_form_encoded(self):
"""Login accepts application/x-www-form-urlencoded (OAuth2)."""
self._create_user("formuser", "formpass")
client = TestClient(app)
resp = client.post(
"/api/auth/login",
data={"username": "formuser", "password": "formpass"},
)
assert resp.status_code == 200
assert "access_token" in resp.json()
def test_login_token_is_valid_jwt(self):
"""The returned token is a valid JWT that can be decoded."""
self._create_user("jwttoken", "jwt123")
client = TestClient(app)
resp = client.post(
"/api/auth/login",
data={"username": "jwttoken", "password": "jwt123"},
)
token = resp.json()["access_token"]
# Verify it's a 3-part JWT
parts = token.split(".")
assert len(parts) == 3
+171
View File
@@ -0,0 +1,171 @@
"""Tests for expired-token rejection across auth endpoints.
Covers two distinct expiry paths:
1. An *access* JWT whose ``exp`` claim is in the past must be rejected
with 401 by ``GET /api/auth/me`` (jose's built-in ``exp`` validation).
2. A *refresh* token whose persisted ``expires_at`` is in the past must be
rejected with 401 by ``POST /api/auth/refresh`` even though its JWT
signature is still valid.
Requires migrations already applied (alembic upgrade head) before running.
"""
import asyncio
from datetime import datetime, timedelta, timezone
import pytest
from fastapi.testclient import TestClient
from jose import jwt
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.database import get_db
from app.models.session import RefreshToken
# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _override_db():
"""Override get_db so the app uses a fresh session per test."""
test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
async def _get_db():
async_session = test_session_factory()
try:
yield async_session
await async_session.commit()
except Exception:
await async_session.rollback()
raise
finally:
await async_session.close()
app.dependency_overrides[get_db] = _get_db
yield
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mint_access_token(exp: datetime) -> str:
"""Mint an RS256 access token with an explicit ``exp`` claim."""
return jwt.encode(
{"sub": "1", "exp": exp},
settings.private_pem,
algorithm=settings.ALGORITHM,
)
def _expire_refresh_token_in_db(token: str) -> None:
"""Set the persisted refresh token's ``expires_at`` into the past."""
async def _do():
engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
from sqlalchemy import update
await session.execute(
update(RefreshToken)
.where(RefreshToken.token == token)
.values(expires_at=datetime.now(timezone.utc) - timedelta(hours=1))
)
await session.commit()
await engine.dispose()
asyncio.run(_do())
def _register_user(client: TestClient, username: str) -> str:
"""Register a user and return its refresh token."""
resp = client.post(
"/api/auth/register",
json={
"username": username,
"email": f"{username}@example.com",
"password": "secret123",
"over_18": True,
},
)
assert resp.status_code == 201
return resp.json()["refresh_token"]
# ---------------------------------------------------------------------------
# Expired access token (GET /api/auth/me)
# ---------------------------------------------------------------------------
class TestExpiredAccessToken:
def test_expired_access_token_is_rejected_by_me(self):
"""An access JWT with a past exp claim returns 401 on /me."""
client = TestClient(app)
expired_token = _mint_access_token(
datetime.now(timezone.utc) - timedelta(minutes=5)
)
resp = client.get(
"/api/auth/me",
headers={"Authorization": f"Bearer {expired_token}"},
)
assert resp.status_code == 401
assert "Could not validate credentials" in resp.json()["detail"]
def test_garbage_access_token_is_rejected_by_me(self):
"""A non-JWT bearer token returns 401 on /me."""
client = TestClient(app)
resp = client.get(
"/api/auth/me",
headers={"Authorization": "Bearer not-a-real-jwt"},
)
assert resp.status_code == 401
def test_missing_bearer_token_returns_401(self):
"""No Authorization header at all returns 401 on /me."""
client = TestClient(app)
resp = client.get("/api/auth/me")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Expired refresh token (POST /api/auth/refresh)
# ---------------------------------------------------------------------------
class TestExpiredRefreshToken:
def test_expired_refresh_token_is_rejected(self):
"""A refresh token whose DB expires_at is in the past returns 401."""
client = TestClient(app)
refresh_token = _register_user(client, "expired_rt_user")
# Force the persisted record to be expired while its JWT signature
# is still valid.
_expire_refresh_token_in_db(refresh_token)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp.status_code == 401
assert "Invalid or expired refresh token" in resp.json()["detail"]
def test_expired_refresh_token_jwt_is_rejected(self):
"""A refresh JWT with a past exp claim is rejected before DB lookup."""
client = TestClient(app)
expired_jwt = jwt.encode(
{"sub": "1", "exp": datetime.now(timezone.utc) - timedelta(minutes=5)},
settings.private_pem,
algorithm=settings.ALGORITHM,
)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": expired_jwt},
)
assert resp.status_code == 401
assert "Invalid or expired refresh token" in resp.json()["detail"]
+220
View File
@@ -0,0 +1,220 @@
"""Tests for the lightweight glTF metadata extraction service."""
import json
import struct
import pytest
from app.services.gltf import (
extract_gltf_metadata,
looks_like_gltf,
_parse_glb,
_parse_gltf_json,
)
# ---------------------------------------------------------------------------
# Test data helpers
# ---------------------------------------------------------------------------
def _make_gltf_json(
meshes: int = 1,
vertices_per_mesh: int = 24,
has_animations: bool = False,
images: list[dict] | None = None,
) -> dict:
"""Build a minimal glTF JSON structure."""
accessors = []
meshes_data = []
for i in range(meshes):
acc_index = len(accessors)
accessors.append({
"bufferView": 0,
"componentType": 5126, # FLOAT
"count": vertices_per_mesh,
"type": "VEC3",
"max": [1.0, 1.0, 1.0],
"min": [-1.0, -1.0, -1.0],
})
meshes_data.append({
"primitives": [{
"attributes": {"POSITION": acc_index},
}],
})
gltf = {
"asset": {"version": "2.0"},
"accessors": accessors,
"meshes": meshes_data,
}
if has_animations:
gltf["animations"] = [{"name": "anim1", "channels": [], "samplers": []}]
if images:
gltf["images"] = images
return gltf
def _make_glb(gltf_json: dict) -> bytes:
"""Wrap *gltf_json* in a minimal GLB binary container."""
json_bytes = json.dumps(gltf_json).encode("utf-8")
json_length = len(json_bytes)
# Pad JSON chunk to 4-byte boundary.
json_padding = (4 - json_length % 4) % 4
json_chunk = json_bytes + b"\x00" * json_padding
json_chunk_type = 0x4E4F534A # "JSON"
# No BIN chunk.
bin_length = 0
bin_chunk_type = 0x004E4942
total_length = 12 + 8 + len(json_chunk) + 8 + bin_length
header = struct.pack("<III", 0x46546C67, 2, total_length) # magic, version, length
json_header = struct.pack("<II", len(json_chunk), json_chunk_type)
if bin_length > 0:
bin_header = struct.pack("<II", bin_length, bin_chunk_type)
else:
bin_header = b""
return header + json_header + json_chunk + bin_header
# ---------------------------------------------------------------------------
# looks_like_gltf
# ---------------------------------------------------------------------------
class TestLooksLikeGltf:
def test_glb_magic(self):
assert looks_like_gltf(b"glTF\x00\x00\x00\x00") is True
def test_gltf_extension(self):
assert looks_like_gltf(b'{"asset":{"version":"2.0"}}', "model.gltf") is True
def test_glb_extension(self):
assert looks_like_gltf(b"some binary", "model.glb") is True
def test_json_with_asset_version(self):
payload = json.dumps({"asset": {"version": "2.0"}}).encode()
assert looks_like_gltf(payload) is True
def test_plain_json_no_asset(self):
payload = json.dumps({"foo": "bar"}).encode()
assert looks_like_gltf(payload) is False
def test_non_json_bytes(self):
assert looks_like_gltf(b"\x89PNG\r\n\x1a\n") is False
# ---------------------------------------------------------------------------
# _parse_glb
# ---------------------------------------------------------------------------
class TestParseGlb:
def test_parses_valid_glb(self):
gltf_json = _make_gltf_json()
glb = _make_glb(gltf_json)
result = _parse_glb(glb)
assert result["asset"]["version"] == "2.0"
assert len(result["meshes"]) == 1
def test_rejects_too_short(self):
with pytest.raises(ValueError, match="too short"):
_parse_glb(b"glTF\x00\x00")
def test_rejects_invalid_magic(self):
with pytest.raises(ValueError, match="invalid GLB magic"):
_parse_glb(b"NOTglTF\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
# ---------------------------------------------------------------------------
# _parse_gltf_json
# ---------------------------------------------------------------------------
class TestParseGltfJson:
def test_parses_valid_gltf(self):
gltf_json = _make_gltf_json()
result = _parse_gltf_json(json.dumps(gltf_json).encode())
assert result["asset"]["version"] == "2.0"
def test_rejects_non_object(self):
with pytest.raises(ValueError, match="root must be an object"):
_parse_gltf_json(b"[1, 2, 3]")
# ---------------------------------------------------------------------------
# extract_gltf_metadata (GLB path)
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataGlb:
def test_basic_glb(self):
gltf_json = _make_gltf_json(meshes=1, vertices_per_mesh=24)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 1
assert meta["vertex_count"] == 24
assert meta["has_animation"] is False
assert meta["bounding_box_min"] == [-1.0, -1.0, -1.0]
assert meta["bounding_box_max"] == [1.0, 1.0, 1.0]
assert meta["texture_refs"] == []
def test_multiple_meshes(self):
gltf_json = _make_gltf_json(meshes=3, vertices_per_mesh=12)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 3
assert meta["vertex_count"] == 36
def test_with_animation(self):
gltf_json = _make_gltf_json(has_animations=True)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["has_animation"] is True
def test_with_external_texture(self):
gltf_json = _make_gltf_json(images=[{"uri": "textures/rock.png"}])
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["texture_refs"] == ["textures/rock.png"]
def test_with_named_texture(self):
gltf_json = _make_gltf_json(images=[{"name": "my_texture"}])
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["texture_refs"] == ["my_texture"]
def test_empty_meshes(self):
gltf_json = _make_gltf_json(meshes=0)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 0
assert meta["vertex_count"] == 0
assert meta["bounding_box_min"] == [0.0, 0.0, 0.0]
assert meta["bounding_box_max"] == [0.0, 0.0, 0.0]
# ---------------------------------------------------------------------------
# extract_gltf_metadata (plain .gltf JSON path)
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataGltfJson:
def test_plain_gltf(self):
gltf_json = _make_gltf_json(meshes=2, vertices_per_mesh=8)
raw = json.dumps(gltf_json).encode()
meta = extract_gltf_metadata(raw)
assert meta["mesh_count"] == 2
assert meta["vertex_count"] == 16
assert meta["has_animation"] is False
def test_plain_gltf_with_animation(self):
gltf_json = _make_gltf_json(has_animations=True)
raw = json.dumps(gltf_json).encode()
meta = extract_gltf_metadata(raw)
assert meta["has_animation"] is True
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataErrors:
def test_invalid_glb(self):
with pytest.raises(ValueError):
extract_gltf_metadata(b"not a glb")
def test_invalid_gltf_json(self):
with pytest.raises(ValueError):
_parse_gltf_json(b"[1, 2, 3]")
+201
View File
@@ -0,0 +1,201 @@
"""Tests for the POST /api/auth/refresh endpoint.
These tests exercise the full stack: register → get refresh token →
refresh → new access token. Requires migrations already applied
(alembic upgrade head) before running.
"""
import pytest
from fastapi.testclient import TestClient
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.database import get_db
# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _override_db():
"""Override get_db so the app uses a fresh session per test."""
test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
async def _get_db():
async_session = test_session_factory()
try:
yield async_session
await async_session.commit()
except Exception:
await async_session.rollback()
raise
finally:
await async_session.close()
app.dependency_overrides[get_db] = _get_db
yield
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Refresh endpoint tests
# ---------------------------------------------------------------------------
class TestRefresh:
"""Tests for POST /api/auth/refresh."""
def _register_user(self, username: str, password: str, email: str = None):
"""Helper to register a user and return the refresh token."""
client = TestClient(app)
if email is None:
email = f"{username}@example.com"
resp = client.post(
"/api/auth/register",
json={
"username": username,
"email": email,
"password": password,
"over_18": True,
},
)
assert resp.status_code == 201
return resp.json()["refresh_token"]
def test_refresh_success(self):
"""A valid refresh token returns a new access token."""
refresh_token = self._register_user("refep_refreshuser", "secret123")
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp.status_code == 200
data = resp.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
# The refresh token in the response is the same one we sent
assert data["refresh_token"] == refresh_token
def test_refresh_returns_different_access_token(self):
"""Refreshing produces a new access token (different from any previous)."""
refresh_token = self._register_user("refep_refreshtoken", "secret123")
client = TestClient(app)
# First refresh
resp1 = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp1.status_code == 200
access_token_1 = resp1.json()["access_token"]
# Second refresh
resp2 = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp2.status_code == 200
access_token_2 = resp2.json()["access_token"]
# Each refresh should produce a new access token
assert access_token_1 != access_token_2
def test_refresh_invalid_token(self):
"""An invalid (garbage) refresh token returns 401."""
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": "not.a.jwt.token"},
)
assert resp.status_code == 401
assert "Invalid or expired refresh token" in resp.json()["detail"]
def test_refresh_empty_body(self):
"""Missing refresh_token field returns 422 validation error."""
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
json={},
)
assert resp.status_code == 422
def test_refresh_missing_field(self):
"""No body at all returns 422."""
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
content="{}",
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 422
def test_refresh_with_login_token(self):
"""A refresh token obtained from login also works for /refresh."""
client = TestClient(app)
# Register first
client.post(
"/api/auth/register",
json={
"username": "refep_loginrefuser",
"email": "loginref@example.com",
"password": "secret123",
"over_18": True,
},
)
# Login to get a refresh token
login_resp = client.post(
"/api/auth/login",
data={"username": "refep_loginrefuser", "password": "secret123"},
)
assert login_resp.status_code == 200
refresh_token = login_resp.json()["refresh_token"]
# Use it to refresh
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp.status_code == 200
assert "access_token" in resp.json()
def test_refresh_access_token_is_valid_jwt(self):
"""The access token returned by /refresh is a valid JWT."""
from jose import jwt
refresh_token = self._register_user("refep_jwttoken", "secret123")
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp.status_code == 200
access_token = resp.json()["access_token"]
# Verify it's a 3-part JWT
parts = access_token.split(".")
assert len(parts) == 3
# Verify it decodes with the public key
payload = jwt.decode(
access_token,
settings.public_pem,
algorithms=[settings.ALGORITHM],
)
assert "sub" in payload
assert "exp" in payload
def test_refresh_expires_minutes_in_response(self):
"""The response includes access_token_expires_minutes."""
refresh_token = self._register_user("refep_expiretest", "secret123")
client = TestClient(app)
resp = client.post(
"/api/auth/refresh",
json={"refresh_token": refresh_token},
)
assert resp.status_code == 200
assert "access_token_expires_minutes" in resp.json()
assert resp.json()["access_token_expires_minutes"] == settings.ACCESS_TOKEN_EXPIRE_MINUTES
+79
View File
@@ -0,0 +1,79 @@
"""Tests for refresh token uniqueness (jti claim) and model column length."""
import uuid
from datetime import datetime, timedelta, timezone
import pytest
from jose import jwt
from app.config import settings
from app.routers.auth import _create_refresh_token
from app.models.session import RefreshToken
class TestRefreshTokenJtiUniqueness:
"""Verify that _create_refresh_token produces unique tokens even when
called with identical data in rapid succession (issue #1 from review)."""
def test_two_calls_produce_different_tokens(self):
"""Two calls with the same sub produce different JWTs because of jti."""
token_a = _create_refresh_token({"sub": "42"})
token_b = _create_refresh_token({"sub": "42"})
assert token_a != token_b
def test_jti_is_valid_uuid(self):
"""The jti claim inside the refresh token is a valid UUID."""
token = _create_refresh_token({"sub": "1"})
payload = jwt.decode(
token,
settings.public_pem,
algorithms=[settings.ALGORITHM],
)
assert "jti" in payload
# Verify it parses as a UUID (raises ValueError if not)
uuid.UUID(payload["jti"])
def test_each_token_has_unique_jti(self):
"""Generate 100 tokens for the same user — all jtis must differ."""
jtis = set()
for _ in range(100):
token = _create_refresh_token({"sub": "999"})
payload = jwt.decode(
token,
settings.public_pem,
algorithms=[settings.ALGORITHM],
)
jtis.add(payload["jti"])
assert len(jtis) == 100
def test_jti_survives_decode(self):
"""The jti claim is present and correct after decode."""
token = _create_refresh_token({"sub": "7"})
payload = jwt.decode(
token,
settings.public_pem,
algorithms=[settings.ALGORITHM],
)
assert payload["sub"] == "7"
assert "jti" in payload
assert "exp" in payload
class TestRefreshTokenModelColumnLength:
"""Verify the model's token column length matches the migration (issue #2)."""
def test_token_column_has_length_512(self):
"""RefreshToken.token column should be String(512), not unbounded."""
from sqlalchemy import inspect
# Get the column definition from the model's table
column = RefreshToken.__table__.columns["token"]
# The type should be String with length 512
assert column.type.length == 512
def test_token_column_type_is_string(self):
"""The token column should be a String type."""
from sqlalchemy import String as SAString
column = RefreshToken.__table__.columns["token"]
assert isinstance(column.type, SAString)