"""Featured-listing expiration tests (Phase 2B enhancement). Covers the 7-day window, expiry-aware "currently featured" checks, automatic cleanup, ordering, replacement-resets-timer, and the UI reflecting expiry. """ 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 ( FEATURE_DURATION_DAYS, cleanup_expired_featured_listings, feature_listing, featured_days_remaining, get_featured_listing, is_listing_currently_featured, search_published_listings, ) from app.models import ( Category, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, Subscription, SubscriptionStatus, User, ) from app.models.base import utcnow class ExpiryTests(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): db.session.add(Subscription( user_id=user.id, plan_name="monthly", status=SubscriptionStatus.ACTIVE, start_date=utcnow(), end_date=utcnow() + timedelta(days=30), 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"): listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category().id, title=f"Book {seller.id}-{idx}", slug=f"book-{seller.id}-{idx}", description="d", price=Decimal(price), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() return listing def _make_featured(self, listing, *, until): """Directly set featured state with an explicit expiry (bypasses service).""" listing.is_featured = True listing.featured_at = utcnow() listing.featured_until = until db.session.commit() # --- duration --- def test_feature_sets_seven_day_window(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) feature_listing(listing, prem) self.assertIsNotNone(listing.featured_until) delta = listing.featured_until - listing.featured_at self.assertEqual(delta, timedelta(days=FEATURE_DURATION_DAYS)) self.assertEqual(featured_days_remaining(listing), FEATURE_DURATION_DAYS) # --- expiry-aware checks --- def test_expired_reads_as_not_currently_featured(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) self._make_featured(listing, until=utcnow() - timedelta(minutes=1)) # already ended self.assertTrue(listing.is_featured) # raw flag still set self.assertFalse(is_listing_currently_featured(listing)) # but not currently featured self.assertIsNone(featured_days_remaining(listing)) self.assertIsNone(get_featured_listing(prem)) # ignores expired def test_live_window_is_currently_featured(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) self._make_featured(listing, until=utcnow() + timedelta(days=3)) self.assertTrue(is_listing_currently_featured(listing)) self.assertEqual(get_featured_listing(prem).id, listing.id) # --- automatic cleanup --- def test_cleanup_clears_only_expired(self): prem = self._user("p@c.edu") self._premium(prem) expired = self._listing(prem, 1) self._make_featured(expired, until=utcnow() - timedelta(seconds=1)) live = self._listing(prem, 2) self._make_featured(live, until=utcnow() + timedelta(days=2)) cleared = cleanup_expired_featured_listings() self.assertEqual(cleared, 1) db.session.refresh(expired) db.session.refresh(live) self.assertFalse(expired.is_featured) self.assertIsNone(expired.featured_until) self.assertIsNone(expired.featured_at) self.assertTrue(live.is_featured) # untouched def test_browse_transparently_expires(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._make_featured(listing, until=utcnow() - timedelta(seconds=1)) # Accessing listings through the service triggers cleanup. search_published_listings(1, sort="newest") db.session.refresh(listing) self.assertFalse(listing.is_featured) # --- ordering --- def test_expired_featured_loses_priority(self): prem = self._user("p@c.edu") self._premium(prem) free = self._user("f@c.edu") # Expired "featured" listing (oldest) — must NOT jump to the top. expired = self._listing(free, 1) expired.created_at = utcnow() - timedelta(days=5) self._make_featured(expired, until=utcnow() - timedelta(seconds=1)) # A genuinely newer normal listing. newer = self._listing(free, 2) newer.created_at = utcnow() db.session.commit() items = search_published_listings(1, sort="newest").items # Newer normal listing comes first; the expired one behaves normally. self.assertEqual(items[0].id, newer.id) def test_live_featured_keeps_priority(self): free = self._user("f@c.edu") live = self._listing(free, 1) live.created_at = utcnow() - timedelta(days=5) # old but featured self._make_featured(live, until=utcnow() + timedelta(days=2)) newer = self._listing(free, 2) newer.created_at = utcnow() db.session.commit() items = search_published_listings(1, sort="newest").items self.assertEqual(items[0].id, live.id) # featured first despite being older # --- replacement resets the timer --- def test_replacement_resets_timer(self): prem = self._user("p@c.edu") self._premium(prem) a = self._listing(prem, 1) feature_listing(a, prem) # Simulate A being close to expiry. a.featured_until = utcnow() + timedelta(hours=1) db.session.commit() b = self._listing(prem, 2) feature_listing(b, prem) db.session.refresh(a) self.assertFalse(a.is_featured) self.assertIsNone(a.featured_until) # B gets a fresh full window. self.assertEqual(b.featured_until - b.featured_at, timedelta(days=FEATURE_DURATION_DAYS)) self.assertEqual(featured_days_remaining(b), FEATURE_DURATION_DAYS) # --- UI --- def test_my_listings_shows_expiry_then_returns_to_feature(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) feature_listing(listing, prem) client = self.app.test_client() client.post("/login", data={"email": "p@c.edu", "password": "Passw0rd!"}) html = client.get("/app/my-listings").get_data(as_text=True) self.assertIn("Expires in", html) # The seller card highlights a featured listing with the unified badge. self.assertIn("mk-badge-featured", html) # Expire it; My Listings must auto-clean and offer to feature again. listing.featured_until = utcnow() - timedelta(seconds=1) db.session.commit() html2 = client.get("/app/my-listings").get_data(as_text=True) self.assertIn("Feature Listing", html2) db.session.refresh(listing) self.assertFalse(listing.is_featured) # cleaned up on access def test_detail_shows_featured_until(self): prem = self._user("p@c.edu") self._premium(prem) listing = self._listing(prem) feature_listing(listing, prem) client = self.app.test_client() html = client.get(f"/app/listings/{listing.public_id}").get_data(as_text=True) self.assertIn("Featured until", html) if __name__ == "__main__": unittest.main()