tests/test_webhooks.py

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")