diff options
| author | Ben Sima <ben@bsima.me> | 2025-11-12 16:52:48 -0500 |
|---|---|---|
| committer | Ben Sima <ben@bsima.me> | 2025-11-12 16:52:48 -0500 |
| commit | 14fdc4b98a9f442df72fb8ff445f4e7f7a24a10e (patch) | |
| tree | 07d9217527c4ad1b41b6b9d3d3e0657a50add88b /Biz/PodcastItLater/Billing.py | |
| parent | 5efd438acf7afe9fbe2d621542bcb74688a4d647 (diff) | |
Add Stripe webhook tests and testing documentation
- Add TestWebhookHandling class with tests for subscription webhooks -
Test normal subscription creation flow with mock data - Test handling
of incomplete webhook data (missing fields) - Add STRIPE_TESTING.md
with comprehensive testing guide:
- How to use Stripe test mode and test cards - How to use Stripe
CLI to trigger test webhooks - Instructions for running unit tests -
Database migration instructions for production - End-to-end testing
workflow - Common troubleshooting issues
Tests verify webhook handling works without calling real Stripe API.
Diffstat (limited to 'Biz/PodcastItLater/Billing.py')
| -rw-r--r-- | Biz/PodcastItLater/Billing.py | 100 |
1 files changed, 100 insertions, 0 deletions
diff --git a/Biz/PodcastItLater/Billing.py b/Biz/PodcastItLater/Billing.py index 9c0e1db..3716660 100644 --- a/Biz/PodcastItLater/Billing.py +++ b/Biz/PodcastItLater/Billing.py @@ -9,9 +9,12 @@ Stripe subscription management and usage enforcement. # : dep pytest # : dep pytest-mock import Biz.PodcastItLater.Core as Core +import Omni.App as App import Omni.Log as Log +import Omni.Test as Test import os import stripe +import sys import typing from datetime import datetime from datetime import timezone @@ -422,3 +425,100 @@ def get_tier_info(tier: str) -> dict[str, typing.Any]: }, } return tier_info.get(tier, tier_info["free"]) # type: ignore[return-value] + + +# Tests +# ruff: noqa: PLR6301, PLW0603, S101 + + +class TestWebhookHandling(Test.TestCase): + """Test Stripe webhook handling.""" + + def setUp(self) -> None: + """Set up test database.""" + Core.Database.init_db() + + def tearDown(self) -> None: + """Clean up test database.""" + Core.Database.teardown() + + def test_subscription_created(self) -> None: + """Test handling subscription.created webhook with mock data.""" + # Create test user + user_id, _token = Core.Database.create_user("test@example.com") + Core.Database.set_user_stripe_customer(user_id, "cus_test123") + + # Mock subscription event + subscription = { + "id": "sub_test123", + "customer": "cus_test123", + "status": "active", + "current_period_start": 1700000000, + "current_period_end": 1702592000, + "cancel_at_period_end": False, + "items": { + "data": [ + { + "price": { + "id": "price_test_pro", + }, + }, + ], + }, + } + + # Temporarily set price mapping for test + global PRICE_TO_TIER + old_mapping = PRICE_TO_TIER.copy() + PRICE_TO_TIER["price_test_pro"] = "pro" + + try: + _update_subscription_state(subscription) + + # Verify user was updated + user = Core.Database.get_user_by_id(user_id) + self.assertIsNotNone(user) + assert user is not None + self.assertEqual(user["plan_tier"], "pro") + self.assertEqual(user["subscription_status"], "active") + self.assertEqual(user["stripe_subscription_id"], "sub_test123") + finally: + PRICE_TO_TIER = old_mapping + + def test_webhook_missing_fields(self) -> None: + """Test handling webhook with missing required fields.""" + # Create test user + user_id, _token = Core.Database.create_user("test@example.com") + Core.Database.set_user_stripe_customer(user_id, "cus_test456") + + # Mock subscription with missing current_period_start + subscription = { + "id": "sub_test456", + "customer": "cus_test456", + "status": "active", + # Missing current_period_start and current_period_end + "cancel_at_period_end": False, + "items": {"data": []}, + } + + # Should not crash, just log warning and return + _update_subscription_state(subscription) + + # User should remain on free tier + user = Core.Database.get_user_by_id(user_id) + self.assertIsNotNone(user) + assert user is not None + self.assertEqual(user["plan_tier"], "free") + + +def main() -> None: + """Run tests.""" + if len(sys.argv) > 1 and sys.argv[1] == "test": + os.environ["AREA"] = "Test" + Test.run(App.Area.Test, [TestWebhookHandling]) + else: + logger.error("Usage: billing.py test") + + +if __name__ == "__main__": + main() |
