"""Campus Overview Version 2 — live marketplace analytics tests. Covers the ``app.admin.campus`` composition service (per-university listing, premium, saved, report, message, meetup, insights and health aggregates plus the merged activity feed) and the enriched Campus Overview route. """ 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 import campus as c from app.auth import verification as v from app.extensions import db from app.marketplace.services import add_favorite from app.premium.services import activate_premium from app.models import ( Category, Conversation, ConversationStatus, Favorite, Listing, ListingCondition, ListingStatus, ListingVisibility, Message, Report, ReportEntityType, ReportReason, ReportStatus, ReviewStatus, User, UserProfile, UserRole, ) from app.models.base import utcnow class _QueryCounter: """Count SQL statements issued within the ``with`` block.""" def __enter__(self): self.n = 0 self._engine = db.engine event.listen(self._engine, "after_cursor_execute", self._cb) return self def __exit__(self, *a): event.remove(self._engine, "after_cursor_execute", self._cb) def _cb(self, *a): self.n += 1 class CampusMarketplaceTests(unittest.TestCase): UNI = "Sharda University" OTHER = "Delhi University" def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() self.admin = self._user("admin@c.edu", UserRole.SUPER_ADMIN, "Admin") self._cats = {} def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() # --- factories ------------------------------------------------------------- def _user(self, email, role=UserRole.STUDENT, name="Student"): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=role) u.profile = UserProfile(full_name=name) db.session.add(u) db.session.commit() return u def _seller(self, email, university=None, name="Seller", verified=True): university = university or self.UNI u = self._user(email, name=name) v.submit_verification(u, "9876543210", university, "ID-" + email) if verified: v.approve_verification(u, self.admin) return u def _category(self, name="Electronics"): if name not in self._cats: cat = Category(name=name, description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() self._cats[name] = cat return self._cats[name] def _listing(self, seller, *, status=ListingStatus.AVAILABLE, category="Electronics", title=None, deleted=False): idx = (db.session.scalar(db.select(db.func.count(Listing.id))) or 0) + 1 listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category(category).id, title=title or f"Item {idx}", slug=f"slug-{seller.id}-{idx}", description="d", price=Decimal("100"), condition=ListingCondition.GOOD, status=status, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, deleted_at=(utcnow() if deleted else None), ) db.session.add(listing) db.session.commit() return listing def _favorite(self, user, listing): db.session.add(Favorite(user_id=user.id, listing_id=listing.id)) db.session.commit() def _report(self, reporter, listing, status=ReportStatus.PENDING): db.session.add(Report( reporter_id=reporter.id, entity_type=ReportEntityType.LISTING, entity_public_id=listing.public_id, listing_id=listing.id, reported_user_id=listing.seller_id, reason=ReportReason.SCAM, status=status, resolved_at=(utcnow() if status in (ReportStatus.RESOLVED, ReportStatus.DISMISSED) else None), created_at=utcnow(), )) db.session.commit() def _conversation(self, listing, buyer, status=ConversationStatus.ACTIVE): convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=listing.seller_id, status=status) db.session.add(convo) db.session.commit() return convo def _message(self, convo, sender): db.session.add(Message(conversation_id=convo.id, sender_id=sender.id, message="hi")) db.session.commit() def _admin_client(self): client = self.app.test_client() client.post("/login", data={"email": "admin@c.edu", "password": "Passw0rd!"}) return client def _url(self, name): return "/admin/verifications/university/" + name # --- listings -------------------------------------------------------------- def test_listing_statistics(self): seller = self._seller("s@c.edu") self._listing(seller, status=ListingStatus.AVAILABLE) self._listing(seller, status=ListingStatus.AVAILABLE) self._listing(seller, status=ListingStatus.SOLD) self._listing(seller, status=ListingStatus.RESERVED) self._listing(seller, status=ListingStatus.EXPIRED) self._listing(seller, status=ListingStatus.AVAILABLE, deleted=True) # excluded # A seller at another university — must not leak in. self._listing(self._seller("o@c.edu", self.OTHER), status=ListingStatus.AVAILABLE) s = c.get_university_listing_statistics(self.UNI) self.assertEqual(s["total"], 5) # soft-deleted excluded self.assertEqual(s["active"], 2) self.assertEqual(s["sold"], 1) self.assertEqual(s["reserved"], 1) self.assertEqual(s["expired"], 1) def test_top_categories(self): seller = self._seller("s@c.edu") for _ in range(3): self._listing(seller, category="Books") for _ in range(2): self._listing(seller, category="Electronics") self._listing(seller, category="Furniture") rows = c.get_university_top_categories(self.UNI) self.assertEqual(rows[0], {"category": "Books", "listings": 3}) self.assertEqual(rows[1], {"category": "Electronics", "listings": 2}) # --- premium --------------------------------------------------------------- def test_premium_statistics(self): # 4 verified students, 1 of them premium. premium_seller = self._seller("p@c.edu") activate_premium(premium_seller) for i in range(3): self._seller(f"v{i}@c.edu") # A premium student at another university must not count here. activate_premium(self._seller("op@c.edu", self.OTHER)) s = c.get_university_premium_statistics(self.UNI) self.assertEqual(s["premium_members"], 1) self.assertEqual(s["verified_students"], 4) self.assertEqual(s["premium_rate"], round(1 / 4 * 100, 1)) # 25.0 # --- saved listings -------------------------------------------------------- def test_saved_statistics(self): seller = self._seller("s@c.edu") buyer = self._user("b@c.edu", name="Buyer") l1 = self._listing(seller) l2 = self._listing(seller) self._favorite(buyer, l1) self._favorite(self._user("b2@c.edu"), l1) self._favorite(buyer, l2) s = c.get_university_saved_statistics(self.UNI) self.assertEqual(s["total_favorites"], 3) self.assertEqual(s["avg_per_listing"], round(3 / 2, 1)) # 1.5 # --- reports --------------------------------------------------------------- def test_report_statistics(self): seller = self._seller("s@c.edu") reporter = self._user("r@c.edu") listing = self._listing(seller) self._report(reporter, listing, ReportStatus.PENDING) self._report(reporter, listing, ReportStatus.UNDER_REVIEW) self._report(reporter, listing, ReportStatus.RESOLVED) s = c.get_university_report_statistics(self.UNI) self.assertEqual(s["total"], 3) self.assertEqual(s["pending"], 2) # PENDING + UNDER_REVIEW = open self.assertEqual(s["resolved"], 1) # --- messages -------------------------------------------------------------- def test_message_statistics(self): seller = self._seller("s@c.edu") buyer = self._user("b@c.edu") l1 = self._listing(seller) l2 = self._listing(seller) active = self._conversation(l1, buyer, ConversationStatus.ACTIVE) closed = self._conversation(l2, buyer, ConversationStatus.CLOSED) self._message(active, buyer) self._message(active, seller) self._message(closed, buyer) s = c.get_university_message_statistics(self.UNI) self.assertEqual(s["total_conversations"], 2) self.assertEqual(s["active_conversations"], 1) self.assertEqual(s["total_messages"], 3) # --- meetups --------------------------------------------------------------- def test_meetup_statistics_unavailable(self): s = c.get_university_meetup_statistics(self.UNI) self.assertFalse(s["available"]) # --- insights -------------------------------------------------------------- def test_top_sellers(self): top = self._seller("top@c.edu", name="Top Seller") mid = self._seller("mid@c.edu", name="Mid Seller") for _ in range(3): self._listing(top, status=ListingStatus.AVAILABLE) self._listing(top, status=ListingStatus.SOLD) self._listing(mid, status=ListingStatus.AVAILABLE) rows = c.get_university_top_sellers(self.UNI) self.assertEqual(rows[0]["name"], "Top Seller") self.assertEqual(rows[0]["public_id"], top.public_id) # links to admin.user_detail self.assertEqual(rows[0]["listings"], 4) self.assertEqual(rows[0]["active"], 3) self.assertEqual(rows[0]["sold"], 1) self.assertEqual(rows[1]["name"], "Mid Seller") def test_popular_categories_with_favorites(self): seller = self._seller("s@c.edu") buyer = self._user("b@c.edu") books = self._listing(seller, category="Books") self._listing(seller, category="Books") self._favorite(buyer, books) rows = c.get_university_popular_categories(self.UNI) self.assertEqual(rows[0]["category"], "Books") self.assertEqual(rows[0]["listings"], 2) self.assertEqual(rows[0]["favorites"], 1) def test_most_saved_listings(self): seller = self._seller("s@c.edu", name="Alice") popular = self._listing(seller, title="Popular") quiet = self._listing(seller, title="Quiet") for i in range(3): self._favorite(self._user(f"f{i}@c.edu"), popular) self._favorite(self._user("q@c.edu"), quiet) rows = c.get_university_most_saved_listings(self.UNI) self.assertEqual(rows[0]["title"], "Popular") self.assertEqual(rows[0]["seller"], "Alice") self.assertEqual(rows[0]["saves"], 3) self.assertEqual(rows[0]["public_id"], popular.public_id) def test_most_reported_listings(self): seller = self._seller("s@c.edu") reporter = self._user("r@c.edu") bad = self._listing(seller, title="Bad") ok = self._listing(seller, title="Ok") self._report(reporter, bad) self._report(reporter, bad) self._report(reporter, ok) rows = c.get_university_most_reported_listings(self.UNI) self.assertEqual(rows[0]["title"], "Bad") self.assertEqual(rows[0]["reports"], 2) # --- health ---------------------------------------------------------------- def test_health_metrics(self): # 2 verified students; 1 premium. 4 listings (2 active). 1 report. s1 = self._seller("s1@c.edu") activate_premium(s1) self._seller("s2@c.edu") self._listing(s1, status=ListingStatus.AVAILABLE) self._listing(s1, status=ListingStatus.AVAILABLE) self._listing(s1, status=ListingStatus.SOLD) self._listing(s1, status=ListingStatus.EXPIRED) self._report(self._user("r@c.edu"), self._listing(s1, status=ListingStatus.AVAILABLE)) h = c.get_university_health_metrics(self.UNI) self.assertEqual(h["premium_adoption"], round(1 / 2 * 100, 1)) # 50.0 self.assertEqual(h["active_listing_rate"], round(3 / 5 * 100, 1)) # 60.0 (3 active of 5) self.assertEqual(h["report_rate"], round(1 / 5 * 100, 1)) # 20.0 self.assertIn("verification_rate", h) # --- activity feed --------------------------------------------------------- def test_activity_feed_merges_and_orders(self): seller = self._seller("s@c.edu", name="Alice") # verification events self._listing(seller, title="Chair") # listing event self._report(self._user("r@c.edu"), self._listing(seller, title="Lamp")) # report event feed = c.get_university_activity_feed(self.UNI) types = {e["type"] for e in feed} self.assertEqual(types, {"verification", "listing", "report"}) # Chronologically sorted, newest first. stamps = [e["created_at"] for e in feed] self.assertEqual(stamps, sorted(stamps, reverse=True)) # --- orchestration + performance ------------------------------------------ def test_get_campus_overview_shape(self): seller = self._seller("s@c.edu") self._listing(seller) data = c.get_campus_overview(self.UNI) for key in ("listing", "top_categories", "premium", "saved", "report", "message", "meetup", "insights", "health", "activity_feed"): self.assertIn(key, data) for key in ("top_sellers", "popular_categories", "most_saved", "most_reported"): self.assertIn(key, data["insights"]) def test_get_campus_overview_query_count_bounded(self): seller = self._seller("s@c.edu") for _ in range(10): self._listing(seller) with _QueryCounter() as qc: c.get_campus_overview(self.UNI) # A fixed set of aggregate queries — must not scale with listing count (no N+1). self.assertLess(qc.n, 30) # --- route ----------------------------------------------------------------- def test_route_renders_v2_sections(self): seller = self._seller("s@c.edu", name="Alice Seller") activate_premium(seller) buyer = self._user("b@c.edu") listing = self._listing(seller, title="Nice Desk") self._favorite(buyer, listing) self._report(buyer, listing) self._message(self._conversation(listing, buyer), buyer) html = self._admin_client().get(self._url(self.UNI)).get_data(as_text=True) self.assertIn("Marketplace Snapshot", html) self.assertIn("Total Listings", html) self.assertIn("Premium Members", html) self.assertIn("Marketplace Insights", html) self.assertIn("Top Sellers", html) self.assertIn("Most Popular Categories", html) self.assertIn("Most Saved Listings", html) self.assertIn("Most Reported Listings", html) self.assertIn("Campus Health", html) self.assertIn("Activity Feed", html) self.assertIn("Alice Seller", html) self.assertIn(f"/admin/users/{seller.public_id}", html) # top seller → admin user detail self.assertIn(f"/admin/listings/{listing.public_id}", html) # listing → admin listing detail self.assertIn("Coming Soon", html) # meetups only def test_route_empty_marketplace_state(self): self._seller("s@c.edu") # verified, but no listings html = self._admin_client().get(self._url(self.UNI)).get_data(as_text=True) self.assertIn("No marketplace activity yet", html) self.assertNotIn("Top Sellers", html) if __name__ == "__main__": unittest.main()