Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Payments working

Browse Source
This commit is contained in:
Arc
2026-05-20 01:56:16 +01:00
parent db89a8f608
commit ebb5e0af2a
8 changed files with 824 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/test_fiat_api.py
+81
View File
@@ -4,6 +4,8 @@ from pytest_mock.plugin import MockerFixture
from lnbits.core.models.misc import SimpleStatus
from lnbits.fiat.base import FiatSubscriptionResponse
from lnbits.fiat.revolut import REVOLUT_WEBHOOK_EVENTS
from lnbits.settings import Settings
class _UnsetSecret:
@@ -144,3 +146,82 @@ async def test_fiat_api_connection_token_validates_provider_configuration(
assert ok.status_code == 200
assert ok.json() == {"secret": "tok_live"}
assert good_provider.await_count == 1
@pytest.mark.anyio
async def test_fiat_api_creates_revolut_webhook(
client: AsyncClient,
superuser_token: str,
settings: Settings,
mocker: MockerFixture,
):
create_webhook = mocker.patch(
"lnbits.core.views.fiat_api.RevolutWallet.create_webhook",
mocker.AsyncMock(
return_value={
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
}
),
)
response = await client.post(
"/api/v1/fiat/revolut/webhook",
headers={"Authorization": f"Bearer {superuser_token}"},
json={
"url": "https://lnbits.example/api/v1/callback/revolut",
"endpoint": "https://sandbox-merchant.revolut.com",
"api_secret_key": "secret_1",
"api_version": "2026-04-20",
},
)
assert response.status_code == 200
assert response.json() == {
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
"already_exists": False,
}
create_webhook.assert_awaited_once_with(
url="https://lnbits.example/api/v1/callback/revolut",
endpoint="https://sandbox-merchant.revolut.com",
api_secret_key="secret_1",
api_version="2026-04-20",
)
assert settings.revolut_payment_webhook_url == (
"https://lnbits.example/api/v1/callback/revolut"
)
assert settings.revolut_webhook_signing_secret == "whsec_1"
@pytest.mark.anyio
async def test_fiat_api_rejects_local_revolut_webhook(
client: AsyncClient,
superuser_token: str,
mocker: MockerFixture,
):
create_webhook = mocker.patch(
"lnbits.core.views.fiat_api.RevolutWallet.create_webhook",
mocker.AsyncMock(
side_effect=ValueError("Revolut webhook URL must be a clearnet URL.")
),
)
response = await client.post(
"/api/v1/fiat/revolut/webhook",
headers={"Authorization": f"Bearer {superuser_token}"},
json={
"url": "http://localhost:5000/api/v1/callback/revolut",
"endpoint": "https://sandbox-merchant.revolut.com",
"api_secret_key": "secret_1",
"api_version": "2026-04-20",
},
)
assert response.status_code == 400
assert response.json()["detail"] == ("Revolut webhook URL must be a clearnet URL.")
create_webhook.assert_awaited_once()
tests/unit/test_fiat_providers.py
+154 -1
View File
@@ -33,7 +33,7 @@ from lnbits.fiat.base import (
FiatStatusResponse,
FiatSubscriptionPaymentOptions,
)
from lnbits.fiat.revolut import RevolutWallet
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
@@ -899,6 +899,110 @@ async def test_revolut_wallet_cancel_subscription(settings: Settings):
assert client.calls[0][0] == "/api/subscriptions/SUBSCRIPTION123/cancel"
@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": 15,
},
),
(
"/api/webhooks",
{
"json": {
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
},
"timeout": 15,
},
),
]
@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": 15,
},
)
]
@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()))
@@ -908,6 +1012,55 @@ def test_check_revolut_signature():
check_revolut_signature(payload, sig, timestamp, secret)
def test_check_revolut_signature_millisecond_timestamp():
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()
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():
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"
check_revolut_signature(
payload, sig, timestamp, secret, tolerance_seconds=100000000
)
@pytest.mark.anyio
async def test_fiat_service_fee(settings: Settings):
# settings.stripe_limits.service_min_amount_sats = 0