"""Phase 2A: premium feature-unlock tests. Covers the two enforcement paths that matter server-side — unlimited messaging and ad-free browsing — plus the capability helpers and badge rendering. Free-user behaviour must be unchanged; premium users must bypass limits even via the API. """ 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.chat.services import ( MessageLimitReachedError, get_messaging_state, send_message, ) from app.extensions import db from app.models import ( Advertisement, AdvertisementPlacement, AdvertisementType, Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, Subscription, SubscriptionStatus, User, ) from app.models.base import utcnow from app.premium.permissions import ( can_send_unlimited_messages, can_view_without_ads, has_premium, should_show_premium_badge, ) class PremiumPhase2ATests(unittest.TestCase): def setUp(self) -> None: self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() # --- helpers --- def _user(self, email: str) -> User: user = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!")) db.session.add(user) db.session.commit() return user def _make_premium(self, user: User, days: int = 30) -> None: 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) -> Category: 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: User, idx: int = 1) -> Listing: cat = self._category() listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=cat.id, title=f"Book {seller.id}-{idx}", slug=f"book-{seller.id}-{idx}", description="d", price=Decimal("50.00"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() return listing def _convo(self, buyer: User, seller: User, listing: Listing) -> Conversation: convo = Conversation( listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id, status=ConversationStatus.ACTIVE, ) db.session.add(convo) db.session.commit() return convo # --- unlimited messaging --- def test_free_user_hits_message_limit(self): """Free behaviour unchanged: the free allowance is still enforced.""" buyer = self._user("b@c.edu") seller = self._user("s@c.edu") convo = self._convo(buyer, seller, self._listing(seller)) limit = self.app.config["FREE_MESSAGE_LIMIT"] for i in range(limit): send_message(convo, buyer, f"hello {i}") with self.assertRaises(MessageLimitReachedError): send_message(convo, buyer, "one message too many") def test_premium_user_never_hits_limit(self): """Premium bypasses the quota server-side and consumes no allowance.""" buyer = self._user("b@c.edu") seller = self._user("s@c.edu") self._make_premium(buyer) convo = self._convo(buyer, seller, self._listing(seller)) limit = self.app.config["FREE_MESSAGE_LIMIT"] for i in range(limit + 5): # well past the free limit send_message(convo, buyer, f"hello {i}") self.assertEqual(convo.messages_used_by(buyer), 0) state = get_messaging_state(convo, buyer) self.assertTrue(state["premium"]) self.assertFalse(state["limit_reached"]) def test_premium_applies_to_existing_conversation(self): """A conversation started while free becomes unlimited once premium.""" buyer = self._user("b@c.edu") seller = self._user("s@c.edu") convo = self._convo(buyer, seller, self._listing(seller)) limit = self.app.config["FREE_MESSAGE_LIMIT"] for i in range(limit): send_message(convo, buyer, f"hello {i}") # Now upgrade — the SAME conversation should no longer be limited. self._make_premium(buyer) send_message(convo, buyer, "still going after upgrade") # must not raise # --- capability helpers --- def test_capability_helpers(self): free = self._user("f@c.edu") prem = self._user("p@c.edu") self._make_premium(prem) for fn in (has_premium, can_send_unlimited_messages, can_view_without_ads, should_show_premium_badge): self.assertFalse(fn(free), fn.__name__) self.assertTrue(fn(prem), fn.__name__) def test_expired_subscription_has_no_premium(self): user = self._user("e@c.edu") db.session.add(Subscription( user_id=user.id, plan_name="monthly", status=SubscriptionStatus.ACTIVE, start_date=utcnow() - timedelta(days=40), end_date=utcnow() - timedelta(days=1), activated_by=None, )) db.session.commit() self.assertFalse(has_premium(user)) # --- advertisements visibility --- def _add_marketplace_ad(self) -> None: db.session.add(Advertisement( title="SUPERSALEADXYZ", advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.MARKETPLACE, is_active=True, priority=10, )) db.session.commit() def test_ads_shown_to_free_user(self): self._add_marketplace_ad() self._user("free@c.edu") client = self.app.test_client() client.post("/login", data={"email": "free@c.edu", "password": "Passw0rd!"}) html = client.get("/marketplace").get_data(as_text=True) self.assertIn("SUPERSALEADXYZ", html) def test_ads_hidden_from_premium_user(self): self._add_marketplace_ad() prem = self._user("prem@c.edu") self._make_premium(prem) client = self.app.test_client() client.post("/login", data={"email": "prem@c.edu", "password": "Passw0rd!"}) html = client.get("/marketplace").get_data(as_text=True) self.assertNotIn("SUPERSALEADXYZ", html) # --- badge rendering --- def test_badge_renders_for_premium_member(self): prem = self._user("member@c.edu") self._make_premium(prem) client = self.app.test_client() client.post("/login", data={"email": "member@c.edu", "password": "Passw0rd!"}) html = client.get("/profile").get_data(as_text=True) self.assertIn("Premium Member", html) def test_no_badge_for_free_member(self): self._user("plain@c.edu") client = self.app.test_client() client.post("/login", data={"email": "plain@c.edu", "password": "Passw0rd!"}) html = client.get("/profile").get_data(as_text=True) self.assertNotIn("premium-badge", html) # --- batch premium resolution (N+1 fix) --- def test_batch_matches_permissions_and_dedups(self): """Batch result is identical to per-user checks, and dedups sellers.""" from app.premium.services import get_premium_seller_ids prem = self._user("prem@c.edu") self._make_premium(prem) free = self._user("free@c.edu") # Two listings from the SAME premium seller + one from a free seller. listings = [self._listing(prem, 1), self._listing(prem, 2), self._listing(free, 1)] ids = get_premium_seller_ids(listings) # Identical to asking the permissions layer per seller. expected = {l.seller_id for l in listings if should_show_premium_badge(l.seller)} self.assertEqual(ids, expected) self.assertEqual(ids, {prem.id}) # free excluded, premium seller once def test_batch_handles_empty_and_zero_premium(self): from app.premium.services import get_premium_seller_ids self.assertEqual(get_premium_seller_ids([]), set()) free = self._user("free@c.edu") listings = [self._listing(free, 1), self._listing(free, 2)] self.assertEqual(get_premium_seller_ids(listings), set()) # no premium def test_batch_uses_single_query(self): from app.premium.services import get_premium_seller_ids prem = self._user("prem@c.edu") self._make_premium(prem) free = self._user("free@c.edu") listings = [self._listing(prem, i) for i in range(4)] + [self._listing(free, 1)] with _SubscriptionQueryCounter() as counter: get_premium_seller_ids(listings) self.assertEqual(counter.count, 1) # one query for the whole page def test_grid_render_has_no_n_plus_1(self): """The subscriptions-query count on the grid is constant, not per-card.""" prem = self._user("prem@c.edu") self._make_premium(prem) free = self._user("free@c.edu") self._listing(prem, 1) self._listing(free, 1) client = self.app.test_client() # anonymous: no per-user ad query # Target the catalog grid directly (an explicit sort switches off the # discovery homepage, which has its own dedicated N+1 test). grid_url = "/marketplace?sort=price_low" with _SubscriptionQueryCounter() as counter: html_small = client.get(grid_url).get_data(as_text=True) small = counter.count # Many more cards from both sellers — query count must not grow. for i in range(2, 8): self._listing(prem, i) self._listing(free, i) with _SubscriptionQueryCounter() as counter: html_large = client.get(grid_url).get_data(as_text=True) large = counter.count self.assertEqual(small, large) # O(1), not O(N) — the real guarantee # A small constant number of subscription queries per page (badge batch # + Phase 2B's ordering subquery), never one-per-card. self.assertLessEqual(large, 2) # Badges still correct: premium seller shown, free seller not. self.assertIn("Premium Seller", html_large) self.assertIn("premium-badge", html_large) def test_grid_badge_only_for_premium_seller(self): prem = self._user("prem@c.edu") self._make_premium(prem) free = self._user("free@c.edu") self._listing(prem, 1) self._listing(free, 1) client = self.app.test_client() # Target the catalog grid directly (explicit sort switches off the # discovery homepage, whose sections would repeat the same listing). html = client.get("/marketplace?sort=price_low").get_data(as_text=True) # Exactly one premium badge (the premium seller's single listing). # 'premium-badge--sm' is emitted once per rendered badge. self.assertEqual(html.count("premium-badge--sm"), 1) class _SubscriptionQueryCounter: """Count SQL statements touching the ``subscriptions`` table.""" def __init__(self): self.count = 0 def __enter__(self): from sqlalchemy import event self._engine = db.engine self._event = event event.listen(self._engine, "after_cursor_execute", self._on_execute) return self def __exit__(self, *exc): self._event.remove(self._engine, "after_cursor_execute", self._on_execute) def _on_execute(self, conn, cursor, statement, parameters, context, executemany): if "subscriptions" in statement.lower(): self.count += 1 if __name__ == "__main__": unittest.main()