76c867fca4
- shared/ = portable Android-origin sources vendored from deferred/desktop-server (app/build.gradle.kts srcDir repointed; PlaybackController.kt excluded as Android-only) - backend/ = bundled-lite engine (SQLite + inline queue); .venv symlinked from the old checkout, PYTHONPATH pins THIS backend's code over any editable install - repoRoot() resolves this project dir (env SHONAR_REPO still wins); desktop-dev.sh watches shared/ + backend/ - Verified: :app:compileKotlin + :app:test green (23 tests); engine boots on :8010, self-migrates, /healthz ok
110 lines
4 KiB
Python
110 lines
4 KiB
Python
"""Auth flow tests: register, login, refresh rotation + reuse detection,
|
|
logout, protected access, account guards."""
|
|
|
|
AUTH = {"email": "test@example.com", "password": "correct-horse-battery"}
|
|
|
|
|
|
async def register(client, email=AUTH["email"], password=AUTH["password"]):
|
|
return await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": password, "display_name": "Tester"},
|
|
)
|
|
|
|
|
|
async def test_register_returns_token_pair(client):
|
|
r = await register(client)
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["token_type"] == "bearer"
|
|
assert body["expires_in"] == 15 * 60
|
|
assert body["access_token"] and body["refresh_token"]
|
|
|
|
|
|
async def test_register_rejects_duplicate_email(client):
|
|
assert (await register(client)).status_code == 201
|
|
r = await register(client)
|
|
assert r.status_code == 409
|
|
|
|
|
|
async def test_register_rejects_weak_password(client):
|
|
r = await client.post(
|
|
"/api/v1/auth/register", json={"email": "x@example.com", "password": "short"}
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_login_success_and_failure(client):
|
|
await register(client)
|
|
r = await client.post(
|
|
"/api/v1/auth/login", json={"email": AUTH["email"], "password": AUTH["password"]}
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["device_id"]
|
|
|
|
bad = await client.post("/api/v1/auth/login", json={"email": AUTH["email"], "password": "***"})
|
|
assert bad.status_code == 401
|
|
# Same generic message either way (no user enumeration via password check)
|
|
bad2 = await client.post(
|
|
"/api/v1/auth/login", json={"email": "nobody@example.com", "password": "***"}
|
|
)
|
|
assert bad2.status_code == 401
|
|
assert bad.json()["detail"] == bad2.json()["detail"]
|
|
|
|
|
|
async def test_me_requires_valid_token(client):
|
|
r = await client.get("/api/v1/auth/me")
|
|
assert r.status_code == 401
|
|
tok = (await register(client)).json()["access_token"]
|
|
r = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {tok}"})
|
|
assert r.status_code == 200
|
|
assert r.json()["email"] == AUTH["email"]
|
|
|
|
|
|
async def test_refresh_rotates_and_detects_reuse(client):
|
|
tok = (await register(client)).json()
|
|
old_refresh = tok["refresh_token"]
|
|
|
|
r = await client.post("/api/v1/auth/refresh", json={"refresh_token": old_refresh})
|
|
assert r.status_code == 200
|
|
new_refresh = r.json()["refresh_token"]
|
|
assert new_refresh != old_refresh
|
|
|
|
# Old token is dead; reuse kills the whole family.
|
|
r2 = await client.post("/api/v1/auth/refresh", json={"refresh_token": old_refresh})
|
|
assert r2.status_code == 401
|
|
|
|
# The replacement is also revoked (family revocation).
|
|
r3 = await client.post("/api/v1/auth/refresh", json={"refresh_token": new_refresh})
|
|
assert r3.status_code == 401
|
|
|
|
|
|
async def test_logout_revokes_refresh(client):
|
|
tok = (await register(client)).json()
|
|
r = await client.post("/api/v1/auth/logout", json={"refresh_token": tok["refresh_token"]})
|
|
assert r.status_code == 204
|
|
r2 = await client.post("/api/v1/auth/refresh", json={"refresh_token": tok["refresh_token"]})
|
|
assert r2.status_code == 401
|
|
|
|
|
|
async def test_access_token_with_refresh_type_rejected(client):
|
|
tok = (await register(client)).json()
|
|
r = await client.get(
|
|
"/api/v1/auth/me", headers={"Authorization": f"Bearer {tok['refresh_token']}"}
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
async def test_delete_account_requires_password(client):
|
|
tok = (await register(client)).json()
|
|
h = {"Authorization": f"Bearer {tok['access_token']}"}
|
|
r = await client.post("/api/v1/auth/delete-account", json={"password": "***"}, headers=h)
|
|
assert r.status_code == 403
|
|
r = await client.post(
|
|
"/api/v1/auth/delete-account", json={"password": AUTH["password"]}, headers=h
|
|
)
|
|
assert r.status_code == 202
|
|
# Deleted account can no longer log in.
|
|
r = await client.post(
|
|
"/api/v1/auth/login", json={"email": AUTH["email"], "password": AUTH["password"]}
|
|
)
|
|
assert r.status_code == 401
|