"""Admin user-management tests (Phase 6.0).""" import os import unittest 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.user_management import get_user_summary, get_users from app.extensions import db from app.models import ( Category, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, User, UserProfile, ) from app.models.enums import AccountStatus, UserRole from app.premium.services import activate_premium, is_user_currently_premium 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 AdminUserManagementTests(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() def _user(self, email, role=UserRole.STUDENT, name=None, status=AccountStatus.ACTIVE): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!"), role=role, account_status=status) 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, views=0): db.session.add(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=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, view_count=views, )) db.session.commit() def _admin_client(self): self._user("admin@c.edu", UserRole.SUPER_ADMIN) client = self.app.test_client() client.post("/login", data={"email": "admin@c.edu", "password": "Passw0rd!"}) return client # --- access --- def test_admin_access(self): client = self._admin_client() self.assertEqual(client.get("/admin/users").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/users").status_code, 403) # --- search --- def test_search_by_name_and_email(self): self._user("alice@c.edu", name="Alice Adams") self._user("bob@c.edu", name="Bob Brown") _, rows = get_users(search="alice") self.assertEqual([r["email"] for r in rows], ["alice@c.edu"]) _, rows = get_users(search="Brown") self.assertEqual([r["email"] for r in rows], ["bob@c.edu"]) # --- filters --- def test_premium_filter(self): prem = self._user("prem@c.edu") activate_premium(prem) self._user("free@c.edu") _, rows = get_users(premium="premium") self.assertEqual([r["email"] for r in rows], ["prem@c.edu"]) _, rows = get_users(premium="free") self.assertEqual([r["email"] for r in rows], ["free@c.edu"]) def test_status_filter(self): self._user("act@c.edu") self._user("sus@c.edu", status=AccountStatus.SUSPENDED) _, rows = get_users(status="suspended") self.assertEqual([r["email"] for r in rows], ["sus@c.edu"]) _, rows = get_users(status="active") self.assertEqual(sorted(r["email"] for r in rows), ["act@c.edu"]) def test_role_filter(self): self._user("stu@c.edu") self._user("adm@c.edu", UserRole.SUPER_ADMIN) _, rows = get_users(role="admin") self.assertEqual([r["email"] for r in rows], ["adm@c.edu"]) # --- sort --- def test_sort_most_listings(self): a = self._user("a@c.edu") b = self._user("b@c.edu") for i in range(3): self._listing(a, i) self._listing(b, 1) _, rows = get_users(sort="most_listings") self.assertEqual(rows[0]["email"], "a@c.edu") self.assertEqual(rows[0]["listings"], 3) def test_sort_newest_oldest(self): first = self._user("first@c.edu") second = self._user("second@c.edu") _, rows = get_users(sort="oldest") self.assertEqual(rows[0]["email"], "first@c.edu") _, rows = get_users(sort="newest") self.assertEqual(rows[0]["email"], "second@c.edu") # --- pagination --- def test_pagination(self): for i in range(25): self._user(f"u{i}@c.edu") pagination, rows = get_users(page=1) self.assertEqual(pagination.per_page, 20) self.assertEqual(len(rows), 20) self.assertEqual(pagination.pages, 2) pagination2, rows2 = get_users(page=2) self.assertEqual(len(rows2), 5) # --- summary --- def test_summary_totals(self): a = self._user("a@c.edu") activate_premium(a) self._user("b@c.edu") self._user("c@c.edu", status=AccountStatus.SUSPENDED) self._listing(a, 1) self._listing(a, 2) s = get_user_summary() self.assertEqual(s["total_users"], 3) self.assertEqual(s["premium_users"], 1) self.assertEqual(s["free_users"], 2) self.assertEqual(s["suspended_users"], 1) self.assertEqual(s["avg_listings_per_user"], round(2 / 3, 1)) # --- action visibility --- def test_action_visibility(self): client = self._admin_client() free = self._user("free@c.edu") html = client.get(f"/admin/users/{free.public_id}").get_data(as_text=True) self.assertIn("Grant Premium", html) self.assertIn("Suspend Account", html) # student, not self activate_premium(free) html = client.get(f"/admin/users/{free.public_id}").get_data(as_text=True) self.assertIn("Remove Premium", html) def test_admin_cannot_be_suspended_in_ui(self): client = self._admin_client() # admin@c.edu is the logged-in admin admin = db.session.scalar(db.select(User).where(User.email == "admin@c.edu")) html = client.get(f"/admin/users/{admin.public_id}").get_data(as_text=True) self.assertNotIn("Suspend Account", html) # cannot suspend self / admin # --- actions perform --- def test_grant_and_remove_premium(self): client = self._admin_client() u = self._user("u@c.edu") client.post(f"/admin/users/{u.public_id}/grant-premium") db.session.expire_all() self.assertTrue(is_user_currently_premium(db.session.get(User, u.id))) client.post(f"/admin/users/{u.public_id}/remove-premium") db.session.expire_all() self.assertFalse(is_user_currently_premium(db.session.get(User, u.id))) def test_suspend_and_reactivate(self): client = self._admin_client() u = self._user("u@c.edu") client.post(f"/admin/users/{u.public_id}/suspend") db.session.expire_all() self.assertEqual(db.session.get(User, u.id).account_status, AccountStatus.SUSPENDED) client.post(f"/admin/users/{u.public_id}/reactivate") db.session.expire_all() self.assertEqual(db.session.get(User, u.id).account_status, AccountStatus.ACTIVE) def test_cannot_suspend_self(self): client = self._admin_client() admin = db.session.scalar(db.select(User).where(User.email == "admin@c.edu")) client.post(f"/admin/users/{admin.public_id}/suspend") db.session.expire_all() self.assertEqual(db.session.get(User, admin.id).account_status, AccountStatus.ACTIVE) # --- performance --- def test_no_n_plus_1(self): for i in range(4): u = self._user(f"u{i}@c.edu") self._listing(u, i) with _QC() as c1: get_users(page=1) small = c1.n for i in range(4, 40): u = self._user(f"u{i}@c.edu") self._listing(u, i) with _QC() as c2: get_users(page=1) self.assertEqual(small, c2.n) # constant regardless of user count if __name__ == "__main__": unittest.main()