feat: JWT auth, user management, audit logging system
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
"""Test fixtures: async client + in-memory SQLite database."""
|
||||
import asyncio
|
||||
from typing import AsyncGenerator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.main import app
|
||||
|
||||
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
loop = asyncio.new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session")
|
||||
async def engine():
|
||||
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield engine
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db(engine) -> AsyncGenerator[AsyncSession, None]:
|
||||
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(engine) -> AsyncGenerator[AsyncClient, None]:
|
||||
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
async def override_get_db():
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||||
yield ac
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Auth API tests."""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
class TestRegister:
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_success(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/register", json={
|
||||
"username": "owner1",
|
||||
"email": "owner@test.local",
|
||||
"password": "Pass1234",
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["username"] == "owner1"
|
||||
assert data["role"] == "owner"
|
||||
assert data["is_approved"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_duplicate(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/register", json={
|
||||
"username": "owner1",
|
||||
"email": "another@test.local",
|
||||
"password": "Pass1234",
|
||||
})
|
||||
assert resp.status_code == 409
|
||||
assert "Username" in resp.json()["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_second_user_is_viewer(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/register", json={
|
||||
"username": "viewer1",
|
||||
"email": "viewer@test.local",
|
||||
"password": "Pass1234",
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["role"] == "viewer"
|
||||
assert data["is_approved"] is False
|
||||
|
||||
|
||||
class TestLogin:
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_success(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/login", json={
|
||||
"username": "owner1",
|
||||
"password": "Pass1234",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_wrong_password(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/login", json={
|
||||
"username": "owner1",
|
||||
"password": "wrongpass",
|
||||
})
|
||||
assert resp.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_locked(self, client: AsyncClient):
|
||||
# Fail 4 more times (already failed once above) → total 5 failures
|
||||
for _ in range(4):
|
||||
await client.post("/api/v1/auth/login", json={
|
||||
"username": "owner1",
|
||||
"password": "wrongpass",
|
||||
})
|
||||
resp = await client.post("/api/v1/auth/login", json={
|
||||
"username": "owner1",
|
||||
"password": "wrongpass",
|
||||
})
|
||||
assert resp.status_code == 423
|
||||
|
||||
|
||||
class TestAuthGuard:
|
||||
@pytest.mark.asyncio
|
||||
async def test_require_auth_no_token(self, client: AsyncClient):
|
||||
resp = await client.post("/api/v1/auth/refresh")
|
||||
assert resp.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_viewer_cannot_manage_users(self, client: AsyncClient):
|
||||
# Login as viewer1 → token
|
||||
login = await client.post("/api/v1/auth/login", json={
|
||||
"username": "viewer1",
|
||||
"password": "Pass1234",
|
||||
})
|
||||
token = login.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
resp = await client.get("/api/v1/users", headers=headers)
|
||||
assert resp.status_code == 403
|
||||
Reference in New Issue
Block a user