feat(auth): get_current_user + login/logout/me + admin user management
This commit is contained in:
@@ -0,0 +1,103 @@
|
||||
"""Admin-only user management endpoints."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import delete
|
||||
|
||||
from cyclone.api import app
|
||||
from cyclone.auth import users
|
||||
from cyclone.db import Session as DbSession
|
||||
from cyclone.db import SessionLocal, User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear():
|
||||
with SessionLocal()() as db:
|
||||
db.execute(delete(DbSession))
|
||||
db.execute(delete(User))
|
||||
db.commit()
|
||||
yield
|
||||
with SessionLocal()() as db:
|
||||
db.execute(delete(DbSession))
|
||||
db.execute(delete(User))
|
||||
db.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_client():
|
||||
client = TestClient(app)
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="root", password="rootpassword1", role="admin")
|
||||
login = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "root", "password": "rootpassword1"},
|
||||
)
|
||||
assert login.status_code == 200
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_client():
|
||||
client = TestClient(app)
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="plain", password="plainpassword1", role="user")
|
||||
login = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "plain", "password": "plainpassword1"},
|
||||
)
|
||||
return client
|
||||
|
||||
|
||||
def test_list_users_as_admin(admin_client):
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="alice", password="hunter2hunter2", role="user")
|
||||
resp = admin_client.get("/api/admin/users")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
usernames = {u["username"] for u in body}
|
||||
assert {"root", "alice"} <= usernames
|
||||
|
||||
|
||||
def test_list_users_as_nonadmin_returns_403(user_client):
|
||||
resp = user_client.get("/api/admin/users")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_create_user_as_admin(admin_client):
|
||||
resp = admin_client.post(
|
||||
"/api/admin/users",
|
||||
json={"username": "newbie", "password": "newbiepassword1", "role": "viewer"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["username"] == "newbie"
|
||||
assert resp.json()["role"] == "viewer"
|
||||
|
||||
|
||||
def test_create_user_rejects_invalid_role(admin_client):
|
||||
resp = admin_client.post(
|
||||
"/api/admin/users",
|
||||
json={"username": "badrole", "password": "hunter2hunter2", "role": "owner"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
def test_patch_user_role_and_password(admin_client):
|
||||
with SessionLocal()() as db:
|
||||
u = users.create(db, username="subject", password="hunter2hunter2", role="viewer")
|
||||
resp = admin_client.patch(
|
||||
f"/api/admin/users/{u.id}",
|
||||
json={"role": "user", "password": "newpassword1"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["role"] == "user"
|
||||
|
||||
|
||||
def test_admin_cannot_demote_self(admin_client):
|
||||
me = admin_client.get("/api/auth/me").json()
|
||||
resp = admin_client.patch(
|
||||
f"/api/admin/users/{me['id']}",
|
||||
json={"role": "viewer"},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
@@ -0,0 +1,111 @@
|
||||
"""API tests for /api/auth/login, /api/auth/logout, /api/auth/me."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import delete
|
||||
|
||||
from cyclone.api import app
|
||||
from cyclone.auth import users
|
||||
from cyclone.db import Session as DbSession
|
||||
from cyclone.db import SessionLocal, User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear():
|
||||
with SessionLocal()() as db:
|
||||
db.execute(delete(DbSession))
|
||||
db.execute(delete(User))
|
||||
db.commit()
|
||||
yield
|
||||
with SessionLocal()() as db:
|
||||
db.execute(delete(DbSession))
|
||||
db.execute(delete(User))
|
||||
db.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seeded_admin(client):
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="admin", password="adminpassword1", role="admin")
|
||||
return client
|
||||
|
||||
|
||||
def test_login_success_returns_user_and_cookie(client):
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="alice", password="hunter2hunter2", role="user")
|
||||
resp = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "alice", "password": "hunter2hunter2"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["username"] == "alice"
|
||||
assert body["role"] == "user"
|
||||
assert "password_hash" not in body
|
||||
assert "cyclone_session" in resp.cookies
|
||||
|
||||
|
||||
def test_login_bad_password_returns_401(client):
|
||||
with SessionLocal()() as db:
|
||||
users.create(db, username="bob", password="hunter2hunter2", role="user")
|
||||
resp = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "bob", "password": "WRONG"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
assert resp.json()["error"] == "invalid_credentials"
|
||||
|
||||
|
||||
def test_login_unknown_user_returns_401(client):
|
||||
resp = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "ghost", "password": "whatever"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
assert resp.json()["error"] == "invalid_credentials"
|
||||
|
||||
|
||||
def test_login_disabled_user_returns_403(client):
|
||||
with SessionLocal()() as db:
|
||||
u = users.create(db, username="carol", password="hunter2hunter2", role="user")
|
||||
users.disable(db, u.id)
|
||||
resp = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "carol", "password": "hunter2hunter2"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["error"] == "account_disabled"
|
||||
|
||||
|
||||
def test_logout_clears_session_and_cookie(seeded_admin):
|
||||
login = seeded_admin.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "admin", "password": "adminpassword1"},
|
||||
)
|
||||
cookie = login.cookies.get("cyclone_session")
|
||||
assert cookie
|
||||
resp = seeded_admin.post("/api/auth/logout", cookies={"cyclone_session": cookie})
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
def test_me_returns_current_user(seeded_admin):
|
||||
login = seeded_admin.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "admin", "password": "adminpassword1"},
|
||||
)
|
||||
cookie = login.cookies.get("cyclone_session")
|
||||
resp = seeded_admin.get("/api/auth/me", cookies={"cyclone_session": cookie})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["username"] == "admin"
|
||||
|
||||
|
||||
def test_me_without_cookie_returns_401(seeded_admin):
|
||||
resp = seeded_admin.get("/api/auth/me")
|
||||
assert resp.status_code == 401
|
||||
Reference in New Issue
Block a user