Dmitry
3fc7a954b9
Префикс /api/v1, operationId = "<tag>_<name>", чтобы Dart-клиент получил методы вроде accountsList, а не list_accounts_api_v1_accounts_get. Ошибки — application/problem+json. Auth: access-JWT на час + refresh на 30 дней, который хранится хэшем и ротируется, логин ограничен по частоте в памяти. Деньги в JSON — всегда позиционные строки (api/schemas/common.py): float в проводе потерял бы копейки, которые NUMERIC(24,10) бережёт. Тестовый harness поднимает свой Postgres через pg_ctl (pytest-postgresql), мигрирует его один раз на сессию и усекает таблицы после каждого теста.
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
async def test_login_refresh_logout_cycle(client, user):
|
|
r = await client.post("/api/v1/auth/login", json=user)
|
|
assert r.status_code == 200, r.text
|
|
pair = r.json()
|
|
assert pair["token_type"] == "bearer"
|
|
|
|
me = await client.get(
|
|
"/api/v1/auth/me", headers={"Authorization": f"Bearer {pair['access_token']}"}
|
|
)
|
|
assert me.status_code == 200
|
|
assert me.json()["email"] == user["email"]
|
|
|
|
# rotation: the refresh token is single-use
|
|
r2 = await client.post("/api/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]})
|
|
assert r2.status_code == 200
|
|
r3 = await client.post("/api/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]})
|
|
assert r3.status_code == 401
|
|
assert r3.headers["content-type"].startswith("application/problem+json")
|
|
|
|
new_refresh = r2.json()["refresh_token"]
|
|
assert (
|
|
await client.post("/api/v1/auth/logout", json={"refresh_token": new_refresh})
|
|
).status_code == 204
|
|
assert (
|
|
await client.post("/api/v1/auth/refresh", json={"refresh_token": new_refresh})
|
|
).status_code == 401
|
|
|
|
|
|
async def test_wrong_password_and_rate_limit(client, user):
|
|
bad = {"email": user["email"], "password": "nope"}
|
|
for _ in range(5):
|
|
r = await client.post("/api/v1/auth/login", json=bad)
|
|
assert r.status_code == 401
|
|
r = await client.post("/api/v1/auth/login", json=bad)
|
|
assert r.status_code == 429
|
|
assert "retry-after" in r.headers
|
|
|
|
|
|
async def test_protected_routes_need_token(client):
|
|
r = await client.get("/api/v1/auth/me")
|
|
assert r.status_code == 401
|
|
r = await client.get("/api/v1/auth/me", headers={"Authorization": "Bearer garbage"})
|
|
assert r.status_code == 401
|