daeecc078d
BillingCore.create_client now returns the flat record keyed as 'id' (main's PR #16/#19 rename). The Stripe test fixtures were reading the old 'client_id' return key, failing 5 tests on current main. Only the return value changed; the 'client_id' parameter to member_payload and the client_id metadata key are untouched.
312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""Hermetic tests for the Stripe adapter seam (issue #3).
|
|
|
|
Mock-only, zero network. Run with:
|
|
python -m pytest tests/test_stripe_client.py -q
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# PEP 420 namespace package: make the project root importable regardless of
|
|
# how pytest was invoked.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from gitea_saas.billing_core import BillingCore # noqa: E402
|
|
from gitea_saas.stripe_client import ( # noqa: E402
|
|
MockStripeClient,
|
|
StripeClient,
|
|
StripeConfigError,
|
|
StripeSignatureError,
|
|
apply_payment_event,
|
|
build_stripe_client,
|
|
)
|
|
|
|
# Explicit mock signature for the hermetic suite. The point of the bounce
|
|
# fix: this constant lives in TESTS ONLY; the production client never has
|
|
# a default signature.
|
|
MOCK_SIG = "test-mock-secret"
|
|
|
|
WEBHOOK_PAYLOAD = {
|
|
"id": "evt_1A2B3C",
|
|
"type": "payment.succeeded",
|
|
"data": {
|
|
"object": {
|
|
"id": "pi_xyz",
|
|
"metadata": {"client_id": 7},
|
|
"amount_total": 2500,
|
|
"currency": "usd",
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def payload_bytes(data=None) -> bytes:
|
|
return json.dumps(data or WEBHOOK_PAYLOAD).encode("utf-8")
|
|
|
|
|
|
def member_payload(event_id, event_type, client_id=None, amount=None):
|
|
return {"id": event_id, "type": event_type,
|
|
"data": {"object": {"metadata": {"client_id": client_id},
|
|
"amount_total": amount}}}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Checkout sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_mock_creates_checkout_session_with_correct_fields():
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
result = client.create_checkout_session(
|
|
client_id=7, amount_minor=2500, currency="usd",
|
|
success_url="https://example.test/success", cancel_url="https://example.test/cancel",
|
|
)
|
|
assert set(result) == {"checkout_url", "session_id"}
|
|
assert result["session_id"] == "cs_mock_7_2500_1"
|
|
assert result["checkout_url"].endswith("/" + result["session_id"])
|
|
assert client.created_checkouts == [{
|
|
"client_id": 7, "amount_minor": 2500, "currency": "usd",
|
|
"session_id": "cs_mock_7_2500_1",
|
|
"checkout_url": f"https://checkout.stripe.com/mock/cs_mock_7_2500_1",
|
|
"success_url": "https://example.test/success",
|
|
"cancel_url": "https://example.test/cancel",
|
|
}]
|
|
# Deterministic: next session id increments.
|
|
second = client.create_checkout_session(7, 2500)
|
|
assert second["session_id"] == "cs_mock_7_2500_2"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Webhook mapping + signature check
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_handler_maps_payment_succeeded_fields():
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
event = client.process_webhook(payload_bytes(), client.webhook_signature)
|
|
assert event == {"event_id": "evt_1A2B3C", "type": "payment.succeeded",
|
|
"client_id": 7, "amount_minor": 2500}
|
|
assert client.processed_events == [event]
|
|
|
|
|
|
def test_bad_signature_raises_signature_error():
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
with pytest.raises(StripeSignatureError):
|
|
client.process_webhook(payload_bytes(), "wrong-signature")
|
|
assert client.processed_events == [] # nothing recorded
|
|
|
|
|
|
def test_non_payment_events_map_without_credit_fields():
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
event = client.process_webhook(
|
|
payload_bytes(member_payload("evt_9", "checkout.session.completed")),
|
|
client.webhook_signature,
|
|
)
|
|
assert event == {"event_id": "evt_9", "type": "checkout.session.completed",
|
|
"client_id": None, "amount_minor": None}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# payment.succeeded -> idempotent credit tie-in
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_payment_succeeded_credits_once_and_duplicate_delivery_is_noop(tmp_path):
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
ok, c = core.create_client("stripe-user")
|
|
assert ok
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
|
|
payload = payload_bytes(member_payload("evt_credit", "payment.succeeded",
|
|
client_id=c["id"], amount=2000))
|
|
ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature))
|
|
assert ok and details["applied"] is True
|
|
assert details["amount_minor"] == 2000
|
|
assert details["balance"] == 2000
|
|
assert details["client_id"] == c["id"]
|
|
|
|
# Same event_id re-delivered: no double credit.
|
|
ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature))
|
|
assert ok and details["applied"] is True
|
|
assert details["replayed"] is True
|
|
assert details["balance"] == 2000
|
|
|
|
ok, ledger = core.list_ledger(c["id"])
|
|
assert ok and ledger["count"] == 1 # one credit row, applied once
|
|
entry = ledger["entries"][0]
|
|
assert entry["kind"] == "credit" and entry["idempotency_key"] == "evt_credit"
|
|
|
|
|
|
def test_apply_payment_event_ignores_non_payment_events(tmp_path):
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
ok, c = core.create_client("stripe-user")
|
|
assert ok
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
event = client.process_webhook(
|
|
payload_bytes(member_payload("evt_1", "checkout.session.completed",
|
|
client_id=c["id"], amount=5000)),
|
|
client.webhook_signature,
|
|
)
|
|
ok, details = apply_payment_event(core, event)
|
|
assert ok and details["applied"] is False
|
|
ok, ledger = core.list_ledger(c["id"])
|
|
assert ok and ledger["count"] == 0
|
|
|
|
|
|
def test_apply_payment_event_unknown_client_structured_error(tmp_path):
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
core.create_client("other")
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
event = client.process_webhook(
|
|
payload_bytes(member_payload("evt_u", "payment.succeeded", client_id=999, amount=100)),
|
|
client.webhook_signature,
|
|
)
|
|
ok, details = apply_payment_event(core, event)
|
|
assert ok is False and details["applied"] is False
|
|
assert details["error"]["code"] == "UNKNOWN_CLIENT"
|
|
|
|
|
|
def test_apply_payment_event_incomplete_event_structured_error(tmp_path):
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
ok, c = core.create_client("stripe-user")
|
|
assert ok
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
# Missing amount_total.
|
|
event = client.process_webhook(
|
|
payload_bytes({"id": "evt_i", "type": "payment.succeeded",
|
|
"data": {"object": {"metadata": {"client_id": c["id"]}}}}),
|
|
client.webhook_signature,
|
|
)
|
|
ok, details = apply_payment_event(core, event)
|
|
assert ok is False and details["error"]["code"] == "INCOMPLETE_EVENT"
|
|
|
|
|
|
def test_apply_payment_event_coerces_string_metadata_client_id(tmp_path):
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
ok, c = core.create_client("stripe-user")
|
|
assert ok
|
|
client = MockStripeClient(webhook_signature=MOCK_SIG)
|
|
# Stripe metadata values arrive as strings; the tie-in must coerce.
|
|
payload = payload_bytes(member_payload("evt_s", "payment.succeeded",
|
|
client_id=str(c["id"]), amount=750))
|
|
event = client.process_webhook(payload, client.webhook_signature)
|
|
ok, details = apply_payment_event(core, event)
|
|
assert ok and details["applied"] is True and details["balance"] == 750
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Factory: fail-closed credential resolution (bounce fix)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_missing_credentials_raise_without_mock_optin(monkeypatch):
|
|
"""No env keys and no explicit mock opt-in -> construction raises,
|
|
naming the missing variables without leaking any value."""
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
|
|
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
|
|
monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False)
|
|
with pytest.raises(StripeConfigError) as excinfo:
|
|
build_stripe_client(secret_key="sk_test_SUPER_SECRET")
|
|
message = str(excinfo.value)
|
|
assert "STRIPE_WEBHOOK_SECRET" in message
|
|
assert "sk_test_SUPER_SECRET" not in message # name only, never the value
|
|
# Neither key at all -> both named.
|
|
with pytest.raises(StripeConfigError) as excinfo2:
|
|
build_stripe_client()
|
|
message2 = str(excinfo2.value)
|
|
assert "STRIPE_SECRET_KEY" in message2 and "STRIPE_WEBHOOK_SECRET" in message2
|
|
|
|
|
|
def test_only_one_of_two_keys_raises(monkeypatch):
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
|
|
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
|
|
monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False)
|
|
monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_abc")
|
|
with pytest.raises(StripeConfigError) as excinfo:
|
|
build_stripe_client()
|
|
assert "STRIPE_WEBHOOK_SECRET" in str(excinfo.value)
|
|
assert "sk_test_abc" not in str(excinfo.value)
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY")
|
|
monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_abc")
|
|
with pytest.raises(StripeConfigError) as excinfo2:
|
|
build_stripe_client()
|
|
assert "STRIPE_SECRET_KEY" in str(excinfo2.value)
|
|
assert "whsec_abc" not in str(excinfo2.value)
|
|
|
|
|
|
def test_live_mode_still_works_when_both_keys_set(monkeypatch):
|
|
monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_abc")
|
|
monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_abc")
|
|
monkeypatch.setenv("SAAS_STRIPE_MODE", "live")
|
|
client = build_stripe_client()
|
|
assert isinstance(client, StripeClient)
|
|
assert client.secret_key == "sk_test_abc"
|
|
assert client.webhook_secret == "whsec_abc"
|
|
|
|
|
|
def test_explicit_mock_optin_with_explicit_signature_works(monkeypatch):
|
|
"""The suite's seam survives: explicit opt-in + explicit signature."""
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
|
|
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
|
|
monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False)
|
|
client = build_stripe_client(mock=True, webhook_signature=MOCK_SIG)
|
|
assert isinstance(client, MockStripeClient)
|
|
assert client.webhook_signature == MOCK_SIG
|
|
# And the seam functions end to end.
|
|
event = client.process_webhook(payload_bytes(), MOCK_SIG)
|
|
assert event["event_id"] == "evt_1A2B3C"
|
|
|
|
|
|
def test_env_mock_mode_requires_explicit_signature(monkeypatch):
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
|
|
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
|
|
monkeypatch.setenv("SAAS_STRIPE_MODE", "mock")
|
|
# Opt-in via env, but no signature -> fail closed (no published default).
|
|
with pytest.raises(StripeConfigError):
|
|
build_stripe_client()
|
|
client = build_stripe_client(webhook_signature=MOCK_SIG)
|
|
assert isinstance(client, MockStripeClient)
|
|
assert client.webhook_signature == MOCK_SIG
|
|
|
|
|
|
def test_mock_without_signature_fails_closed():
|
|
# No default exists any more: omitting the signature is a hard
|
|
# TypeError at construction (fail-closed; nothing to inherit).
|
|
with pytest.raises(TypeError):
|
|
MockStripeClient() # noqa: E1120 - deliberately missing the arg
|
|
with pytest.raises(StripeConfigError):
|
|
MockStripeClient(webhook_signature=None)
|
|
with pytest.raises(StripeConfigError):
|
|
MockStripeClient(webhook_signature="")
|
|
with pytest.raises(StripeConfigError):
|
|
build_stripe_client(mock=True)
|
|
|
|
|
|
def test_forged_webhook_regression_balance_unchanged(tmp_path, monkeypatch):
|
|
"""Bounce regression: the exact forged-webhook scenario fails closed.
|
|
|
|
With credentials absent (and no mock opt-in) construction RAISES, so a
|
|
deployment can never silently degrade to a public-constant verifier.
|
|
Even on an explicitly-opted-in mock, a webhook signed with the old
|
|
published constant is rejected and the balance stays exactly 0 --
|
|
assert the balance, not just the exception, so a future refactor
|
|
cannot credit and then raise.
|
|
"""
|
|
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
|
|
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
|
|
monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False)
|
|
with pytest.raises(StripeConfigError):
|
|
build_stripe_client()
|
|
|
|
core = BillingCore(tmp_path / "billing.db")
|
|
ok, c = core.create_client("target")
|
|
assert ok
|
|
# An operator-chosen mock (this deployment's own secret, not a constant).
|
|
client = MockStripeClient(webhook_signature="operator-chosen-secret")
|
|
forged = payload_bytes(member_payload("evt_forged", "payment.succeeded",
|
|
client_id=c["id"], amount=500000))
|
|
with pytest.raises(StripeSignatureError):
|
|
event = client.process_webhook(forged, "mock-webhook-secret") # old constant
|
|
apply_payment_event(core, event) # pragma: no cover - never reached
|
|
ok, balance = core.get_balance(c["id"])
|
|
assert ok and balance["balance"] == 0 # unchanged: nothing was credited
|