"""Stripe billing: the webhook is the source of truth, folded into the local row.
No live Stripe calls anywhere. `is_enabled()` gates them off, `sync_from_event`
and `construct_event` are exercised with plain event dicts - the shape Stripe
delivers - and the two functions that would reach the network are given a stand-in
client. So this file needs neither a Stripe account nor the `stripe` package
installed, which is the same property the app itself has.
"""
import pytest
from app.core import billing
from app.core.context import Invalid
from db.db_declarations import Organization, Subscription
def _org_with_sub(session, tier="free", status="active", slug="acme", **kw):
org = Organization(name=slug.title(), slug=slug)
session.add(org)
session.commit()
sub = Subscription(org_id=org.id, tier=tier, status=status, **kw)
session.add(sub)
session.commit()
return org, sub
@pytest.fixture(autouse=True)
def _cold_price_cache():
"""No test inherits another's cached amounts."""
billing.clear_price_cache()
yield
billing.clear_price_cache()
# --- the switch -------------------------------------------------------------
def test_billing_is_off_without_a_key(monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "")
assert billing.is_enabled() is False
# And the one call that would need the SDK refuses rather than importing it.
with pytest.raises(Invalid):
billing._client()
def test_billing_is_on_with_a_key(monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "sk_test_x")
assert billing.is_enabled() is True
def test_an_unconfigured_price_refuses_rather_than_starting_a_checkout(monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_SILVER", "")
with pytest.raises(Invalid):
billing.price_for_tier("silver")
def test_the_webhook_needs_a_configured_secret(monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_WEBHOOK_SECRET", "")
with pytest.raises(Invalid):
billing.construct_event(b"{}", "sig")
# --- the events -------------------------------------------------------------
def test_checkout_completed_sets_the_tier(session):
org, sub = _org_with_sub(session)
event = {"type": "checkout.session.completed",
"data": {"object": {
"client_reference_id": str(org.id),
"customer": "cus_1", "subscription": "sub_1",
"metadata": {"org_id": str(org.id), "tier": "silver"}}}}
assert billing.sync_from_event(session, event) is True
session.refresh(sub)
assert (sub.tier, sub.status) == ("silver", "active")
assert sub.stripe_customer_id == "cus_1"
assert sub.stripe_subscription_id == "sub_1"
def test_applying_the_same_event_twice_changes_nothing(session):
"""Stripe delivers at least once and retries. Two deliveries must converge on
one row rather than producing two."""
org, sub = _org_with_sub(session)
event = {"type": "checkout.session.completed",
"data": {"object": {
"client_reference_id": str(org.id),
"customer": "cus_1", "subscription": "sub_1",
"metadata": {"tier": "silver"}}}}
billing.sync_from_event(session, event)
billing.sync_from_event(session, event)
session.refresh(sub)
assert sub.tier == "silver"
assert session.query(Subscription).filter_by(org_id=org.id).count() == 1
def test_subscription_updated_maps_the_price_back_to_a_tier(session, monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_GOLD", "price_g")
_, sub = _org_with_sub(session, tier="silver", stripe_customer_id="cus_2")
event = {"type": "customer.subscription.updated",
"data": {"object": {
"id": "sub_2", "customer": "cus_2", "status": "active",
"current_period_end": 2000000000,
"items": {"data": [{"price": {"id": "price_g"}}]}}}}
assert billing.sync_from_event(session, event) is True
session.refresh(sub)
assert sub.tier == "gold"
assert sub.stripe_price_id == "price_g"
assert sub.current_period_end is not None
def test_a_price_we_do_not_know_leaves_the_tier_alone(session, monkeypatch):
"""A Price created in the Stripe dashboard and never configured here. Better
to keep the tier the customer had than to guess at one."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_SILVER", "price_s")
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_GOLD", "")
_, sub = _org_with_sub(session, tier="silver", stripe_customer_id="cus_3")
event = {"type": "customer.subscription.updated",
"data": {"object": {
"id": "sub_3", "customer": "cus_3", "status": "active",
"items": {"data": [{"price": {"id": "price_unknown"}}]}}}}
assert billing.sync_from_event(session, event) is True
session.refresh(sub)
assert sub.tier == "silver"
def test_subscription_deleted_reverts_to_free_and_keeps_the_customer(session):
_, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_4",
stripe_subscription_id="sub_4",
cancel_at_period_end=True)
event = {"type": "customer.subscription.deleted",
"data": {"object": {"id": "sub_4", "customer": "cus_4"}}}
assert billing.sync_from_event(session, event) is True
session.refresh(sub)
assert (sub.tier, sub.status) == ("free", "canceled")
assert sub.stripe_subscription_id is None
# Kept, so a later re-subscribe reuses the same Stripe customer.
assert sub.stripe_customer_id == "cus_4"
# Cleared, or a re-subscribe would start out claiming it was already cancelled.
assert sub.cancel_at_period_end is False
def test_a_failed_payment_marks_the_row_past_due(session):
_, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_6")
event = {"type": "invoice.payment_failed",
"data": {"object": {"customer": "cus_6"}}}
assert billing.sync_from_event(session, event) is True
session.refresh(sub)
assert sub.status == "past_due"
# And the entitlement is gone without the row being deleted, so the billing
# page can still explain what happened.
from app.core import plans
assert plans.effective_tier(sub) == "free"
def test_an_event_we_do_not_act_on_is_a_no_op(session):
"""False, not an exception: the endpoint still answers 200, because an
unhandled event is not an error and Stripe should not retry it."""
_org_with_sub(session)
event = {"type": "customer.updated", "data": {"object": {"id": "cus_x"}}}
assert billing.sync_from_event(session, event) is False
def test_an_event_for_an_unknown_customer_is_refused_quietly(session):
event = {"type": "invoice.payment_failed",
"data": {"object": {"customer": "cus_nobody"}}}
assert billing.sync_from_event(session, event) is False
# --- cancelling from inside the product -------------------------------------
class _FakeStripe:
"""The one call set_cancel_at_period_end makes, recorded. Stripe's own
response is echoed back rather than the argument, so a test can prove the row
is written from what Stripe did and not from what we asked for."""
def __init__(self, reply=None):
self.modified = []
self.reply = reply
outer = self
class Subscription:
@staticmethod
def modify(sub_id, **kw):
outer.modified.append((sub_id, kw))
return outer.reply if outer.reply is not None else {
"cancel_at_period_end": kw.get("cancel_at_period_end"),
"current_period_end": 2000000000,
}
self.Subscription = Subscription
def _stripe(monkeypatch, **kw):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "sk_test_x")
fake = _FakeStripe(**kw)
monkeypatch.setattr(billing, "_client", lambda: fake)
return fake
def test_cancel_asks_stripe_to_stop_at_the_period_end(session, monkeypatch):
org, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_5",
stripe_subscription_id="sub_5")
fake = _stripe(monkeypatch)
out = billing.set_cancel_at_period_end(session, org, True)
assert fake.modified == [("sub_5", {"cancel_at_period_end": True})]
assert out["cancel_at_period_end"] is True
session.refresh(sub)
assert sub.cancel_at_period_end is True
def test_resume_is_the_same_call_with_the_flag_off(session, monkeypatch):
org, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_5",
stripe_subscription_id="sub_5",
cancel_at_period_end=True)
fake = _stripe(monkeypatch)
billing.set_cancel_at_period_end(session, org, False)
assert fake.modified == [("sub_5", {"cancel_at_period_end": False})]
session.refresh(sub)
assert sub.cancel_at_period_end is False
def test_the_row_follows_stripe_and_not_the_argument(session, monkeypatch):
"""Written from Stripe's response, so the page reflects what Stripe actually
did rather than what we asked for - and the webhook that follows finds the
row already agreeing with it."""
org, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_5",
stripe_subscription_id="sub_5")
_stripe(monkeypatch, reply={"cancel_at_period_end": False,
"current_period_end": 2000000000})
billing.set_cancel_at_period_end(session, org, True)
session.refresh(sub)
assert sub.cancel_at_period_end is False
def test_cancelling_a_plan_that_is_not_paid_is_refused(session, monkeypatch):
"""There is nothing at Stripe to modify, and reporting success for a
cancellation that never happened is worse than refusing."""
org, _ = _org_with_sub(session)
_stripe(monkeypatch)
with pytest.raises(Invalid):
billing.set_cancel_at_period_end(session, org, True)
def test_a_cancellation_taken_at_stripe_shows_up_here(session):
"""The portal and the Stripe dashboard can cancel too. The flag is read off
the event rather than only set by our own call."""
_, sub = _org_with_sub(session, tier="gold", stripe_customer_id="cus_7",
stripe_subscription_id="sub_7")
event = {"type": "customer.subscription.updated",
"data": {"object": {
"id": "sub_7", "customer": "cus_7", "status": "active",
"cancel_at_period_end": True,
"current_period_end": 2000000000,
"items": {"data": []}}}}
billing.sync_from_event(session, event)
session.refresh(sub)
assert sub.cancel_at_period_end is True
def test_a_winding_down_plan_still_entitles_until_it_lapses(session):
"""The customer paid through a date and keeps what they paid for. Entitlement
deliberately does not read cancel_at_period_end."""
from app.core import plans
_, sub = _org_with_sub(session, tier="gold", status="active",
cancel_at_period_end=True)
assert plans.effective_tier(sub) == "gold"
# --- ending everything ------------------------------------------------------
class _FakeAccount:
"""Subscription.delete and Customer.delete, recorded. `missing` makes the
call raise the error Stripe raises for an object that is already gone."""
def __init__(self, missing=()):
self.deleted = []
outer = self
def _delete(kind):
def go(obj_id):
outer.deleted.append((kind, obj_id))
if kind in missing:
raise type("StripeError", (Exception,),
{"code": "resource_missing"})()
return go
self.Subscription = type("S", (), {"delete": staticmethod(_delete("sub"))})
self.Customer = type("C", (), {"delete": staticmethod(_delete("cus"))})
def test_cancel_everything_ends_the_subscription_then_the_customer(session, monkeypatch):
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "sk_test_x")
org, _ = _org_with_sub(session, tier="gold", stripe_customer_id="cus_8",
stripe_subscription_id="sub_8")
fake = _FakeAccount()
monkeypatch.setattr(billing, "_client", lambda: fake)
billing.cancel_everything(session, org)
# In that order: the state Stripe ends in is one this code asked for, not a
# side effect of deleting the customer.
assert fake.deleted == [("sub", "sub_8"), ("cus", "cus_8")]
def test_cancel_everything_tolerates_what_is_already_gone(session, monkeypatch):
"""Closing an account twice, or closing one whose subscription was cancelled
from the dashboard yesterday, must reach the same end state."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "sk_test_x")
org, _ = _org_with_sub(session, tier="gold", stripe_customer_id="cus_9",
stripe_subscription_id="sub_9")
monkeypatch.setattr(billing, "_client",
lambda: _FakeAccount(missing=("sub", "cus")))
billing.cancel_everything(session, org) # does not raise
def test_cancel_everything_is_a_no_op_for_a_free_tenant(session, monkeypatch):
"""Closing a free account must not require a Stripe key to be set."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "")
org, _ = _org_with_sub(session)
def _boom():
raise AssertionError("Stripe must not be reached here.")
monkeypatch.setattr(billing, "_client", _boom)
billing.cancel_everything(session, org)
# --- the price a card shows -------------------------------------------------
class _FakePrices:
"""Stand-in for stripe.Price, counting retrievals so the cache is testable."""
def __init__(self, prices, fail=False):
self.prices = prices
self.fail = fail
self.calls = 0
def retrieve(self, price_id):
self.calls += 1
if self.fail:
raise RuntimeError("Stripe is having a day")
return self.prices[price_id]
def _stub_prices(monkeypatch, fake):
"""Swap the configured Stripe client for `fake` and start from a cold cache."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "sk_test_x")
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_SILVER", "price_s")
monkeypatch.setattr(settings.Config, "STRIPE_PRICE_GOLD", "")
monkeypatch.setattr(billing, "_client", lambda: type("C", (), {"Price": fake}))
billing.clear_price_cache()
return fake
def test_no_live_prices_without_stripe(monkeypatch):
"""The common deployment: no key, no calls, and the declared prices stand."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "")
assert billing.live_prices() == {}
def test_a_live_price_is_read_and_then_cached(monkeypatch):
fake = _stub_prices(monkeypatch, _FakePrices({
"price_s": {"unit_amount": 3500, "currency": "chf",
"recurring": {"interval": "month"}}}))
assert billing.live_prices() == {
"silver": {"amount": 3500, "currency": "CHF", "interval": "month"}}
assert fake.calls == 1
# No page load should wait on a Stripe round-trip twice.
billing.live_prices()
assert fake.calls == 1
def test_a_stripe_failure_falls_back_rather_than_raising(monkeypatch):
"""A stale number is bad; a broken billing page is worse."""
_stub_prices(monkeypatch, _FakePrices({}, fail=True))
assert billing.live_prices() == {}
def test_a_price_with_no_flat_amount_is_skipped(monkeypatch):
"""A tiered or metered price has no single number to put on a card."""
_stub_prices(monkeypatch, _FakePrices({
"price_s": {"unit_amount": None, "currency": "chf"}}))
assert billing.live_prices() == {}
def test_stripe_wins_over_the_declared_amount(monkeypatch):
"""What the card says and what the checkout charges cannot drift apart while
Stripe is answering."""
from app.core import plans
_stub_prices(monkeypatch, _FakePrices({
"price_s": {"unit_amount": 3500, "currency": "chf",
"recurring": {"interval": "month"}}}))
tiers = plans.catalogue()
declared_gold = tiers["gold"]["price"]["amount"]
billing.apply_live_prices(tiers)
assert tiers["silver"]["price"]["amount"] == 3500
# Gold has no configured price id, so its declared amount stands.
assert tiers["gold"]["price"]["amount"] == declared_gold
# --- over HTTP --------------------------------------------------------------
def test_a_member_is_refused_every_billing_mutation(account):
"""The link is hidden for a member, and hiding is decoration: the routes
behind it refuse on their own."""
from tests.conftest import csrf_for
client, _, _ = account
csrf = csrf_for(client)
for path in ("/api/billing/checkout", "/api/billing/portal",
"/api/billing/cancel", "/api/billing/resume"):
r = client.post(path, json={"tier": "silver"},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 403, (path, r.get_json())
def test_an_owner_passes_the_bar_and_meets_the_config(app_client, monkeypatch):
"""Past the permission check, an unconfigured deployment says so plainly
rather than erroring - the same 'unset config = feature off' stance."""
import settings
from tests.conftest import csrf_for, sign_in_as_owner
monkeypatch.setattr(settings.Config, "STRIPE_SECRET_KEY", "")
client, _ = sign_in_as_owner(app_client)
r = client.post("/api/billing/checkout", json={"tier": "silver"},
headers={"X-CSRF-Token": csrf_for(client)})
assert r.status_code == 400, r.get_json()
assert "not available" in r.get_json()["error"]
def test_the_billing_page_renders(app_client):
from tests.conftest import sign_in_as_owner
client, _ = sign_in_as_owner(app_client)
r = client.get("/billing")
assert r.status_code == 200
assert b"billing.js" in r.data
def test_the_webhook_is_reachable_without_a_session(app_client, monkeypatch):
"""It holds no cookie, so the guard lets it past to authenticate itself - and
a bad signature is a 400, not a redirect to /login and not a 500 Stripe would
retry."""
import settings
monkeypatch.setattr(settings.Config, "STRIPE_WEBHOOK_SECRET", "")
r = app_client.post("/webhooks/stripe", data=b"{}",
headers={"Stripe-Signature": "nonsense",
"Content-Type": "application/json"})
assert r.status_code == 400, r.get_json()