"""Premium subscription expiry tests (Phase 3). Premium is modelled on the existing ``subscriptions`` table (Phase 1), so these mirror the Featured-Listing expiry tests but for premium: activation window, expiry-aware checks, bulk cleanup, permissions failing after expiry, days remaining, timer reset, and — importantly — that an existing featured listing survives premium expiry (independent timers). """ 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_listing, is_listing_currently_featured 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, has_premium, max_featured_listings from app.premium.services import ( PREMIUM_DURATION_DAYS, activate_premium, cleanup_expired_premium_users, expire_premium, get_active_subscription, is_user_currently_premium, premium_days_remaining, ) class Phase3PremiumExpiryTests(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 # Helper fixture: a premium user always gets a real subscription window # (premium_until == subscription.end_date) via the service. def _premium_user(self, email): user = self._user(email) activate_premium(user) return user 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): listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=self._category().id, title=f"Book {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 # --- activation --- def test_activation_sets_window(self): user = self._user("a@c.edu") sub = activate_premium(user) self.assertEqual(sub.status, SubscriptionStatus.ACTIVE) self.assertIsNotNone(sub.start_date) self.assertEqual(sub.end_date - sub.start_date, timedelta(days=PREMIUM_DURATION_DAYS)) self.assertTrue(is_user_currently_premium(user)) def test_days_remaining(self): user = self._premium_user("a@c.edu") self.assertEqual(premium_days_remaining(user), PREMIUM_DURATION_DAYS) # Non-premium user -> 0. self.assertEqual(premium_days_remaining(self._user("b@c.edu")), 0) # --- expiry-aware checks --- def test_expired_returns_false(self): user = self._user("a@c.edu") # Active status but end_date already passed -> not currently premium. db.session.add(Subscription( user_id=user.id, plan_name="monthly", status=SubscriptionStatus.ACTIVE, start_date=utcnow() - timedelta(days=40), end_date=utcnow() - timedelta(minutes=1), )) db.session.commit() self.assertFalse(is_user_currently_premium(user)) self.assertEqual(premium_days_remaining(user), 0) def test_expire_premium_marks_expired(self): user = self._premium_user("a@c.edu") self.assertTrue(is_user_currently_premium(user)) n = expire_premium(user) self.assertEqual(n, 1) self.assertFalse(is_user_currently_premium(user)) sub = db.session.scalar(db.select(Subscription).where(Subscription.user_id == user.id)) self.assertEqual(sub.status, SubscriptionStatus.EXPIRED) # history preserved # --- bulk cleanup --- def test_cleanup_expires_only_lapsed(self): lapsed = self._user("lapsed@c.edu") db.session.add(Subscription( user_id=lapsed.id, plan_name="monthly", status=SubscriptionStatus.ACTIVE, start_date=utcnow() - timedelta(days=40), end_date=utcnow() - timedelta(seconds=1), )) live = self._premium_user("live@c.edu") db.session.commit() expired = cleanup_expired_premium_users() self.assertEqual(expired, 1) lapsed_sub = db.session.scalar(db.select(Subscription).where(Subscription.user_id == lapsed.id)) self.assertEqual(lapsed_sub.status, SubscriptionStatus.EXPIRED) self.assertTrue(is_user_currently_premium(live)) # untouched def test_cleanup_runs_on_premium_page(self): user = self._user("u@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(seconds=1), )) db.session.commit() client = self.app.test_client() client.post("/login", data={"email": "u@c.edu", "password": "Passw0rd!"}) client.get("/app/premium/") # entry point runs cleanup sub = db.session.scalar(db.select(Subscription).where(Subscription.user_id == user.id)) self.assertEqual(sub.status, SubscriptionStatus.EXPIRED) # --- permissions after expiry --- def test_permissions_fail_after_expiry(self): user = self._premium_user("a@c.edu") self.assertTrue(can_feature_listing(user)) self.assertEqual(max_featured_listings(user), 1) expire_premium(user) self.assertFalse(has_premium(user)) self.assertFalse(can_feature_listing(user)) self.assertEqual(max_featured_listings(user), 0) # --- re-activation resets the timer --- def test_reactivation_resets_timer(self): user = self._premium_user("a@c.edu") first = get_active_subscription(user) first.end_date = utcnow() + timedelta(hours=1) # near expiry db.session.commit() new = activate_premium(user) self.assertEqual(new.end_date - new.start_date, timedelta(days=PREMIUM_DURATION_DAYS)) self.assertEqual(premium_days_remaining(user), PREMIUM_DURATION_DAYS) # Old subscription is retired (single active), not deleted. db.session.refresh(first) self.assertEqual(first.status, SubscriptionStatus.CANCELLED) actives = db.session.scalars(db.select(Subscription).where( Subscription.user_id == user.id, Subscription.status == SubscriptionStatus.ACTIVE)).all() self.assertEqual(len(actives), 1) # --- featured listing survives premium expiry (independent timers) --- def test_featured_listing_survives_premium_expiry(self): user = self._premium_user("a@c.edu") listing = self._listing(user) feature_listing(listing, user) # featured for 7 days self.assertTrue(is_listing_currently_featured(listing)) expire_premium(user) # premium lapses immediately self.assertFalse(has_premium(user)) # The already-featured listing keeps its own 7-day window. db.session.refresh(listing) self.assertTrue(listing.is_featured) self.assertTrue(is_listing_currently_featured(listing)) # But the (now non-premium) user cannot feature ANOTHER listing. other = self._listing(user, 2) from app.marketplace.services import FeatureListingError with self.assertRaises(FeatureListingError): feature_listing(other, user) # --- UI --- def test_profile_shows_expiry(self): self._premium_user("a@c.edu") client = self.app.test_client() client.post("/login", data={"email": "a@c.edu", "password": "Passw0rd!"}) html = client.get("/profile").get_data(as_text=True) self.assertIn("Expires in", html) self.assertIn("Premium until", html) def test_premium_page_shows_until_and_days(self): self._premium_user("a@c.edu") client = self.app.test_client() client.post("/login", data={"email": "a@c.edu", "password": "Passw0rd!"}) html = client.get("/app/premium/").get_data(as_text=True) # The Phase 10 hub surfaces the same info via the floating status card. self.assertIn("valid until", html) # "Premium valid until" self.assertIn("remaining", html) # "N days remaining" if __name__ == "__main__": unittest.main()