1106 lines
40 KiB
Python
1106 lines
40 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_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_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.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_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)
|
|
settings.lnbits_allowed_currencies = []
|
|
settings.paypal_enabled = False
|
|
settings.square_enabled = False
|
|
yield
|
|
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
|
|
|
|
|
|
@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"]
|
|
|
|
|
|
@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.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.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_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.tasks.internal_invoice_queue.put", AsyncMock())
|
|
|
|
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_awaited_once_with("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.PENDING,
|
|
fiat_provider="stripe",
|
|
extra={"fiat_checking_id": "stripe_checking_id"},
|
|
),
|
|
skip_internal_payment_notifications=True,
|
|
)
|
|
assert queue_put.await_count == 1
|
|
|
|
|
|
@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."
|