"""Premium Seller Analytics tests.""" 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.models import ( Category, CompletedTransaction, Conversation, Listing, ListingCondition, ListingStatus, ListingVisibility, Offer, OfferStatus, Review, ReviewStatus, User, UserReputation, ) from app.models.base import utcnow from app.marketplace.analytics import get_dashboard_analytics from app.premium.services import activate_premium class SellerAnalyticsTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() self.cat = Category(name="Electronics", description="d", is_active=True, display_order=1) db.session.add(self.cat) db.session.commit() def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() # --- factories --- 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 _login(self, email): c = self.app.test_client() c.post("/login", data={"email": email, "password": "Passw0rd!"}) return c def _rep(self, user, **kwargs): db.session.add(UserReputation(user_id=user.id, **kwargs)) db.session.commit() def _listing(self, seller, title="Item", *, views=0, favorites=0, status=ListingStatus.AVAILABLE, idx=None, category=None): idx = idx if idx is not None else (db.session.scalar(db.select(db.func.count(Listing.id))) + 1) listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=(category or self.cat).id, title=title, slug=f"slug-{seller.id}-{idx}", description="d", price=Decimal("100"), 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 _completed_sale(self, seller, buyer, listing, rating=5): convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id) db.session.add(convo) db.session.commit() offer = Offer(listing_id=listing.id, conversation_id=convo.id, buyer_id=buyer.id, seller_id=seller.id, amount=Decimal("90"), status=OfferStatus.COMPLETED) db.session.add(offer) db.session.commit() tx = CompletedTransaction(listing_id=listing.id, offer_id=offer.id, buyer_id=buyer.id, seller_id=seller.id, completed_at=utcnow()) db.session.add(tx) db.session.commit() db.session.add(Review(transaction_id=tx.id, reviewer_id=buyer.id, reviewee_id=seller.id, rating=rating)) db.session.commit() return tx # --- access control (route) --- def test_premium_sees_analytics(self): seller = self._premium("s@c.edu") self._listing(seller, "Item", views=100) html = self._login("s@c.edu").get("/app/my-listings").get_data(as_text=True) self.assertIn("Premium Insights", html) self.assertIn("Conversion Rate", html) self.assertNotIn("Upgrade to Premium", html) def test_free_sees_upgrade_card(self): seller = self._user("s@c.edu") self._listing(seller, "Item") html = self._login("s@c.edu").get("/app/my-listings").get_data(as_text=True) self.assertIn("Premium Insights", html) self.assertIn("Upgrade to Premium", html) self.assertNotIn("Conversion Rate", html) # analytics body hidden for free # --- metrics --- def test_conversion_rate_guards_zero_and_computes(self): seller = self._user("s@c.edu") # no views yet → guarded to 0 self.assertEqual(get_dashboard_analytics(seller)["summary"]["conversion_rate"], 0.0) self._listing(seller, "Item", views=200) self._rep(seller, completed_sales=2, average_rating=Decimal("4.8"), review_count=3) self.assertEqual(get_dashboard_analytics(seller)["summary"]["conversion_rate"], 1.0) # 2/200 def test_best_listing_is_highest_scoring(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") # score = views*1 + favorites*4 + messages*5 + offers*8 + (sale ? 50) self._listing(seller, "Weak", views=40, idx=1) # score 40 strong = self._listing(seller, "Strong", views=50, favorites=5, idx=2) # 50+20+50(sale)=120 self._completed_sale(seller, buyer, strong) best = get_dashboard_analytics(seller)["best_listing"] self.assertEqual(best["title"], "Strong") def test_weekly_activity(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") listing = self._listing(seller, "Item") # recent conversation + offer count toward this week convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id) db.session.add(convo); db.session.commit() db.session.add(Offer(listing_id=listing.id, conversation_id=convo.id, buyer_id=buyer.id, seller_id=seller.id, amount=Decimal("50"), status=OfferStatus.PENDING)) db.session.commit() weekly = get_dashboard_analytics(seller)["weekly"] self.assertEqual(weekly["messages"], 1) self.assertEqual(weekly["offers"], 1) def test_recommendations_are_capped_and_relevant(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") strong = self._listing(seller, "MacBook", views=100, favorites=30) self._completed_sale(seller, buyer, strong) self._rep(seller, completed_sales=1, average_rating=Decimal("4.9"), review_count=5) recs = get_dashboard_analytics(seller)["recommendations"] self.assertLessEqual(len(recs), 3) self.assertTrue(any("MacBook" in r for r in recs)) # best listing insight self.assertTrue(any("rate you" in r for r in recs)) # rating insight def test_completed_sale_and_rating_in_table(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") sold = self._listing(seller, "Sold item", views=10, status=ListingStatus.SOLD) self._completed_sale(seller, buyer, sold, rating=4) rows = get_dashboard_analytics(seller)["listings"] row = next(r for r in rows if r["title"] == "Sold item") self.assertTrue(row["completed_sale"]) self.assertEqual(row["rating"], 4) # --- empty state --- def test_empty_state_for_new_seller(self): data = get_dashboard_analytics(self._user("new@c.edu")) self.assertFalse(data["has_data"]) self.assertEqual(data["listings"], []) self.assertIsNone(data["best_listing"]) if __name__ == "__main__": unittest.main()