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

test: on_paid listeners using a mock (#3802)

Browse Source
This commit is contained in:
dni ⚡
2026-03-19 11:29:05 +02:00
committed by GitHub
parent 92fc9a130f
commit e1e2112461
3 changed files with 71 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/helpers.py
-4
View File
@@ -6,10 +6,6 @@ from pydantic import BaseModel
from lnbits.wallets import get_funding_source, set_funding_source
class FakeError(Exception):
pass
class DbTestModel(BaseModel):
id: int
name: str
tests/regtest/test_real_invoice.py
+56 -58
View File
@@ -2,6 +2,7 @@ import asyncio
import hashlib
import pytest
from pytest_mock.plugin import MockerFixture
from lnbits import bolt11
from lnbits.core.crud import get_standalone_payment, update_payment
@@ -18,7 +19,7 @@ from lnbits.exceptions import PaymentError
from lnbits.tasks import create_task, wait_for_paid_invoices
from lnbits.wallets import get_funding_source
from ..helpers import FakeError, is_fake, is_regtest
from ..helpers import is_fake, is_regtest
from .helpers import (
cancel_invoice,
get_real_invoice,
@@ -138,7 +139,9 @@ async def test_pay_real_invoice_mainnet(
@pytest.mark.anyio
@pytest.mark.skipif(is_fake, reason="this only works in regtest")
async def test_create_real_invoice(client, adminkey_headers_from, inkey_headers_from):
async def test_create_real_invoice(
client, adminkey_headers_from, inkey_headers_from, mocker: MockerFixture
):
prev_balance = await get_node_balance_sats()
create_invoice = CreateInvoice(out=False, amount=1000, memo="test")
response = await client.post(
@@ -156,34 +159,32 @@ async def test_create_real_invoice(client, adminkey_headers_from, inkey_headers_
payment_status = response.json()
assert not payment_status["paid"]
async def on_paid(payment: Payment):
on_paid_mock = mocker.AsyncMock()
create_task(wait_for_paid_invoices("test_create_invoice", on_paid_mock)())
assert payment.payment_hash == invoice["payment_hash"]
assert payment.checking_id == invoice["checking_id"]
response = await client.get(
f'/api/v1/payments/{invoice["payment_hash"]}', headers=inkey_headers_from
)
assert response.status_code < 300
payment_status = response.json()
assert payment_status["paid"]
await asyncio.sleep(1)
balance = await get_node_balance_sats()
fee = abs(payment_status.get("details", {}).get("fee", 0) // 1000)
assert balance - prev_balance == create_invoice.amount - fee
assert payment_status.get("preimage") is not None
# exit out of infinite loop
raise FakeError()
task = create_task(wait_for_paid_invoices("test_create_invoice", on_paid)())
pay_real_invoice(invoice["bolt11"])
# wait for the task to exit
with pytest.raises(FakeError):
await task
await asyncio.sleep(1)
assert on_paid_mock.call_count == 1
payment = on_paid_mock.call_args_list[0][0][0]
assert payment.payment_hash == invoice["payment_hash"]
assert payment.checking_id == invoice["checking_id"]
response = await client.get(
f'/api/v1/payments/{invoice["payment_hash"]}', headers=inkey_headers_from
)
assert response.status_code < 300
payment_status = response.json()
assert payment_status["paid"]
await asyncio.sleep(1)
balance = await get_node_balance_sats()
fee = abs(payment_status.get("details", {}).get("fee", 0) // 1000)
assert balance - prev_balance == create_invoice.amount - fee
assert payment_status.get("preimage") is not None
@pytest.mark.anyio
@@ -367,7 +368,7 @@ async def test_pay_hold_invoice_check_pending_and_fail_cancel_payment_task_in_me
@pytest.mark.anyio
@pytest.mark.skipif(is_fake, reason="this only works in regtest")
async def test_receive_real_invoice_set_pending_and_check_state(
client, adminkey_headers_from, inkey_headers_from
client, adminkey_headers_from, inkey_headers_from, mocker: MockerFixture
):
"""
1. We create a real invoice
@@ -391,39 +392,36 @@ async def test_receive_real_invoice_set_pending_and_check_state(
payment_status = response.json()
assert not payment_status["paid"]
async def on_paid(payment: Payment):
on_paid_mock = mocker.AsyncMock()
create_task(wait_for_paid_invoices("test_create_invoice", on_paid_mock)())
assert payment.payment_hash == invoice["payment_hash"]
assert payment.checking_id == invoice["checking_id"]
response = await client.get(
f'/api/v1/payments/{invoice["payment_hash"]}', headers=inkey_headers_from
)
assert response.status_code < 300
payment_status = response.json()
assert payment_status["paid"]
assert payment
# set the incoming invoice to pending
payment.status = PaymentState.PENDING
await update_payment(payment)
payment_pending = await get_standalone_payment(
invoice["payment_hash"], incoming=True
)
assert payment_pending
assert payment_pending.success is False
assert payment_pending.failed is False
# exit out of infinite loop
raise FakeError()
task = create_task(wait_for_paid_invoices("test_create_invoice", on_paid)())
pay_real_invoice(invoice["bolt11"])
with pytest.raises(FakeError):
await task
await asyncio.sleep(1)
assert on_paid_mock.call_count == 1
payment = on_paid_mock.call_args_list[0][0][0]
assert payment.payment_hash == invoice["payment_hash"]
assert payment.checking_id == invoice["checking_id"]
response = await client.get(
f'/api/v1/payments/{invoice["payment_hash"]}', headers=inkey_headers_from
)
assert response.status_code < 300
payment_status = response.json()
assert payment_status["paid"]
# set the incoming invoice to pending
payment.status = PaymentState.PENDING
await update_payment(payment)
payment_pending = await get_standalone_payment(
invoice["payment_hash"], incoming=True
)
assert payment_pending
assert payment_pending.success is False
assert payment_pending.failed is False
@pytest.mark.anyio
tests/unit/test_pay_invoice.py
+15 -21
View File
@@ -10,16 +10,12 @@ from pytest_mock.plugin import MockerFixture
from lnbits.core.crud import create_wallet, get_standalone_payment, get_wallet
from lnbits.core.crud.payments import get_payment, get_payments_paginated
from lnbits.core.models import Payment, PaymentState, Wallet
from lnbits.core.models import PaymentState, Wallet
from lnbits.core.services import create_invoice, create_user_account, pay_invoice
from lnbits.core.services.payments import update_wallet_balance
from lnbits.exceptions import InvoiceError, PaymentError
from lnbits.settings import Settings
from lnbits.tasks import (
create_permanent_task,
internal_invoice_listener,
register_invoice_listener,
)
from lnbits.tasks import create_task, wait_for_paid_invoices
from lnbits.wallets.base import PaymentResponse
from lnbits.wallets.fake import FakeWallet
@@ -230,13 +226,13 @@ async def test_pay_for_extension(to_wallet: Wallet, settings: Settings):
@pytest.mark.anyio
async def test_notification_for_internal_payment(to_wallet: Wallet):
async def test_notification_for_internal_payment(
to_wallet: Wallet, mocker: MockerFixture
):
test_name = "test_notification_for_internal_payment"
create_permanent_task(internal_invoice_listener)
invoice_queue: asyncio.Queue = asyncio.Queue()
register_invoice_listener(invoice_queue, test_name)
on_paid_mock = mocker.AsyncMock()
create_task(wait_for_paid_invoices(test_name, on_paid_mock)())
payment = await create_invoice(
wallet_id=to_wallet.id,
amount=123,
@@ -248,17 +244,15 @@ async def test_notification_for_internal_payment(to_wallet: Wallet):
)
await asyncio.sleep(1)
while True:
_payment: Payment = invoice_queue.get_nowait() # raises if queue empty
assert _payment
if _payment.memo == test_name:
assert _payment.status == PaymentState.SUCCESS.value
assert _payment.bolt11 == payment.bolt11
assert _payment.amount == 123_000
updated_payment = await get_payment(_payment.checking_id)
assert updated_payment.webhook_status == "404"
assert on_paid_mock.call_count == 1
_payment = on_paid_mock.call_args_list[0][0][0]
break # we found our payment, success
assert _payment.memo == test_name
assert _payment.status == PaymentState.SUCCESS.value
assert _payment.bolt11 == payment.bolt11
assert _payment.amount == 123_000
updated_payment = await get_payment(_payment.checking_id)
assert updated_payment.webhook_status == "404"
@pytest.mark.anyio