jcv-dev
551b42eab5
Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
302 lines
12 KiB
Python
302 lines
12 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from infrasynth.billing.gateways.base import (
|
|
BasePaymentGateway,
|
|
CheckoutSessionResult,
|
|
WebhookResult,
|
|
)
|
|
from infrasynth.billing.gateways.mercadopago import MercadoPagoGateway
|
|
from infrasynth.billing.gateways.stripe import StripeGateway
|
|
from infrasynth.billing.gateways.wompi import WompiGateway
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
class _FakePlan:
|
|
slug = "pro"
|
|
name = "Pro"
|
|
price_amount = 100
|
|
price_currency = "USD"
|
|
external_id = "price_123"
|
|
|
|
|
|
class _FakeUser:
|
|
email = "user@example.com"
|
|
pk = 7
|
|
|
|
|
|
class _FakeSubscription:
|
|
external_id = "sub_123"
|
|
|
|
|
|
class _FakeInvoice:
|
|
external_id = "inv_123"
|
|
|
|
|
|
class TestBasePaymentGateway:
|
|
def test_is_abstract(self):
|
|
with pytest.raises(TypeError):
|
|
BasePaymentGateway()
|
|
|
|
def test_result_dataclasses(self):
|
|
result = CheckoutSessionResult("s1", "https://x", "cs")
|
|
assert result.session_id == "s1"
|
|
webhook = WebhookResult("evt", True, {})
|
|
assert webhook.is_handled is True
|
|
|
|
|
|
class TestStripeGateway:
|
|
def test_missing_credentials_raise(self):
|
|
gateway = StripeGateway({})
|
|
with pytest.raises(ValueError, match="API key"):
|
|
gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
|
|
def test_create_checkout_session(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
with mock.patch("stripe.checkout.Session.create") as create:
|
|
create.return_value = mock.Mock(id="cs_1", url="https://stripe.com/checkout", client_secret="cs_sec")
|
|
result = gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
assert result.session_id == "cs_1"
|
|
assert result.checkout_url == "https://stripe.com/checkout"
|
|
create.assert_called_once()
|
|
kwargs = create.call_args.kwargs
|
|
assert kwargs["mode"] == "subscription"
|
|
assert kwargs["line_items"] == [{"price": "price_123", "quantity": 1}]
|
|
assert kwargs["customer_email"] == "user@example.com"
|
|
|
|
def test_create_checkout_without_external_id_raises(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
plan = _FakePlan()
|
|
plan.external_id = ""
|
|
with pytest.raises(ValueError, match="external price ID"):
|
|
gateway.create_checkout_session(plan, _FakeUser())
|
|
|
|
def test_handle_webhook(self):
|
|
gateway = StripeGateway({"api_key": "sk_test", "webhook_secret": "whsec_1"})
|
|
with mock.patch("stripe.Webhook.construct_event") as construct:
|
|
construct.return_value = {
|
|
"type": "checkout.session.completed",
|
|
"data": {"object": {"id": "cs_1"}},
|
|
}
|
|
result = gateway.handle_webhook(
|
|
{"id": "cs_1"},
|
|
{"Stripe-Signature": "t=1,v1=sig"},
|
|
)
|
|
assert result.event_type == "checkout.session.completed"
|
|
assert result.is_handled is True
|
|
assert result.data["id"] == "cs_1"
|
|
|
|
def test_handle_webhook_without_secret_raises(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
with pytest.raises(ValueError, match="webhook secret"):
|
|
gateway.handle_webhook({}, {})
|
|
|
|
def test_cancel_subscription(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
with mock.patch("stripe.Subscription.cancel") as cancel:
|
|
assert gateway.cancel_subscription(_FakeSubscription()) is True
|
|
cancel.assert_called_once_with("sub_123")
|
|
|
|
def test_cancel_without_external_id(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
subscription = _FakeSubscription()
|
|
subscription.external_id = ""
|
|
assert gateway.cancel_subscription(subscription) is False
|
|
|
|
def test_sync_subscription_maps_status(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
with mock.patch("stripe.Subscription.retrieve") as retrieve:
|
|
retrieve.return_value = {
|
|
"status": "past_due",
|
|
"current_period_start": 1700000000,
|
|
"current_period_end": 1702592000,
|
|
"cancel_at_period_end": True,
|
|
"canceled_at": None,
|
|
"trial_end": None,
|
|
"metadata": {"x": "1"},
|
|
}
|
|
data = gateway.sync_subscription(_FakeSubscription())
|
|
assert data["status"] == "past_due"
|
|
assert data["cancel_at_period_end"] is True
|
|
assert data["current_period_end"] is not None
|
|
assert data["metadata"] == {"x": "1"}
|
|
|
|
def test_get_invoice(self):
|
|
gateway = StripeGateway({"api_key": "sk_test"})
|
|
with mock.patch("stripe.Invoice.retrieve") as retrieve:
|
|
retrieve.return_value = {
|
|
"id": "inv_1",
|
|
"status": "paid",
|
|
"amount_due": 12000,
|
|
"currency": "usd",
|
|
"paid_at": 1700000000,
|
|
"lines": {"data": [{"description": "Pro", "amount": 12000, "quantity": 1}]},
|
|
}
|
|
data = gateway.get_invoice(_FakeInvoice())
|
|
assert data["status"] == "paid"
|
|
assert data["amount"] == 120.0
|
|
assert data["currency"] == "USD"
|
|
assert data["line_items"][0]["description"] == "Pro"
|
|
|
|
def test_health_check(self):
|
|
assert StripeGateway({"api_key": "k"}).health_check() is True
|
|
assert StripeGateway({}).health_check() is False
|
|
|
|
|
|
class TestMercadoPagoGateway:
|
|
def test_missing_credentials_raise(self):
|
|
gateway = MercadoPagoGateway({})
|
|
with pytest.raises(ValueError, match="access token"):
|
|
gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
|
|
def test_create_checkout_session(self):
|
|
gateway = MercadoPagoGateway({"access_token": "APP_USR-token"})
|
|
with mock.patch("mercadopago.SDK") as sdk_cls:
|
|
sdk_cls.return_value.preference().create.return_value = {
|
|
"status": 201,
|
|
"response": {
|
|
"id": "pref_1",
|
|
"init_point": "https://mercadopago.com/checkout",
|
|
},
|
|
}
|
|
result = gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
assert result.session_id == "pref_1"
|
|
assert result.checkout_url == "https://mercadopago.com/checkout"
|
|
preference = sdk_cls.return_value.preference().create.call_args.args[0]
|
|
assert preference["items"][0]["title"] == "Pro"
|
|
assert preference["items"][0]["unit_price"] == 100
|
|
|
|
def test_create_checkout_error_status_raises(self):
|
|
gateway = MercadoPagoGateway({"access_token": "tok"})
|
|
with mock.patch("mercadopago.SDK") as sdk_cls:
|
|
sdk_cls.return_value.preference().create.return_value = {
|
|
"status": 400,
|
|
"response": {"message": "bad"},
|
|
}
|
|
with pytest.raises(ValueError, match="MercadoPago error"):
|
|
gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
|
|
def test_handle_webhook(self):
|
|
gateway = MercadoPagoGateway({"access_token": "tok"})
|
|
result = gateway.handle_webhook({"type": "payment", "data": {"id": "pay_1"}}, {})
|
|
assert result.event_type == "payment"
|
|
assert result.is_handled is True
|
|
assert result.data == {"id": "pay_1"}
|
|
|
|
def test_cancel_subscription(self):
|
|
gateway = MercadoPagoGateway({"access_token": "tok"})
|
|
with mock.patch("mercadopago.SDK") as sdk_cls:
|
|
sdk_cls.return_value.preapproval().update.return_value = {"status": 200}
|
|
assert gateway.cancel_subscription(_FakeSubscription()) is True
|
|
sdk_cls.return_value.preapproval().update.assert_called_once_with("sub_123", {"status": "cancelled"})
|
|
|
|
def test_sync_subscription(self):
|
|
gateway = MercadoPagoGateway({"access_token": "tok"})
|
|
with mock.patch("mercadopago.SDK") as sdk_cls:
|
|
sdk_cls.return_value.preapproval().get.return_value = {
|
|
"status": 200,
|
|
"response": {"status": "cancelled", "metadata": {"a": 1}},
|
|
}
|
|
data = gateway.sync_subscription(_FakeSubscription())
|
|
assert data["status"] == "cancelled"
|
|
assert data["metadata"] == {"a": 1}
|
|
|
|
def test_health_check(self):
|
|
assert MercadoPagoGateway({"access_token": "t"}).health_check() is True
|
|
assert MercadoPagoGateway({}).health_check() is False
|
|
|
|
|
|
class TestWompiGateway:
|
|
def test_missing_credentials_raise(self):
|
|
gateway = WompiGateway({})
|
|
with pytest.raises(ValueError, match="public key"):
|
|
gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
|
|
def test_create_checkout_session(self):
|
|
gateway = WompiGateway({"public_key": "pub_test"})
|
|
with mock.patch("infrasynth.billing.gateways.wompi.requests.post") as post:
|
|
post.return_value.status_code = 201
|
|
post.return_value.json.return_value = {"data": {"id": "plink_1", "url": "https://checkout.wompi.co/link"}}
|
|
result = gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
assert result.session_id == "plink_1"
|
|
assert result.checkout_url == "https://checkout.wompi.co/link"
|
|
payload = post.call_args.kwargs["json"]
|
|
assert payload["amount_in_cents"] == 100
|
|
assert payload["currency"] == "usd"
|
|
|
|
def test_create_checkout_http_error_raises(self):
|
|
gateway = WompiGateway({"public_key": "pub_test"})
|
|
with mock.patch("infrasynth.billing.gateways.wompi.requests.post") as post:
|
|
post.return_value.status_code = 400
|
|
post.return_value.text = "invalid"
|
|
with pytest.raises(ValueError, match="400"):
|
|
gateway.create_checkout_session(_FakePlan(), _FakeUser())
|
|
|
|
def test_handle_webhook_unverified_without_secret(self):
|
|
gateway = WompiGateway({})
|
|
result = gateway.handle_webhook({"event": "transaction.updated", "data": {"id": "t_1"}}, {})
|
|
assert result.event_type == "transaction.updated"
|
|
assert result.is_handled is True
|
|
assert result.data == {"id": "t_1"}
|
|
|
|
def test_handle_webhook_signature_verification(self):
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
payload = {"event": "transaction.updated", "data": {"id": "t_1"}}
|
|
gateway = WompiGateway({"webhook_secret": "whsec"})
|
|
raw = json.dumps(payload, separators=(",", ":"))
|
|
signature = hmac.new(b"whsec", raw.encode(), hashlib.sha256).hexdigest()
|
|
result = gateway.handle_webhook(payload, {"x-signature": signature})
|
|
assert result.is_handled is True
|
|
|
|
result = gateway.handle_webhook(payload, {"x-signature": "tampered"})
|
|
assert result.is_handled is False
|
|
|
|
def test_cancel_subscription_voids_transaction(self):
|
|
gateway = WompiGateway({"secret_key": "sk_test"})
|
|
with mock.patch("infrasynth.billing.gateways.wompi.requests.post") as post:
|
|
post.return_value.status_code = 200
|
|
assert gateway.cancel_subscription(_FakeSubscription()) is True
|
|
post.assert_called_once_with(
|
|
"https://sandbox.wompi.co/v1/transactions/sub_123/void",
|
|
headers={"Authorization": "Bearer sk_test"},
|
|
timeout=30,
|
|
)
|
|
|
|
def test_cancel_without_secret_returns_false(self):
|
|
gateway = WompiGateway({})
|
|
assert gateway.cancel_subscription(_FakeSubscription()) is False
|
|
|
|
def test_sync_subscription(self):
|
|
gateway = WompiGateway({"public_key": "pub"})
|
|
with mock.patch("infrasynth.billing.gateways.wompi.requests.get") as get:
|
|
get.return_value.status_code = 200
|
|
get.return_value.json.return_value = {"data": {"status": "APPROVED", "metadata": {"m": 1}}}
|
|
data = gateway.sync_subscription(_FakeSubscription())
|
|
assert data["status"] == "active"
|
|
assert data["metadata"] == {"m": 1}
|
|
|
|
def test_get_invoice(self):
|
|
gateway = WompiGateway({"public_key": "pub"})
|
|
with mock.patch("infrasynth.billing.gateways.wompi.requests.get") as get:
|
|
get.return_value.status_code = 200
|
|
get.return_value.json.return_value = {
|
|
"data": {
|
|
"id": "t_1",
|
|
"status": "VOIDED",
|
|
"amount_in_cents": 9900,
|
|
"currency": "cop",
|
|
}
|
|
}
|
|
data = gateway.get_invoice(_FakeInvoice())
|
|
assert data["status"] == "VOIDED"
|
|
assert data["amount"] == 99.0
|
|
assert data["currency"] == "COP"
|
|
|
|
def test_health_check(self):
|
|
assert WompiGateway({"public_key": "p"}).health_check() is True
|
|
assert WompiGateway({}).health_check() is False
|