"""Fair rotation for rewarded advertisement selection. ``get_rewarded_ad`` previously only randomised within the single top priority, so it always served the highest-priority ad when priorities differed. It now performs weighted random selection across the whole eligible pool: every eligible ad can be served, higher priority appears more often, and none of the surrounding reward flow (eligibility filtering, token minting, analytics) changes. """ import os import unittest from datetime import timedelta os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from app import create_app from app.advertisements.reward_token import verify_reward_token from app.advertisements.services import get_active_rewarded_ad, get_rewarded_ad from app.extensions import db from app.models import ( Advertisement, AdvertisementPlacement, AdvertisementType, User, UserRole, ) from app.models.base import utcnow class RewardedAdSelectionTests(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 _ad(self, title, *, priority=5, active=True, start_at=None, end_at=None, placement=AdvertisementPlacement.CHAT): ad = Advertisement( title=title, advertisement_type=AdvertisementType.REWARDED, placement=placement, image_path=f"/static/uploads/advertisements/{title}.png", is_active=active, priority=priority, start_at=start_at, end_at=end_at, ) db.session.add(ad) db.session.commit() return ad def _titles_over_runs(self, runs=200): return {get_rewarded_ad().title for _ in range(runs)} # --- single ad ------------------------------------------------------------- def test_single_ad_always_selected(self): self._ad("Solo") self.assertEqual(self._titles_over_runs(), {"Solo"}) def test_no_eligible_ads_returns_none(self): self.assertIsNone(get_rewarded_ad()) def test_alias_delegates_to_selector(self): self._ad("Solo") self.assertEqual(get_active_rewarded_ad().title, "Solo") # --- rotation across multiple ads ----------------------------------------- def test_two_equal_ads_both_selected(self): self._ad("A", priority=5) self._ad("B", priority=5) self.assertEqual(self._titles_over_runs(), {"A", "B"}) def test_distinct_priorities_all_appear(self): # The old logic would only ever serve the priority-10 ad; every eligible ad # must now appear at least once across many runs. self._ad("High", priority=10) self._ad("Mid", priority=5) self._ad("Low", priority=1) self.assertEqual(self._titles_over_runs(runs=400), {"High", "Mid", "Low"}) def test_higher_priority_appears_more_often(self): self._ad("Heavy", priority=20) self._ad("Light", priority=1) counts = {"Heavy": 0, "Light": 0} for _ in range(1500): counts[get_rewarded_ad().title] += 1 # Weighted 20:1 — Heavy must dominate by a wide margin (loose bound avoids # flakiness while still proving priority influences frequency). self.assertGreater(counts["Heavy"], counts["Light"] * 3) def test_zero_weight_ignored_when_others_positive(self): # Weighted-selection semantics (Display Weight spec): a weight of 0 means # "don't show" while another ad has positive weight. self._ad("Zero", priority=0) self._ad("Normal", priority=5) self.assertEqual(self._titles_over_runs(), {"Normal"}) def test_all_zero_weights_equal_chance(self): # When every weight is 0 there is nothing to weight by → equal chance, so # an all-zero configuration never fails or hides every ad. self._ad("A", priority=0) self._ad("B", priority=0) self.assertEqual(self._titles_over_runs(), {"A", "B"}) # --- eligibility filtering is unchanged ----------------------------------- def test_inactive_never_selected(self): self._ad("Live", active=True) self._ad("Off", active=False) self.assertEqual(self._titles_over_runs(), {"Live"}) def test_expired_never_selected(self): now = utcnow() self._ad("Live") self._ad("Expired", end_at=now - timedelta(days=1)) self.assertEqual(self._titles_over_runs(), {"Live"}) def test_not_yet_started_never_selected(self): now = utcnow() self._ad("Live") self._ad("Future", start_at=now + timedelta(days=1)) self.assertEqual(self._titles_over_runs(), {"Live"}) def test_non_rewarded_placement_never_selected(self): self._ad("Reward") self._ad("Banner", placement=AdvertisementPlacement.LANDING) self.assertEqual(self._titles_over_runs(), {"Reward"}) def test_banner_at_chat_placement_never_selected(self): # Defence in depth: a banner that somehow sits at the CHAT placement must # never leak into the rewarded flow — selection also filters by type. self._ad("Reward") banner = Advertisement( title="BannerAtChat", advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.CHAT, image_path="/static/uploads/advertisements/b.png", is_active=True, priority=50, ) db.session.add(banner) db.session.commit() self.assertEqual(self._titles_over_runs(), {"Reward"}) class RewardedAdSelectionFlowTests(unittest.TestCase): """The selected ad flows through the modal, token and analytics unchanged.""" def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() from app.advertisements.rate_limit import reset_rate_limits from app.advertisements.reward_token import reset_replay_guard reset_replay_guard() reset_rate_limits() def tearDown(self): from app.advertisements.rate_limit import reset_rate_limits reset_rate_limits() db.session.remove() db.drop_all() self.ctx.pop() def _seed_conversation(self): from decimal import Decimal from app.models import ( Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, ) buyer = User(email="buyer@c.edu", password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=UserRole.STUDENT) seller = User(email="seller@c.edu", password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=UserRole.STUDENT) db.session.add_all([buyer, seller]) db.session.commit() cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() listing = Listing( sku="SKU-1", seller_id=seller.id, category_id=cat.id, title="Book", slug="book-1", description="d", price=Decimal("50"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() 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 buyer, convo def _ad(self): ad = Advertisement( title="Watch", advertisement_type=AdvertisementType.REWARDED, placement=AdvertisementPlacement.CHAT, image_path="/static/uploads/advertisements/w.png", is_active=True, priority=10, ) db.session.add(ad) db.session.commit() return ad def test_reward_token_binds_to_selected_ad(self): buyer, convo = self._seed_conversation() ad = self._ad() client = self.app.test_client() client.post("/login", data={"email": "buyer@c.edu", "password": "Passw0rd!"}) html = client.get(f"/app/chat/{convo.public_id}").get_data(as_text=True) # The single eligible ad must be the one embedded in the modal. self.assertIn(ad.public_id, html) # The minted token verifies to that same ad. resp = client.get(f"/app/chat/{convo.public_id}/reward-token", query_string={"ad": ad.public_id}) payload = verify_reward_token(resp.get_json()["token"], user_id=buyer.id) self.assertEqual(payload["aid"], ad.public_id) def test_impression_recorded_for_selected_ad(self): buyer, convo = self._seed_conversation() ad = self._ad() client = self.app.test_client() client.post("/login", data={"email": "buyer@c.edu", "password": "Passw0rd!"}) client.get(f"/app/chat/{convo.public_id}/reward-token", query_string={"ad": ad.public_id}) from app.models import RewardedAdEvent events = db.session.scalars( db.select(RewardedAdEvent).filter_by(advertisement_id=ad.id) ).all() self.assertTrue(events) self.assertTrue(all(e.advertisement_id == ad.id for e in events)) if __name__ == "__main__": unittest.main()