06/16 Phase 6 foundation: admin blueprint, audit log, user management

Adds the admin backend foundation: AuditLog + Setting models/migration,
an admin-only dashboard with KPI cards, and user management (search,
ban/suspend/activate, tier override, trust adjust) with audit logging
on every write action. Smoke suite extended to 64 checks.
This commit is contained in:
2026-06-16 11:45:00 -04:00
parent 62b04803ad
commit 2e9025fe55
18 changed files with 680 additions and 1 deletions
+120
View File
@@ -478,6 +478,125 @@ def _phase5(app):
print("Phase 5 route renders: ok")
def _phase6(app):
"""Phase 6 foundation: audit log, settings, admin dashboard + user mgmt."""
import re
from app.models.user import User
from app.models.plan import Plan
from app.models.listing import Listing
from app.models.audit import AuditLog
from app.models.enums import Role, UserStatus, TrustEventType
from app.services import admin_users as usvc
from app.services import admin_dashboard as dash
from app.services.settings import get_setting, set_setting
with app.app_context():
admin = User(email="admin@example.com", display_name="Admin",
role=Role.admin, email_verified=True)
admin.set_password("AdminPass123")
admin.tier_id = Plan.query.filter_by(slug="free").first().id
db.session.add(admin); db.session.commit()
target = User.query.filter_by(email="t@example.com").first()
# --- settings round-trip ---
assert get_setting("flag_threshold", 5) == 5
set_setting("flag_threshold", 3)
db.session.commit()
assert get_setting("flag_threshold") == 3
print("settings round-trip: ok")
# --- dashboard KPIs return sane values ---
assert dash.active_listings_count() >= 0
assert dash.new_users_count(30) >= 2
assert dash.mrr_cents() >= 0
assert dash.revenue_30d_cents() >= 0
assert dash.flag_queue_depth() >= 0
print("dashboard KPI helpers: ok")
# --- ban + audit log ---
usvc.set_status(target, UserStatus.banned, actor=admin)
db.session.commit()
target_fresh = db.session.get(User, target.id)
assert target_fresh.status == UserStatus.banned
assert not target_fresh.is_active
log = AuditLog.query.filter_by(target_type="user", target_id=target.id,
action="user.banned").first()
assert log is not None and log.actor_id == admin.id
print("ban + audit log: ok")
# --- tier override + audit log ---
pro_plan = Plan.query.filter_by(slug="pro").first()
usvc.set_tier(target, pro_plan, actor=admin)
db.session.commit()
assert db.session.get(User, target.id).tier_id == pro_plan.id
assert AuditLog.query.filter_by(action="user.tier_override",
target_id=target.id).count() == 1
print("tier override + audit log: ok")
# --- trust adjust + audit log ---
score_before = target.trust_score
usvc.adjust_trust(target, TrustEventType.payment, 10, actor=admin)
db.session.commit()
assert db.session.get(User, target.id).trust_score == score_before + 10
assert AuditLog.query.filter_by(action="user.trust_adjust",
target_id=target.id).count() == 1
print("trust adjust + audit log: ok")
# --- reactivate so later route checks reflect a normal account ---
usvc.set_status(target, UserStatus.active, actor=admin)
db.session.commit()
# --- route checks ---
c = app.test_client(); B = "https://localhost"
def csrf(html):
return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', html).group(1)
def login(email, password):
r = c.get("/auth/login", base_url=B)
tok = csrf(r.get_data(as_text=True))
return c.post("/auth/login", base_url=B,
data={"csrf_token": tok, "email": email, "password": password},
headers={"Referer": B + "/auth/login"}, follow_redirects=True)
# non-admin gets 403
login("t@example.com", "NewPass456")
assert c.get("/admin", base_url=B).status_code == 403
print("non-admin /admin -> 403: ok")
c.get("/auth/logout", base_url=B)
# admin gets 200 on dashboard/users/user_detail
login("admin@example.com", "AdminPass123")
with app.app_context():
target_id = User.query.filter_by(email="t@example.com").first().id
for path in ("/admin", "/admin/users", f"/admin/users/{target_id}"):
code = c.get(path, base_url=B).status_code
assert code == 200, f"{path} -> {code}"
print(f"{code} {path}")
print("Phase 6 route renders: ok")
# ban via route, confirm banned user's login is rejected
r = c.get(f"/admin/users/{target_id}", base_url=B)
tok = csrf(r.get_data(as_text=True))
c.post(f"/admin/users/{target_id}/ban", base_url=B,
data={"csrf_token": tok},
headers={"Referer": B + f"/admin/users/{target_id}"},
follow_redirects=True)
with app.app_context():
assert db.session.get(User, target_id).status == UserStatus.banned
c.get("/auth/logout", base_url=B)
r = login("t@example.com", "NewPass456")
assert "suspended" in r.get_data(as_text=True).lower()
print("banned user login rejected: ok")
# restore target to active for cleanliness (not strictly needed, smoke ends here)
with app.app_context():
u = User.query.filter_by(email="t@example.com").first()
u.status = UserStatus.active
db.session.commit()
def run():
app = create_app()
with app.app_context():
@@ -577,6 +696,7 @@ def run():
_phase3(app)
_phase4(app)
_phase5(app)
_phase6(app)
print("\nALL SMOKE CHECKS PASSED")