104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""Integration tests for auth endpoints using in-memory DuckDB."""
|
|
from backend.app.main import app
|
|
from backend.app import db as db_module
|
|
from httpx import AsyncClient, ASGITransport
|
|
import os
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
os.environ.setdefault("JWT_SECRET", "test-secret-key-for-testing-only")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh_db():
|
|
"""Reset the DB singleton to a fresh in-memory DB for each test."""
|
|
db_module._conn = None
|
|
db_module.init_db(":memory:")
|
|
yield
|
|
db_module.close_db()
|
|
db_module._conn = None
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(fresh_db):
|
|
"""HTTP client wired to the app; explicitly depends on fresh_db to guarantee ordering."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
async def test_register_success(client):
|
|
resp = await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["email"] == "user@example.com"
|
|
assert data["role"] == "user"
|
|
assert "id" in data
|
|
|
|
|
|
async def test_register_duplicate_email(client):
|
|
payload = {"email": "dup@example.com", "password": "secret123"}
|
|
await client.post("/auth/register", json=payload)
|
|
resp = await client.post("/auth/register", json=payload)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
async def test_login_success(client):
|
|
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
resp = await client.post("/auth/login", json={"email": "user@example.com", "password": "secret123"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
|
|
async def test_login_wrong_password(client):
|
|
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
resp = await client.post("/auth/login", json={"email": "user@example.com", "password": "wrong"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_login_unknown_user(client):
|
|
resp = await client.post("/auth/login", json={"email": "nobody@example.com", "password": "x"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_refresh_success(client):
|
|
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
login = await client.post("/auth/login", json={"email": "user@example.com", "password": "secret123"})
|
|
refresh_token = login.json()["refresh_token"]
|
|
|
|
resp = await client.post("/auth/refresh", json={"refresh_token": refresh_token})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
# New refresh token must differ (rotation)
|
|
assert data["refresh_token"] != refresh_token
|
|
|
|
|
|
async def test_refresh_revoked_token(client):
|
|
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
login = await client.post("/auth/login", json={"email": "user@example.com", "password": "secret123"})
|
|
refresh_token = login.json()["refresh_token"]
|
|
|
|
# Use once (rotates)
|
|
await client.post("/auth/refresh", json={"refresh_token": refresh_token})
|
|
# Try to reuse old token — must fail
|
|
resp = await client.post("/auth/refresh", json={"refresh_token": refresh_token})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_logout_success(client):
|
|
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
|
|
login = await client.post("/auth/login", json={"email": "user@example.com", "password": "secret123"})
|
|
refresh_token = login.json()["refresh_token"]
|
|
|
|
resp = await client.post("/auth/logout", json={"refresh_token": refresh_token})
|
|
assert resp.status_code == 204
|
|
|
|
# Refresh after logout must fail
|
|
resp2 = await client.post("/auth/refresh", json={"refresh_token": refresh_token})
|
|
assert resp2.status_code == 401
|