"""Smart discovery & recommendation tests (Phase 8).""" import os import unittest from datetime import timedelta 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.extensions import db from app.marketplace.discovery import ( get_browse_categories, get_homepage_sections, get_recently_viewed, get_recommended_for_you, get_similar_listings, get_trending_listings, ) from app.models import ( Category, Listing, ListingCondition, ListingStatus, ListingView, ListingVisibility, Report, ReportEntityType, ReportReason, ReportStatus, ReviewStatus, User, ) from app.models.base import utcnow from app.models.enums import UserRole class _QC: def __init__(self): self.n = 0 def __enter__(self): self._e = db.engine event.listen(self._e, "after_cursor_execute", self._cb) return self def __exit__(self, *a): event.remove(self._e, "after_cursor_execute", self._cb) def _cb(self, *a): self.n += 1 class _Anon: is_authenticated = False id = None class DiscoveryTests(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() # --- factories --- def _user(self, email): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!"), role=UserRole.STUDENT) 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, *, category="Books", price="50", views=0, favs=0, condition=ListingCondition.GOOD, featured=False, status=ListingStatus.AVAILABLE): cat = self._category(category) l = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=cat.id, title=f"Item {idx}", slug=f"item-{seller.id}-{idx}", description="d", price=Decimal(price), condition=condition, status=status, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, view_count=views, favorite_count=favs, ) if featured: l.is_featured = True l.featured_at = utcnow() l.featured_until = utcnow() + timedelta(days=7) db.session.add(l) db.session.commit() return l def _view(self, listing, viewer=None, session_id="sess", when=None): db.session.add(ListingView( listing_id=listing.id, viewer_id=(viewer.id if viewer else None), session_id=session_id, viewed_at=when or utcnow(), )) db.session.commit() def _report(self, listing, reporter): 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=ReportStatus.PENDING, created_at=utcnow(), )) db.session.commit() # --- trending --- def test_trending_ranking(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") hot = self._listing(seller, 1, favs=10) cold = self._listing(seller, 2) for _ in range(30): self._view(hot, viewer) trending = get_trending_listings() self.assertEqual(trending[0].id, hot.id) self.assertLess(trending.index(hot), trending.index(cold)) def test_trending_report_penalty(self): seller = self._user("s@c.edu") reporter = self._user("r@c.edu") # Two otherwise-equal listings; the reported one should rank lower. clean = self._listing(seller, 1, favs=1) flagged = self._listing(seller, 2, favs=1) self._report(flagged, reporter) trending = get_trending_listings() self.assertLess(trending.index(clean), trending.index(flagged)) # --- recommendations --- def test_recommendation_excludes_own_and_viewed(self): viewer = self._user("v@c.edu") other = self._user("o@c.edu") viewed = self._listing(other, 1) self._view(viewed, viewer) # already viewed -> excluded own = self._listing(viewer, 2) # own listing -> excluded candidate = self._listing(other, 3) # should be recommended recs = get_recommended_for_you(viewer) ids = {l.id for l in recs} self.assertIn(candidate.id, ids) self.assertNotIn(viewed.id, ids) self.assertNotIn(own.id, ids) def test_anonymous_recommendations_fall_back_to_trending(self): seller = self._user("s@c.edu") self._listing(seller, 1, favs=5) self._listing(seller, 2) anon_recs = get_recommended_for_you(_Anon()) trending = get_trending_listings() self.assertEqual([l.id for l in anon_recs], [l.id for l in trending]) # --- recently viewed --- def test_recently_viewed_ordering_and_dedup(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") l1 = self._listing(seller, 1) l2 = self._listing(seller, 2) base = utcnow() self._view(l1, viewer, when=base - timedelta(minutes=10)) self._view(l2, viewer, when=base - timedelta(minutes=5)) self._view(l1, viewer, when=base) # l1 viewed again, most recent recent = get_recently_viewed(viewer) self.assertEqual([l.id for l in recent], [l1.id, l2.id]) # deduped, newest first # --- similar --- def test_similar_listings(self): seller = self._user("s@c.edu") target = self._listing(seller, 1, category="Books", price="100") same = self._listing(seller, 2, category="Books", price="110") other_cat = self._listing(seller, 3, category="Electronics", price="105") similar = get_similar_listings(target) ids = {l.id for l in similar} self.assertIn(same.id, ids) self.assertNotIn(target.id, ids) # excludes itself self.assertNotIn(other_cat.id, ids) # different category # --- categories --- def test_category_counts(self): seller = self._user("s@c.edu") self._listing(seller, 1, category="Books") self._listing(seller, 2, category="Books") self._listing(seller, 3, category="Electronics") cats = {c["category"].name: c["count"] for c in get_browse_categories()} self.assertEqual(cats["Books"], 2) self.assertEqual(cats["Electronics"], 1) # A category with no available listings is omitted. self._category("Empty") self.assertNotIn("Empty", {c["category"].name for c in get_browse_categories()}) # --- homepage aggregate --- def test_homepage_section_generation(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") self._listing(seller, 1, featured=True) self._listing(seller, 2) sections = get_homepage_sections(viewer) self.assertEqual( set(sections.keys()), {"trending", "featured", "featured_exists", "recommendations", "popular_this_week", "new_arrivals", "recently_viewed", "categories"}, ) self.assertTrue(sections["featured"]) self.assertTrue(sections["featured_exists"]) self.assertTrue(sections["new_arrivals"]) # --- performance --- def test_no_n_plus_1(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") for i in range(5): self._listing(seller, i) with _QC() as c1: get_homepage_sections(viewer) small = c1.n for i in range(5, 30): self._listing(seller, i) with _QC() as c2: get_homepage_sections(viewer) self.assertEqual(small, c2.n) # constant regardless of catalog size if __name__ == "__main__": unittest.main()