"""Historical / trend analytics tests (Phase 4.2). Verifies the daily 30-day series (zero-filled), trend summary cards, unique / returning visitor stats, peak day/hour traffic, conversion funnel percentages, data-driven insights and chart JSON — all at the service layer. """ import os import unittest from datetime import datetime, time, timedelta from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from app import create_app from app.extensions import db from app.models import ( Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingView, ListingVisibility, Offer, OfferStatus, ReviewStatus, User, ) from app.models.base import utcnow from app.premium.analytics import _build_insights, _conversion, get_premium_dashboard from app.premium.services import activate_premium class Phase42TrendTests(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="x") db.session.add(u) 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="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): 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("50.00"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, view_count=views, favorite_count=favorites, ) db.session.add(listing) db.session.commit() return listing def _view(self, listing, *, viewer=None, session_id=None, when=None): db.session.add(ListingView( listing_id=listing.id, viewer_id=viewer.id if viewer else None, session_id=session_id, ip_hash=None if viewer else "iphash", viewed_at=when or utcnow(), )) db.session.commit() def _dash(self, user): return get_premium_dashboard(user) # --- daily series --- def test_daily_series_length_and_zero_fill(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._view(listing, session_id="a", when=utcnow()) self._view(listing, session_id="b", when=utcnow() - timedelta(days=5)) daily = self._dash(seller)["daily_views"] self.assertEqual(len(daily), 30) self.assertTrue(all(set(d.keys()) == {"date", "views"} for d in daily)) self.assertEqual(sum(d["views"] for d in daily), 2) self.assertEqual(daily[-1]["views"], 1) # today # A day with no views is present with 0 (zero-filled). self.assertTrue(any(d["views"] == 0 for d in daily)) # Dates are strictly consecutive. dates = [datetime.fromisoformat(d["date"]).date() for d in daily] self.assertEqual(dates[-1] - dates[0], timedelta(days=29)) # --- window counts --- def test_views_today_7d_30d(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._view(listing, session_id="t", when=utcnow()) self._view(listing, session_id="d3", when=utcnow() - timedelta(days=3)) self._view(listing, session_id="d20", when=utcnow() - timedelta(days=20)) self._view(listing, session_id="d40", when=utcnow() - timedelta(days=40)) # outside window t = self._dash(seller)["trends"] self.assertEqual(t["views_today"], 1) self.assertEqual(t["views_7d"], 2) # today + 3d self.assertEqual(t["views_30d"], 3) # excludes the 40-day-old view # --- visitor stats --- def test_unique_and_returning_visitors(self): seller = self._user("s@c.edu") viewer_a = self._user("a@c.edu") viewer_b = self._user("b@c.edu") listing = self._listing(seller) # A: two views (returning). B: one. Anon session S: two (returning). self._view(listing, viewer=viewer_a, when=utcnow()) self._view(listing, viewer=viewer_a, when=utcnow() - timedelta(days=2)) self._view(listing, viewer=viewer_b, when=utcnow()) self._view(listing, session_id="S", when=utcnow()) self._view(listing, session_id="S", when=utcnow() - timedelta(days=1)) t = self._dash(seller)["trends"] self.assertEqual(t["unique_visitors"], 3) # A, B, S self.assertEqual(t["returning_visitors"], 2) # A and S self.assertEqual(t["returning_pct"], round(2 / 3 * 100, 1)) self.assertEqual(t["new_pct"], round(100 - t["returning_pct"], 1)) # --- peak day / hour --- def test_peak_day_and_hour(self): seller = self._user("s@c.edu") listing = self._listing(seller) today = utcnow().date() saturday = today while saturday.weekday() != 5: # 5 = Saturday saturday -= timedelta(days=1) when = datetime.combine(saturday, time(19, 0)) # 7 PM for i in range(3): self._view(listing, session_id=f"s{i}", when=when) traffic = self._dash(seller)["traffic"] self.assertEqual(traffic["most_active_day"], "Saturday") self.assertEqual(traffic["peak_hour"], 19) self.assertEqual(traffic["peak_hour_label"], "7 PM") # --- conversion --- def test_conversion_percentages(self): funnel = _conversion({ "total_views": 1000, "total_favorites": 100, "total_conversations": 40, "total_offers": 20, }) by_label = {s["label"]: s for s in funnel} self.assertIsNone(by_label["Views"]["pct"]) self.assertEqual(by_label["Favorites"]["pct"], 10.0) # 100/1000 self.assertEqual(by_label["Chats"]["pct"], 40.0) # 40/100 self.assertEqual(by_label["Offers"]["pct"], 50.0) # 20/40 def test_conversion_divide_by_zero_safe(self): funnel = _conversion({ "total_views": 0, "total_favorites": 0, "total_conversations": 0, "total_offers": 0, }) self.assertTrue(all(s["pct"] in (None, 0.0) for s in funnel)) # --- insights --- def test_insights_generation(self): traffic = {"most_active_day": "Saturday", "least_active_day": "Sunday", "peak_hour": 19, "peak_hour_label": "7 PM"} visitors = {"unique_visitors": 100, "returning_visitors": 18, "returning_pct": 18.0} comparison = {"multiplier": 3.4} top_rows = [{"title": "Gaming Laptop", "views": 410}] summary = {"total_views": 1000} insights = _build_insights(summary, comparison, traffic, visitors, top_rows) joined = " ".join(insights) self.assertIn("3.4× more views", joined) self.assertIn("Saturday generates the highest traffic", joined) self.assertIn("between 7 PM and 8 PM", joined) self.assertIn("41% of all views", joined) # 410/1000 self.assertIn("Returning visitors represent 18.0% of your audience", joined) def test_insights_empty_when_no_data(self): seller = self._user("s@c.edu") # no listings, no views self.assertEqual(self._dash(seller)["insights"], []) # --- chart JSON structure --- def test_chart_json_structure(self): seller = self._user("s@c.edu") l1 = self._listing(seller, 1, views=100, favorites=10) self._view(l1, session_id="x", when=utcnow()) dash = self._dash(seller) # Daily series shape. self.assertEqual(len(dash["daily_views"]), 30) self.assertIn("date", dash["daily_views"][0]) self.assertIn("views", dash["daily_views"][0]) # Bar-chart series still present. self.assertEqual(len(dash["charts"]["favorites"]), len(dash["charts"]["labels"])) self.assertEqual(len(dash["charts"]["offers"]), len(dash["charts"]["labels"])) # New sections present. for key in ("daily_views", "trends", "traffic", "conversion", "insights"): self.assertIn(key, dash) # --- integration: page renders with a premium user --- def test_analytics_page_renders_with_trends(self): seller = self._user("s@c.edu") seller.password_hash = __import__("werkzeug.security", fromlist=["generate_password_hash"]).generate_password_hash("Passw0rd!") db.session.commit() activate_premium(seller) listing = self._listing(seller, 1, views=50, favorites=5) self._view(listing, session_id="x", when=utcnow()) client = self.app.test_client() client.post("/login", data={"email": "s@c.edu", "password": "Passw0rd!"}) html = client.get("/app/premium/analytics").get_data(as_text=True) self.assertEqual(client.get("/app/premium/analytics").status_code, 200) self.assertIn("Last 30 days", html) self.assertIn("Conversion funnel", html) self.assertIn("dailyViewsChart", html) if __name__ == "__main__": unittest.main()