1915 lines
67 KiB
Python
1915 lines
67 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from base64 import b64encode
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from pytest_mock.plugin import MockerFixture
|
|
|
|
from lnbits.core.crud.payments import get_payment, get_payments
|
|
from lnbits.core.crud.users import get_user
|
|
from lnbits.core.crud.wallets import create_wallet
|
|
from lnbits.core.models.payments import CreateInvoice, Payment, PaymentState
|
|
from lnbits.core.models.users import User
|
|
from lnbits.core.models.wallets import Wallet
|
|
from lnbits.core.services import check_payment_status, payments
|
|
from lnbits.core.services.fiat_providers import (
|
|
check_fiat_status,
|
|
check_revolut_signature,
|
|
check_square_signature,
|
|
check_stripe_signature,
|
|
handle_fiat_payment_confirmation,
|
|
verify_paypal_webhook,
|
|
)
|
|
from lnbits.core.services.fiat_providers import (
|
|
test_connection as fiat_provider_connection,
|
|
)
|
|
from lnbits.core.services.users import create_user_account
|
|
from lnbits.fiat.base import (
|
|
FiatInvoiceResponse,
|
|
FiatPaymentStatus,
|
|
FiatStatusResponse,
|
|
FiatSubscriptionPaymentOptions,
|
|
)
|
|
from lnbits.fiat.revolut import REVOLUT_WEBHOOK_EVENTS, RevolutWallet
|
|
from lnbits.fiat.square import SquareWallet
|
|
from lnbits.settings import Settings
|
|
from tests.helpers import get_random_string
|
|
|
|
|
|
class MockHTTPResponse:
|
|
def __init__(self, json_data=None, error: Exception | None = None):
|
|
self._json_data = json_data or {}
|
|
self._error = error
|
|
|
|
def raise_for_status(self):
|
|
if self._error:
|
|
raise self._error
|
|
|
|
def json(self):
|
|
return self._json_data
|
|
|
|
|
|
class MockHTTPClient:
|
|
def __init__(self, responses: list[MockHTTPResponse]):
|
|
self._responses = responses
|
|
self.calls: list[tuple[str, dict]] = []
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def post(self, path: str, **kwargs):
|
|
self.calls.append((path, kwargs))
|
|
return self._responses.pop(0)
|
|
|
|
async def get(self, path: str, **kwargs):
|
|
self.calls.append((path, kwargs))
|
|
return self._responses.pop(0)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fiat_provider_test_settings(settings: Settings):
|
|
original_lnbits_running = settings.lnbits_running
|
|
original_allowed_currencies = settings.lnbits_allowed_currencies
|
|
original_paypal_enabled = settings.paypal_enabled
|
|
original_square_enabled = settings.square_enabled
|
|
original_square_api_endpoint = settings.square_api_endpoint
|
|
original_square_access_token = settings.square_access_token
|
|
original_square_location_id = settings.square_location_id
|
|
original_square_api_version = settings.square_api_version
|
|
original_square_payment_success_url = settings.square_payment_success_url
|
|
original_square_payment_webhook_url = settings.square_payment_webhook_url
|
|
original_square_webhook_signature_key = settings.square_webhook_signature_key
|
|
original_square_limits = settings.square_limits.copy(deep=True)
|
|
original_revolut_enabled = settings.revolut_enabled
|
|
original_revolut_api_endpoint = settings.revolut_api_endpoint
|
|
original_revolut_api_secret_key = settings.revolut_api_secret_key
|
|
original_revolut_api_version = settings.revolut_api_version
|
|
original_revolut_payment_success_url = settings.revolut_payment_success_url
|
|
original_revolut_payment_webhook_url = settings.revolut_payment_webhook_url
|
|
original_revolut_webhook_signing_secret = settings.revolut_webhook_signing_secret
|
|
original_revolut_limits = settings.revolut_limits.copy(deep=True)
|
|
settings.lnbits_allowed_currencies = []
|
|
settings.paypal_enabled = False
|
|
settings.square_enabled = False
|
|
settings.revolut_enabled = False
|
|
yield
|
|
settings.lnbits_running = original_lnbits_running
|
|
settings.lnbits_allowed_currencies = original_allowed_currencies
|
|
settings.paypal_enabled = original_paypal_enabled
|
|
settings.square_enabled = original_square_enabled
|
|
settings.square_api_endpoint = original_square_api_endpoint
|
|
settings.square_access_token = original_square_access_token
|
|
settings.square_location_id = original_square_location_id
|
|
settings.square_api_version = original_square_api_version
|
|
settings.square_payment_success_url = original_square_payment_success_url
|
|
settings.square_payment_webhook_url = original_square_payment_webhook_url
|
|
settings.square_webhook_signature_key = original_square_webhook_signature_key
|
|
settings.square_limits = original_square_limits
|
|
settings.revolut_enabled = original_revolut_enabled
|
|
settings.revolut_api_endpoint = original_revolut_api_endpoint
|
|
settings.revolut_api_secret_key = original_revolut_api_secret_key
|
|
settings.revolut_api_version = original_revolut_api_version
|
|
settings.revolut_payment_success_url = original_revolut_payment_success_url
|
|
settings.revolut_payment_webhook_url = original_revolut_payment_webhook_url
|
|
settings.revolut_webhook_signing_secret = original_revolut_webhook_signing_secret
|
|
settings.revolut_limits = original_revolut_limits
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_missing_provider():
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider=None
|
|
)
|
|
with pytest.raises(ValueError, match="Fiat provider is required"):
|
|
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_provider_not_enabled(settings: Settings):
|
|
settings.stripe_enabled = False
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="notarealprovider"
|
|
)
|
|
with pytest.raises(
|
|
ValueError, match="Fiat provider 'notarealprovider' is not enabled"
|
|
):
|
|
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_with_sat_unit(settings: Settings):
|
|
settings.stripe_enabled = True
|
|
invoice_data = CreateInvoice(
|
|
unit="sat", amount=1.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
with pytest.raises(ValueError, match="Fiat provider cannot be used with satoshis"):
|
|
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_allowed_users(
|
|
to_user: User, settings: Settings
|
|
):
|
|
|
|
settings.stripe_enabled = False
|
|
settings.stripe_limits.allowed_users = []
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == []
|
|
|
|
settings.stripe_enabled = True
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["stripe"]
|
|
|
|
settings.stripe_limits.allowed_users = ["some_other_user_id"]
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == []
|
|
|
|
settings.stripe_limits.allowed_users.append(to_user.id)
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["stripe"]
|
|
|
|
settings.stripe_enabled = False
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == []
|
|
|
|
settings.square_enabled = True
|
|
settings.square_limits.allowed_users = []
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["square"]
|
|
|
|
settings.square_limits.allowed_users = ["some_other_user_id"]
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == []
|
|
|
|
settings.square_limits.allowed_users.append(to_user.id)
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["square"]
|
|
|
|
settings.square_enabled = False
|
|
settings.revolut_enabled = True
|
|
settings.revolut_limits.allowed_users = []
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["revolut"]
|
|
|
|
settings.revolut_limits.allowed_users = ["some_other_user_id"]
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == []
|
|
|
|
settings.revolut_limits.allowed_users.append(to_user.id)
|
|
user = await get_user(to_user.id)
|
|
assert user
|
|
assert user.fiat_providers == ["revolut"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_fiat_limits_fail(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
|
|
settings.stripe_enabled = True
|
|
settings.stripe_limits.service_min_amount_sats = 0
|
|
settings.stripe_limits.service_max_amount_sats = 105
|
|
settings.stripe_limits.service_faucet_wallet_id = None
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
|
)
|
|
with pytest.raises(ValueError, match="Maximum amount is 105 sats for 'stripe'."):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
settings.stripe_limits.service_min_amount_sats = 1001
|
|
settings.stripe_limits.service_max_amount_sats = 10000
|
|
|
|
with pytest.raises(ValueError, match="Minimum amount is 1001 sats for 'stripe'."):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
settings.stripe_limits.service_min_amount_sats = 10
|
|
settings.stripe_limits.service_max_amount_sats = 10000
|
|
settings.stripe_limits.service_max_fee_sats = 100
|
|
|
|
with pytest.raises(
|
|
ValueError, match="Fiat provider 'stripe' service fee wallet missing."
|
|
):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
settings.stripe_limits.service_fee_wallet_id = "not_a_real_wallet_id"
|
|
|
|
with pytest.raises(
|
|
ValueError, match="Fiat provider 'stripe' service fee wallet not found."
|
|
):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
settings.stripe_limits.service_fee_wallet_id = to_wallet.id
|
|
settings.stripe_limits.service_faucet_wallet_id = "not_a_real_wallet_id"
|
|
|
|
with pytest.raises(
|
|
ValueError, match="Fiat provider 'stripe' faucet wallet not found."
|
|
):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
user = await create_user_account()
|
|
wallet = await create_wallet(user_id=user.id)
|
|
settings.stripe_limits.service_faucet_wallet_id = wallet.id
|
|
|
|
with pytest.raises(
|
|
ValueError, match="The amount exceeds the 'stripe'faucet wallet balance."
|
|
):
|
|
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_provider_fails(
|
|
settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.stripe_enabled = True
|
|
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=2.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=False,
|
|
error_message="Failed to create invoice",
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
|
)
|
|
|
|
user = await create_user_account()
|
|
wallet = await create_wallet(user_id=user.id)
|
|
with pytest.raises(ValueError, match="Cannot create payment request for 'stripe'."):
|
|
await payments.create_fiat_invoice(wallet.id, invoice_data)
|
|
|
|
wallet_payments = await get_payments(wallet_id=wallet.id)
|
|
assert len(wallet_payments) == 1
|
|
assert wallet_payments[0].status == PaymentState.FAILED
|
|
assert wallet_payments[0].amount == 2000000
|
|
assert wallet_payments[0].fee == 0
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_fiat_invoice_success(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.stripe_enabled = True
|
|
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
|
settings.stripe_limits.service_min_amount_sats = 0
|
|
settings.stripe_limits.service_max_amount_sats = 0
|
|
settings.stripe_limits.service_faucet_wallet_id = None
|
|
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=True,
|
|
checking_id=f"session_123_{get_random_string(10)}",
|
|
payment_request="https://stripe.com/pay/session_123",
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
|
)
|
|
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
assert payment.status == PaymentState.PENDING
|
|
assert payment.amount == 1000_000
|
|
assert payment.fiat_provider == "stripe"
|
|
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
|
assert (
|
|
payment.extra.get("fiat_payment_request")
|
|
== "https://stripe.com/pay/session_123"
|
|
)
|
|
assert payment.checking_id.startswith("fiat_stripe_")
|
|
assert payment.fee <= 0
|
|
|
|
status = await check_payment_status(payment)
|
|
assert status.success is False
|
|
assert status.pending is True
|
|
|
|
fiat_mock_status = FiatPaymentStatus(paid=True, fee=123)
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.get_invoice_status",
|
|
AsyncMock(return_value=fiat_mock_status),
|
|
)
|
|
status = await check_payment_status(payment)
|
|
assert status.paid is True
|
|
assert status.success is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_square_fiat_invoice_success(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.square_enabled = True
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_limits.service_min_amount_sats = 0
|
|
settings.square_limits.service_max_amount_sats = 0
|
|
settings.square_limits.service_faucet_wallet_id = None
|
|
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="square"
|
|
)
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=True,
|
|
checking_id="order_123",
|
|
payment_request="https://square.link/u/session_123",
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.SquareWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000),
|
|
)
|
|
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
assert payment.status == PaymentState.PENDING
|
|
assert payment.fiat_provider == "square"
|
|
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
|
assert payment.checking_id.startswith("fiat_square_order_123")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_create_invoice(settings: Settings):
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
settings.square_payment_success_url = "https://lnbits.example/success"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"payment_link": {
|
|
"order_id": "ORDER123",
|
|
"url": "https://square.link/u/abc123",
|
|
}
|
|
}
|
|
)
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.create_invoice(
|
|
amount=1.23,
|
|
payment_hash="hash123",
|
|
currency="USD",
|
|
memo="LNbits Square invoice",
|
|
extra={"checkout": {"metadata": {"source": "test"}}},
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert response.checking_id == "order_ORDER123"
|
|
assert response.payment_request == "https://square.link/u/abc123"
|
|
assert client.calls[0][0] == "/v2/online-checkout/payment-links"
|
|
payload = client.calls[0][1]["json"]
|
|
assert payload["idempotency_key"] == "hash123"
|
|
assert payload["order"]["location_id"] == "LOC123"
|
|
assert payload["order"]["metadata"]["payment_hash"] == "hash123"
|
|
assert payload["order"]["metadata"]["alan_action"] == "invoice"
|
|
assert payload["order"]["metadata"]["source"] == "test"
|
|
assert payload["order"]["line_items"][0]["base_price_money"]["amount"] == 123
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_create_subscription(settings: Settings):
|
|
settings.lnbits_running = False
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
settings.square_payment_success_url = "https://lnbits.example/success"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"object": {
|
|
"type": "SUBSCRIPTION_PLAN_VARIATION",
|
|
"id": "PLAN_VARIATION_123",
|
|
"subscription_plan_variation_data": {
|
|
"phases": [
|
|
{
|
|
"ordinal": 0,
|
|
"pricing": {
|
|
"type": "STATIC",
|
|
"price_money": {
|
|
"amount": 1500,
|
|
"currency": "USD",
|
|
},
|
|
},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"payment_link": {
|
|
"id": "plink_123",
|
|
"url": "https://square.link/u/sub_123",
|
|
}
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
memo="Monthly Gold",
|
|
tag="gold",
|
|
extra={"link": "link-1"},
|
|
success_url="https://lnbits.example/subscription-success",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert response.checkout_session_url == "https://square.link/u/sub_123"
|
|
assert response.subscription_request_id is not None
|
|
assert client.calls[0][0] == "/v2/catalog/object/PLAN_VARIATION_123"
|
|
assert client.calls[1][0] == "/v2/online-checkout/payment-links"
|
|
payload = client.calls[1][1]["json"]
|
|
assert payload["idempotency_key"] == response.subscription_request_id
|
|
assert payload["quick_pay"]["location_id"] == "LOC123"
|
|
assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"}
|
|
assert payload["checkout_options"] == {
|
|
"redirect_url": "https://lnbits.example/subscription-success",
|
|
"subscription_plan_id": "PLAN_VARIATION_123",
|
|
}
|
|
metadata = json.loads(payload["payment_note"])
|
|
assert metadata[:3] == ["wallet_1", "gold", response.subscription_request_id]
|
|
assert metadata[3:] == ["link-1", "Monthly Gold"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_create_subscription_from_plan_id(settings: Settings):
|
|
settings.lnbits_running = False
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
settings.square_payment_success_url = "https://lnbits.example/success"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"object": {
|
|
"type": "SUBSCRIPTION_PLAN",
|
|
"id": "PLAN123",
|
|
"subscription_plan_data": {
|
|
"name": "LNbits Test Weekly Personal Plan",
|
|
"subscription_plan_variations": [
|
|
{
|
|
"type": "SUBSCRIPTION_PLAN_VARIATION",
|
|
"id": "PLAN_VARIATION_123",
|
|
"subscription_plan_variation_data": {
|
|
"name": "LNbits Test Weekly Personal Plan",
|
|
"phases": [
|
|
{
|
|
"uid": "PHASE123",
|
|
"cadence": "WEEKLY",
|
|
"ordinal": 0,
|
|
"pricing": {"type": "RELATIVE"},
|
|
}
|
|
],
|
|
"subscription_plan_id": "PLAN123",
|
|
},
|
|
}
|
|
],
|
|
"eligible_item_ids": ["ITEM123"],
|
|
"all_items": False,
|
|
},
|
|
}
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"object": {
|
|
"type": "ITEM",
|
|
"id": "ITEM123",
|
|
"item_data": {
|
|
"name": "LNbits Test Weekly Personal Plan",
|
|
"variations": [
|
|
{
|
|
"type": "ITEM_VARIATION",
|
|
"id": "ITEM_VARIATION_123",
|
|
"item_variation_data": {
|
|
"item_id": "ITEM123",
|
|
"name": "Regular",
|
|
"pricing_type": "FIXED_PRICING",
|
|
"price_money": {
|
|
"amount": 1500,
|
|
"currency": "USD",
|
|
},
|
|
},
|
|
}
|
|
],
|
|
},
|
|
}
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"payment_link": {
|
|
"id": "plink_123",
|
|
"url": "https://square.link/u/sub_123",
|
|
}
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN123",
|
|
1,
|
|
FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
memo="Weekly Plan",
|
|
success_url="https://lnbits.example/success",
|
|
),
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/v2/catalog/object/PLAN123"
|
|
assert client.calls[1][0] == "/v2/catalog/object/ITEM123"
|
|
assert client.calls[2][0] == "/v2/online-checkout/payment-links"
|
|
payload = client.calls[2][1]["json"]
|
|
assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"}
|
|
assert payload["checkout_options"] == {
|
|
"redirect_url": "https://lnbits.example/success",
|
|
"subscription_plan_id": "PLAN_VARIATION_123",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_create_subscription_invoice(settings: Settings):
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
|
|
wallet = SquareWallet()
|
|
|
|
response = await wallet.create_invoice(
|
|
amount=15,
|
|
payment_hash="hash123",
|
|
currency="USD",
|
|
memo="Square subscription payment",
|
|
extra={
|
|
"fiat_method": "subscription",
|
|
"subscription": {
|
|
"checking_id": "payment_PAYMENT123",
|
|
"payment_request": "https://square.example/invoice",
|
|
},
|
|
},
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert response.checking_id == "payment_PAYMENT123"
|
|
assert response.payment_request == "https://square.example/invoice"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_cancel_subscription(settings: Settings):
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient([MockHTTPResponse(json_data={"subscription": {}})])
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.cancel_subscription("SUBSCRIPTION123", "wallet_1")
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/v2/subscriptions/SUBSCRIPTION123/cancel"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_cancel_subscription_by_request_id(
|
|
settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient([MockHTTPResponse(json_data={"subscription": {}})])
|
|
wallet.client = client # type: ignore[assignment]
|
|
payment = Payment(
|
|
checking_id="fiat_square_payment_PAYMENT123",
|
|
payment_hash="hash123",
|
|
wallet_id="wallet_1",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="lnbc1square",
|
|
fiat_provider="square",
|
|
extra={"subscription_request_id": "REQUEST123"},
|
|
external_id="SUBSCRIPTION123",
|
|
)
|
|
get_payments_mock = mocker.patch(
|
|
"lnbits.core.crud.payments.get_payments",
|
|
AsyncMock(side_effect=[[], [payment]]),
|
|
)
|
|
|
|
response = await wallet.cancel_subscription("REQUEST123", "wallet_1")
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/v2/subscriptions/SUBSCRIPTION123/cancel"
|
|
assert get_payments_mock.await_count == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_square_wallet_get_invoice_status(settings: Settings):
|
|
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
|
|
settings.square_access_token = "square-token"
|
|
settings.square_location_id = "LOC123"
|
|
settings.square_api_version = "2026-01-22"
|
|
|
|
wallet = SquareWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"order": {
|
|
"id": "ORDER123",
|
|
"state": "COMPLETED",
|
|
"tenders": [{"payment_id": "PAYMENT123"}],
|
|
}
|
|
}
|
|
),
|
|
MockHTTPResponse(json_data={"payment": {"status": "COMPLETED"}}),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
status = await wallet.get_invoice_status("fiat_square_order_ORDER123")
|
|
|
|
assert status.success is True
|
|
assert client.calls[0][0] == "/v2/orders/ORDER123"
|
|
assert client.calls[1][0] == "/v2/payments/PAYMENT123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_wallet_revolut_fiat_invoice_success(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.revolut_enabled = True
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_limits.service_min_amount_sats = 0
|
|
settings.revolut_limits.service_max_amount_sats = 0
|
|
settings.revolut_limits.service_faucet_wallet_id = None
|
|
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="revolut"
|
|
)
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=True,
|
|
checking_id="order_ORDER123",
|
|
payment_request="https://checkout.revolut.com/payment-link/ORDER123",
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.RevolutWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000),
|
|
)
|
|
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
assert payment.status == PaymentState.PENDING
|
|
assert payment.fiat_provider == "revolut"
|
|
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
|
assert payment.checking_id.startswith("fiat_revolut_order_ORDER123")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_invoice(settings: Settings):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
settings.revolut_payment_success_url = "https://lnbits.example/success"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/abc123",
|
|
}
|
|
)
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.create_invoice(
|
|
amount=1.23,
|
|
payment_hash="hash123",
|
|
currency="USD",
|
|
memo="LNbits Revolut invoice",
|
|
extra={"checkout": {"metadata": {"source": "test"}}},
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert response.checking_id == "order_ORDER123"
|
|
assert (
|
|
response.payment_request == "https://checkout.revolut.com/payment-link/abc123"
|
|
)
|
|
assert client.calls[0][0] == "/api/orders"
|
|
payload = client.calls[0][1]["json"]
|
|
assert payload["amount"] == 123
|
|
assert payload["currency"] == "USD"
|
|
assert payload["metadata"]["payment_hash"] == "hash123"
|
|
assert payload["metadata"]["alan_action"] == "invoice"
|
|
assert payload["metadata"]["source"] == "test"
|
|
assert payload["redirect_url"] == "https://lnbits.example/success"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_invoice_uses_currency_minor_units(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER_JPY",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/jpy",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER_KWD",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/kwd",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
await wallet.create_invoice(
|
|
amount=123,
|
|
payment_hash="hash_jpy",
|
|
currency="JPY",
|
|
)
|
|
await wallet.create_invoice(
|
|
amount=1.234,
|
|
payment_hash="hash_kwd",
|
|
currency="KWD",
|
|
)
|
|
|
|
assert client.calls[0][1]["json"]["amount"] == 123
|
|
assert client.calls[1][1]["json"]["amount"] == 1234
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_get_invoice_status(settings: Settings):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient([MockHTTPResponse(json_data={"state": "COMPLETED"})])
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
status = await wallet.get_invoice_status("fiat_revolut_order_ORDER123")
|
|
|
|
assert status.success is True
|
|
assert client.calls[0][0] == "/api/orders/ORDER123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription(settings: Settings):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
settings.revolut_payment_success_url = "https://lnbits.example/subscription-success"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"customers": [
|
|
{
|
|
"id": "CUSTOMER123",
|
|
"email": "customer@example.com",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "SUBSCRIPTION123",
|
|
"setup_order_id": "ORDER123",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
memo="Monthly Gold",
|
|
tag="gold",
|
|
customer_email="customer@example.com",
|
|
extra={"link": "link-1"},
|
|
success_url="https://lnbits.example/subscription-success",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
subscription_request_id = payment_options.subscription_request_id
|
|
assert response.ok is True
|
|
assert response.subscription_request_id == "SUBSCRIPTION123"
|
|
assert subscription_request_id is not None
|
|
assert (
|
|
response.checkout_session_url
|
|
== "https://checkout.revolut.com/payment-link/sub_123"
|
|
)
|
|
assert client.calls[0][0] == "/api/customers"
|
|
assert client.calls[0][1]["params"] == {"limit": 500}
|
|
assert client.calls[0][1]["timeout"] == 30
|
|
assert client.calls[1][0] == "/api/subscriptions"
|
|
payload = client.calls[1][1]["json"]
|
|
assert payload["plan_variation_id"] == "PLAN_VARIATION_123"
|
|
assert payload["customer_id"] == "CUSTOMER123"
|
|
assert client.calls[1][1]["timeout"] == 30
|
|
assert client.calls[1][1]["headers"]["Idempotency-Key"] == (subscription_request_id)
|
|
assert payload["setup_order_redirect_url"] == (
|
|
"https://lnbits.example/subscription-success"
|
|
)
|
|
reference = json.loads(payload["external_reference"])
|
|
assert reference["wallet_id"] == "wallet_1"
|
|
assert reference["tag"] == "gold"
|
|
assert reference["subscription_request_id"] == subscription_request_id
|
|
assert reference["memo"] == "Monthly Gold"
|
|
assert reference["extra"]["link"] == "link-1"
|
|
assert client.calls[2][0] == "/api/orders/ORDER123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription_uses_customer_email(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"customers": [
|
|
{
|
|
"id": "CUSTOMER123",
|
|
"email": "customer@example.com",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "SUBSCRIPTION123",
|
|
"setup_order_id": "ORDER123",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
customer_email="customer@example.com",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/api/customers"
|
|
assert client.calls[0][1]["params"] == {"limit": 500}
|
|
assert client.calls[0][1]["timeout"] == 30
|
|
assert client.calls[1][0] == "/api/subscriptions"
|
|
assert client.calls[1][1]["json"]["customer_id"] == "CUSTOMER123"
|
|
assert client.calls[1][1]["timeout"] == 30
|
|
assert client.calls[2][0] == "/api/orders/ORDER123"
|
|
assert client.calls[2][1]["timeout"] == 30
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription_uses_paginated_customer_email(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"next_page_token": "PAGE2",
|
|
"customers": [
|
|
{
|
|
"id": "OTHER_CUSTOMER",
|
|
"email": "other@example.com",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"customers": [
|
|
{
|
|
"id": "CUSTOMER123",
|
|
"email": "customer@example.com",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "SUBSCRIPTION123",
|
|
"setup_order_id": "ORDER123",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
customer_email="customer@example.com",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/api/customers"
|
|
assert client.calls[0][1]["params"] == {"limit": 500}
|
|
assert client.calls[1][0] == "/api/customers"
|
|
assert client.calls[1][1]["params"] == {
|
|
"limit": 500,
|
|
"page_token": "PAGE2",
|
|
}
|
|
assert client.calls[1][1]["timeout"] == 30
|
|
assert client.calls[2][0] == "/api/subscriptions"
|
|
assert client.calls[2][1]["json"]["customer_id"] == "CUSTOMER123"
|
|
assert client.calls[3][0] == "/api/orders/ORDER123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription_creates_customer(settings: Settings):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"next_page_token": "PAGE2",
|
|
"customers": [
|
|
{
|
|
"id": "OTHER_CUSTOMER",
|
|
"email": "other@example.com",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
MockHTTPResponse(json_data={"customers": []}),
|
|
MockHTTPResponse(json_data={"id": "CUSTOMER123"}),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "SUBSCRIPTION123",
|
|
"setup_order_id": "ORDER123",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
customer_email="customer@example.com",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/api/customers"
|
|
assert client.calls[0][1]["params"] == {"limit": 500}
|
|
assert client.calls[1][0] == "/api/customers"
|
|
assert client.calls[1][1]["params"] == {
|
|
"limit": 500,
|
|
"page_token": "PAGE2",
|
|
}
|
|
assert client.calls[2][0] == "/api/customers"
|
|
assert client.calls[2][1]["json"] == {"email": "customer@example.com"}
|
|
assert client.calls[2][1]["timeout"] == 30
|
|
assert client.calls[3][0] == "/api/subscriptions"
|
|
assert client.calls[3][1]["json"]["customer_id"] == "CUSTOMER123"
|
|
assert client.calls[4][0] == "/api/orders/ORDER123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription_stops_customer_lookup_after_20_pages(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
customer_pages = [
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"next_page_token": f"PAGE{page + 2}",
|
|
"customers": [
|
|
{
|
|
"id": f"OTHER_CUSTOMER_{page}",
|
|
"email": f"other-{page}@example.com",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
for page in range(20)
|
|
]
|
|
client = MockHTTPClient(
|
|
[
|
|
*customer_pages,
|
|
MockHTTPResponse(json_data={"id": "CUSTOMER123"}),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "SUBSCRIPTION123",
|
|
"setup_order_id": "ORDER123",
|
|
}
|
|
),
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"id": "ORDER123",
|
|
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(
|
|
wallet_id="wallet_1",
|
|
customer_email="customer@example.com",
|
|
)
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is True
|
|
assert [call[0] for call in client.calls[:20]] == ["/api/customers"] * 20
|
|
assert client.calls[0][1]["params"] == {"limit": 500}
|
|
assert client.calls[19][1]["params"] == {
|
|
"limit": 500,
|
|
"page_token": "PAGE20",
|
|
}
|
|
assert client.calls[20][0] == "/api/customers"
|
|
assert client.calls[20][1]["json"] == {"email": "customer@example.com"}
|
|
assert client.calls[21][0] == "/api/subscriptions"
|
|
assert client.calls[21][1]["json"]["customer_id"] == "CUSTOMER123"
|
|
assert client.calls[22][0] == "/api/orders/ORDER123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_subscription_requires_customer_email(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient([])
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
payment_options = FiatSubscriptionPaymentOptions(wallet_id="wallet_1")
|
|
|
|
response = await wallet.create_subscription(
|
|
"PLAN_VARIATION_123", 1, payment_options
|
|
)
|
|
|
|
assert response.ok is False
|
|
assert response.error_message == "Revolut subscriptions require customer_email."
|
|
assert client.calls == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_cancel_subscription(settings: Settings):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"external_reference": json.dumps({"wallet_id": "wallet_1"}),
|
|
}
|
|
),
|
|
MockHTTPResponse(json_data={}),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.cancel_subscription("SUBSCRIPTION123", "wallet_1")
|
|
|
|
assert response.ok is True
|
|
assert client.calls[0][0] == "/api/subscriptions/SUBSCRIPTION123"
|
|
assert client.calls[1][0] == "/api/subscriptions/SUBSCRIPTION123/cancel"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_cancel_subscription_checks_wallet_id(
|
|
settings: Settings,
|
|
):
|
|
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
|
|
settings.revolut_api_secret_key = "revolut-secret"
|
|
settings.revolut_api_version = "2026-04-20"
|
|
|
|
wallet = RevolutWallet()
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
json_data={
|
|
"external_reference": json.dumps({"wallet_id": "wallet_2"}),
|
|
}
|
|
),
|
|
]
|
|
)
|
|
wallet.client = client # type: ignore[assignment]
|
|
|
|
response = await wallet.cancel_subscription("SUBSCRIPTION123", "wallet_1")
|
|
|
|
assert response.ok is False
|
|
assert response.error_message == "Subscription not found."
|
|
assert client.calls == [
|
|
(
|
|
"/api/subscriptions/SUBSCRIPTION123",
|
|
{"timeout": 30},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_create_webhook(mocker: MockerFixture):
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse({"webhooks": []}),
|
|
MockHTTPResponse(
|
|
{
|
|
"id": "webhook_1",
|
|
"url": "https://lnbits.example/api/v1/callback/revolut",
|
|
"events": REVOLUT_WEBHOOK_EVENTS,
|
|
"signing_secret": "whsec_1",
|
|
}
|
|
),
|
|
]
|
|
)
|
|
async_client = mocker.patch("lnbits.fiat.revolut.httpx.AsyncClient")
|
|
async_client.return_value = client
|
|
|
|
response = await RevolutWallet.create_webhook(
|
|
url="https://lnbits.example/api/v1/callback/revolut",
|
|
endpoint="https://sandbox-merchant.revolut.com",
|
|
api_secret_key="revolut-secret",
|
|
api_version="2026-04-20",
|
|
)
|
|
|
|
assert response["signing_secret"] == "whsec_1"
|
|
async_client.assert_called_once()
|
|
assert async_client.call_args.kwargs["base_url"] == (
|
|
"https://sandbox-merchant.revolut.com"
|
|
)
|
|
assert async_client.call_args.kwargs["headers"]["Authorization"] == (
|
|
"Bearer revolut-secret"
|
|
)
|
|
assert client.calls == [
|
|
(
|
|
"/api/webhooks",
|
|
{
|
|
"timeout": 30,
|
|
},
|
|
),
|
|
(
|
|
"/api/webhooks",
|
|
{
|
|
"json": {
|
|
"url": "https://lnbits.example/api/v1/callback/revolut",
|
|
"events": REVOLUT_WEBHOOK_EVENTS,
|
|
},
|
|
"timeout": 30,
|
|
},
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_reuses_existing_webhook(mocker: MockerFixture):
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(
|
|
{
|
|
"webhooks": [
|
|
{
|
|
"id": "webhook_1",
|
|
"url": "https://lnbits.example/api/v1/callback/revolut",
|
|
"events": REVOLUT_WEBHOOK_EVENTS,
|
|
"signing_secret": "whsec_1",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
]
|
|
)
|
|
async_client = mocker.patch("lnbits.fiat.revolut.httpx.AsyncClient")
|
|
async_client.return_value = client
|
|
|
|
response = await RevolutWallet.create_webhook(
|
|
url="https://lnbits.example/api/v1/callback/revolut",
|
|
endpoint="https://sandbox-merchant.revolut.com",
|
|
api_secret_key="revolut-secret",
|
|
api_version="2026-04-20",
|
|
)
|
|
|
|
assert response["already_exists"] is True
|
|
assert response["signing_secret"] == "whsec_1"
|
|
assert client.calls == [
|
|
(
|
|
"/api/webhooks",
|
|
{
|
|
"timeout": 30,
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revolut_wallet_rejects_local_webhook_url():
|
|
with pytest.raises(ValueError, match="clearnet URL"):
|
|
await RevolutWallet.create_webhook(
|
|
url="http://localhost:5000/api/v1/callback/revolut",
|
|
endpoint="https://sandbox-merchant.revolut.com",
|
|
api_secret_key="revolut-secret",
|
|
api_version="2026-04-20",
|
|
)
|
|
|
|
|
|
def test_check_revolut_signature():
|
|
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
|
|
timestamp = str(int(time.time() * 1000))
|
|
secret = "revolut-secret"
|
|
signed_payload = b"v1." + timestamp.encode() + b"." + payload
|
|
sig = "v1=" + hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
|
|
|
|
check_revolut_signature(payload, sig, timestamp, secret)
|
|
|
|
|
|
def test_check_revolut_signature_rejects_payload_only_signature():
|
|
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
|
|
timestamp = str(int(time.time() * 1000))
|
|
secret = "revolut-secret"
|
|
sig = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
|
|
|
|
with pytest.raises(ValueError, match="signature verification failed"):
|
|
check_revolut_signature(payload, sig, timestamp, secret)
|
|
|
|
|
|
def test_check_revolut_signature_v1_header():
|
|
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
|
|
timestamp = str(int(time.time() * 1000))
|
|
secret = "revolut-secret"
|
|
signed_payload = b"v1." + timestamp.encode() + b"." + payload
|
|
sig = "v1=" + hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
|
|
|
|
check_revolut_signature(payload, sig, timestamp, secret)
|
|
|
|
|
|
def test_check_revolut_signature_multiple_v1_headers():
|
|
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
|
|
timestamp = str(int(time.time() * 1000))
|
|
secret = "revolut-secret"
|
|
signed_payload = b"v1." + timestamp.encode() + b"." + payload
|
|
valid_sig = (
|
|
"v1=" + hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
|
|
)
|
|
sig_header = f"v1=deadbeef,{valid_sig}"
|
|
|
|
check_revolut_signature(payload, sig_header, timestamp, secret)
|
|
|
|
|
|
def test_check_revolut_signature_docs_vector(mocker: MockerFixture):
|
|
payload = (
|
|
b'{"data":{"id":"645a7696-22f3-aa47-9c74-cbae0449cc46",'
|
|
b'"new_state":"completed","old_state":"pending",'
|
|
b'"request_id":"app_charges-9f5d5eb3-1e06-46c5-b1c0-3914763e0bcb"},'
|
|
b'"event":"TransactionStateChanged",'
|
|
b'"timestamp":"2023-05-09T16:36:38.028960Z"}'
|
|
)
|
|
timestamp = "1683650202360"
|
|
secret = "wsk_r59a4HfWVAKycbCaNO1RvgCJec02gRd8"
|
|
sig = "v1=bca326fb378d0da7f7c490ad584a8106bab9723d8d9cdd0d50b4c5b3be3837c0"
|
|
|
|
# This is a fixed vector straight from Revolut's docs, so its timestamp is
|
|
# necessarily in the past. Freeze time to it instead of growing
|
|
# tolerance_seconds indefinitely as real time marches on.
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.time.time",
|
|
return_value=int(timestamp) / 1000,
|
|
)
|
|
check_revolut_signature(payload, sig, timestamp, secret)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_fiat_service_fee(settings: Settings):
|
|
# settings.stripe_limits.service_min_amount_sats = 0
|
|
amount_msats = 100_000
|
|
fee = payments.service_fee_fiat(amount_msats, "no_such_fiat_provider")
|
|
assert fee == 0
|
|
|
|
settings.stripe_limits.service_fee_wallet_id = None
|
|
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
|
assert fee == 0
|
|
|
|
settings.stripe_limits.service_fee_wallet_id = "wallet_id"
|
|
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
|
assert fee == 0
|
|
|
|
settings.stripe_limits.service_max_fee_sats = 5
|
|
settings.stripe_limits.service_fee_percent = 20
|
|
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
|
assert fee == 5000
|
|
|
|
fee = payments.service_fee_fiat(-amount_msats, "stripe")
|
|
assert fee == 5000
|
|
|
|
settings.stripe_limits.service_max_fee_sats = 5
|
|
settings.stripe_limits.service_fee_percent = 3
|
|
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
|
assert fee == 3000
|
|
|
|
fee = payments.service_fee_fiat(-amount_msats, "stripe")
|
|
assert fee == 3000
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_handle_fiat_payment_confirmation(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
user = await create_user_account()
|
|
service_fee_wallet = await create_wallet(user_id=user.id)
|
|
faucet_wallet = await create_wallet(user_id=user.id)
|
|
await payments.update_wallet_balance(wallet=faucet_wallet, amount=100_000_000)
|
|
|
|
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
|
|
settings.stripe_enabled = True
|
|
settings.stripe_limits.service_min_amount_sats = 0
|
|
settings.stripe_limits.service_max_amount_sats = 0
|
|
|
|
settings.stripe_limits.service_fee_percent = 20
|
|
settings.stripe_limits.service_fee_wallet_id = service_fee_wallet.id
|
|
settings.stripe_limits.service_faucet_wallet_id = faucet_wallet.id
|
|
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=True,
|
|
checking_id=f"session_1000_{get_random_string(10)}",
|
|
payment_request="https://stripe.com/pay/session_1000",
|
|
)
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=10000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
|
)
|
|
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
assert payment.status == PaymentState.PENDING
|
|
assert payment.amount == 10_000_000
|
|
|
|
await handle_fiat_payment_confirmation(payment)
|
|
# await asyncio.sleep(1) # Simulate async delay
|
|
|
|
service_fee_payments = await get_payments(wallet_id=service_fee_wallet.id)
|
|
assert len(service_fee_payments) == 1
|
|
assert service_fee_payments[0].amount == 2_000_000
|
|
assert service_fee_payments[0].fee == 0
|
|
assert service_fee_payments[0].status == PaymentState.SUCCESS
|
|
assert service_fee_payments[0].fiat_provider is None
|
|
|
|
faucet_wallet_payments = await get_payments(wallet_id=faucet_wallet.id)
|
|
|
|
# Background tasks may create more payments, so we check for at least 2
|
|
# One for the service fee, one for the top-up)
|
|
assert len(faucet_wallet_payments) >= 2
|
|
faucet_payment = next(
|
|
(p for p in faucet_wallet_payments if p.payment_hash == payment.payment_hash),
|
|
None,
|
|
)
|
|
assert faucet_payment
|
|
assert faucet_payment.amount == -10_000_000
|
|
assert faucet_payment.fee == 0
|
|
assert faucet_payment.status == PaymentState.SUCCESS
|
|
assert faucet_payment.fiat_provider is None
|
|
assert (
|
|
faucet_payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
|
)
|
|
assert (
|
|
faucet_payment.extra.get("fiat_payment_request")
|
|
== fiat_mock_response.payment_request
|
|
)
|
|
assert faucet_payment.checking_id.startswith("internal_fiat_stripe_")
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [b'{"id": "evt_test"}', b"{}", b""])
|
|
def test_check_stripe_signature_success(payload):
|
|
secret = "whsec_testsecret"
|
|
sig_header, _, _ = _make_stripe_sig_header(payload, secret)
|
|
# Should not raise
|
|
check_stripe_signature(payload, sig_header, secret)
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [b'{"id": "evt_test"}'])
|
|
def test_check_stripe_signature_missing_header(payload):
|
|
secret = "whsec_testsecret"
|
|
with pytest.raises(ValueError, match="Stripe-Signature header is missing"):
|
|
check_stripe_signature(payload, None, secret)
|
|
|
|
|
|
def test_check_stripe_signature_missing_secret():
|
|
payload = b'{"id": "evt_test"}'
|
|
sig_header, _, _ = _make_stripe_sig_header(payload, "whsec_testsecret")
|
|
with pytest.raises(ValueError, match="Stripe webhook cannot be verified"):
|
|
check_stripe_signature(payload, sig_header, None)
|
|
|
|
|
|
def test_check_stripe_signature_invalid_signature():
|
|
payload = b'{"id": "evt_test"}'
|
|
secret = "whsec_testsecret"
|
|
_, timestamp, _ = _make_stripe_sig_header(payload, secret)
|
|
# Tamper with signature
|
|
bad_sig_header = f"t={timestamp},v1=deadbeef"
|
|
with pytest.raises(ValueError, match="Stripe signature verification failed"):
|
|
check_stripe_signature(payload, bad_sig_header, secret)
|
|
|
|
|
|
def test_check_stripe_signature_old_timestamp():
|
|
payload = b'{"id": "evt_test"}'
|
|
secret = "whsec_testsecret"
|
|
old_timestamp = int(time.time()) - 10000 # way outside default tolerance
|
|
sig_header, _, _ = _make_stripe_sig_header(payload, secret, timestamp=old_timestamp)
|
|
with pytest.raises(ValueError, match="Timestamp outside tolerance"):
|
|
check_stripe_signature(payload, sig_header, secret)
|
|
|
|
|
|
def test_check_stripe_signature_future_timestamp():
|
|
payload = b'{"id": "evt_test"}'
|
|
secret = "whsec_testsecret"
|
|
future_timestamp = int(time.time()) + 10000
|
|
sig_header, _, _ = _make_stripe_sig_header(
|
|
payload, secret, timestamp=future_timestamp
|
|
)
|
|
with pytest.raises(ValueError, match="Timestamp outside tolerance"):
|
|
check_stripe_signature(payload, sig_header, secret)
|
|
|
|
|
|
def test_check_stripe_signature_malformed_header():
|
|
payload = b'{"id": "evt_test"}'
|
|
secret = "whsec_testsecret"
|
|
# Missing v1 part
|
|
bad_header = "t=1234567890"
|
|
with pytest.raises(Exception): # noqa: B017
|
|
check_stripe_signature(payload, bad_header, secret)
|
|
# Missing t part
|
|
bad_header2 = "v1=abcdef"
|
|
with pytest.raises(Exception): # noqa: B017
|
|
check_stripe_signature(payload, bad_header2, secret)
|
|
# Not split by =
|
|
bad_header3 = "t1234567890,v1abcdef"
|
|
with pytest.raises(Exception): # noqa: B017
|
|
check_stripe_signature(payload, bad_header3, secret)
|
|
|
|
|
|
def test_check_stripe_signature_non_utf8_payload():
|
|
secret = "whsec_testsecret"
|
|
payload = b"\xff\xfe\xfd" # not valid utf-8
|
|
timestamp = int(time.time())
|
|
# This will raise UnicodeDecodeError inside check_stripe_signature
|
|
signed_payload = f"{timestamp}." + payload.decode(errors="ignore")
|
|
signature = hmac.new(
|
|
secret.encode(), signed_payload.encode(), hashlib.sha256
|
|
).hexdigest()
|
|
sig_header = f"t={timestamp},v1={signature}"
|
|
with pytest.raises(UnicodeDecodeError):
|
|
check_stripe_signature(payload, sig_header, secret)
|
|
|
|
|
|
def test_check_square_signature_success():
|
|
payload = b'{"type":"payment.updated"}'
|
|
secret = "signature-key"
|
|
notification_url = "https://lnbits.example/api/v1/callback/square"
|
|
signature = b64encode(
|
|
hmac.new(
|
|
key=secret.encode(),
|
|
msg=notification_url.encode() + payload,
|
|
digestmod=hashlib.sha256,
|
|
).digest()
|
|
).decode()
|
|
|
|
check_square_signature(payload, signature, secret, notification_url)
|
|
|
|
|
|
def test_check_square_signature_rejects_invalid_signature():
|
|
with pytest.raises(ValueError, match="Square signature verification failed."):
|
|
check_square_signature(
|
|
b'{"type":"payment.updated"}',
|
|
"invalid-signature",
|
|
"signature-key",
|
|
"https://lnbits.example/api/v1/callback/square",
|
|
)
|
|
|
|
|
|
# Helper to generate a valid Stripe signature header
|
|
def _make_stripe_sig_header(payload, secret, timestamp=None):
|
|
if timestamp is None:
|
|
timestamp = int(time.time())
|
|
signed_payload = f"{timestamp}.{payload.decode()}"
|
|
signature = hmac.new(
|
|
secret.encode(), signed_payload.encode(), hashlib.sha256
|
|
).hexdigest()
|
|
return f"t={timestamp},v1={signature}", timestamp, signature
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_check_fiat_status_handles_internal_states(mocker: MockerFixture):
|
|
pending_payment = Payment(
|
|
checking_id="external_payment",
|
|
payment_hash="hash_pending",
|
|
wallet_id="wallet_id",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="bolt11",
|
|
status=PaymentState.PENDING,
|
|
)
|
|
success_payment = Payment(
|
|
checking_id="fiat_success",
|
|
payment_hash="hash_success",
|
|
wallet_id="wallet_id",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="bolt11",
|
|
status=PaymentState.SUCCESS,
|
|
fiat_provider="stripe",
|
|
)
|
|
failed_payment = Payment(
|
|
checking_id="fiat_failed",
|
|
payment_hash="hash_failed",
|
|
wallet_id="wallet_id",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="bolt11",
|
|
status=PaymentState.FAILED,
|
|
fiat_provider="stripe",
|
|
)
|
|
|
|
assert (await check_fiat_status(pending_payment)).pending is True
|
|
assert (await check_fiat_status(success_payment)).success is True
|
|
assert (await check_fiat_status(failed_payment)).failed is True
|
|
|
|
provider = mocker.Mock()
|
|
provider.get_invoice_status = AsyncMock(return_value=FiatPaymentStatus(paid=True))
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.get_fiat_provider",
|
|
AsyncMock(return_value=provider),
|
|
)
|
|
queue_put = mocker.patch(
|
|
"lnbits.task_manager.task_manager.internal_invoice_queue.put_nowait"
|
|
)
|
|
|
|
success_status = await check_fiat_status(
|
|
Payment(
|
|
checking_id="fiat_pending",
|
|
payment_hash="hash_queue",
|
|
wallet_id="wallet_id",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="bolt11",
|
|
status=PaymentState.PENDING,
|
|
fiat_provider="stripe",
|
|
extra={"fiat_checking_id": "stripe_checking_id"},
|
|
)
|
|
)
|
|
|
|
assert success_status.success is True
|
|
queue_put.assert_called_once()
|
|
assert queue_put.call_args[0][0].checking_id == "fiat_pending"
|
|
|
|
await check_fiat_status(
|
|
Payment(
|
|
checking_id="fiat_pending_skip",
|
|
payment_hash="hash_skip",
|
|
wallet_id="wallet_id",
|
|
amount=1000,
|
|
fee=0,
|
|
bolt11="bolt11",
|
|
status=PaymentState.SUCCESS,
|
|
fiat_provider="stripe",
|
|
extra={"fiat_checking_id": "stripe_checking_id"},
|
|
)
|
|
)
|
|
assert queue_put.call_count == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_check_fiat_status_persists_successful_payment(
|
|
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.stripe_enabled = True
|
|
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
|
settings.stripe_limits.service_min_amount_sats = 0
|
|
settings.stripe_limits.service_max_amount_sats = 0
|
|
settings.stripe_limits.service_fee_wallet_id = None
|
|
settings.stripe_limits.service_faucet_wallet_id = None
|
|
|
|
invoice_data = CreateInvoice(
|
|
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
|
)
|
|
fiat_mock_response = FiatInvoiceResponse(
|
|
ok=True,
|
|
checking_id=f"session_paid_{get_random_string(10)}",
|
|
payment_request="https://stripe.com/pay/session_paid",
|
|
)
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.create_invoice",
|
|
AsyncMock(return_value=fiat_mock_response),
|
|
)
|
|
mocker.patch(
|
|
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
|
AsyncMock(return_value=1000),
|
|
)
|
|
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
|
assert payment.status == PaymentState.PENDING
|
|
|
|
mocker.patch(
|
|
"lnbits.fiat.StripeWallet.get_invoice_status",
|
|
AsyncMock(return_value=FiatPaymentStatus(paid=True)),
|
|
)
|
|
queue_put = mocker.patch(
|
|
"lnbits.task_manager.task_manager.internal_invoice_queue.put_nowait"
|
|
)
|
|
|
|
status = await check_fiat_status(payment)
|
|
|
|
assert status.success is True
|
|
assert payment.status == PaymentState.SUCCESS
|
|
updated_payment = await get_payment(payment.checking_id)
|
|
assert updated_payment.status == PaymentState.SUCCESS
|
|
queue_put.assert_called_once_with(payment)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_verify_paypal_webhook_requires_configuration(settings: Settings):
|
|
settings.paypal_webhook_id = None
|
|
|
|
with pytest.raises(
|
|
ValueError, match="PayPal webhook cannot be verified. Missing webhook ID."
|
|
):
|
|
await verify_paypal_webhook({}, b"{}")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_verify_paypal_webhook_requires_headers(settings: Settings):
|
|
settings.paypal_webhook_id = "webhook-id"
|
|
|
|
with pytest.raises(
|
|
ValueError, match="PayPal webhook cannot be verified. Missing headers."
|
|
):
|
|
await verify_paypal_webhook({}, b"{}")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_verify_paypal_webhook_success(settings: Settings, mocker: MockerFixture):
|
|
settings.paypal_webhook_id = "webhook-id"
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(json_data={"access_token": "token"}),
|
|
MockHTTPResponse(json_data={"verification_status": "SUCCESS"}),
|
|
]
|
|
)
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.httpx.AsyncClient",
|
|
return_value=client,
|
|
)
|
|
|
|
await verify_paypal_webhook(
|
|
{
|
|
"PAYPAL-TRANSMISSION-ID": "tx-id",
|
|
"PAYPAL-TRANSMISSION-TIME": "2024-01-01T00:00:00Z",
|
|
"PAYPAL-TRANSMISSION-SIG": "signature",
|
|
"PAYPAL-CERT-URL": "https://cert.example.com",
|
|
"PAYPAL-AUTH-ALGO": "SHA256withRSA",
|
|
},
|
|
b'{"id":"event-1"}',
|
|
)
|
|
|
|
assert client.calls[0][0] == "/v1/oauth2/token"
|
|
assert client.calls[1][0] == "/v1/notifications/verify-webhook-signature"
|
|
assert client.calls[1][1]["headers"]["Authorization"] == "Bearer token"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_verify_paypal_webhook_raises_on_failed_verification(
|
|
settings: Settings, mocker: MockerFixture
|
|
):
|
|
settings.paypal_webhook_id = "webhook-id"
|
|
client = MockHTTPClient(
|
|
[
|
|
MockHTTPResponse(json_data={"access_token": "token"}),
|
|
MockHTTPResponse(json_data={"verification_status": "FAILURE"}),
|
|
]
|
|
)
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.httpx.AsyncClient",
|
|
return_value=client,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="PayPal webhook cannot be verified."):
|
|
await verify_paypal_webhook(
|
|
{
|
|
"PAYPAL-TRANSMISSION-ID": "tx-id",
|
|
"PAYPAL-TRANSMISSION-TIME": "2024-01-01T00:00:00Z",
|
|
"PAYPAL-TRANSMISSION-SIG": "signature",
|
|
"PAYPAL-CERT-URL": "https://cert.example.com",
|
|
"PAYPAL-AUTH-ALGO": "SHA256withRSA",
|
|
},
|
|
b'{"id":"event-1"}',
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_test_connection_reports_provider_status(mocker: MockerFixture):
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.get_fiat_provider",
|
|
AsyncMock(return_value=None),
|
|
)
|
|
missing_status = await fiat_provider_connection("stripe")
|
|
assert missing_status.success is False
|
|
assert missing_status.message == "Fiat provider 'stripe' not found."
|
|
|
|
provider = mocker.Mock()
|
|
provider.status = AsyncMock(
|
|
return_value=FiatStatusResponse(error_message="bad key")
|
|
)
|
|
mocker.patch(
|
|
"lnbits.core.services.fiat_providers.get_fiat_provider",
|
|
AsyncMock(return_value=provider),
|
|
)
|
|
error_status = await fiat_provider_connection("stripe")
|
|
assert error_status.success is False
|
|
assert error_status.message == "Cconnection test failed: bad key"
|
|
|
|
provider.status = AsyncMock(return_value=FiatStatusResponse(balance=21.0))
|
|
success_status = await fiat_provider_connection("stripe")
|
|
assert success_status.success is True
|
|
assert success_status.message == "Connection test successful. Balance: 21.0."
|