90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""
|
|
Tests for user management endpoints.
|
|
"""
|
|
import uuid
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper
|
|
# ---------------------------------------------------------------------------
|
|
async def _create_user(client, admin_headers, *, role="user"):
|
|
"""Register a unique user and return their id."""
|
|
uname = f"usr_{uuid.uuid4().hex[:10]}"
|
|
r = await client.post("/api/auth/register", json={
|
|
"first_name": "U", "last_name": "Test",
|
|
"username": uname, "email": f"{uname}@test.com",
|
|
"password": "Usr12345!", "role": role,
|
|
}, headers=admin_headers)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
# ── /api/users/me ──────────────────────────────────────────────────────────────
|
|
|
|
async def test_get_me(client, user_headers):
|
|
r = await client.get("/api/users/me", headers=user_headers)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
for field in ("id", "username", "role", "auth_provider", "must_change_password"):
|
|
assert field in body, f"Missing field: {field}"
|
|
|
|
|
|
# ── /api/users (list) ─────────────────────────────────────────────────────────
|
|
|
|
async def test_list_users_admin(client, admin_headers):
|
|
r = await client.get("/api/users", headers=admin_headers)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert isinstance(body, list)
|
|
assert len(body) > 0
|
|
assert "auth_provider" in body[0]
|
|
|
|
|
|
async def test_list_users_non_admin(client, user_headers):
|
|
r = await client.get("/api/users", headers=user_headers)
|
|
assert r.status_code == 403
|
|
|
|
|
|
# ── /api/users/{uid} (update) ─────────────────────────────────────────────────
|
|
|
|
async def test_update_user_role(client, admin_headers):
|
|
uid = await _create_user(client, admin_headers)
|
|
r = await client.put(f"/api/users/{uid}", json={"role": "viewer"}, headers=admin_headers)
|
|
assert r.status_code == 200
|
|
assert r.json().get("ok") is True
|
|
|
|
|
|
async def test_update_user_timezone(client, admin_headers):
|
|
uid = await _create_user(client, admin_headers)
|
|
r = await client.put(f"/api/users/{uid}", json={"timezone": "UTC"}, headers=admin_headers)
|
|
assert r.status_code == 200
|
|
assert r.json().get("ok") is True
|
|
|
|
|
|
# ── /api/users/{uid} (delete / deactivate) ────────────────────────────────────
|
|
|
|
async def test_delete_user(client, admin_headers):
|
|
uid = await _create_user(client, admin_headers)
|
|
r = await client.delete(f"/api/users/{uid}", headers=admin_headers)
|
|
assert r.status_code == 200
|
|
assert r.json().get("ok") is True
|
|
|
|
|
|
# ── /api/users/me/timezone ─────────────────────────────────────────────────────
|
|
|
|
async def test_get_user_timezone(client, user_headers):
|
|
r = await client.get("/api/users/me/timezone", headers=user_headers)
|
|
assert r.status_code == 200
|
|
assert "timezone" in r.json()
|
|
|
|
|
|
async def test_set_user_timezone(client, user_headers):
|
|
r = await client.put("/api/users/me/timezone", json={"timezone": "UTC"}, headers=user_headers)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body.get("ok") is True
|
|
assert body.get("timezone") == "UTC"
|