268 lines
15 KiB
Python
Executable File
268 lines
15 KiB
Python
Executable File
import json
|
|
from datetime import datetime, timedelta, timezone as dt_timezone
|
|
from decimal import Decimal
|
|
from unittest.mock import patch
|
|
|
|
from django.conf import settings
|
|
from django.db import IntegrityError, transaction
|
|
from django.test import Client, SimpleTestCase, TestCase, override_settings
|
|
from django.utils import timezone
|
|
|
|
from members import billing, shkeeper
|
|
from members.models import Account, CryptoInvoice, ManagedGroup, Subscription
|
|
from .test_membership import DN, GROUP, OTHER_GROUP, TIERS, account, paged
|
|
|
|
CRYPTO_TIERS = {key: {**value, "usd_amount": "5.00" if key == "supporter" else "15.00"} for key, value in TIERS.items()}
|
|
SHKEEPER_SETTINGS = {
|
|
"SHKEEPER_ENABLED": True, "SHKEEPER_URL": "https://pay.example.org",
|
|
"SHKEEPER_API_KEY": "test-only-key", "SHKEEPER_CA_CERT_FILE": "",
|
|
"SHKEEPER_CURRENCIES": ["BTC", "XMR"], "MEMBERSHIP_TIERS": CRYPTO_TIERS,
|
|
}
|
|
|
|
|
|
@override_settings(**SHKEEPER_SETTINGS)
|
|
class SHKeeperTests(TestCase):
|
|
def setUp(self):
|
|
self.member = account()
|
|
session = self.client.session
|
|
session["account_id"] = str(self.member.pk)
|
|
session.save()
|
|
|
|
def invoice(self, **kwargs):
|
|
return CryptoInvoice.objects.create(account=self.member, tier="supporter", usd_amount="5.00", crypto="BTC", **kwargs)
|
|
|
|
def creation(self):
|
|
return {"status": "success", "id": 61, "wallet": "bc1-test-wallet", "amount": "0.0001", "recalculate_after": 0}
|
|
|
|
def remote(self, invoice, **changes):
|
|
return {"status": "success", "invoices": [{
|
|
"external_id": shkeeper.external_id(invoice), "fiat": "USD",
|
|
"amount_fiat": "5.00", "balance_fiat": "5.00", "status": "PAID",
|
|
"txs": [{"status": "CONFIRMED", "addr": "bc1-test-wallet", "crypto": "BTC", "amount": "0.0001", "txid": "confirmed-tx"}], **changes,
|
|
}]}
|
|
|
|
def prepared(self):
|
|
return self.invoice(provider_id="61", wallet="bc1-test-wallet", crypto_amount="0.0001", status="UNPAID")
|
|
|
|
def login_other(self):
|
|
other = Account.objects.create(uid="bob", email="bob@example.org", ldap_dn="uid=bob,dc=example", ldap_uuid="uuid-bob")
|
|
session = self.client.session
|
|
session["account_id"] = str(other.pk)
|
|
session.save()
|
|
|
|
def test_crypto_buttons_only_when_configured(self):
|
|
self.assertContains(self.client.get("/members/"), "Pay with crypto", count=2)
|
|
with override_settings(SHKEEPER_ENABLED=False):
|
|
self.assertNotContains(self.client.get("/members/"), "Pay with crypto")
|
|
self.assertEqual(self.client.get("/members/crypto/checkout/supporter/").status_code, 404)
|
|
|
|
def test_checkout_form_csrf_login_and_currency_allowlist(self):
|
|
self.assertContains(self.client.get("/members/crypto/checkout/supporter/"), "not an automatically renewing")
|
|
self.assertEqual(self.client.post("/members/crypto/checkout/supporter/", {"crypto": "../../evil"}).status_code, 400)
|
|
self.assertFalse(CryptoInvoice.objects.exists())
|
|
strict = Client(enforce_csrf_checks=True)
|
|
self.assertEqual(strict.post("/members/crypto/checkout/supporter/", {"crypto": "BTC"}).status_code, 403)
|
|
self.client.post("/members/logout/")
|
|
self.assertRedirects(self.client.get("/members/crypto/checkout/supporter/"), "/members/")
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_checkout_uses_server_price_and_stable_uuid(self, request):
|
|
request.return_value = self.creation()
|
|
response = self.client.post("/members/crypto/checkout/supporter/", {"crypto": "BTC", "usd_amount": "0.01", "account": "victim"})
|
|
invoice = CryptoInvoice.objects.get()
|
|
self.assertRedirects(response, f"/members/crypto/invoice/{invoice.pk}/")
|
|
self.assertEqual(invoice.usd_amount, Decimal("5.00"))
|
|
self.assertEqual(invoice.account, self.member)
|
|
method, path, params = request.call_args.args
|
|
self.assertEqual((method, path), ("POST", "/api/v1/BTC/payment_request"))
|
|
self.assertEqual(params["amount"], "5.00")
|
|
self.assertEqual(params["external_id"], shkeeper.external_id(invoice))
|
|
self.assertEqual(params["callback_url"], settings.PUBLIC_URL + "/members/shkeeper/callback/")
|
|
self.assertIsNone(invoice.credited_at)
|
|
self.assertEqual(ManagedGroup.objects.count(), 2)
|
|
# Another crypto/tier request reopens the previous immutable order.
|
|
self.assertRedirects(self.client.post("/members/crypto/checkout/patron/", {"crypto": "XMR"}), f"/members/crypto/invoice/{invoice.pk}/")
|
|
self.assertEqual(request.call_count, 1)
|
|
self.assertEqual(CryptoInvoice.objects.count(), 1)
|
|
|
|
@patch("members.shkeeper.request", side_effect=OSError("timeout"))
|
|
def test_creation_timeout_keeps_reserved_order_for_worker(self, request):
|
|
self.assertContains(self.client.post("/members/crypto/checkout/supporter/", {"crypto": "BTC"}), "checkout unavailable")
|
|
invoice = CryptoInvoice.objects.get()
|
|
self.assertEqual(invoice.status, "creating")
|
|
first_id = request.call_args.args[2]["external_id"]
|
|
request.side_effect = None
|
|
request.return_value = self.creation()
|
|
shkeeper.prepare_invoice(self.member.pk, invoice.pk)
|
|
self.assertEqual(request.call_args.args[2]["external_id"], first_id)
|
|
|
|
def test_only_one_uncredited_order_constraint(self):
|
|
self.invoice()
|
|
with self.assertRaises(IntegrityError), transaction.atomic():
|
|
self.invoice()
|
|
|
|
def test_invoice_is_account_scoped_and_get_never_calls_provider(self):
|
|
invoice = self.prepared()
|
|
with patch("members.shkeeper.request") as request:
|
|
response = self.client.get(f"/members/crypto/invoice/{invoice.pk}/")
|
|
self.assertContains(response, "bc1-test-wallet")
|
|
self.assertContains(response, "Bronze invoice")
|
|
request.assert_not_called()
|
|
self.login_other()
|
|
self.assertEqual(self.client.get(f"/members/crypto/invoice/{invoice.pk}/").status_code, 404)
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_spoofed_callback_cannot_grant_or_trigger_network(self, request):
|
|
invoice = self.prepared()
|
|
for _ in range(3):
|
|
self.assertEqual(self.client.post("/members/shkeeper/callback/", json.dumps({"external_id": shkeeper.external_id(invoice), "paid": True, "status": "PAID", "balance_fiat": "999999"}), content_type="application/json").status_code, 200)
|
|
request.assert_not_called()
|
|
invoice.refresh_from_db()
|
|
self.assertIsNone(invoice.credited_at)
|
|
self.assertEqual(invoice.status, "UNPAID")
|
|
self.assertEqual(self.client.get("/members/shkeeper/callback/").status_code, 405)
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_canonical_paid_once_repeated_callbacks_do_not_extend(self, request):
|
|
invoice = self.prepared()
|
|
request.return_value = self.remote(invoice)
|
|
with transaction.atomic():
|
|
shkeeper.refresh_invoice(invoice)
|
|
first_end = invoice.access_end
|
|
self.assertEqual(invoice.access_end, shkeeper.add_month(invoice.access_start))
|
|
shkeeper.refresh_invoice(invoice)
|
|
self.assertEqual(invoice.access_end, first_end)
|
|
self.assertEqual(invoice.status, "PAID")
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_partial_unpaid_unconfirmed_underpaid_cannot_grant(self, request):
|
|
invoice = self.prepared()
|
|
for changes in ({"status": "UNPAID", "balance_fiat": "0"}, {"status": "PARTIAL", "balance_fiat": "1"}, {"balance_fiat": "4.99"}, {"txs": []}, {"txs": [{"status": "UNCONFIRMED"}]}, {"txs": [{"status": "CONFIRMED", "addr": "wrong", "crypto": "BTC", "txid": "x"}]}):
|
|
request.return_value = self.remote(invoice, **changes)
|
|
shkeeper.refresh_invoice(invoice)
|
|
self.assertIsNone(invoice.credited_at)
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_mismatched_invalid_amount_or_ambiguous_data_rejected(self, request):
|
|
invoice = self.prepared()
|
|
for changes in ({"external_id": "another-order"}, {"fiat": "EUR"}, {"amount_fiat": "0.01"}, {"amount_fiat": "NaN"}, {"balance_fiat": "Infinity"}, {"status": "unknown"}):
|
|
request.return_value = self.remote(invoice, **changes)
|
|
with self.assertRaises(ValueError):
|
|
shkeeper.refresh_invoice(invoice)
|
|
data = self.remote(invoice)
|
|
data["invoices"] *= 2
|
|
request.return_value = data
|
|
with self.assertRaises(ValueError):
|
|
shkeeper.refresh_invoice(invoice)
|
|
invoice.refresh_from_db()
|
|
self.assertIsNone(invoice.credited_at)
|
|
|
|
@patch("members.shkeeper.request")
|
|
def test_renewal_stacks_calendar_months_and_overpayment_once(self, request):
|
|
first = self.prepared()
|
|
request.return_value = self.remote(first)
|
|
shkeeper.refresh_invoice(first)
|
|
second = self.prepared()
|
|
request.return_value = self.remote(second, status="OVERPAID", balance_fiat="50")
|
|
shkeeper.refresh_invoice(second)
|
|
self.assertEqual(second.access_start, first.access_end)
|
|
self.assertEqual(second.access_end, shkeeper.add_month(first.access_end))
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.shkeeper.request")
|
|
def test_paid_crypto_grant_expiry_and_outage(self, request, ldap):
|
|
invoice = self.prepared()
|
|
request.return_value = self.remote(invoice)
|
|
self.assertTrue(billing.sync_account(self.member.pk))
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
self.member.refresh_from_db()
|
|
self.assertEqual(self.member.membership_level, 1)
|
|
request.side_effect = OSError("SHKeeper offline")
|
|
self.assertFalse(billing.sync_account(self.member.pk))
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
CryptoInvoice.objects.update(access_end=timezone.now() - timedelta(seconds=1))
|
|
billing.sync_account(self.member.pk, refresh_stripe=False)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
self.member.refresh_from_db()
|
|
self.assertEqual(self.member.membership_level, 0)
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.shkeeper.request")
|
|
def test_reorg_revokes_and_recovery_does_not_restart_month(self, request, ldap):
|
|
invoice = self.prepared()
|
|
request.return_value = self.remote(invoice)
|
|
billing.sync_account(self.member.pk)
|
|
invoice.refresh_from_db()
|
|
original_end = invoice.access_end
|
|
request.return_value = self.remote(invoice, status="UNPAID", balance_fiat="0", txs=[])
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
request.return_value = self.remote(invoice)
|
|
billing.sync_account(self.member.pk)
|
|
invoice.refresh_from_db()
|
|
self.assertEqual(invoice.access_end, original_end)
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.refresh_billing", side_effect=OSError("Stripe offline"))
|
|
@patch("members.shkeeper.request")
|
|
def test_stripe_failure_does_not_prevent_crypto_and_union(self, request, stripe, ldap):
|
|
invoice = self.prepared()
|
|
request.return_value = self.remote(invoice)
|
|
Subscription.objects.create(account=self.member, stripe_id="sub_other", tier="patron", status="active", paid_until=timezone.now() + timedelta(days=1))
|
|
self.assertFalse(billing.sync_account(self.member.pk))
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP, OTHER_GROUP})
|
|
with override_settings(MEMBERSHIP_TIERS={}):
|
|
billing.sync_account(self.member.pk, refresh_stripe=False)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
self.assertIn(GROUP, ldap.call_args.args[2])
|
|
|
|
@patch("members.shkeeper.request")
|
|
@patch("members.views.billing.client")
|
|
def test_active_stripe_and_open_checkout_block_crypto(self, stripe, request):
|
|
self.member.stripe_customer = "cus_alice"
|
|
self.member.save()
|
|
stripe.return_value.subscriptions.list.return_value = paged([{"status": "active"}])
|
|
self.assertContains(self.client.post("/members/crypto/checkout/supporter/", {"crypto": "BTC"}), "Recurring membership already exists")
|
|
stripe.return_value.subscriptions.list.return_value = paged([])
|
|
Account.objects.filter(pk=self.member.pk).update(checkout_id="cs_open")
|
|
stripe.return_value.checkout.sessions.retrieve.return_value = {"status": "open"}
|
|
self.assertContains(self.client.post("/members/crypto/checkout/supporter/", {"crypto": "BTC"}), "Stripe checkout is still open")
|
|
request.assert_not_called()
|
|
self.assertFalse(CryptoInvoice.objects.exists())
|
|
|
|
@patch("members.views.billing.client")
|
|
def test_unfinished_or_active_crypto_blocks_stripe(self, stripe):
|
|
invoice = self.prepared()
|
|
self.assertContains(self.client.post("/members/checkout/supporter/"), "Prepaid membership already exists")
|
|
stripe.assert_not_called()
|
|
CryptoInvoice.objects.filter(pk=invoice.pk).update(credited_at=timezone.now(), access_end=timezone.now() + timedelta(days=1))
|
|
self.assertContains(self.client.post("/members/crypto/checkout/patron/", {"crypto": "BTC"}), "Another prepaid tier is active")
|
|
|
|
|
|
@override_settings(**SHKEEPER_SETTINGS)
|
|
class SHKeeperTransportTests(SimpleTestCase):
|
|
@patch("members.shkeeper.requests.request")
|
|
def test_api_auth_tls_timeout_no_redirects_and_limits(self, request):
|
|
response = request.return_value.__enter__.return_value
|
|
response.status_code = 200
|
|
response.iter_content.return_value = [b'{"status":"success","invoices":[]}']
|
|
self.assertEqual(shkeeper.request("GET", "/api/v1/invoices/order")["invoices"], [])
|
|
self.assertEqual(request.call_args.args[:2], ("GET", "https://pay.example.org/api/v1/invoices/order"))
|
|
self.assertEqual(request.call_args.kwargs["headers"]["X-Shkeeper-Api-Key"], "test-only-key")
|
|
self.assertIs(request.call_args.kwargs["verify"], True)
|
|
self.assertFalse(request.call_args.kwargs["allow_redirects"])
|
|
self.assertEqual(request.call_args.kwargs["timeout"], (5, 15))
|
|
response.status_code = 302
|
|
with self.assertRaises(ValueError):
|
|
shkeeper.request("GET", "/api/v1/invoices/order")
|
|
response.status_code = 200
|
|
response.iter_content.return_value = [b"x" * 1048577]
|
|
with self.assertRaises(ValueError):
|
|
shkeeper.request("GET", "/api/v1/invoices/order")
|
|
|
|
def test_calendar_month_handles_leap_year_and_december(self):
|
|
january = datetime(2028, 1, 31, tzinfo=dt_timezone.utc)
|
|
self.assertEqual(shkeeper.add_month(january), datetime(2028, 2, 29, tzinfo=dt_timezone.utc))
|
|
self.assertEqual(shkeeper.add_month(datetime(2027, 12, 15, tzinfo=dt_timezone.utc)), datetime(2028, 1, 15, tzinfo=dt_timezone.utc))
|