"""Premium analytics dashboard tests (Phase 4). Exercises access control and every section of ``get_premium_dashboard`` — totals, featured card, expired-featured exclusion, top-listing ordering, the featured-vs- average comparison (incl. divide-by-zero), chart JSON, and recent-activity order. All calculations are verified at the service layer. """ 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.extensions import db from app.marketplace.services import feature_listing from app.models import ( Category, Conversation, ConversationStatus, Favorite, Listing, ListingCondition, ListingStatus, ListingVisibility, Offer, OfferStatus, ReviewStatus, User, ) from app.models.base import utcnow from app.premium.analytics import get_premium_dashboard from app.premium.services import activate_premium class Phase4AnalyticsTests(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): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!")) db.session.add(u) db.session.commit() return u def _premium(self, email): u = self._user(email) activate_premium(u) 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="B", 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, price="50.00"): 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(price), 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, when=None): c = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id, status=ConversationStatus.ACTIVE) db.session.add(c) db.session.commit() if when: c.created_at = when db.session.commit() return c def _offer(self, convo, amount, when=None): o = Offer(listing_id=convo.listing_id, conversation_id=convo.id, buyer_id=convo.buyer_id, seller_id=convo.seller_id, amount=Decimal(amount), status=OfferStatus.PENDING) db.session.add(o) db.session.commit() if when: o.created_at = when db.session.commit() return o def _favorite(self, user, listing, when=None): f = Favorite(user_id=user.id, listing_id=listing.id) db.session.add(f) db.session.commit() if when: f.created_at = when db.session.commit() return f # --- access --- def test_premium_user_can_access(self): self._premium("p@c.edu") client = self.app.test_client() client.post("/login", data={"email": "p@c.edu", "password": "Passw0rd!"}) resp = client.get("/app/premium/analytics") self.assertEqual(resp.status_code, 200) self.assertIn("listing performance", resp.get_data(as_text=True).lower()) def test_non_premium_redirected(self): self._user("f@c.edu") client = self.app.test_client() client.post("/login", data={"email": "f@c.edu", "password": "Passw0rd!"}) resp = client.get("/app/premium/analytics") self.assertEqual(resp.status_code, 302) self.assertIn("/app/premium", resp.headers.get("Location", "")) # --- summary totals --- def test_summary_totals(self): seller = self._premium("p@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=5, status=ListingStatus.SOLD) c1 = self._convo(buyer, seller, l1) c2 = self._convo(buyer, seller, l2) self._offer(c1, "40") self._offer(c1, "42") self._offer(c2, "30") s = get_premium_dashboard(seller)["summary"] self.assertEqual(s["total_listings"], 2) self.assertEqual(s["active_listings"], 1) # l2 is SOLD self.assertEqual(s["total_views"], 150) self.assertEqual(s["total_favorites"], 15) self.assertEqual(s["total_offers"], 3) self.assertEqual(s["total_conversations"], 2) self.assertEqual(s["featured_listings"], 0) # --- featured card --- def test_featured_card_appears(self): seller = self._premium("p@c.edu") l1 = self._listing(seller, 1, views=100, favorites=8) feature_listing(l1, seller) buyer = self._user("b@c.edu") c1 = self._convo(buyer, seller, l1) self._offer(c1, "45") dash = get_premium_dashboard(seller) self.assertIsNotNone(dash["featured"]) self.assertEqual(dash["featured"]["title"], l1.title) self.assertEqual(dash["featured"]["views"], 100) self.assertEqual(dash["featured"]["offers"], 1) self.assertEqual(dash["featured"]["conversations"], 1) self.assertEqual(dash["featured"]["days_remaining"], 7) self.assertEqual(dash["summary"]["featured_listings"], 1) def test_expired_featured_not_shown(self): seller = self._premium("p@c.edu") l1 = self._listing(seller, 1, views=100) feature_listing(l1, seller) l1.featured_until = utcnow() - timedelta(seconds=1) # expired db.session.commit() dash = get_premium_dashboard(seller) self.assertIsNone(dash["featured"]) self.assertEqual(dash["summary"]["featured_listings"], 0) self.assertIsNone(dash["comparison"]) # no featured -> no comparison # --- top listings ordering --- def test_top_listings_ordering(self): seller = self._premium("p@c.edu") self._listing(seller, 1, views=10) self._listing(seller, 2, views=90) self._listing(seller, 3, views=50) rows = get_premium_dashboard(seller)["top_listings"] self.assertEqual([r["views"] for r in rows], [90, 50, 10]) # --- comparison --- def test_comparison_calculation(self): seller = self._premium("p@c.edu") featured = self._listing(seller, 1, views=100) self._listing(seller, 2, views=50) # avg = (100+50)/2 = 75 feature_listing(featured, seller) cmp = get_premium_dashboard(seller)["comparison"] self.assertIsNotNone(cmp) self.assertEqual(cmp["average_views"], 75) self.assertEqual(cmp["featured_views"], 100) self.assertEqual(cmp["multiplier"], round(100 / 75, 1)) def test_comparison_divide_by_zero_safe(self): seller = self._premium("p@c.edu") # no listings at all dash = get_premium_dashboard(seller) self.assertIsNone(dash["comparison"]) self.assertEqual(dash["summary"]["total_listings"], 0) self.assertEqual(dash["charts"]["labels"], []) # --- charts --- def test_chart_json_generation(self): seller = self._premium("p@c.edu") buyer = self._user("b@c.edu") l1 = self._listing(seller, 1, views=100, favorites=10) l2 = self._listing(seller, 2, views=40, favorites=4) c1 = self._convo(buyer, seller, l1) self._offer(c1, "45") charts = get_premium_dashboard(seller)["charts"] self.assertEqual(len(charts["labels"]), 2) self.assertEqual(charts["views"], [100, 40]) # most-viewed first self.assertEqual(charts["favorites"], [10, 4]) self.assertEqual(charts["offers"], [1, 0]) # l1 has one offer # --- recent activity ordering --- def test_recent_activity_ordering(self): seller = self._premium("p@c.edu") buyer = self._user("b@c.edu") listing = self._listing(seller, 1) convo = self._convo(buyer, seller, listing, when=utcnow() - timedelta(days=3)) self._offer(convo, "30", when=utcnow() - timedelta(hours=2)) self._favorite(buyer, listing, when=utcnow() - timedelta(minutes=10)) activity = get_premium_dashboard(seller)["recent_activity"] self.assertLessEqual(len(activity), 5) types = [a["type"] for a in activity] self.assertEqual(types, ["favorite", "offer", "chat"]) # newest first if __name__ == "__main__": unittest.main()