app.bettersight.io/backend/tests/services/test_billing_service.py

"""
Stripe webhook handling is a 100%-coverage path per CLAUDE.md §18.
These tests exercise the pure data-transformation logic in
billing_service (no real Stripe API calls are made).
"""
from services import billing_service
from repositories.tenant_repository import tenant_repository


def test_checkout_completed_with_upgrade_metadata_updates_tenant():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5,
    })

    billing_service.handle_checkout_completed({
        'metadata': {'tenant_id': tenant['id'], 'target_tier': 'discover'},
        'subscription': 'sub_123',
    })

    updated = tenant_repository.get_by_id(tenant['id'])
    assert updated['tier'] == 'discover'
    assert updated['max_seats'] == 8
    assert updated['stripe_subscription_id'] == 'sub_123'
    assert updated['status'] == 'trial'
    assert updated['trial_ends_at'] is not None


def test_checkout_completed_without_metadata_is_ignored_not_raised():
    # New-signup checkout sessions (no upgrade metadata) are out of
    # scope for this handler — manual onboarding handles new tenants.
    # Must not raise.
    billing_service.handle_checkout_completed({'metadata': {}})


def test_subscription_updated_changes_tier_when_present_in_metadata():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5,
    })

    billing_service.handle_subscription_updated({'metadata': {'tier': 'intelligence'}}, tenant['id'])

    updated = tenant_repository.get_by_id(tenant['id'])
    assert updated['tier'] == 'intelligence'
    assert updated['max_seats'] == 10


def test_subscription_updated_no_op_without_tier_metadata():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5,
    })

    billing_service.handle_subscription_updated({'metadata': {}}, tenant['id'])

    updated = tenant_repository.get_by_id(tenant['id'])
    assert updated['tier'] == 'analyse'


def test_subscription_deleted_marks_tenant_inactive():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5,
    })

    billing_service.handle_subscription_deleted(tenant['id'])

    assert tenant_repository.get_by_id(tenant['id'])['status'] == 'inactive'


def test_upgrade_checkout_rejects_unconfigured_tier(monkeypatch):
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5,
    })
    monkeypatch.delenv('STRIPE_PRICE_DISCOVER', raising=False)

    try:
        billing_service.create_upgrade_checkout(tenant['id'], 'discover')
        assert False, 'expected ValueError'
    except ValueError as e:
        assert 'No Stripe price configured' in str(e)


class _FakeCheckoutSession:
    url = 'https://checkout.stripe.com/fake-session'


def _fake_stripe(monkeypatch, captured):
    """
    Stands in for services.billing_service._stripe() — captures the
    kwargs checkout.Session.create() was called with instead of
    hitting the real Stripe API, so the customer/customer_email
    branching can be asserted directly.
    """
    fake = type('_FakeStripe', (), {
        'checkout': type('_Checkout', (), {
            'Session': type('_Session', (), {
                'create': staticmethod(lambda **kwargs: captured.update(kwargs) or _FakeCheckoutSession())
            })
        })
    })
    monkeypatch.setattr(billing_service, '_stripe', lambda: fake)


def test_upgrade_checkout_omits_customer_when_tenant_has_none(monkeypatch):
    """
    Regression test — Stripe rejects an empty-string `customer` param
    outright (raises InvalidRequestError), and every trial/manually-
    onboarded tenant (CLAUDE.md §32) starts with stripe_customer_id=""
    rather than unset. This must be omitted, not passed empty.
    """
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'email': 'pm@gadventures.com',
        'tier': 'analyse', 'status': 'trial', 'max_seats': 5, 'stripe_customer_id': '',
    })
    monkeypatch.setenv('STRIPE_PRICE_DISCOVER', 'price_123')
    captured = {}
    _fake_stripe(monkeypatch, captured)

    url = billing_service.create_upgrade_checkout(tenant['id'], 'discover')

    assert url == 'https://checkout.stripe.com/fake-session'
    assert 'customer' not in captured
    assert captured['customer_email'] == 'pm@gadventures.com'


def test_upgrade_checkout_uses_existing_customer_id_when_present(monkeypatch):
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'email': 'pm@gadventures.com',
        'tier': 'analyse', 'status': 'active', 'max_seats': 5, 'stripe_customer_id': 'cus_existing123',
    })
    monkeypatch.setenv('STRIPE_PRICE_DISCOVER', 'price_123')
    captured = {}
    _fake_stripe(monkeypatch, captured)

    billing_service.create_upgrade_checkout(tenant['id'], 'discover')

    assert captured['customer'] == 'cus_existing123'
    assert 'customer_email' not in captured


def test_checkout_completed_captures_new_stripe_customer_id():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'trial', 'max_seats': 5, 'stripe_customer_id': '',
    })

    billing_service.handle_checkout_completed({
        'metadata': {'tenant_id': tenant['id'], 'target_tier': 'discover'},
        'subscription': 'sub_123',
        'customer': 'cus_new456',
    })

    assert tenant_repository.get_by_id(tenant['id'])['stripe_customer_id'] == 'cus_new456'


def test_checkout_completed_does_not_clobber_customer_id_when_event_lacks_one():
    tenant = tenant_repository.create({
        'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
        'status': 'active', 'max_seats': 5, 'stripe_customer_id': 'cus_existing123',
    })

    billing_service.handle_checkout_completed({
        'metadata': {'tenant_id': tenant['id'], 'target_tier': 'intelligence'},
        'subscription': 'sub_123',
    })

    assert tenant_repository.get_by_id(tenant['id'])['stripe_customer_id'] == 'cus_existing123'