115 lines
4.4 KiB
Python
115 lines
4.4 KiB
Python
"""Audit trail: business actions land in audit_log attributed to the verified user, the
|
|
history survives a closing's deletion, and /api/audit is readable by admins only."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import auth as auth_mod
|
|
from app.api.main import app
|
|
from app.db import models
|
|
from app.db.database import SessionLocal, init_db
|
|
from tests.test_excel_export import make_amazon_xlsx
|
|
|
|
_TMP = tempfile.mkdtemp(prefix="ar_audit_test_")
|
|
|
|
|
|
def _entries(session_id: int) -> list[models.AuditLog]:
|
|
db = SessionLocal()
|
|
try:
|
|
return (db.query(models.AuditLog)
|
|
.filter(models.AuditLog.session_id == session_id)
|
|
.order_by(models.AuditLog.id).all())
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def clean_users():
|
|
"""Users flip AR_AUTH=auto to 'required' for the WHOLE shared test DB — always remove
|
|
them again so the rest of the suite keeps running unauthenticated."""
|
|
init_db()
|
|
yield
|
|
db = SessionLocal()
|
|
try:
|
|
db.query(models.User).delete()
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
auth_mod.invalidate_users_cache()
|
|
|
|
|
|
def _add_user(username: str, display: str, password: str, admin: bool = False) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
db.add(models.User(username=username, display_name=display,
|
|
password_hash=auth_mod.hash_password(password),
|
|
is_active=True, is_admin=admin))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
auth_mod.invalidate_users_cache()
|
|
|
|
|
|
def test_actions_recorded_and_history_survives_delete():
|
|
init_db()
|
|
path = os.path.join(_TMP, "USA audit.xlsx")
|
|
make_amazon_xlsx(path, order_rows=4)
|
|
with TestClient(app) as c:
|
|
sid = c.post("/api/sessions", json={
|
|
"name": "audit-me", "month_end_date": "2027-03-31",
|
|
"allow_duplicate": True}).json()["id"]
|
|
with open(path, "rb") as fh:
|
|
r = c.post(f"/api/sessions/{sid}/files",
|
|
files={"files": ("USA audit.xlsx", fh)})
|
|
assert r.status_code == 200
|
|
|
|
rows = _entries(sid)
|
|
assert [e.action for e in rows][:2] == ["session_create", "file_upload"]
|
|
assert "USA audit.xlsx" in rows[1].detail
|
|
# With auth off (no users) the row is still written, just unattributed.
|
|
assert rows[1].username == ""
|
|
|
|
# Deleting the closing records the deletion and KEEPS the history (no FK).
|
|
assert c.delete(f"/api/sessions/{sid}").status_code == 200
|
|
actions = [e.action for e in _entries(sid)]
|
|
assert "session_delete" in actions and "session_create" in actions
|
|
|
|
|
|
def test_audit_endpoint_admin_only_and_logins_attributed(clean_users):
|
|
_add_user("admin@x.com", "Admin A", "pw-longenough", admin=True)
|
|
_add_user("user@x.com", "User U", "pw-longenough2")
|
|
with TestClient(app) as c:
|
|
tok_admin = c.post("/api/auth/login", json={
|
|
"username": "admin@x.com", "password": "pw-longenough"}).json()["token"]
|
|
tok_user = c.post("/api/auth/login", json={
|
|
"username": "user@x.com", "password": "pw-longenough2"}).json()["token"]
|
|
|
|
assert c.get("/api/audit").status_code == 401 # not signed in
|
|
r = c.get("/api/audit", headers={"Authorization": f"Bearer {tok_user}"})
|
|
assert r.status_code == 403 # not an admin
|
|
r = c.get("/api/audit", headers={"Authorization": f"Bearer {tok_admin}"})
|
|
assert r.status_code == 200
|
|
|
|
logins = [e for e in r.json()["entries"] if e["action"] == "login"]
|
|
assert {e["username"] for e in logins} >= {"admin@x.com", "user@x.com"}
|
|
|
|
|
|
def test_login_and_me_carry_admin_flag(clean_users):
|
|
_add_user("admin2@x.com", "Admin B", "pw-longenough", admin=True)
|
|
_add_user("user2@x.com", "User V", "pw-longenough2")
|
|
with TestClient(app) as c:
|
|
res = c.post("/api/auth/login", json={
|
|
"username": "admin2@x.com", "password": "pw-longenough"}).json()
|
|
assert res["user"]["is_admin"] is True
|
|
me = c.get("/api/auth/me",
|
|
headers={"Authorization": f"Bearer {res['token']}"}).json()
|
|
assert me["is_admin"] is True
|
|
|
|
res = c.post("/api/auth/login", json={
|
|
"username": "user2@x.com", "password": "pw-longenough2"}).json()
|
|
assert res["user"]["is_admin"] is False
|