"""Phase 2B: featured listings + premium ordering tests. Covers the single-featured rule, feature replacement, permissions/security, catalog ordering (featured first, premium tier without changing relevance), badge rendering and the absence of an ordering N+1. """ 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 ( FeatureListingError, feature_listing, get_featured_listing, is_listing_featured, mark_listing_as_sold, search_published_listings, soft_delete_listing, unfeature_listing, ) from app.models import ( Category, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, Subscription, SubscriptionStatus, User, ) from app.models.base import utcnow from app.premium.permissions import can_feature_listing, max_featured_listings class _SubQueryCounter: def __init__(self): self.count = 0 def __enter__(self): from sqlalchemy import event self._event = event self._engine = db.engine event.listen(self._engine, "after_cursor_execute", self._cb) return self def __exit__(self, *exc): self._event.remove(self._engine, "after_cursor_execute", self._cb) def _cb(self, conn, cursor, statement, params, context, executemany): if "subscriptions" in statement.lower(): self.count += 1 class Phase2BTests(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, user, days=30): db.session.add(Subscription( user_id=user.id, plan_name="monthly", status=SubscriptionStatus.ACTIVE, start_date=utcnow(), end_date=utcnow() + timedelta(days=days), activated_by=None, )) db.session.commit() 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, price="50.00", featured=False, days_old=0, status=ListingStatus.AVAILABLE, title=None): listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category().id, title=title or f"Book {seller.id}-{idx}", slug=f"book-{seller.id}-{idx}", description="d", price=Decimal(price), condition=ListingCondition.GOOD, status=status, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() listing.created_at = utcnow() - timedelta(days=days_old) if featured: listing.is_featured = True listing.featured_at = utcnow() - timedelta(days=days_old) listing.featured_until = utcnow() + timedelta(days=7) # live window db.session.commit() return listing # --- permissions --- def test_permissions_free_vs_premium(self): free = self._user("f@c.edu") prem = self._user("p@c.edu") self._premium(prem) self.assertFalse(can_feature_listing(free)) self.assertTrue(can_feature_listing(prem)) self.assertEqual(max_featured_listings(free), 0) self.assertEqual(max_featured_listings(prem), 1) # --- feature service: security --- def test_free_user_cannot_feature(self): free = self._user("f@c.edu") listing = self._listing(free) with self.assertRaises(FeatureListingError): feature_listing(listing, free) self.assertFalse(listing.is_featured) def test_cannot_feature_others_listing(self): prem = self._user("p@c.edu") self._premium(prem) other = self._user("o@c.edu") listing = self._listing(other) with self.assertRaises(FeatureListingError): feature_listing(listing, prem) def test_cannot_feature_sold_or_deleted(self): prem = self._user("p@c.edu") self._premium(prem) sold = self._listing(prem, 1, status=ListingStatus.SOLD) with self.assertRaises(FeatureListingError): feature_listing(sold, prem) deleted = self._listing(prem, 2) deleted.deleted_at = utcnow() db.session.commit() with self.assertRaises(FeatureListingError): feature_listing(deleted, prem) # --- single featured + replacement --- def test_premium_features_one_and_replaces(self): prem = self._user("p@c.edu") self._premium(prem) a = self._listing(prem, 1) b = self._listing(prem, 2) replaced = feature_listing(a, prem) self.assertIsNone(replaced) self.assertTrue(a.is_featured) self.assertIsNotNone(a.featured_at) self.assertEqual(get_featured_listing(prem).id, a.id) # Featuring B must auto-unfeature A (exactly one featured at a time). replaced = feature_listing(b, prem) self.assertIsNotNone(replaced) self.assertEqual(replaced.id, a.id) db.session.refresh(a) self.assertFalse(a.is_featured) self.assertTrue(b.is_featured) featured = db.session.scalars( db.select(Listing).where(Listing.seller_id == prem.id, Listing.is_featured.is_(True)) ).all() self.assertEqual(len(featured), 1) def test_is_featured_and_unfeature(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) feature_listing(listing, prem) self.assertTrue(is_listing_featured(listing)) unfeature_listing(listing) self.assertFalse(is_listing_featured(listing)) self.assertIsNone(listing.featured_at) def test_sold_clears_featured(self): prem = self._user("p@c.edu") self._premium(prem) buyer = self._user("b@c.edu") listing = self._listing(prem) feature_listing(listing, prem) listing.status = ListingStatus.RESERVED # sold requires reserved first db.session.commit() mark_listing_as_sold(listing, prem) self.assertFalse(listing.is_featured) def test_delete_clears_featured(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) feature_listing(listing, prem) soft_delete_listing(listing, prem) self.assertFalse(listing.is_featured) # --- ordering --- def test_ordering_featured_then_premium_then_free(self): prem = self._user("p@c.edu") self._premium(prem) free = self._user("f@c.edu") # A: free seller but FEATURED, oldest -> must be first. a = self._listing(free, 1, featured=True, days_old=5) # B: premium seller, not featured, middle age -> second (premium tier). b = self._listing(prem, 1, days_old=3) # C: free seller, not featured, newest -> last. c = self._listing(free, 2, days_old=1) items = search_published_listings(1, sort="newest").items self.assertEqual([l.id for l in items], [a.id, b.id, c.id]) def test_explicit_price_sort_preserves_relevance_with_premium_tiebreak(self): prem = self._user("p@c.edu") self._premium(prem) free = self._user("f@c.edu") f = self._listing(prem, 1, price="99.00", featured=True) # featured -> first p1 = self._listing(free, 1, price="10.00") # cheapest p2 = self._listing(prem, 2, price="20.00") # tie 20, premium p3 = self._listing(free, 2, price="20.00") # tie 20, free items = search_published_listings(1, sort="price_low").items # Featured first; then price asc; premium wins the 20.00 tie. self.assertEqual([l.id for l in items], [f.id, p1.id, p2.id, p3.id]) def test_search_filter_still_works(self): prem = self._user("p@c.edu") self._premium(prem) self._listing(prem, 1, featured=True, title="Featured Widget") self._listing(prem, 2, title="Unrelated Gadget") items = search_published_listings(1, q="Widget").items titles = [l.title for l in items] self.assertIn("Featured Widget", titles) self.assertNotIn("Unrelated Gadget", titles) def test_ordering_has_no_n_plus_1(self): free = self._user("f@c.edu") prem = self._user("p@c.edu") self._premium(prem) for i in range(3): self._listing(free, i) self._listing(prem, i) with _SubQueryCounter() as c1: search_published_listings(1, sort="newest").items small = c1.count for i in range(3, 9): self._listing(free, i) self._listing(prem, i) with _SubQueryCounter() as c2: search_published_listings(1, sort="newest").items self.assertEqual(small, c2.count) # constant, not per-listing self.assertLessEqual(c2.count, 1) # one embedded subquery per page # --- route + badge rendering --- def test_route_free_user_redirected_to_premium(self): free = self._user("f@c.edu") listing = self._listing(free) client = self.app.test_client() client.post("/login", data={"email": "f@c.edu", "password": "Passw0rd!"}) resp = client.post(f"/app/listings/{listing.public_id}/feature") self.assertEqual(resp.status_code, 302) self.assertIn("/app/premium", resp.headers.get("Location", "")) db.session.refresh(listing) self.assertFalse(listing.is_featured) # never featured server-side def test_route_premium_features_listing(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) client = self.app.test_client() client.post("/login", data={"email": "p@c.edu", "password": "Passw0rd!"}) resp = client.post(f"/app/listings/{listing.public_id}/feature", follow_redirects=True) self.assertEqual(resp.status_code, 200) db.session.refresh(listing) self.assertTrue(listing.is_featured) self.assertIn("now featured", resp.get_data(as_text=True)) def test_featured_badge_renders_on_grid(self): prem = self._user("p@c.edu") self._premium(prem) self._listing(prem, 1, featured=True) client = self.app.test_client() html = client.get("/marketplace").get_data(as_text=True) self.assertIn("featured-badge", html) # gold featured pill self.assertIn("premium-badge", html) # premium seller badge too (both show) if __name__ == "__main__": unittest.main()