"""Rate limiting for reward-related endpoints.""" import os import unittest 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.advertisements.rate_limit import check_rate_limit, reset_rate_limits from app.advertisements.reward_token import issue_reward_token, reset_replay_guard from app.chat.services import get_messaging_state from app.extensions import db from app.models import ( Advertisement, AdvertisementPlacement, AdvertisementType, Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, RewardedAdEvent, User, ) class RewardRateLimitTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() reset_rate_limits() reset_replay_guard() def tearDown(self): reset_rate_limits() reset_replay_guard() db.session.remove() db.drop_all() self.ctx.pop() # --- unit: the limiter primitive ------------------------------------------ def test_allows_up_to_limit(self): self.assertEqual([check_rate_limit("b", 1, limit=3, window_seconds=60) for _ in range(3)], [True, True, True]) def test_blocks_beyond_limit(self): for _ in range(3): check_rate_limit("b", 1, limit=3, window_seconds=60) self.assertFalse(check_rate_limit("b", 1, limit=3, window_seconds=60)) def test_recovers_after_reset(self): for _ in range(3): check_rate_limit("b", 1, limit=3, window_seconds=60) self.assertFalse(check_rate_limit("b", 1, limit=3, window_seconds=60)) reset_rate_limits() # simulates the window elapsing self.assertTrue(check_rate_limit("b", 1, limit=3, window_seconds=60)) def test_identities_and_buckets_independent(self): for _ in range(3): check_rate_limit("b", 1, limit=3, window_seconds=60) self.assertTrue(check_rate_limit("b", 2, limit=3, window_seconds=60)) # other user self.assertTrue(check_rate_limit("other", 1, limit=3, window_seconds=60)) # other bucket # --- factories ------------------------------------------------------------- def _user(self, email): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True) db.session.add(u) db.session.commit() return u def _convo(self, buyer, seller): cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() listing = Listing( sku=f"SKU-{seller.id}", seller_id=seller.id, category_id=cat.id, title="Book", slug=f"book-{seller.id}", 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 convo def _ad(self): ad = Advertisement(title="Ad", advertisement_type=AdvertisementType.REWARDED, placement=AdvertisementPlacement.CHAT, is_active=True, priority=10, image_path="/x.png") db.session.add(ad) db.session.commit() return ad def _login(self, email): c = self.app.test_client() c.post("/login", data={"email": email, "password": "Passw0rd!"}) return c def _remaining(self, convo, user): db.session.refresh(convo) return get_messaging_state(convo, user)["total_remaining"] # --- integration: normal usage is never blocked --------------------------- def test_normal_usage_not_blocked(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() client = self._login("b@c.edu") # A couple of normal mints + one claim — well under the generous defaults. self.assertEqual(client.get(f"/app/chat/{convo.public_id}/reward-token?ad={ad.public_id}").status_code, 200) before = self._remaining(convo, buyer) client.post(f"/app/chat/{convo.public_id}/watch-ad", data={"reward_token": issue_reward_token(buyer.id, ad.public_id)}, follow_redirects=True) self.assertEqual(self._remaining(convo, buyer) - before, self.app.config["MESSAGES_PER_REWARDED_AD"]) # --- integration: mint endpoint throttled --------------------------------- def test_token_endpoint_rate_limited(self): self.app.config["REWARD_TOKEN_RATE_MAX"] = 3 buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() client = self._login("b@c.edu") url = f"/app/chat/{convo.public_id}/reward-token?ad={ad.public_id}" codes = [client.get(url).status_code for _ in range(5)] self.assertEqual(codes[:3], [200, 200, 200]) self.assertEqual(codes[3], 429) self.assertEqual(client.get(url).get_json().get("error"), "rate_limited") def test_token_endpoint_recovers_after_reset(self): self.app.config["REWARD_TOKEN_RATE_MAX"] = 2 buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() client = self._login("b@c.edu") url = f"/app/chat/{convo.public_id}/reward-token?ad={ad.public_id}" client.get(url); client.get(url) self.assertEqual(client.get(url).status_code, 429) reset_rate_limits() # window elapses self.assertEqual(client.get(url).status_code, 200) # --- integration: claim endpoint throttled -------------------------------- def test_claim_endpoint_rate_limited(self): self.app.config["REWARD_CLAIM_RATE_MAX"] = 3 buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() client = self._login("b@c.edu") before = self._remaining(convo, buyer) for _ in range(3): # three valid claims → all granted client.post(f"/app/chat/{convo.public_id}/watch-ad", data={"reward_token": issue_reward_token(buyer.id, ad.public_id)}, follow_redirects=True) granted = self._remaining(convo, buyer) # Fourth claim is rate-limited: no reward, and a rate_limit rejection logged. client.post(f"/app/chat/{convo.public_id}/watch-ad", data={"reward_token": issue_reward_token(buyer.id, ad.public_id)}, follow_redirects=True) self.assertEqual(granted - before, 3 * self.app.config["MESSAGES_PER_REWARDED_AD"]) self.assertEqual(self._remaining(convo, buyer), granted) # no extra reward rate_rejects = db.session.scalar(db.select(db.func.count(RewardedAdEvent.id)).where( RewardedAdEvent.event_type == "reward_rejected", RewardedAdEvent.reason == "rate_limit")) self.assertEqual(rate_rejects, 1) if __name__ == "__main__": unittest.main()