import os import pytest from unittest.mock import patch from fastapi.testclient import TestClient from app.main import app client = TestClient(app) # Skip all tests in this module if Stripe webhook secret is not set STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET") if not STRIPE_WEBHOOK_SECRET: pytest.skip( "Stripe webhook not configured – skipping webhook tests", allow_module_level=True) @pytest.fixture def mock_stripe_webhook(): with patch("stripe.Webhook.construct_event") as mock: yield mock def test_webhook_missing_secret(monkeypatch): monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "") response = client.post( "/webhooks/stripe", json={}, headers={"stripe-signature": "test"} ) assert response.status_code == 500 assert "Stripe not configured" in response.json()["detail"] def test_webhook_invalid_payload(mock_stripe_webhook, monkeypatch): monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_test") mock_stripe_webhook.side_effect = ValueError("Invalid payload") response = client.post( "/webhooks/stripe", json={}, headers={"stripe-signature": "test"} ) assert response.status_code == 400 assert "Invalid payload" in response.json()["detail"] def test_webhook_invalid_signature(mock_stripe_webhook, monkeypatch): monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_test") mock_stripe_webhook.side_effect = Exception("Invalid signature") response = client.post( "/webhooks/stripe", json={}, headers={"stripe-signature": "test"} ) assert response.status_code == 400 assert "Invalid signature" in response.json()["detail"] def test_webhook_checkout_completed(monkeypatch): monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_test") monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test") with patch("stripe.Webhook.construct_event") as mock_construct, \ patch("app.core.usage_tracker.update_key_tier") as mock_update: mock_construct.return_value = { "type": "checkout.session.completed", "data": { "object": { "client_reference_id": "test_key", "metadata": { "api_key": "test_key"}}}} response = client.post( "/webhooks/stripe", json={}, headers={"stripe-signature": "test"} ) assert response.status_code == 200 mock_update.assert_called_once_with("test_key", "pro") def test_webhook_subscription_deleted(monkeypatch): monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_test") monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test") with patch("stripe.Webhook.construct_event") as mock_construct, \ patch("app.core.usage_tracker.update_key_tier") as mock_update: mock_construct.return_value = { "type": "customer.subscription.deleted", "data": {"object": {"metadata": {"api_key": "test_key"}}} } response = client.post( "/webhooks/stripe", json={}, headers={"stripe-signature": "test"} ) assert response.status_code == 200 mock_update.assert_called_once_with("test_key", "free")