"""Admin analytics dashboard tests (Phase 5). Covers access control, platform totals, 30-day trend generation (zero-filled), top-listing ordering, weighted top-seller ranking, premium/advertisement/health metrics, growth insights and chart JSON. """ import os import unittest from datetime import timedelta from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from app import create_app from app.admin.analytics import _seller_score, get_admin_dashboard from app.extensions import db from app.models import ( Advertisement, AdvertisementPlacement, AdvertisementType, Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingView, ListingVisibility, Message, Offer, OfferStatus, ReviewStatus, Subscription, SubscriptionStatus, User, ) from app.models.base import utcnow from app.models.enums import UserRole from app.premium.services import activate_premium class AdminAnalyticsTests(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): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!"), role=role) db.session.add(u) db.session.commit() return u def _category(self, name="Books"): cat = db.session.scalar(db.select(Category).filter_by(name=name)) if cat is None: cat = Category(name=name, 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, favorites=0, status=ListingStatus.AVAILABLE, category="Books"): listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category(category).id, title=f"Item {seller.id}-{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, view_count=views, favorite_count=favorites, ) db.session.add(listing) db.session.commit() return listing def _convo(self, buyer, seller, listing): c = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id, status=ConversationStatus.ACTIVE) db.session.add(c) db.session.commit() return c def _offer(self, convo): o = Offer(listing_id=convo.listing_id, conversation_id=convo.id, buyer_id=convo.buyer_id, seller_id=convo.seller_id, amount=Decimal("40"), status=OfferStatus.PENDING) db.session.add(o) db.session.commit() return o def _view(self, listing, session_id="s", when=None): db.session.add(ListingView(listing_id=listing.id, session_id=session_id, ip_hash="h", viewed_at=when or utcnow())) db.session.commit() # --- access --- def test_admin_can_access(self): self._user("a@c.edu", UserRole.SUPER_ADMIN) client = self.app.test_client() client.post("/login", data={"email": "a@c.edu", "password": "Passw0rd!"}) resp = client.get("/admin/analytics") self.assertEqual(resp.status_code, 200) self.assertIn("Platform Analytics", resp.get_data(as_text=True)) def test_non_admin_denied(self): self._user("s@c.edu", UserRole.STUDENT) client = self.app.test_client() client.post("/login", data={"email": "s@c.edu", "password": "Passw0rd!"}) self.assertEqual(client.get("/admin/analytics").status_code, 403) # --- platform totals --- def test_platform_totals(self): seller = self._user("sell@c.edu") buyer = self._user("buy@c.edu") l1 = self._listing(seller, 1, views=100, favorites=5) l2 = self._listing(seller, 2, views=50, status=ListingStatus.SOLD) convo = self._convo(buyer, seller, l1) self._offer(convo) db.session.add(Message(conversation_id=convo.id, sender_id=buyer.id, message="hi")) db.session.commit() o = get_admin_dashboard()["overview"] self.assertEqual(o["total_users"], 2) self.assertEqual(o["total_listings"], 2) self.assertEqual(o["active_listings"], 1) self.assertEqual(o["sold_listings"], 1) self.assertEqual(o["total_views"], 150) self.assertEqual(o["total_favorites"], 5) self.assertEqual(o["total_conversations"], 1) self.assertEqual(o["total_messages"], 1) self.assertEqual(o["total_offers"], 1) # --- trends --- def test_trend_generation_and_zero_fill(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._view(listing, "a", when=utcnow()) self._view(listing, "b", when=utcnow() - timedelta(days=4)) dash = get_admin_dashboard() views_series = dash["activity"]["views"] self.assertEqual(len(views_series), 30) self.assertTrue(all(set(d) == {"date", "count"} for d in views_series)) self.assertEqual(sum(d["count"] for d in views_series), 2) self.assertTrue(any(d["count"] == 0 for d in views_series)) # zero-filled # New users trend counts the seller created today. self.assertEqual(len(dash["activity"]["new_users"]), 30) self.assertGreaterEqual(sum(d["count"] for d in dash["activity"]["new_users"]), 1) # --- top listings ordering --- def test_top_listings_ordering(self): seller = self._user("s@c.edu") self._listing(seller, 1, views=10) self._listing(seller, 2, views=90) self._listing(seller, 3, views=50) rows = get_admin_dashboard()["top_listings"] self.assertEqual([r["views"] for r in rows], [90, 50, 10]) self.assertIn("seller", rows[0]) # --- top sellers ranking --- def test_seller_score_formula(self): self.assertEqual(_seller_score(100, 6, 2), 100 + 6 * 5 + 2 * 10) # 150 def test_top_sellers_ranking(self): a = self._user("a@c.edu") b = self._user("b@c.edu") c = self._user("c@c.edu") self._listing(a, 1, views=100, favorites=0) # score 100 self._listing(b, 1, views=0, favorites=50) # score 250 self._listing(c, 1, views=200, favorites=0) # score 200 activate_premium(b) rows = get_admin_dashboard()["top_sellers"] self.assertEqual([r["name"] for r in rows], ["b@c.edu", "c@c.edu", "a@c.edu"]) self.assertEqual(rows[0]["score"], 250) self.assertTrue(rows[0]["is_premium"]) self.assertFalse(rows[1]["is_premium"]) # --- premium --- def test_premium_statistics(self): active = self._user("act@c.edu") activate_premium(active) expired = self._user("exp@c.edu") db.session.add(Subscription(user_id=expired.id, plan_name="monthly", status=SubscriptionStatus.EXPIRED, start_date=utcnow() - timedelta(days=40), end_date=utcnow() - timedelta(days=1))) self._user("free@c.edu") db.session.commit() p = get_admin_dashboard()["premium"] self.assertEqual(p["active_members"], 1) self.assertEqual(p["expired_members"], 1) self.assertEqual(p["adoption_pct"], round(1 / 3 * 100, 1)) self.assertEqual(p["conversion_pct"], round(1 / 3 * 100, 1)) # --- advertisements --- def test_advertisement_statistics(self): now = utcnow() db.session.add(Advertisement(title="A", advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.SIDEBAR, is_active=True, priority=1, impression_count=200, click_count=10)) db.session.add(Advertisement(title="B", advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.SIDEBAR, is_active=True, priority=1, end_at=now - timedelta(days=1), impression_count=0, click_count=0)) db.session.commit() ad = get_admin_dashboard()["advertisements"] self.assertEqual(ad["total"], 2) self.assertEqual(ad["active"], 1) self.assertEqual(ad["expired"], 1) self.assertEqual(ad["impressions"], 200) self.assertEqual(ad["clicks"], 10) self.assertEqual(ad["ctr"], 5.0) # 10/200 # --- health --- def test_marketplace_health(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") l1 = self._listing(seller, 1, views=100, favorites=10) l2 = self._listing(seller, 2, views=50, favorites=6) convo = self._convo(buyer, seller, l1) db.session.add(Message(conversation_id=convo.id, sender_id=buyer.id, message="a")) db.session.add(Message(conversation_id=convo.id, sender_id=seller.id, message="b")) db.session.commit() h = get_admin_dashboard()["health"] self.assertEqual(h["avg_views_per_listing"], 75.0) # (100+50)/2 self.assertEqual(h["avg_favorites_per_listing"], 8.0) # (10+6)/2 self.assertEqual(h["avg_conversations_per_listing"], 0.5) # 1/2 self.assertEqual(h["avg_messages_per_conversation"], 2.0) # 2/1 def test_health_divide_by_zero_safe(self): h = get_admin_dashboard()["health"] # no data self.assertEqual(h["avg_views_per_listing"], 0.0) self.assertEqual(h["premium_pct"], 0.0) # --- insights --- def test_growth_insights(self): seller = self._user("s@c.edu") listing = self._listing(seller, 1, views=42, category="Electronics") self._view(listing, "s", when=utcnow()) insights = get_admin_dashboard()["insights"] joined = " ".join(insights) self.assertIn("new users this month", joined) self.assertIn("Electronics received the highest traffic", joined) self.assertIn("Average listing receives 42 views", joined) # --- chart JSON --- def test_chart_json_structure(self): seller = self._user("s@c.edu") self._listing(seller, 1, views=10) charts = get_admin_dashboard()["charts"] for key in ("users", "listings", "views", "messages", "offers"): self.assertEqual(len(charts[key]["labels"]), 30) self.assertEqual(len(charts[key]["values"]), 30) for key in ("top_categories", "top_sellers", "status", "premium_free"): self.assertIn("labels", charts[key]) self.assertIn("values", charts[key]) self.assertEqual(len(charts[key]["labels"]), len(charts[key]["values"])) self.assertEqual(charts["premium_free"]["labels"], ["Premium", "Free"]) if __name__ == "__main__": unittest.main()