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

test: more tests

Browse Source
This commit is contained in:
Vlad Stan
2026-03-23 12:32:39 +02:00
parent 9d9ab48345
commit ca2f78d25b
13 changed files with 2595 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_services_notifications.py
+598
View File
@@ -0,0 +1,598 @@
import asyncio
from datetime import datetime, timezone
from http import HTTPStatus
from types import SimpleNamespace
from unittest.mock import MagicMock
from uuid import uuid4
import httpx
import pytest
from pywebpush import WebPushException
from pytest_mock.plugin import MockerFixture
from lnbits.core.crud import (
create_account,
create_payment,
create_wallet,
create_webpush_subscription,
get_payment,
get_webpush_subscription,
update_payment,
update_wallet,
)
from lnbits.core.models import Account, CreatePayment, Payment, PaymentState, Wallet
from lnbits.core.models.notifications import NotificationType
from lnbits.core.models.users import UserExtra, UserNotifications
from lnbits.core.models.wallets import (
WalletPermission,
WalletSharePermission,
WalletShareStatus,
)
from lnbits.core.services.notifications import (
dispatch_webhook,
enqueue_admin_notification,
enqueue_user_notification,
process_next_notification,
send_admin_notification,
send_chat_payment_notification,
send_email,
send_email_notification,
send_notification,
send_nostr_notification,
send_nostr_notifications,
send_payment_notification,
send_payment_push_notification,
send_push_notification,
send_telegram_message,
send_telegram_notification,
send_user_notification,
send_ws_payment_notification,
)
from lnbits.settings import Settings
class MockHTTPClient:
def __init__(self, post_response=None, post_exception=None):
self.post_response = post_response
self.post_exception = post_exception
self.posts: list[tuple[str, dict]] = []
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
async def post(self, url, **kwargs):
self.posts.append((url, kwargs))
if self.post_exception:
raise self.post_exception
return self.post_response
async def _create_wallet(
notifications: UserNotifications | None = None,
*,
name: str | None = None,
) -> Wallet:
account = Account(
id=uuid4().hex,
username=f"user_{uuid4().hex[:8]}",
extra=UserExtra(notifications=notifications or UserNotifications()),
)
await create_account(account)
return await create_wallet(
user_id=account.id,
wallet_name=name or f"wallet_{account.id[:8]}",
)
async def _create_payment(
wallet: Wallet,
*,
amount_msat: int = 2_000,
status: PaymentState = PaymentState.SUCCESS,
webhook: str | None = None,
webhook_status: str | None = None,
memo: str | None = "memo",
extra: dict | None = None,
) -> Payment:
checking_id = f"checking_{uuid4().hex[:8]}"
payment = await create_payment(
checking_id=checking_id,
data=CreatePayment(
wallet_id=wallet.id,
payment_hash=uuid4().hex,
bolt11=f"bolt11-{checking_id}",
amount_msat=amount_msat,
memo=memo or "",
webhook=webhook,
extra=extra or {},
),
status=status,
)
if webhook_status is not None:
payment.webhook_status = webhook_status
await update_payment(payment)
return await get_payment(checking_id)
@pytest.mark.anyio
async def test_enqueue_and_process_notifications(settings: Settings, mocker: MockerFixture):
queue: asyncio.Queue = asyncio.Queue()
admin_mock = mocker.patch(
"lnbits.core.services.notifications.send_admin_notification",
mocker.AsyncMock(),
)
user_mock = mocker.patch(
"lnbits.core.services.notifications.send_user_notification",
mocker.AsyncMock(),
)
mocker.patch("lnbits.core.services.notifications.notifications_queue", queue)
mocker.patch(
"lnbits.core.services.notifications._is_message_type_enabled",
return_value=True,
)
enqueue_admin_notification(NotificationType.settings_update, {"username": "alice"})
await process_next_notification()
assert admin_mock.await_count == 1
assert admin_mock.await_args.args[0].startswith(f"[{settings.lnbits_site_title}]")
assert "alice" in admin_mock.await_args.args[0]
assert admin_mock.await_args.args[1] == NotificationType.settings_update.value
user_notifications = UserNotifications(email_address="alice@example.com")
enqueue_user_notification(
NotificationType.text_message,
{"message": "hello"},
user_notifications,
)
await process_next_notification()
assert user_mock.await_count == 1
assert user_mock.await_args.args[0] == user_notifications
assert "hello" in user_mock.await_args.args[1]
assert user_mock.await_args.args[2] == NotificationType.text_message.value
@pytest.mark.anyio
async def test_send_admin_and_user_notification_use_expected_targets(
settings: Settings, mocker: MockerFixture
):
send_mock = mocker.patch(
"lnbits.core.services.notifications.send_notification",
mocker.AsyncMock(),
)
original_chat_id = settings.lnbits_telegram_notifications_chat_id
original_identifiers = list(settings.lnbits_nostr_notifications_identifiers)
original_emails = list(settings.lnbits_email_notifications_to_emails)
try:
settings.lnbits_telegram_notifications_chat_id = "chat-id"
settings.lnbits_nostr_notifications_identifiers = ["alice@example.com"]
settings.lnbits_email_notifications_to_emails = ["admin@example.com"]
await send_admin_notification("hello", "settings_update")
await send_user_notification(
UserNotifications(
telegram_chat_id="user-chat",
nostr_identifier="bob@example.com",
email_address="bob@example.com",
),
"hello user",
"text_message",
)
finally:
settings.lnbits_telegram_notifications_chat_id = original_chat_id
settings.lnbits_nostr_notifications_identifiers = original_identifiers
settings.lnbits_email_notifications_to_emails = original_emails
assert send_mock.await_args_list[0].args == (
"chat-id",
["alice@example.com"],
["admin@example.com"],
"hello",
"settings_update",
)
assert send_mock.await_args_list[1].args == (
"user-chat",
["bob@example.com"],
["bob@example.com"],
"hello user",
"text_message",
)
@pytest.mark.anyio
async def test_send_notification_uses_available_channels_and_swallows_exceptions(
settings: Settings, mocker: MockerFixture
):
original_email_enabled = settings.lnbits_email_notifications_enabled
try:
settings.lnbits_email_notifications_enabled = True
mocker.patch.object(
type(settings),
"is_telegram_notifications_configured",
return_value=True,
)
mocker.patch.object(
type(settings),
"is_nostr_notifications_configured",
return_value=True,
)
telegram_mock = mocker.patch(
"lnbits.core.services.notifications.send_telegram_notification",
mocker.AsyncMock(side_effect=Exception("telegram boom")),
)
nostr_mock = mocker.patch(
"lnbits.core.services.notifications.send_nostr_notifications",
mocker.AsyncMock(return_value=["alice@example.com"]),
)
email_mock = mocker.patch(
"lnbits.core.services.notifications.send_email_notification",
mocker.AsyncMock(side_effect=Exception("email boom")),
)
await send_notification(
"chat-id",
["alice@example.com"],
["alice@example.com"],
"hello",
"text_message",
)
finally:
settings.lnbits_email_notifications_enabled = original_email_enabled
telegram_mock.assert_awaited_once()
nostr_mock.assert_awaited_once()
email_mock.assert_awaited_once()
@pytest.mark.anyio
async def test_send_nostr_notifications_and_single_notification(
mocker: MockerFixture,
):
send_mock = mocker.patch(
"lnbits.core.services.notifications.send_nostr_notification",
mocker.AsyncMock(side_effect=[None, Exception("boom"), None]),
)
result = await send_nostr_notifications(["ok-1", "bad", "ok-2"], "hello")
assert result == ["ok-1", "ok-2"]
assert send_mock.await_count == 3
fetch_mock = mocker.patch(
"lnbits.core.services.notifications.fetch_nip5_details",
mocker.AsyncMock(return_value=("pubkey", ["wss://relay"])),
)
normalize_mock = mocker.patch(
"lnbits.core.services.notifications.normalize_private_key",
return_value="server-private-key",
)
dm_mock = mocker.patch(
"lnbits.core.services.notifications.send_nostr_dm",
mocker.AsyncMock(),
)
await send_nostr_notification("alice@example.com", "hello")
fetch_mock.assert_awaited_once_with("alice@example.com")
normalize_mock.assert_called_once()
dm_mock.assert_awaited_once_with(
"server-private-key",
"pubkey",
"hello",
["wss://relay"],
)
@pytest.mark.anyio
async def test_send_telegram_message_and_wrapper(
settings: Settings, mocker: MockerFixture
):
response = httpx.Response(
200,
request=httpx.Request("POST", "https://api.telegram.org"),
json={"ok": True},
)
client = MockHTTPClient(post_response=response)
mocker.patch(
"lnbits.core.services.notifications.httpx.AsyncClient",
return_value=client,
)
result = await send_telegram_message("token", "chat-id", "hello")
assert result == {"ok": True}
assert client.posts[0][0].endswith("/bottoken/sendMessage")
original_token = settings.lnbits_telegram_notifications_access_token
try:
settings.lnbits_telegram_notifications_access_token = "wrapper-token"
wrapper_mock = mocker.patch(
"lnbits.core.services.notifications.send_telegram_message",
mocker.AsyncMock(return_value={"ok": True}),
)
await send_telegram_notification("chat-id", "hello")
finally:
settings.lnbits_telegram_notifications_access_token = original_token
wrapper_mock.assert_awaited_once_with("wrapper-token", "chat-id", "hello")
@pytest.mark.anyio
async def test_send_email_notification_and_send_email(
settings: Settings, mocker: MockerFixture
):
original_email_enabled = settings.lnbits_email_notifications_enabled
try:
settings.lnbits_email_notifications_enabled = False
disabled = await send_email_notification(["alice@example.com"], "hello")
assert disabled["status"] == "error"
settings.lnbits_email_notifications_enabled = True
send_email_mock = mocker.patch(
"lnbits.core.services.notifications.send_email",
mocker.AsyncMock(return_value=True),
)
enabled = await send_email_notification(["alice@example.com"], "hello")
assert enabled == {"status": "ok"}
send_email_mock.assert_awaited_once()
finally:
settings.lnbits_email_notifications_enabled = original_email_enabled
smtp_server = MagicMock()
smtp_context = MagicMock()
smtp_context.__enter__.return_value = smtp_server
smtp_context.__exit__.return_value = None
mocker.patch(
"lnbits.core.services.notifications.smtplib.SMTP",
return_value=smtp_context,
)
assert (
await send_email(
"smtp.example.com",
587,
"",
"password",
"from@example.com",
["to@example.com"],
"Subject",
"Body",
)
is True
)
smtp_server.starttls.assert_called_once()
smtp_server.login.assert_called_once_with("from@example.com", "password")
smtp_server.sendmail.assert_called_once()
with pytest.raises(ValueError, match="Invalid from email address"):
await send_email(
"smtp.example.com",
587,
"user",
"password",
"bad-email",
["to@example.com"],
"Subject",
"Body",
)
with pytest.raises(ValueError, match="No email addresses provided"):
await send_email(
"smtp.example.com",
587,
"user",
"password",
"from@example.com",
[],
"Subject",
"Body",
)
@pytest.mark.anyio
async def test_dispatch_webhook_marks_missing_invalid_and_failed_requests(
mocker: MockerFixture,
):
wallet = await _create_wallet()
payment = await _create_payment(wallet, webhook=None)
await dispatch_webhook(payment)
assert (await get_payment(payment.checking_id)).webhook_status == "-1"
invalid_payment = await _create_payment(wallet, webhook="https://invalid.example")
invalid_client = MockHTTPClient(
post_response=httpx.Response(
200,
request=httpx.Request("POST", invalid_payment.webhook),
json={"ok": True},
)
)
mocker.patch(
"lnbits.core.services.notifications.check_callback_url",
side_effect=ValueError("blocked"),
)
mocker.patch(
"lnbits.core.services.notifications.httpx.AsyncClient",
return_value=invalid_client,
)
await dispatch_webhook(invalid_payment)
assert (await get_payment(invalid_payment.checking_id)).webhook_status in {"-1", "200"}
error_payment = await _create_payment(wallet, webhook="https://error.example")
mocker.patch(
"lnbits.core.services.notifications.check_callback_url",
return_value=None,
)
mocker.patch(
"lnbits.core.services.notifications.httpx.AsyncClient",
return_value=MockHTTPClient(
post_response=httpx.Response(
500,
request=httpx.Request("POST", error_payment.webhook),
)
),
)
await dispatch_webhook(error_payment)
assert (await get_payment(error_payment.checking_id)).webhook_status == "500"
request_payment = await _create_payment(wallet, webhook="https://request.example")
mocker.patch(
"lnbits.core.services.notifications.httpx.AsyncClient",
return_value=MockHTTPClient(
post_exception=httpx.RequestError(
"boom",
request=httpx.Request("POST", request_payment.webhook),
)
),
)
await dispatch_webhook(request_payment)
assert (await get_payment(request_payment.checking_id)).webhook_status == "-1"
@pytest.mark.anyio
async def test_send_payment_notification_fans_out_to_shared_wallet_and_webhook(
mocker: MockerFixture,
):
wallet = await _create_wallet(name="Primary Wallet")
shared_wallet = await _create_wallet(name="Shared Wallet")
wallet.extra.shared_with = [
WalletSharePermission(
request_id="share-1",
username="bob",
shared_with_wallet_id=shared_wallet.id,
permissions=[WalletPermission.VIEW_PAYMENTS],
status=WalletShareStatus.APPROVED,
)
]
await update_wallet(wallet)
payment = await _create_payment(wallet, webhook="https://webhook.example")
ws_mock = mocker.patch(
"lnbits.core.services.notifications.send_ws_payment_notification",
mocker.AsyncMock(),
)
chat_mock = mocker.patch(
"lnbits.core.services.notifications.send_chat_payment_notification",
mocker.AsyncMock(),
)
push_mock = mocker.patch(
"lnbits.core.services.notifications.send_payment_push_notification",
mocker.AsyncMock(),
)
dispatch_mock = mocker.patch(
"lnbits.core.services.notifications.dispatch_webhook",
mocker.AsyncMock(),
)
await send_payment_notification(wallet, payment)
assert [call.args[0].id for call in ws_mock.await_args_list] == [
wallet.id,
shared_wallet.id,
]
chat_mock.assert_awaited_once_with(wallet, payment)
push_mock.assert_awaited_once_with(wallet, payment)
dispatch_mock.assert_awaited_once_with(payment)
@pytest.mark.anyio
async def test_send_ws_payment_notification_and_chat_notifications(
settings: Settings, mocker: MockerFixture
):
user_notifications = UserNotifications(
telegram_chat_id="chat-id",
nostr_identifier="alice@example.com",
email_address="alice@example.com",
incoming_payments_sats=1,
outgoing_payments_sats=1,
)
wallet = await _create_wallet(user_notifications)
payment = await _create_payment(
wallet,
amount_msat=-2_000,
extra={"wallet_fiat_currency": "USD", "wallet_fiat_amount": 5.25},
)
websocket_mock = mocker.patch(
"lnbits.core.services.notifications.websocket_manager.send",
mocker.AsyncMock(),
)
await send_ws_payment_notification(wallet, payment)
assert [call.args[0] for call in websocket_mock.await_args_list] == [
wallet.inkey,
wallet.adminkey,
payment.payment_hash,
]
original_outgoing = settings.lnbits_notification_outgoing_payment_amount_sats
original_incoming = settings.lnbits_notification_incoming_payment_amount_sats
try:
settings.lnbits_notification_outgoing_payment_amount_sats = 1
settings.lnbits_notification_incoming_payment_amount_sats = 1
admin_mock = mocker.patch(
"lnbits.core.services.notifications.enqueue_admin_notification"
)
user_mock = mocker.patch(
"lnbits.core.services.notifications.enqueue_user_notification"
)
await send_chat_payment_notification(wallet, payment)
finally:
settings.lnbits_notification_outgoing_payment_amount_sats = original_outgoing
settings.lnbits_notification_incoming_payment_amount_sats = original_incoming
assert admin_mock.call_args.args[0] == NotificationType.outgoing_payment
assert "`5.25`*USD* / " in admin_mock.call_args.args[1]["fiat_value_fmt"]
assert user_mock.call_args.args[0] == NotificationType.outgoing_payment
@pytest.mark.anyio
async def test_send_payment_push_notification_and_cleanup_gone_subscriptions(
settings: Settings, mocker: MockerFixture
):
wallet = await _create_wallet()
payment = await _create_payment(wallet, amount_msat=2_000, memo="Thanks")
endpoint = f"https://push.example/{uuid4().hex}"
subscription = await create_webpush_subscription(
endpoint,
wallet.user,
'{"endpoint":"https://push.example"}',
"push.example",
)
send_push_mock = mocker.patch(
"lnbits.core.services.notifications.send_push_notification",
mocker.AsyncMock(),
)
await send_payment_push_notification(wallet, payment)
assert send_push_mock.await_args.args[0].endpoint == subscription.endpoint
assert send_push_mock.await_args.args[1] == f"LNbits: {wallet.name}"
assert "received 2 sats" in send_push_mock.await_args.args[2]
assert send_push_mock.await_args.args[3] == (
f"https://{subscription.host}/wallet?usr={wallet.user}&wal={wallet.id}"
)
original_privkey = settings.lnbits_webpush_privkey
try:
settings.lnbits_webpush_privkey = ""
exc = WebPushException("gone")
exc.response = SimpleNamespace(status_code=HTTPStatus.GONE, text="gone")
mocker.patch(
"lnbits.core.services.notifications.webpush",
side_effect=exc,
)
await send_push_notification(subscription, "Title", "Body")
finally:
settings.lnbits_webpush_privkey = original_privkey
assert await get_webpush_subscription(subscription.endpoint, wallet.user) is None