Nora
8fc3d9adda
- New test_existing_endpoints_require_auth.py: spot-check that existing /api/* endpoints now require auth (gated via Depends(matrix_gate)) when AUTH_DISABLED is False. Health remains public. - conftest.py: flip AUTH_DISABLED=True for the suite so the legacy pre-auth tests keep passing without login. Auth tests flip it back off via their own autouse fixture (now patched to use monkeypatch for cleanup). Verified: 53 auth tests pass; 222 pre-existing non-auth failures are unchanged.
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
"""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(monkeypatch):
|
|
# conftest sets AUTH_DISABLED=True so pre-auth tests keep passing
|
|
# without a login. The auth-admin tests need the real auth path
|
|
# exercised, so flip it back off here.
|
|
monkeypatch.setattr("cyclone.auth.deps.AUTH_DISABLED", False)
|
|
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
|