"""Admin audit-log tests (Phase 6.2).""" import os import unittest from datetime import timedelta from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from sqlalchemy import event from werkzeug.security import generate_password_hash from app import create_app from app.admin.analytics import get_recent_admin_activity from app.admin.audit import ( get_audit_logs, get_audit_summary, record_admin_action, ) from app.extensions import db from app.models import ( AdminAuditLog, Advertisement, Category, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, User, UserProfile, ) from app.models.base import utcnow from app.models.enums import AdvertisementPlacement, AdvertisementType, UserRole _IP = "203.0.113.9" class _QC: def __init__(self): self.n = 0 def __enter__(self): self._e = db.engine event.listen(self._e, "after_cursor_execute", self._cb) return self def __exit__(self, *a): event.remove(self._e, "after_cursor_execute", self._cb) def _cb(self, *a): self.n += 1 class AdminAuditTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() # --- factories --- def _user(self, email, role=UserRole.STUDENT, name=None): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!"), role=role) db.session.add(u) db.session.commit() if name: db.session.add(UserProfile(user_id=u.id, full_name=name)) db.session.commit() return u def _category(self): cat = db.session.scalar(db.select(Category).filter_by(name="Books")) if cat is None: cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() return cat def _listing(self, seller, idx=1, status=ListingStatus.AVAILABLE): listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category().id, title=f"Item {idx}", slug=f"item-{seller.id}-{idx}", description="d", price=Decimal("50.00"), condition=ListingCondition.GOOD, status=status, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() return listing def _ad(self, title="Promo"): ad = Advertisement(title=title, advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.SIDEBAR, is_active=True, priority=1) db.session.add(ad) db.session.commit() return ad def _admin(self, email="admin@c.edu", name="Admin One"): return self._user(email, UserRole.SUPER_ADMIN, name=name) def _admin_client(self, admin=None): admin = admin or self._admin() client = self.app.test_client() client.post("/login", data={"email": admin.email, "password": "Passw0rd!"}) return client def _actions(self): return [x.action for x in db.session.scalars( db.select(AdminAuditLog).order_by(AdminAuditLog.id)).all()] # --- access --- def test_admin_access(self): client = self._admin_client() self.assertEqual(client.get("/admin/audit-logs").status_code, 200) def test_non_admin_denied(self): self._user("s@c.edu") client = self.app.test_client() client.post("/login", data={"email": "s@c.edu", "password": "Passw0rd!"}) self.assertEqual(client.get("/admin/audit-logs").status_code, 403) # --- automatic logging: premium + user --- def test_logs_premium_and_user_actions(self): admin = self._admin() client = self._admin_client(admin) target = self._user("t@c.edu") client.post(f"/admin/users/{target.public_id}/grant-premium") client.post(f"/admin/users/{target.public_id}/remove-premium") client.post(f"/admin/users/{target.public_id}/suspend") client.post(f"/admin/users/{target.public_id}/reactivate") self.assertEqual( self._actions(), ["premium_granted", "premium_removed", "user_suspended", "user_reactivated"], ) # --- automatic logging: listing (all six) --- def test_logs_all_listing_actions(self): client = self._admin_client() seller = self._user("seller@c.edu") listing = self._listing(seller) pid = listing.public_id for act in ["feature", "unfeature", "reserved", "sold", "delete", "restore"]: client.post(f"/admin/listings/{pid}/action", data={"action": act}) self.assertEqual(self._actions(), [ "listing_featured", "listing_unfeatured", "listing_marked_reserved", "listing_marked_sold", "listing_deleted", "listing_restored", ]) def test_logs_bulk_listing_once_with_count(self): client = self._admin_client() seller = self._user("seller@c.edu") a = self._listing(seller, 1) b = self._listing(seller, 2) client.post("/admin/listings/bulk", data={"action": "delete", "selected": [a.public_id, b.public_id]}) logs = db.session.scalars(db.select(AdminAuditLog)).all() self.assertEqual(len(logs), 1) self.assertEqual(logs[0].action, "bulk_listing") self.assertEqual(logs[0].details_data["count"], 2) # --- automatic logging: advertisements --- def test_logs_advertisement_route_actions(self): client = self._admin_client() ad = self._ad() client.post(f"/admin/advertisements/{ad.public_id}/enable") client.post(f"/admin/advertisements/{ad.public_id}/disable") client.post(f"/admin/advertisements/{ad.public_id}/delete") self.assertEqual(self._actions(), ["ad_activated", "ad_deactivated", "ad_deleted"]) def test_logs_ad_create_update_via_service(self): admin = self._admin() ad = self._ad("Launch") record_admin_action(admin, "ad_created", ad) record_admin_action(admin, "ad_updated", ad) rows = db.session.scalars(db.select(AdminAuditLog).order_by(AdminAuditLog.id)).all() self.assertEqual([r.action for r in rows], ["ad_created", "ad_updated"]) self.assertEqual(rows[0].entity_type, "advertisement") self.assertEqual(rows[0].entity_name, "Launch") # --- search --- def test_search(self): admin = self._admin("boss@c.edu", name="Boss Admin") listing = self._listing(self._user("s@c.edu")) record_admin_action(admin, "listing_featured", listing) record_admin_action(admin, "premium_granted", self._user("other@c.edu", name="Other Person")) # entity name _, rows = get_audit_logs(search=listing.title) self.assertEqual([r["action"] for r in rows], ["listing_featured"]) # entity public id _, rows = get_audit_logs(search=listing.public_id) self.assertEqual([r["action"] for r in rows], ["listing_featured"]) # admin email _, rows = get_audit_logs(search="boss@c.edu") self.assertEqual(len(rows), 2) # --- filters --- def test_filter_by_action_and_entity_type(self): admin = self._admin() listing = self._listing(self._user("s@c.edu")) record_admin_action(admin, "listing_featured", listing) record_admin_action(admin, "premium_granted", self._user("u@c.edu")) _, rows = get_audit_logs(action="listing_featured") self.assertEqual([r["action"] for r in rows], ["listing_featured"]) _, rows = get_audit_logs(entity_type="user") self.assertEqual([r["action"] for r in rows], ["premium_granted"]) def test_filter_by_admin_and_date_range(self): a1 = self._admin("a1@c.edu", name="A One") a2 = self._admin("a2@c.edu", name="A Two") record_admin_action(a1, "user_suspended", self._user("x@c.edu")) record_admin_action(a2, "user_reactivated", self._user("y@c.edu")) _, rows = get_audit_logs(admin=a1.public_id) self.assertEqual([r["action"] for r in rows], ["user_suspended"]) # date range excluding everything (future window) future = utcnow() + timedelta(days=1) _, rows = get_audit_logs(date_from=future) self.assertEqual(rows, []) # --- pagination --- def test_pagination(self): admin = self._admin() u = self._user("u@c.edu") for _ in range(25): record_admin_action(admin, "user_suspended", u) pagination, rows = get_audit_logs(page=1) self.assertEqual(pagination.per_page, 20) self.assertEqual(len(rows), 20) self.assertEqual(pagination.pages, 2) _, rows2 = get_audit_logs(page=2) self.assertEqual(len(rows2), 5) # --- summary --- def test_summary_totals(self): admin = self._admin() listing = self._listing(self._user("s@c.edu")) record_admin_action(admin, "premium_granted", self._user("p@c.edu")) record_admin_action(admin, "listing_featured", listing) record_admin_action(admin, "listing_deleted", listing) record_admin_action(admin, "ad_created", self._ad()) record_admin_action(admin, "bulk_listing", "3 listings") s = get_audit_summary() self.assertEqual(s["total_events"], 5) self.assertEqual(s["today_events"], 5) self.assertEqual(s["premium_actions"], 1) self.assertEqual(s["listing_actions"], 2) self.assertEqual(s["advertisement_actions"], 1) self.assertEqual(s["bulk_actions"], 1) # --- append-only --- def test_audit_is_append_only(self): # The model carries no updated_at column (rows are never modified). self.assertNotIn("updated_at", AdminAuditLog.__table__.columns) # No write endpoint exists for audit logs — only a GET reader. methods = set() for rule in self.app.url_map.iter_rules(): if rule.endpoint.startswith("admin.audit"): methods |= (rule.methods - {"HEAD", "OPTIONS"}) self.assertEqual(methods, {"GET"}) # --- privacy: no raw IP --- def test_no_raw_ip_stored(self): admin = self._admin() client = self.app.test_client() client.post("/login", data={"email": admin.email, "password": "Passw0rd!"}, environ_base={"REMOTE_ADDR": _IP}) target = self._user("t@c.edu") client.post(f"/admin/users/{target.public_id}/suspend", environ_base={"REMOTE_ADDR": _IP}) log = db.session.scalar(db.select(AdminAuditLog).where(AdminAuditLog.action == "user_suspended")) self.assertIsNotNone(log.ip_hash) self.assertNotEqual(log.ip_hash, _IP) self.assertEqual(len(log.ip_hash), 64) # SHA-256 hex digest self.assertNotIn(_IP, log.ip_hash) # --- dashboard recent activity --- def test_dashboard_recent_activity(self): admin = self._admin() for _ in range(7): record_admin_action(admin, "user_suspended", self._user(f"u{_}@c.edu")) recent = get_recent_admin_activity(limit=5) self.assertEqual(len(recent), 5) # capped self.assertEqual(recent[0]["action"], "user_suspended") self.assertEqual(recent[0]["admin_name"], "Admin One") # --- performance --- def test_no_n_plus_1(self): a1 = self._admin("a1@c.edu", name="A One") a2 = self._admin("a2@c.edu", name="A Two") u = self._user("u@c.edu") for i in range(4): record_admin_action(a1 if i % 2 else a2, "user_suspended", u) with _QC() as c1: get_audit_logs(page=1) small = c1.n for i in range(4, 40): record_admin_action(a1 if i % 2 else a2, "user_suspended", u) with _QC() as c2: get_audit_logs(page=1) self.assertEqual(small, c2.n) # constant regardless of row count if __name__ == "__main__": unittest.main()