From 0b5c8cdee2ddb477943a3edcd6e3aa96a376089b Mon Sep 17 00:00:00 2001 From: Vlad Stan Date: Mon, 9 Mar 2026 11:14:44 +0200 Subject: [PATCH] test: more api tests --- tests/api/test_admin_api.py | 109 ++++++ tests/api/test_asset_api.py | 157 ++++++++ tests/api/test_audit_api.py | 64 ++++ tests/api/test_auth_api.py | 163 ++++++++ tests/api/test_callback_api.py | 173 +++++++++ tests/api/test_extension_api.py | 461 +++++++++++++++++++++++ tests/api/test_extensions_builder_api.py | 149 ++++++++ tests/api/test_fiat_api.py | 138 +++++++ tests/api/test_lnurl_api.py | 145 +++++++ tests/api/test_node_api.py | 350 +++++++++++++++++ tests/api/test_payment_api.py | 187 +++++++++ tests/api/test_tinyurl_api.py | 71 ++++ tests/api/test_user_api.py | 100 +++++ tests/api/test_wallet_api.py | 184 +++++++++ tests/api/test_websocket_api.py | 31 ++ 15 files changed, 2482 insertions(+) create mode 100644 tests/api/test_asset_api.py create mode 100644 tests/api/test_audit_api.py create mode 100644 tests/api/test_auth_api.py create mode 100644 tests/api/test_callback_api.py create mode 100644 tests/api/test_extension_api.py create mode 100644 tests/api/test_extensions_builder_api.py create mode 100644 tests/api/test_fiat_api.py create mode 100644 tests/api/test_lnurl_api.py create mode 100644 tests/api/test_node_api.py create mode 100644 tests/api/test_payment_api.py create mode 100644 tests/api/test_tinyurl_api.py create mode 100644 tests/api/test_user_api.py create mode 100644 tests/api/test_wallet_api.py create mode 100644 tests/api/test_websocket_api.py diff --git a/tests/api/test_admin_api.py b/tests/api/test_admin_api.py index 8e8dfa03c..29de36fc4 100644 --- a/tests/api/test_admin_api.py +++ b/tests/api/test_admin_api.py @@ -1,6 +1,9 @@ +from pathlib import Path + import pytest from httpx import AsyncClient +from lnbits.server import server_restart from lnbits.settings import Settings @@ -49,3 +52,109 @@ async def test_admin_update_noneditable_settings( headers={"Authorization": f"Bearer {superuser_token}"}, ) assert response.status_code == 400 + + +@pytest.mark.anyio +async def test_admin_audit_monitor_and_test_email( + client: AsyncClient, superuser_token: str, mocker +): + mocker.patch( + "lnbits.core.views.admin_api.get_balance_delta", + mocker.AsyncMock( + return_value={"lnbits_balance_sats": 21, "node_balance_sats": 13} + ), + ) + mocker.patch( + "lnbits.core.views.admin_api.send_email_notification", + mocker.AsyncMock(return_value={"status": "queued"}), + ) + + audit = await client.get( + "/admin/api/v1/audit", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert audit.status_code == 200 + assert audit.json()["lnbits_balance_sats"] == 21 + + monitor = await client.get( + "/admin/api/v1/monitor", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert monitor.status_code == 200 + assert "invoice_listeners" in monitor.json() + + test_email = await client.get( + "/admin/api/v1/testemail", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert test_email.status_code == 200 + assert test_email.json()["status"] == "queued" + + +@pytest.mark.anyio +async def test_admin_partial_reset_restart_and_backup( + client: AsyncClient, + superuser_token: str, + settings: Settings, + tmp_path, +): + response = await client.patch( + "/admin/api/v1/settings", + json={"lnbits_site_title": "PATCHED TITLE"}, + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert response.status_code == 200 + assert response.json()["status"] == "Success" + + default_value = await client.get( + "/admin/api/v1/settings/default", + params={"field_name": "lnbits_site_title"}, + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert default_value.status_code == 200 + assert "default_value" in default_value.json() + + backup_path = Path("lnbits-backup.zip") + original_data_folder = settings.lnbits_data_folder + + try: + data_folder = tmp_path / "backup_data" + data_folder.mkdir(parents=True, exist_ok=True) + (data_folder / "sample.txt").write_text("backup me") + settings.lnbits_data_folder = str(data_folder) + + backup = await client.get( + "/admin/api/v1/backup", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert backup.status_code == 200 + assert backup.headers["content-type"] == "application/zip" + assert backup.content.startswith(b"PK") + assert backup_path.is_file() + finally: + settings.lnbits_data_folder = original_data_folder + backup_path.unlink(missing_ok=True) + + server_restart.clear() + restart = await client.get( + "/admin/api/v1/restart", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert restart.status_code == 200 + assert restart.json()["status"] == "Success" + assert server_restart.is_set() is True + server_restart.clear() + + +@pytest.mark.anyio +async def test_admin_delete_settings_requires_superuser( + client: AsyncClient, superuser_token: str +): + server_restart.clear() + response = await client.delete( + "/admin/api/v1/settings", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert response.status_code == 200 + assert server_restart.is_set() is True + server_restart.clear() diff --git a/tests/api/test_asset_api.py b/tests/api/test_asset_api.py new file mode 100644 index 000000000..b68e05308 --- /dev/null +++ b/tests/api/test_asset_api.py @@ -0,0 +1,157 @@ +from io import BytesIO +from uuid import uuid4 + +import pytest +from fastapi import UploadFile +from httpx import AsyncClient +from PIL import Image +from starlette.datastructures import Headers + +from lnbits.core.crud.assets import get_user_asset +from lnbits.core.services.assets import create_user_asset + + +def _png_bytes() -> bytes: + image = Image.new("RGB", (32, 32), color="blue") + buffer = BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() + + +def _upload_file(contents: bytes, filename: str, content_type: str) -> UploadFile: + return UploadFile( + BytesIO(contents), + filename=filename, + headers=Headers({"content-type": content_type}), + ) + + +async def _user_headers(client: AsyncClient, user_id: str) -> dict[str, str]: + response = await client.post("/api/v1/auth/usr", json={"usr": user_id}) + client.cookies.clear() + access_token = response.json()["access_token"] + return { + "Authorization": f"Bearer {access_token}", + "Content-type": "application/json", + } + + +@pytest.mark.anyio +async def test_asset_api_upload_list_update_and_delete( + client: AsyncClient, + user_headers_from: dict[str, str], +): + upload = await client.post( + "/api/v1/assets?public_asset=false", + headers={"Authorization": user_headers_from["Authorization"]}, + files={"file": ("note.txt", b"hello world", "text/plain")}, + ) + assert upload.status_code == 200 + asset = upload.json() + assert asset["name"] == "note.txt" + assert asset["is_public"] is False + + page = await client.get("/api/v1/assets/paginated", headers=user_headers_from) + assert page.status_code == 200 + assert any(item["id"] == asset["id"] for item in page.json()["data"]) + + info = await client.get(f"/api/v1/assets/{asset['id']}", headers=user_headers_from) + assert info.status_code == 200 + assert info.json()["name"] == "note.txt" + + data = await client.get( + f"/api/v1/assets/{asset['id']}/data", headers=user_headers_from + ) + assert data.status_code == 200 + assert data.content == b"hello world" + assert data.headers["content-disposition"] == 'inline; filename="note.txt"' + + updated = await client.put( + f"/api/v1/assets/{asset['id']}", + headers=user_headers_from, + json={"name": "renamed.txt", "is_public": True}, + ) + assert updated.status_code == 200 + assert updated.json()["name"] == "renamed.txt" + assert updated.json()["is_public"] is True + + public_data = await client.get(f"/api/v1/assets/{asset['id']}/data") + assert public_data.status_code == 200 + assert public_data.content == b"hello world" + + deleted = await client.delete( + f"/api/v1/assets/{asset['id']}", headers=user_headers_from + ) + assert deleted.status_code == 200 + assert deleted.json()["success"] is True + + missing = await client.get(f"/api/v1/assets/{asset['id']}", headers=user_headers_from) + assert missing.status_code == 404 + + +@pytest.mark.anyio +async def test_asset_api_enforces_visibility_and_supports_admin_updates( + client: AsyncClient, + from_user, + to_user, + superuser_token: str, +): + private_asset = await create_user_asset( + from_user.id, + _upload_file(_png_bytes(), f"private_{uuid4().hex[:8]}.png", "image/png"), + is_public=False, + ) + other_user_headers = await _user_headers(client, to_user.id) + + anonymous = await client.get(f"/api/v1/assets/{private_asset.id}/data") + assert anonymous.status_code == 404 + + wrong_user = await client.get( + f"/api/v1/assets/{private_asset.id}/data", headers=other_user_headers + ) + assert wrong_user.status_code == 404 + + admin_updated = await client.put( + f"/api/v1/assets/{private_asset.id}", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={"is_public": True, "name": "admin-visible.png"}, + ) + assert admin_updated.status_code == 200 + assert admin_updated.json()["is_public"] is True + assert admin_updated.json()["name"] == "admin-visible.png" + + thumbnail = await client.get(f"/api/v1/assets/{private_asset.id}/thumbnail") + assert thumbnail.status_code == 200 + assert thumbnail.content + assert thumbnail.headers["content-type"] == "image/png" + + +@pytest.mark.anyio +async def test_asset_api_validates_uploads_and_missing_assets( + client: AsyncClient, + user_headers_from: dict[str, str], +): + invalid = await client.post( + "/api/v1/assets", + headers={"Authorization": user_headers_from["Authorization"]}, + files={"file": ("payload.exe", b"boom", "application/x-msdownload")}, + ) + assert invalid.status_code == 400 + assert "not allowed" in invalid.json()["detail"] + + missing = await client.delete( + f"/api/v1/assets/{uuid4().hex}", + headers=user_headers_from, + ) + assert missing.status_code == 404 + + missing_thumb = await client.get(f"/api/v1/assets/{uuid4().hex}/thumbnail") + assert missing_thumb.status_code == 404 + + stored = await create_user_asset( + "missing-user-check", + _upload_file(b"content", "content.txt", "text/plain"), + is_public=True, + ) + fetched = await get_user_asset("missing-user-check", stored.id) + assert fetched is not None diff --git a/tests/api/test_audit_api.py b/tests/api/test_audit_api.py new file mode 100644 index 000000000..23080a68b --- /dev/null +++ b/tests/api/test_audit_api.py @@ -0,0 +1,64 @@ +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest +from httpx import AsyncClient + +from lnbits.core.crud.audit import create_audit_entry +from lnbits.core.models import AuditEntry + + +@pytest.mark.anyio +async def test_audit_api_requires_admin(client: AsyncClient, user_headers_from): + response = await client.get("/audit/api/v1", headers=user_headers_from) + assert response.status_code == 403 + + +@pytest.mark.anyio +async def test_audit_api_returns_entries_and_stats( + client: AsyncClient, + superuser_token: str, +): + component = f"audit_component_{uuid4().hex[:8]}" + await create_audit_entry( + AuditEntry( + component=component, + ip_address="127.0.0.1", + user_id=uuid4().hex, + path="/api/v1/test", + request_method="GET", + response_code="200", + duration=0.12, + created_at=datetime.now(timezone.utc), + ) + ) + await create_audit_entry( + AuditEntry( + component=component, + ip_address="127.0.0.2", + user_id=uuid4().hex, + path="/api/v1/test", + request_method="POST", + response_code="400", + duration=2.5, + created_at=datetime.now(timezone.utc), + ) + ) + headers = {"Authorization": f"Bearer {superuser_token}"} + + page = await client.get(f"/audit/api/v1?component={component}", headers=headers) + assert page.status_code == 200 + page_data = page.json() + assert page_data["total"] == 2 + assert {item["request_method"] for item in page_data["data"]} == {"GET", "POST"} + + stats = await client.get( + f"/audit/api/v1/stats?component={component}", + headers=headers, + ) + assert stats.status_code == 200 + payload = stats.json() + assert {item["field"] for item in payload["request_method"]} == {"GET", "POST"} + assert {item["field"] for item in payload["response_code"]} == {"200", "400"} + assert payload["component"][0]["field"] == component + assert payload["long_duration"][0]["field"] == "/api/v1/test" diff --git a/tests/api/test_auth_api.py b/tests/api/test_auth_api.py new file mode 100644 index 000000000..bbd635bd3 --- /dev/null +++ b/tests/api/test_auth_api.py @@ -0,0 +1,163 @@ +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from fastapi.responses import RedirectResponse +from httpx import AsyncClient + +from lnbits.core.crud.users import get_account, update_account +from lnbits.core.views.auth_api import get_account_by_email +from lnbits.core.models.users import Account +from lnbits.core.services.users import create_user_account +from lnbits.settings import Settings + + +class _FakeSSO: + def __init__(self, userinfo: object | None = None, state: str = ""): + self.userinfo = userinfo + self.state = state + self.redirect_uri: str | None = None + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + async def get_login_redirect(self, state: str): + self.state = state + return RedirectResponse("https://example.com/sso/login") + + async def verify_and_process(self, _request): + return self.userinfo + + +@pytest.mark.anyio +async def test_auth_api_logout_and_update_ui_customization( + http_client: AsyncClient, +): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + + response = await http_client.patch( + f"/api/v1/auth/ui?usr={user.id}", + json={"theme": "amber", "walletLayout": "grid"}, + ) + assert response.status_code == 200 + assert response.json()["ui_customization"]["theme"] == "amber" + assert response.json()["ui_customization"]["walletLayout"] == "grid" + + logout = await http_client.post("/api/v1/auth/logout") + assert logout.status_code == 200 + assert logout.json()["status"] == "success" + assert "cookie_access_token=" in logout.headers["set-cookie"] + + +@pytest.mark.anyio +async def test_auth_api_sso_login_and_callback( + http_client: AsyncClient, mocker +): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + + provider = "github" + login_sso = _FakeSSO() + mocker.patch("lnbits.core.views.auth_api._new_sso", return_value=login_sso) + + response = await http_client.get( + f"/api/v1/auth/{provider}", params={"user_id": user.id} + ) + assert response.status_code == 307 + assert response.headers["location"] == "https://example.com/sso/login" + assert login_sso.redirect_uri == f"{http_client.base_url}/api/v1/auth/github/token" + assert login_sso.state + + email = f"sso_{uuid4().hex[:8]}@lnbits.com" + callback_sso = _FakeSSO(userinfo=SimpleNamespace(email=email), state="") + mocker.patch("lnbits.core.views.auth_api._new_sso", return_value=callback_sso) + + callback = await http_client.get(f"/api/v1/auth/{provider}/token") + assert callback.status_code == 307 + assert callback.headers["location"] == "/wallet" + + account = await get_account_by_email(email, active_only=False) + assert account is not None + assert account.email == email + assert account.extra.email_verified is True + + +@pytest.mark.anyio +async def test_auth_api_first_install_success_and_validation( + http_client: AsyncClient, settings: Settings +): + superuser = await get_account(settings.super_user, active_only=False) + assert superuser is not None + + original_username = superuser.username + original_password_hash = superuser.password_hash + original_first_install = settings.first_install + original_first_install_token = settings.first_install_token + + first_install_token = f"install_{uuid4().hex[:8]}" + new_username = f"reinstall_{uuid4().hex[:8]}" + + try: + settings.first_install = True + settings.first_install_token = first_install_token + + missing_token = await http_client.put( + "/api/v1/auth/first_install", + json={ + "username": new_username, + "password": "secret1234", + "password_repeat": "secret1234", + }, + ) + assert missing_token.status_code == 401 + assert missing_token.json()["detail"] == "Missing first_install_token." + + success = await http_client.put( + "/api/v1/auth/first_install", + json={ + "username": new_username, + "password": "secret1234", + "password_repeat": "secret1234", + "first_install_token": first_install_token, + }, + ) + assert success.status_code == 200 + assert success.json()["access_token"] + + updated_superuser = await get_account(settings.super_user, active_only=False) + assert updated_superuser is not None + assert updated_superuser.username == new_username + assert settings.first_install is False + + forbidden = await http_client.put( + "/api/v1/auth/first_install", + json={ + "username": f"blocked_{uuid4().hex[:8]}", + "password": "secret1234", + "password_repeat": "secret1234", + }, + ) + assert forbidden.status_code == 403 + assert forbidden.json()["detail"] == "This is not your first install" + finally: + restored_superuser = await get_account(settings.super_user, active_only=False) + assert restored_superuser is not None + restored_superuser.username = original_username + restored_superuser.password_hash = original_password_hash + await update_account(restored_superuser) + settings.first_install = original_first_install + settings.first_install_token = original_first_install_token diff --git a/tests/api/test_callback_api.py b/tests/api/test_callback_api.py new file mode 100644 index 000000000..d45a6ba82 --- /dev/null +++ b/tests/api/test_callback_api.py @@ -0,0 +1,173 @@ +import json +from uuid import uuid4 + +import pytest +from httpx import AsyncClient + +from lnbits.core.models import Account, CreateInvoice +from lnbits.core.services.payments import create_wallet_invoice +from lnbits.core.services.users import create_user_account +from lnbits.core.views.callback_api import ( + handle_paypal_event, + handle_stripe_event, +) + + +@pytest.mark.anyio +async def test_callback_api_generic_webhook_handler_routes_providers( + http_client: AsyncClient, mocker +): + stripe_mock = mocker.patch( + "lnbits.core.views.callback_api.handle_stripe_event", mocker.AsyncMock() + ) + paypal_mock = mocker.patch( + "lnbits.core.views.callback_api.handle_paypal_event", mocker.AsyncMock() + ) + mocker.patch("lnbits.core.views.callback_api.check_stripe_signature") + mocker.patch( + "lnbits.core.views.callback_api.verify_paypal_webhook", mocker.AsyncMock() + ) + + stripe = await http_client.post( + "/api/v1/callback/stripe", + headers={"Stripe-Signature": "sig"}, + json={"id": "evt_1", "type": "payment_intent.succeeded"}, + ) + assert stripe.status_code == 200 + assert stripe.json()["success"] is True + stripe_mock.assert_awaited_once() + + paypal = await http_client.post( + "/api/v1/callback/paypal", + json={"id": "evt_2", "event_type": "CHECKOUT.ORDER.APPROVED"}, + ) + assert paypal.status_code == 200 + assert paypal.json()["success"] is True + paypal_mock.assert_awaited_once() + + unknown = await http_client.post("/api/v1/callback/unknown", json={"id": "evt_3"}) + assert unknown.status_code == 200 + assert unknown.json()["success"] is False + + +@pytest.mark.anyio +async def test_callback_api_handles_paid_events_with_real_payments(mocker): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = user.wallets[0] + payment = await create_wallet_invoice( + wallet.id, CreateInvoice(out=False, amount=11, memo="fiat callback") + ) + + fiat_status_mock = mocker.patch( + "lnbits.core.views.callback_api.check_fiat_status", mocker.AsyncMock() + ) + + await handle_stripe_event( + { + "id": "evt_stripe", + "type": "payment_intent.succeeded", + "data": { + "object": { + "object": "payment_intent", + "metadata": {"payment_hash": payment.payment_hash}, + } + }, + } + ) + await handle_paypal_event( + { + "id": "evt_paypal", + "event_type": "CHECKOUT.ORDER.APPROVED", + "resource": { + "purchase_units": [{"invoice_id": payment.payment_hash}], + }, + } + ) + await handle_stripe_event({"id": "evt_unhandled", "type": "customer.created"}) + + assert fiat_status_mock.await_count == 2 + + +@pytest.mark.anyio +async def test_callback_api_handles_subscription_flows_and_validation(mocker): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = user.wallets[0] + payment = await create_wallet_invoice( + wallet.id, CreateInvoice(out=False, amount=15, memo="subscription") + ) + + create_fiat_invoice_mock = mocker.patch( + "lnbits.core.views.callback_api.create_fiat_invoice", + mocker.AsyncMock(return_value=payment), + ) + fiat_status_mock = mocker.patch( + "lnbits.core.views.callback_api.check_fiat_status", mocker.AsyncMock() + ) + + await handle_stripe_event( + { + "id": "evt_invoice_paid", + "type": "invoice.paid", + "data": { + "object": { + "id": "invoice_1", + "currency": "usd", + "amount_paid": 500, + "hosted_invoice_url": "https://stripe.example/invoice", + "customer_email": "alice@example.com", + "lines": {"data": [{"description": "Gold Plan"}]}, + "parent": { + "type": "subscription_details", + "subscription_details": { + "metadata": { + "alan_action": "subscription", + "wallet_id": wallet.id, + "tag": "gold", + "memo": "Monthly Gold", + "extra": json.dumps({"plan": "gold"}), + } + }, + }, + } + }, + } + ) + create_fiat_invoice_mock.assert_awaited() + fiat_status_mock.assert_awaited() + + await handle_paypal_event( + { + "id": "evt_sale_completed", + "event_type": "PAYMENT.SALE.COMPLETED", + "resource": { + "id": "sale_1", + "billing_agreement_id": "agreement_1", + "amount": {"currency": "USD", "total": "7.50"}, + "custom_id": json.dumps( + [wallet.id, "vip", "subscription_1", "link-1", "VIP Plan"] + ), + }, + } + ) + assert create_fiat_invoice_mock.await_count == 2 + + with pytest.raises(ValueError, match="PayPal subscription event missing custom metadata."): + await handle_paypal_event( + { + "id": "evt_bad_sale", + "event_type": "PAYMENT.SALE.COMPLETED", + "resource": {"amount": {"currency": "USD", "total": "5.00"}}, + } + ) diff --git a/tests/api/test_extension_api.py b/tests/api/test_extension_api.py new file mode 100644 index 000000000..e053676e1 --- /dev/null +++ b/tests/api/test_extension_api.py @@ -0,0 +1,461 @@ +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from fastapi import HTTPException +from starlette.requests import Request + +from lnbits.core.crud.db_versions import get_db_version, update_migration_version +from lnbits.core.crud.extensions import ( + create_installed_extension, + get_installed_extension, + get_user_extension, +) +from lnbits.core.crud.users import get_account +from lnbits.core.crud.wallets import create_wallet +from lnbits.core.models import Account, CreateInvoice +from lnbits.core.models.extensions import ( + CreateExtension, + CreateExtensionReview, + Extension, + ExtensionConfig, + ExtensionMeta, + ExtensionRelease, + InstallableExtension, + PayToEnableInfo, + ReleasePaymentInfo, + UserExtensionInfo, +) +from lnbits.core.models.users import AccountId +from lnbits.core.services.payments import create_wallet_invoice +from lnbits.core.services.users import create_user_account +from lnbits.core.views.extension_api import ( + api_activate_extension, + api_deactivate_extension, + api_disable_extension, + api_enable_extension, + api_extension_details, + api_get_user_extensions, + api_install_extension, + api_uninstall_extension, + api_update_pay_to_enable, + create_extension_review, + delete_extension_db, + extensions, + get_extension_release, + get_extension_releases, + get_extension_reviews, + get_extension_reviews_tags, + get_pay_to_enable_invoice, + get_pay_to_install_invoice, +) + + +def _release(ext_id: str, version: str = "1.0.0") -> ExtensionRelease: + return ExtensionRelease( + name=ext_id, + version=version, + archive=f"https://example.com/{ext_id}.zip", + source_repo="org/repo", + hash=f"hash-{ext_id}", + details_link=f"https://example.com/{ext_id}/details.json", + repo=f"https://github.com/org/{ext_id}", + icon=f"/{ext_id}/static/icon.png", + pay_link=f"https://pay.example/{ext_id}", + ) + + +def _installable_extension( + ext_id: str, + *, + active: bool = True, + pay_to_enable: PayToEnableInfo | None = None, + dependencies: list[str] | None = None, + payments: list[ReleasePaymentInfo] | None = None, +) -> InstallableExtension: + release = _release(ext_id) + return InstallableExtension( + id=ext_id, + name=f"Extension {ext_id}", + version=release.version, + active=active, + short_description="Demo extension", + icon=release.icon, + meta=ExtensionMeta( + installed_release=release, + pay_to_enable=pay_to_enable, + dependencies=dependencies or [], + payments=payments or [], + ), + ) + + +class _MockHTTPResponse: + def __init__( + self, + *, + json_data=None, + text: str = "", + status_code: int = 200, + is_error: bool = False, + ): + self._json_data = json_data + self.text = text + self.status_code = status_code + self.is_error = is_error + + def json(self): + return self._json_data + + def raise_for_status(self): + if self.status_code >= 400: + raise ValueError(self.text or "request failed") + + +class _MockHTTPClient: + def __init__(self, responses: dict[str, _MockHTTPResponse]): + self.responses = responses + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, url: str): + return self.responses[url] + + async def post(self, url: str, json=None): + return self.responses[url] + + +@pytest.mark.anyio +async def test_extension_api_install_details_and_release_endpoints(mocker): + ext_id = f"ext_{uuid4().hex[:8]}" + release = _release(ext_id) + create_data = CreateExtension( + ext_id=ext_id, + archive=release.archive, + source_repo=release.source_repo, + version=release.version, + ) + + mocker.patch.object( + InstallableExtension, + "get_extension_release", + mocker.AsyncMock(return_value=release), + ) + mocker.patch( + "lnbits.core.views.extension_api.install_extension", + mocker.AsyncMock(return_value=Extension(code=ext_id, is_valid=True)), + ) + activate_mock = mocker.patch( + "lnbits.core.views.extension_api.activate_extension", mocker.AsyncMock() + ) + + installed = await api_install_extension(create_data) + assert installed.code == ext_id + activate_mock.assert_awaited_once() + + mocker.patch.object( + InstallableExtension, + "get_extension_releases", + mocker.AsyncMock(return_value=[release]), + ) + mocker.patch.object( + ExtensionRelease, + "fetch_release_details", + mocker.AsyncMock(return_value={"description": "Extension details"}), + ) + details = await api_extension_details(ext_id, release.details_link or "") + assert details["description"] == "Extension details" + assert details["icon"] == release.icon + assert details["repo"] == release.repo + + installed_ext = _installable_extension( + ext_id, + payments=[ + ReleasePaymentInfo( + amount=55, + pay_link=release.pay_link, + payment_hash=f"payment_{uuid4().hex[:8]}", + ) + ], + ) + await create_installed_extension(installed_ext) + releases = await get_extension_releases(ext_id) + assert releases[0].paid_sats == 55 + + config = ExtensionConfig( + name=ext_id, + short_description="Config", + min_lnbits_version="0.1.0", + max_lnbits_version=None, + ) + mocker.patch.object( + ExtensionConfig, + "fetch_github_release_config", + mocker.AsyncMock(return_value=config), + ) + release_info = await get_extension_release("org", ext_id, "v1.0.0") + assert release_info["is_version_compatible"] is True + + +@pytest.mark.anyio +async def test_extension_api_pay_to_enable_and_catalog_views(mocker, admin_user): + regular_user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + admin_account = await get_account(admin_user.id) + assert admin_account is not None + admin_wallet = await create_wallet( + user_id=admin_account.id, wallet_name="extension sales" + ) + + ext_id = f"paid_{uuid4().hex[:8]}" + await create_installed_extension( + _installable_extension( + ext_id, + pay_to_enable=PayToEnableInfo(required=True, amount=10, wallet=admin_wallet.id), + ) + ) + + updated = await api_update_pay_to_enable( + ext_id, + PayToEnableInfo(required=True, amount=21, wallet=admin_wallet.id), + account=admin_account, + ) + assert updated.success is True + stored_extension = await get_installed_extension(ext_id) + assert stored_extension is not None + assert stored_extension.meta is not None + assert stored_extension.meta.pay_to_enable is not None + assert stored_extension.meta.pay_to_enable.amount == 21 + + enable_invoice = await create_wallet_invoice( + admin_wallet.id, CreateInvoice(out=False, amount=21, memo="enable extension") + ) + mocker.patch( + "lnbits.core.views.extension_api.create_invoice", + mocker.AsyncMock(return_value=enable_invoice), + ) + invoice_response = await get_pay_to_enable_invoice( + ext_id, + PayToEnableInfo(amount=21), + account_id=AccountId(id=regular_user.id), + ) + assert invoice_response["payment_hash"] == enable_invoice.payment_hash + + user_ext = await get_user_extension(regular_user.id, ext_id) + assert user_ext is not None + assert user_ext.extra is not None + assert user_ext.extra.payment_hash_to_enable == enable_invoice.payment_hash + + mocker.patch( + "lnbits.core.views.extension_api.get_valid_extensions", + mocker.AsyncMock(return_value=[Extension(code=ext_id, is_valid=True)]), + ) + mocker.patch( + "lnbits.core.views.extension_api.check_transaction_status", + mocker.AsyncMock(return_value=SimpleNamespace(paid=True)), + ) + + enabled = await api_enable_extension(ext_id, AccountId(id=regular_user.id)) + assert enabled.success is True + user_ext = await get_user_extension(regular_user.id, ext_id) + assert user_ext is not None + assert user_ext.active is True + assert user_ext.extra == UserExtensionInfo( + payment_hash_to_enable=enable_invoice.payment_hash, + paid_to_enable=True, + ) + + disabled = await api_disable_extension(ext_id, AccountId(id=regular_user.id)) + assert disabled.success is True + disabled_again = await api_disable_extension(ext_id, AccountId(id=regular_user.id)) + assert disabled_again.success is True + assert "already disabled" in disabled_again.message + + mocker.patch( + "lnbits.core.views.extension_api.get_valid_extensions", + mocker.AsyncMock( + return_value=[ + Extension(code=ext_id, is_valid=True, name="Paid Extension"), + Extension(code="other", is_valid=True), + ] + ), + ) + visible_extensions = await api_get_user_extensions(AccountId(id=regular_user.id)) + assert [ext.code for ext in visible_extensions] == [ext_id] + + catalog_entry = _installable_extension( + ext_id, + pay_to_enable=PayToEnableInfo(required=True, amount=21, wallet=admin_wallet.id), + ) + mocker.patch.object( + InstallableExtension, + "get_installable_extensions", + mocker.AsyncMock(return_value=[catalog_entry]), + ) + catalog = await extensions(AccountId(id=regular_user.id)) + catalog_item = next(item for item in catalog if item["id"] == ext_id) + assert catalog_item["payToEnable"]["wallet"] is None + + +@pytest.mark.anyio +async def test_extension_api_activate_uninstall_install_invoice_and_cleanup(mocker): + base_ext = f"base_{uuid4().hex[:8]}" + dependent_ext = f"dependent_{uuid4().hex[:8]}" + uninstall_ext = f"uninstall_{uuid4().hex[:8]}" + db_ext = f"db_{uuid4().hex[:8]}" + + await create_installed_extension(_installable_extension(base_ext)) + await create_installed_extension( + _installable_extension(dependent_ext, dependencies=[base_ext]) + ) + await create_installed_extension(_installable_extension(uninstall_ext)) + + mocker.patch( + "lnbits.core.views.extension_api.get_valid_extensions", + mocker.AsyncMock( + return_value=[ + Extension(code=base_ext, is_valid=True, name="Base"), + Extension(code=dependent_ext, is_valid=True, name="Dependent"), + Extension(code=uninstall_ext, is_valid=True, name="Remove"), + ] + ), + ) + + with pytest.raises(HTTPException, match="depends on this one"): + await api_uninstall_extension(base_ext) + + uninstall_mock = mocker.patch( + "lnbits.core.views.extension_api.uninstall_extension", mocker.AsyncMock() + ) + uninstalled = await api_uninstall_extension(uninstall_ext) + assert uninstalled.success is True + uninstall_mock.assert_awaited_once_with(uninstall_ext) + + mocker.patch( + "lnbits.core.views.extension_api.get_valid_extension", + mocker.AsyncMock(return_value=Extension(code=base_ext, is_valid=True)), + ) + activate_mock = mocker.patch( + "lnbits.core.views.extension_api.activate_extension", mocker.AsyncMock() + ) + deactivate_mock = mocker.patch( + "lnbits.core.views.extension_api.deactivate_extension", mocker.AsyncMock() + ) + activated = await api_activate_extension(base_ext) + assert activated.success is True + deactivated = await api_deactivate_extension(base_ext) + assert deactivated.success is True + activate_mock.assert_awaited_once() + deactivate_mock.assert_awaited_once() + + owner = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = owner.wallets[0] + install_invoice = await create_wallet_invoice( + wallet.id, CreateInvoice(out=False, amount=33, memo="install extension") + ) + release = _release(base_ext, version="2.0.0") + payment_info = ReleasePaymentInfo( + amount=33, + pay_link=release.pay_link, + payment_hash=install_invoice.payment_hash, + payment_request=install_invoice.bolt11, + ) + mocker.patch.object( + InstallableExtension, + "get_extension_release", + mocker.AsyncMock(return_value=release), + ) + mocker.patch.object( + ExtensionRelease, + "fetch_release_payment_info", + mocker.AsyncMock(return_value=payment_info), + ) + invoice = await get_pay_to_install_invoice( + base_ext, + CreateExtension( + ext_id=base_ext, + archive=release.archive, + source_repo=release.source_repo, + version=release.version, + cost_sats=33, + ), + ) + assert invoice.payment_hash == install_invoice.payment_hash + + await update_migration_version(None, db_ext, 1) + drop_mock = mocker.patch( + "lnbits.core.views.extension_api.drop_extension_db", mocker.AsyncMock() + ) + deleted = await delete_extension_db(db_ext) + assert deleted.success is True + drop_mock.assert_awaited_once_with(ext_id=db_ext) + assert await get_db_version(db_ext) is None + + +@pytest.mark.anyio +async def test_extension_api_review_endpoints(mocker): + ext_id = f"review_{uuid4().hex[:8]}" + request = Request( + { + "type": "http", + "method": "GET", + "path": f"/api/v1/extension/reviews/{ext_id}", + "query_string": b"offset=0&limit=5", + "headers": [], + } + ) + mock_client = _MockHTTPClient( + { + "https://demo.lnbits.com/paidreviews/api/v1/AdFzLjzuKFLsdk4Bcnff6r/tags": _MockHTTPResponse( + json_data=[{"tag": "good", "avg_rating": 900, "review_count": 3}] + ), + f"https://demo.lnbits.com/paidreviews/api/v1/AdFzLjzuKFLsdk4Bcnff6r/reviews/{ext_id}?offset=0&limit=5": _MockHTTPResponse( + json_data={ + "data": [ + { + "id": "1", + "name": "Alice", + "tag": "good", + "rating": 950, + "comment": "solid", + } + ], + "total": 1, + } + ), + "https://demo.lnbits.com/paidreviews/api/v1/AdFzLjzuKFLsdk4Bcnff6r/reviews": _MockHTTPResponse( + json_data={ + "payment_hash": f"hash_{uuid4().hex[:8]}", + "payment_request": "lnbc1review", + } + ), + } + ) + mocker.patch("lnbits.core.views.extension_api.httpx.AsyncClient", return_value=mock_client) + + tags = await get_extension_reviews_tags() + assert tags[0].tag == "good" + + reviews = await get_extension_reviews(ext_id, request) + assert reviews.total == 1 + assert reviews.data[0].comment == "solid" + + payment_request = await create_extension_review( + CreateExtensionReview(tag=ext_id, name="Alice", rating=900, comment="Great") + ) + assert payment_request.payment_hash.startswith("hash_") diff --git a/tests/api/test_extensions_builder_api.py b/tests/api/test_extensions_builder_api.py new file mode 100644 index 000000000..f07bb4377 --- /dev/null +++ b/tests/api/test_extensions_builder_api.py @@ -0,0 +1,149 @@ +from pathlib import Path +from uuid import uuid4 + +import pytest + +from lnbits.core.crud.extensions import create_user_extension, get_user_extension +from lnbits.core.crud.users import get_account +from lnbits.core.models.extensions import ( + Extension, + ExtensionMeta, + ExtensionRelease, + UserExtension, +) +from lnbits.core.models.extensions_builder import ( + ActionFields, + ClientDataFields, + DataField, + DataFields, + ExtensionData, + OwnerDataFields, + PublicPageFields, + SettingsFields, +) +from lnbits.core.models.users import AccountId +from lnbits.core.views.extensions_builder_api import ( + api_build_extension, + api_delete_extension_builder_data, + api_deploy_extension, + api_preview_extension, +) +from lnbits.settings import Settings + + +def _extension_data(ext_id: str = "demoext") -> ExtensionData: + return ExtensionData( + id=ext_id, + name="Demo Extension", + stub_version="0.1.0", + short_description="Generated extension", + owner_data=DataFields( + name="OwnerData", + fields=[DataField(name="wallet_id", type="wallet")], + ), + client_data=DataFields( + name="ClientData", + fields=[DataField(name="amount", type="int")], + ), + settings_data=SettingsFields(name="SettingsData", fields=[]), + public_page=PublicPageFields( + owner_data_fields=OwnerDataFields(), + client_data_fields=ClientDataFields(), + action_fields=ActionFields(), + ), + ) + + +def _release(ext_id: str) -> ExtensionRelease: + return ExtensionRelease( + name=ext_id, + version="0.1.0", + archive=f"https://example.com/{ext_id}.zip", + source_repo="org/repo", + is_github_release=False, + hash=f"hash-{ext_id}", + icon=f"/{ext_id}/static/image/{ext_id}.png", + ) + + +@pytest.mark.anyio +async def test_extensions_builder_api_build_preview_and_cleanup( + tmp_path, settings: Settings, mocker, from_user +): + ext_id = f"builder_{uuid4().hex[:8]}" + data = _extension_data(ext_id) + build_dir = tmp_path / "build" + build_dir.mkdir(parents=True, exist_ok=True) + (build_dir / "index.txt").write_text("hello") + + original_data_folder = settings.lnbits_data_folder + build_mock = mocker.patch( + "lnbits.core.views.extensions_builder_api.build_extension_from_data", + mocker.AsyncMock(return_value=(_release(ext_id), build_dir)), + ) + clean_mock = mocker.patch("lnbits.core.views.extensions_builder_api.clean_extension_builder_data") + + try: + settings.lnbits_data_folder = str(tmp_path) + + build_response = await api_build_extension(data) + assert Path(build_response.path).is_file() + assert build_response.filename == f"{ext_id}.zip" + + preview = await api_preview_extension(data, AccountId(id=from_user.id)) + assert preview.success is True + assert ext_id in preview.message + + cleaned = await api_delete_extension_builder_data() + assert cleaned.success is True + clean_mock.assert_called_once() + assert build_mock.await_count == 2 + finally: + settings.lnbits_data_folder = original_data_folder + + +@pytest.mark.anyio +async def test_extensions_builder_api_deploy_updates_user_extension( + tmp_path, settings: Settings, mocker, admin_user +): + ext_id = f"deploy_{uuid4().hex[:8]}" + data = _extension_data(ext_id) + account = await get_account(admin_user.id) + assert account is not None + + build_root = tmp_path / "deploy-root" / ext_id + build_root.mkdir(parents=True, exist_ok=True) + (build_root / "manifest.json").write_text("{}") + + original_data_folder = settings.lnbits_data_folder + await create_user_extension( + UserExtension(user=account.id, extension=ext_id, active=False) + ) + + mocker.patch( + "lnbits.core.views.extensions_builder_api.build_extension_from_data", + mocker.AsyncMock(return_value=(_release(ext_id), build_root)), + ) + install_mock = mocker.patch( + "lnbits.core.views.extensions_builder_api.install_extension", + mocker.AsyncMock(return_value=Extension(code=ext_id, is_valid=True)), + ) + activate_mock = mocker.patch( + "lnbits.core.views.extensions_builder_api.activate_extension", + mocker.AsyncMock(), + ) + + try: + settings.lnbits_data_folder = str(tmp_path) + deployed = await api_deploy_extension(data, account=account) + finally: + settings.lnbits_data_folder = original_data_folder + + assert deployed.success is True + assert ext_id in deployed.message + install_mock.assert_awaited_once() + activate_mock.assert_awaited_once() + + user_ext = await get_user_extension(account.id, ext_id) + assert user_ext is not None + assert user_ext.active is True diff --git a/tests/api/test_fiat_api.py b/tests/api/test_fiat_api.py new file mode 100644 index 000000000..a673ba93b --- /dev/null +++ b/tests/api/test_fiat_api.py @@ -0,0 +1,138 @@ +from typing import Any + +import pytest +from httpx import AsyncClient +from pytest_mock.plugin import MockerFixture + +from lnbits.core.models.misc import SimpleStatus +from lnbits.fiat.base import FiatSubscriptionResponse + + +class FakeStripeWallet: + def __init__(self, secret: str | None = "secret"): + self._secret = secret + + async def create_terminal_connection_token(self) -> dict[str, str]: + if self._secret is None: + return {} + return {"secret": self._secret} + + +@pytest.mark.anyio +async def test_fiat_api_test_provider_and_subscription_lifecycle( + client: AsyncClient, + superuser_token: str, + adminkey_headers_from: dict[str, str], + from_wallet, + mocker: MockerFixture, +): + test_connection = mocker.patch( + "lnbits.core.views.fiat_api.test_connection", + mocker.AsyncMock(return_value=SimpleStatus(success=True, message="ok")), + ) + response = await client.put( + "/api/v1/fiat/check/stripe", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert response.status_code == 200 + assert response.json()["success"] is True + test_connection.assert_awaited_once_with("stripe") + + provider = mocker.Mock() + provider.create_subscription = mocker.AsyncMock( + return_value=FiatSubscriptionResponse( + ok=True, + subscription_request_id="sub-1", + checkout_session_url="https://stripe.example/checkout", + ) + ) + provider.cancel_subscription = mocker.AsyncMock( + return_value=FiatSubscriptionResponse(ok=True, subscription_request_id="sub-1") + ) + get_provider = mocker.patch( + "lnbits.core.views.fiat_api.get_fiat_provider", + mocker.AsyncMock(return_value=provider), + ) + + mismatch = await client.post( + "/api/v1/fiat/stripe/subscription", + headers=adminkey_headers_from, + json={ + "subscription_id": "sub-1", + "quantity": 2, + "payment_options": {"wallet_id": "wrong-wallet"}, + }, + ) + assert mismatch.status_code == 403 + + created = await client.post( + "/api/v1/fiat/stripe/subscription", + headers=adminkey_headers_from, + json={ + "subscription_id": "sub-1", + "quantity": 2, + "payment_options": {"memo": "hello", "wallet_id": from_wallet.id}, + }, + ) + assert created.status_code == 200 + assert created.json()["checkout_session_url"] == "https://stripe.example/checkout" + provider.create_subscription.assert_awaited_once() + assert provider.create_subscription.await_args.args[2].wallet_id == from_wallet.id + + cancelled = await client.delete( + "/api/v1/fiat/stripe/subscription/sub-1", + headers=adminkey_headers_from, + ) + assert cancelled.status_code == 200 + provider.cancel_subscription.assert_awaited_once_with("sub-1", from_wallet.id) + assert get_provider.await_count == 3 + + +@pytest.mark.anyio +async def test_fiat_api_connection_token_validates_provider_configuration( + client: AsyncClient, + superuser_token: str, + mocker: MockerFixture, +): + headers = {"Authorization": f"Bearer {superuser_token}"} + + not_found = mocker.patch( + "lnbits.core.views.fiat_api.get_fiat_provider", + mocker.AsyncMock(return_value=None), + ) + missing = await client.post("/api/v1/fiat/stripe/connection_token", headers=headers) + assert missing.status_code == 404 + assert not_found.await_count == 1 + + unsupported_provider = mocker.patch( + "lnbits.core.views.fiat_api.get_fiat_provider", + mocker.AsyncMock(return_value=object()), + ) + unsupported = await client.post( + "/api/v1/fiat/paypal/connection_token", headers=headers + ) + assert unsupported.status_code == 400 + assert unsupported_provider.await_count == 1 + + mocker.patch("lnbits.core.views.fiat_api.StripeWallet", FakeStripeWallet) + bad_wallet = FakeStripeWallet(secret=None) + bad_provider = mocker.patch( + "lnbits.core.views.fiat_api.get_fiat_provider", + mocker.AsyncMock(return_value=bad_wallet), + ) + no_secret = await client.post( + "/api/v1/fiat/stripe/connection_token", headers=headers + ) + assert no_secret.status_code == 500 + assert no_secret.json()["detail"] == "Failed to create connection token" + assert bad_provider.await_count == 1 + + good_wallet = FakeStripeWallet(secret="tok_live") + good_provider = mocker.patch( + "lnbits.core.views.fiat_api.get_fiat_provider", + mocker.AsyncMock(return_value=good_wallet), + ) + ok = await client.post("/api/v1/fiat/stripe/connection_token", headers=headers) + assert ok.status_code == 200 + assert ok.json() == {"secret": "tok_live"} + assert good_provider.await_count == 1 diff --git a/tests/api/test_lnurl_api.py b/tests/api/test_lnurl_api.py new file mode 100644 index 000000000..92cbd41ff --- /dev/null +++ b/tests/api/test_lnurl_api.py @@ -0,0 +1,145 @@ +from uuid import uuid4 + +import pytest +from bolt11.types import MilliSatoshi +from fastapi import HTTPException +from lnurl import ( + LnAddress, + LnurlAuthResponse, + LnurlErrorResponse, + LnurlException, + LnurlPayActionResponse, + LnurlPayResponse, + LnurlResponseException, +) +from lnurl.models import MessageAction +from lnurl.types import CallbackUrl, LightningInvoice, LnurlPayMetadata +from pydantic import parse_obj_as + +from lnbits.core.models import Account, CreateInvoice +from lnbits.core.models.lnurl import CreateLnurlPayment, LnurlScan +from lnbits.core.models.wallets import KeyType, WalletTypeInfo +from lnbits.core.services.payments import create_wallet_invoice +from lnbits.core.views.lnurl_api import ( + api_lnurlscan, + api_lnurlscan_post, + api_payments_pay_lnurl, + api_perform_lnurlauth, +) +from lnbits.core.services.users import create_user_account + +TEST_BOLT11 = ( + "lnbc1pnsu5z3pp57getmdaxhg5kc9yh2a2qsh7cjf4gnccgkw0qenm8vsqv50w7s" + "ygqdqj0fjhymeqv9kk7atwwscqzzsxqyz5vqsp5e2yyqcp0a3ujeesp24ya0glej" + "srh703md8mrx0g2lyvjxy5w27ss9qxpqysgqyjreasng8a086kpkczv48er5c6l5" + "73aym6ynrdl9nkzqnag49vt3sjjn8qdfq5cr6ha0vrdz5c5r3v4aghndly0hplmv" + "6hjxepwp93cq398l3s" +) + + +def _pay_response() -> LnurlPayResponse: + return LnurlPayResponse( + callback=parse_obj_as(CallbackUrl, "https://example.com/callback"), + minSendable=MilliSatoshi(1_000), + maxSendable=MilliSatoshi(10_000), + metadata=LnurlPayMetadata( + '[["text/plain","Test payment"],["text/identifier","alice@example.com"]]' + ), + ) + + +@pytest.mark.anyio +async def test_lnurl_api_scan_routes_validate_and_forward(mocker): + pay_response = _pay_response() + mocker.patch( + "lnbits.core.views.lnurl_api.lnurl_handle", + mocker.AsyncMock(return_value=pay_response), + ) + + scanned = await api_lnurlscan("lnurl1example") + assert scanned.callback == pay_response.callback + + scanned_post = await api_lnurlscan_post(scan=LnurlScan(lnurl=LnAddress("alice@example.com"))) + assert scanned_post.callback == pay_response.callback + + mocker.patch( + "lnbits.core.views.lnurl_api.lnurl_handle", + mocker.AsyncMock(return_value=LnurlErrorResponse(reason="blocked callback")), + ) + with pytest.raises(HTTPException, match="blocked callback"): + await api_lnurlscan("lnurl1blocked") + + mocker.patch( + "lnbits.core.views.lnurl_api.lnurl_handle", + mocker.AsyncMock(side_effect=LnurlException("invalid lnurl")), + ) + with pytest.raises(HTTPException, match="invalid lnurl"): + await api_lnurlscan("lnurl1invalid") + + +@pytest.mark.anyio +async def test_lnurl_api_auth_and_pay_flow(mocker): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = user.wallets[0] + wallet_info = WalletTypeInfo(key_type=KeyType.admin, wallet=wallet) + pay_response = _pay_response() + payment = await create_wallet_invoice( + wallet.id, CreateInvoice(out=False, amount=21, memo="lnurl") + ) + + auth_response = LnurlAuthResponse( + callback=parse_obj_as(CallbackUrl, "https://example.com/auth"), + k1="k1-value", + ) + mocker.patch( + "lnbits.core.views.lnurl_api.lnurlauth", + mocker.AsyncMock(return_value=auth_response), + ) + authenticated = await api_perform_lnurlauth(auth_response, wallet_info) + assert authenticated.k1 == "k1-value" + + mocker.patch( + "lnbits.core.views.lnurl_api.lnurlauth", + mocker.AsyncMock(side_effect=LnurlResponseException("denied")), + ) + with pytest.raises(HTTPException, match="denied"): + await api_perform_lnurlauth(auth_response, wallet_info) + + action_response = LnurlPayActionResponse( + pr=LightningInvoice(TEST_BOLT11), + disposable=False, + successAction=MessageAction(message="paid"), + ) + fetch_mock = mocker.patch( + "lnbits.core.views.lnurl_api.fetch_lnurl_pay_request", + mocker.AsyncMock(return_value=(pay_response, action_response)), + ) + pay_mock = mocker.patch( + "lnbits.core.views.lnurl_api.pay_invoice", + mocker.AsyncMock(return_value=payment), + ) + + paid = await api_payments_pay_lnurl( + CreateLnurlPayment(res=pay_response, amount=2_000, unit="USD", comment="thanks"), + wallet_info, + ) + assert paid.payment_hash == payment.payment_hash + fetch_mock.assert_awaited_once() + pay_mock.assert_awaited_once() + assert pay_mock.await_args is not None + assert pay_mock.await_args.kwargs["extra"] == { + "stored": True, + "success_action": action_response.successAction.json(), + "comment": "thanks", + "fiat_currency": "USD", + "fiat_amount": 2.0, + } + + with pytest.raises(HTTPException, match="Missing LNURL or LnurlPayResponse data."): + await api_payments_pay_lnurl(CreateLnurlPayment(amount=1), wallet_info) diff --git a/tests/api/test_node_api.py b/tests/api/test_node_api.py new file mode 100644 index 000000000..8c78f2581 --- /dev/null +++ b/tests/api/test_node_api.py @@ -0,0 +1,350 @@ +from typing import Any +from uuid import uuid4 + +import httpx +import pytest +from httpx import AsyncClient +from pytest_mock.plugin import MockerFixture + +from lnbits.core.views import node_api +from lnbits.db import Filters, Page +from lnbits.nodes.base import ( + ChannelBalance, + ChannelPoint, + ChannelState, + NodeChannel, + NodeFees, + NodeInfoResponse, + NodeInvoice, + NodePayment, + NodePeerInfo, + PublicNodeInfo, +) +from lnbits.settings import Settings +from lnbits.wallets.base import Feature + + +class FakeNode: + def __init__(self): + self.channel = NodeChannel( + id="chan-1", + short_id="123x1x0", + peer_id="peer-1", + name="Peer One", + color="#ffffff", + state=ChannelState.ACTIVE, + balance=ChannelBalance(local_msat=1000, remote_msat=2000, total_msat=3000), + point=ChannelPoint(funding_txid="ab" * 32, output_index=1), + fee_ppm=10, + fee_base_msat=1000, + ) + self.peer = NodePeerInfo(id="peer-1", alias="Peer One", addresses=["127.0.0.1"]) + self.info = NodeInfoResponse( + id="node-id", + backend_name="FakeNode", + alias="Fake Alias", + color="#ffffff", + num_peers=1, + blockheight=1, + channel_stats={ + "counts": {ChannelState.ACTIVE: 1}, + "avg_size": 3000, + "biggest_size": 3000, + "smallest_size": 3000, + "total_capacity": 3000, + }, + addresses=["127.0.0.1:9735"], + onchain_balance_sat=1, + onchain_confirmed_sat=1, + fees=NodeFees(total_msat=0), + balance_msat=3000, + ) + self.fees_updated: tuple[str, int | None, int | None] | None = None + + async def get_public_info(self) -> PublicNodeInfo: + return PublicNodeInfo(**self.info.dict()) + + async def get_info(self) -> NodeInfoResponse: + return self.info + + async def get_channels(self) -> list[NodeChannel]: + return [self.channel] + + async def get_channel(self, channel_id: str) -> NodeChannel | None: + return self.channel if channel_id == self.channel.id else None + + async def open_channel( + self, + peer_id: str, + funding_amount: int, + push_amount: int | None = None, + fee_rate: int | None = None, + ) -> ChannelPoint: + assert peer_id == "peer-1" + assert funding_amount == 10_000 + assert push_amount == 100 + assert fee_rate == 5 + return ChannelPoint(funding_txid="cd" * 32, output_index=0) + + async def close_channel( + self, + short_id: str | None = None, + point: ChannelPoint | None = None, + force: bool = False, + ) -> list[NodeChannel]: + assert short_id == self.channel.short_id + assert point is None + assert force is True + return [self.channel] + + async def set_channel_fee( + self, channel_id: str, fee_base_msat: int | None, fee_ppm: int | None + ) -> None: + self.fees_updated = (channel_id, fee_base_msat, fee_ppm) + + async def get_payments(self, filters: Filters[Any]) -> Page[NodePayment]: + return Page( + data=[ + NodePayment( + pending=False, + amount=1, + fee=0, + memo="payment", + time=1, + preimage="11" * 32, + payment_hash="22" * 32, + ) + ], + total=1, + ) + + async def get_invoices(self, filters: Filters[Any]) -> Page[NodeInvoice]: + return Page( + data=[ + NodeInvoice( + pending=False, + amount=1, + memo="invoice", + bolt11="lnbc1dummy", + preimage="11" * 32, + payment_hash="33" * 32, + ) + ], + total=1, + ) + + async def get_peers(self) -> list[NodePeerInfo]: + return [self.peer] + + async def connect_peer(self, uri: str) -> dict[str, str]: + return {"uri": uri} + + async def disconnect_peer(self, peer_id: str) -> dict[str, str]: + return {"peer_id": peer_id} + + async def get_id(self) -> str: + return "fake-node-id" + + +class MockHTTPResponse: + def __init__(self, json_data: dict[str, Any], status_error: Exception | None = None): + self._json_data = json_data + self._status_error = status_error + + def raise_for_status(self) -> None: + if self._status_error: + raise self._status_error + + def json(self) -> dict[str, Any]: + return self._json_data + + +class MockHTTPClient: + def __init__(self, response: MockHTTPResponse): + self.response = response + self.calls: list[str] = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def get(self, url: str, timeout: int): + self.calls.append(url) + return self.response + + +@pytest.mark.anyio +async def test_node_api_dependency_guards(settings: Settings, mocker: MockerFixture): + original_node_ui = settings.lnbits_node_ui + original_public = settings.lnbits_public_node_ui + try: + settings.lnbits_node_ui = True + funding_source = type("FundingSource", (), {})() + funding_source.features = [] + funding_source.__node_cls__ = None + mocker.patch( + "lnbits.core.views.node_api.get_funding_source", + return_value=funding_source, + ) + with pytest.raises(Exception) as excinfo: + node_api.require_node() + assert excinfo.value.status_code == 501 + + node_enabled_source = type("FundingSource", (), {})() + node_enabled_source.features = [Feature.nodemanager] + node_enabled_source.__node_cls__ = lambda wallet: "fake-node" + mocker.patch( + "lnbits.core.views.node_api.get_funding_source", + return_value=node_enabled_source, + ) + settings.lnbits_node_ui = False + with pytest.raises(Exception) as disabled: + node_api.require_node() + assert disabled.value.status_code == 503 + + settings.lnbits_node_ui = True + assert node_api.require_node() == "fake-node" + + settings.lnbits_public_node_ui = False + with pytest.raises(Exception) as public_disabled: + node_api.check_public() + assert public_disabled.value.status_code == 503 + finally: + settings.lnbits_node_ui = original_node_ui + settings.lnbits_public_node_ui = original_public + + +@pytest.mark.anyio +async def test_node_api_route_functions_with_fake_node( + settings: Settings, + mocker: MockerFixture, +): + fake_node = FakeNode() + original_transactions = settings.lnbits_node_ui_transactions + settings.lnbits_node_ui_transactions = True + rank_response = MockHTTPResponse( + { + "noderank": { + "capacity": 1, + "channelcount": 2, + "age": 3, + "growth": 4, + "availability": 5, + } + } + ) + mocker.patch( + "lnbits.core.views.node_api.httpx.AsyncClient", + return_value=MockHTTPClient(rank_response), + ) + + try: + assert await node_api.api_get_ok() is None + + public_info = await node_api.api_get_public_info(node=fake_node) + assert public_info.backend_name == "FakeNode" + + info = await node_api.api_get_info(node=fake_node) + assert info is not None + assert info.id == "node-id" + + channels = await node_api.api_get_channels(node=fake_node) + assert channels is not None + assert channels[0].id == "chan-1" + + channel = await node_api.api_get_channel("chan-1", node=fake_node) + assert channel is not None + assert channel.peer_id == "peer-1" + + created = await node_api.api_create_channel( + node=fake_node, + peer_id="peer-1", + funding_amount=10_000, + push_amount=100, + fee_rate=5, + ) + assert created.output_index == 0 + + deleted = await node_api.api_delete_channel( + short_id="123x1x0", + funding_txid=None, + output_index=None, + force=True, + node=fake_node, + ) + assert deleted is not None + assert deleted[0].id == "chan-1" + + await node_api.api_set_channel_fees( + "chan-1", + node=fake_node, + fee_ppm=42, + fee_base_msat=7, + ) + assert fake_node.fees_updated == ("chan-1", 7, 42) + + payments = await node_api.api_get_payments(node=fake_node, filters=Filters()) + assert payments is not None + assert payments.total == 1 + + invoices = await node_api.api_get_invoices(node=fake_node, filters=Filters()) + assert invoices is not None + assert invoices.total == 1 + + peers = await node_api.api_get_peers(node=fake_node) + assert peers[0].id == "peer-1" + + connect = await node_api.api_connect_peer( + uri="peer-1@127.0.0.1:9735", node=fake_node + ) + assert connect["uri"] == "peer-1@127.0.0.1:9735" + + disconnect = await node_api.api_disconnect_peer("peer-1", node=fake_node) + assert disconnect["peer_id"] == "peer-1" + + rank = await node_api.api_get_1ml_stats(node=fake_node) + assert rank is not None + assert rank["channelcount"] == 2 + finally: + settings.lnbits_node_ui_transactions = original_transactions + + +@pytest.mark.anyio +async def test_node_api_transactions_and_rank_errors( + settings: Settings, + mocker: MockerFixture, +): + fake_node = FakeNode() + original_transactions = settings.lnbits_node_ui_transactions + settings.lnbits_node_ui_transactions = False + + request = httpx.Request("GET", f"https://1ml.com/node/{uuid4().hex}/json") + mocker.patch( + "lnbits.core.views.node_api.httpx.AsyncClient", + return_value=MockHTTPClient( + MockHTTPResponse( + {}, + status_error=httpx.HTTPStatusError( + "not found", request=request, response=httpx.Response(404) + ), + ) + ), + ) + + try: + with pytest.raises(Exception) as payments: + await node_api.api_get_payments(node=fake_node, filters=Filters()) + assert payments.value.status_code == 503 + + with pytest.raises(Exception) as invoices: + await node_api.api_get_invoices(node=fake_node, filters=Filters()) + assert invoices.value.status_code == 503 + + with pytest.raises(Exception) as rank: + await node_api.api_get_1ml_stats(node=fake_node) + assert rank.value.status_code == 404 + assert rank.value.detail == "Node not found on 1ml.com" + finally: + settings.lnbits_node_ui_transactions = original_transactions diff --git a/tests/api/test_payment_api.py b/tests/api/test_payment_api.py new file mode 100644 index 000000000..d4f7eab21 --- /dev/null +++ b/tests/api/test_payment_api.py @@ -0,0 +1,187 @@ +import json +from hashlib import sha256 +from uuid import uuid4 + +import pytest +from fastapi import HTTPException + +from lnbits.core.crud.payments import create_payment +from lnbits.core.models import Account, CreateInvoice, PaymentState +from lnbits.core.models.payments import CancelInvoice, CreatePayment, SettleInvoice +from lnbits.core.models.users import AccountId +from lnbits.core.models.wallets import KeyType, WalletTypeInfo +from lnbits.core.services.payments import create_wallet_invoice +from lnbits.core.views.payment_api import ( + api_all_payments_paginated, + api_payments_cancel, + api_payments_counting_stats, + api_payments_daily_stats, + api_payments_fee_reserve, + api_payments_settle, + api_payments_wallets_stats, +) +from lnbits.core.services.users import create_user_account +from lnbits.db import Filters +from lnbits.wallets.base import InvoiceResponse + +ZERO_AMOUNT_INVOICE = ( + "lnbc1pnsu5z3pp57getmdaxhg5kc9yh2a2qsh7cjf4gnccgkw0qenm8vsqv50w7s" + "ygqdqj0fjhymeqv9kk7atwwscqzzsxqyz5vqsp5e2yyqcp0a3ujeesp24ya0glej" + "srh703md8mrx0g2lyvjxy5w27ss9qxpqysgqyjreasng8a086kpkczv48er5c6l5" + "73aym6ynrdl9nkzqnag49vt3sjjn8qdfq5cr6ha0vrdz5c5r3v4aghndly0hplmv" + "6hjxepwp93cq398l3s" +) + + +async def _create_payment( + wallet_id: str, + *, + amount_msat: int, + status: PaymentState = PaymentState.SUCCESS, + payment_hash: str | None = None, + tag: str | None = None, +) -> str: + checking_id = f"checking_{uuid4().hex[:8]}" + await create_payment( + checking_id=checking_id, + data=CreatePayment( + wallet_id=wallet_id, + payment_hash=payment_hash or uuid4().hex, + bolt11=f"bolt11_{checking_id}", + amount_msat=amount_msat, + memo=f"payment_{checking_id}", + extra={"tag": tag} if tag else {}, + ), + status=status, + ) + return checking_id + + +@pytest.mark.anyio +async def test_payment_api_stats_and_all_paginated(admin_user): + first_user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + second_user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + first_wallet = first_user.wallets[0] + second_wallet = second_user.wallets[0] + + await _create_payment(first_wallet.id, amount_msat=2_000, tag="coffee") + await _create_payment(first_wallet.id, amount_msat=-1_000, tag="coffee") + await _create_payment(second_wallet.id, amount_msat=5_000, tag="books") + + count_stats = await api_payments_counting_stats( + count_by="tag", + filters=Filters(limit=20), + account_id=AccountId(id=first_user.id), + ) + assert any(item.field == "coffee" for item in count_stats) + assert all(item.field != "books" for item in count_stats) + + wallet_stats = await api_payments_wallets_stats( + filters=Filters(limit=20), account_id=AccountId(id=first_user.id) + ) + assert any(item.wallet_id == first_wallet.id for item in wallet_stats) + assert all(item.wallet_id != second_wallet.id for item in wallet_stats) + + daily_stats = await api_payments_daily_stats( + account_id=AccountId(id=first_user.id), + filters=Filters(limit=20), + ) + assert daily_stats + assert daily_stats[0].payments_count >= 1 + + regular_page = await api_all_payments_paginated( + filters=Filters(limit=20), account_id=AccountId(id=first_user.id) + ) + assert regular_page.total >= 2 + assert all(payment.wallet_id == first_wallet.id for payment in regular_page.data) + + admin_page = await api_all_payments_paginated( + filters=Filters(limit=50), account_id=AccountId(id=admin_user.id) + ) + wallet_ids = {payment.wallet_id for payment in admin_page.data} + assert first_wallet.id in wallet_ids + assert second_wallet.id in wallet_ids + + +@pytest.mark.anyio +async def test_payment_api_fee_reserve_and_hold_invoice_actions(mocker): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = user.wallets[0] + + invoice = await create_wallet_invoice( + wallet.id, CreateInvoice(out=False, amount=42, memo="reserve") + ) + reserve = await api_payments_fee_reserve(invoice.bolt11) + assert json.loads(reserve.body)["fee_reserve"] >= 0 + + with pytest.raises(HTTPException, match="Invoice has no amount."): + await api_payments_fee_reserve(ZERO_AMOUNT_INVOICE) + + preimage = "11" * 32 + payment_hash = sha256(bytes.fromhex(preimage)).hexdigest() + await _create_payment( + wallet.id, + amount_msat=1_000, + payment_hash=payment_hash, + status=PaymentState.PENDING, + ) + settle_mock = mocker.patch( + "lnbits.core.views.payment_api.settle_hold_invoice", + mocker.AsyncMock( + return_value=InvoiceResponse( + ok=True, + checking_id="settled", + preimage=preimage, + ) + ), + ) + + settled = await api_payments_settle( + SettleInvoice(preimage=preimage), + WalletTypeInfo(key_type=KeyType.admin, wallet=wallet), + ) + assert settled.success is True + settle_mock.assert_awaited_once() + + cancel_hash = (uuid4().hex * 2)[:64] + await _create_payment( + wallet.id, + amount_msat=2_000, + payment_hash=cancel_hash, + status=PaymentState.PENDING, + ) + cancel_mock = mocker.patch( + "lnbits.core.views.payment_api.cancel_hold_invoice", + mocker.AsyncMock( + return_value=InvoiceResponse( + ok=False, + checking_id="cancelled", + error_message="cancelled", + ) + ), + ) + + cancelled = await api_payments_cancel( + CancelInvoice(payment_hash=cancel_hash), + WalletTypeInfo(key_type=KeyType.admin, wallet=wallet), + ) + assert cancelled.failed is True + cancel_mock.assert_awaited_once() diff --git a/tests/api/test_tinyurl_api.py b/tests/api/test_tinyurl_api.py new file mode 100644 index 000000000..931ccf5ef --- /dev/null +++ b/tests/api/test_tinyurl_api.py @@ -0,0 +1,71 @@ +from http import HTTPStatus + +import pytest +from httpx import AsyncClient + + +@pytest.mark.anyio +async def test_tinyurl_api_create_get_redirect_and_delete( + client: AsyncClient, + adminkey_headers_from: dict[str, str], + inkey_headers_from: dict[str, str], + inkey_headers_to: dict[str, str], +): + created = await client.post( + "/api/v1/tinyurl", + params={"url": "https://example.com/landing", "endless": "true"}, + headers=adminkey_headers_from, + ) + assert created.status_code == HTTPStatus.OK + tinyurl = created.json() + assert tinyurl["url"] == "https://example.com/landing" + assert tinyurl["endless"] is True + + fetched = await client.get( + f"/api/v1/tinyurl/{tinyurl['id']}", + headers=inkey_headers_from, + ) + assert fetched.status_code == HTTPStatus.OK + assert fetched.json()["id"] == tinyurl["id"] + + wrong_wallet = await client.get( + f"/api/v1/tinyurl/{tinyurl['id']}", + headers=inkey_headers_to, + ) + assert wrong_wallet.status_code == HTTPStatus.NOT_FOUND + assert wrong_wallet.json()["detail"] == "Unable to fetch tinyurl" + + redirect = await client.get(f"/t/{tinyurl['id']}") + assert redirect.status_code == HTTPStatus.TEMPORARY_REDIRECT + assert redirect.headers["location"] == "https://example.com/landing" + + deleted = await client.delete( + f"/api/v1/tinyurl/{tinyurl['id']}", + headers=adminkey_headers_from, + ) + assert deleted.status_code == HTTPStatus.OK + assert deleted.json()["deleted"] is True + + missing_redirect = await client.get(f"/t/{tinyurl['id']}") + assert missing_redirect.status_code == HTTPStatus.NOT_FOUND + + +@pytest.mark.anyio +async def test_tinyurl_api_reuses_existing_entries_for_same_wallet( + client: AsyncClient, + adminkey_headers_from: dict[str, str], +): + first = await client.post( + "/api/v1/tinyurl", + params={"url": "https://example.com/reused"}, + headers=adminkey_headers_from, + ) + second = await client.post( + "/api/v1/tinyurl", + params={"url": "https://example.com/reused"}, + headers=adminkey_headers_from, + ) + + assert first.status_code == HTTPStatus.OK + assert second.status_code == HTTPStatus.OK + assert first.json()["id"] == second.json()["id"] diff --git a/tests/api/test_user_api.py b/tests/api/test_user_api.py new file mode 100644 index 000000000..ad0c5ce67 --- /dev/null +++ b/tests/api/test_user_api.py @@ -0,0 +1,100 @@ +from uuid import uuid4 + +import pytest +from httpx import AsyncClient + +from lnbits.core.crud.wallets import create_wallet, get_wallet, get_wallets +from lnbits.core.models import UpdateBalance +from lnbits.core.models.users import Account +from lnbits.core.services.users import create_user_account +from lnbits.core.views.user_api import api_users_create_user_wallet +from lnbits.settings import settings + + +@pytest.mark.anyio +async def test_user_api_toggle_admin_and_update_balance( + http_client: AsyncClient, superuser_token: str +): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + wallet = user.wallets[0] + + promote = await http_client.put( + f"/users/api/v1/user/{user.id}/admin", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert promote.status_code == 200 + assert settings.is_admin_user(user.id) is True + + demote = await http_client.put( + f"/users/api/v1/user/{user.id}/admin", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert demote.status_code == 200 + assert settings.is_admin_user(user.id) is False + + balance = await http_client.put( + "/users/api/v1/balance", + headers={"Authorization": f"Bearer {superuser_token}"}, + json=UpdateBalance(id=wallet.id, amount=7).dict(), + ) + assert balance.status_code == 200 + assert balance.json()["success"] is True + + updated_wallet = await get_wallet(wallet.id) + assert updated_wallet is not None + assert updated_wallet.balance == 7 + + +@pytest.mark.anyio +async def test_user_api_get_wallets_and_delete_all_wallets( + http_client: AsyncClient, superuser_token: str +): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + extra_wallet = await create_wallet(user_id=user.id, wallet_name="spare") + + wallets = await http_client.get( + f"/users/api/v1/user/{user.id}/wallet", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert wallets.status_code == 200 + wallet_ids = {wallet["id"] for wallet in wallets.json()} + assert extra_wallet.id in wallet_ids + + deleted = await http_client.delete( + f"/users/api/v1/user/{user.id}/wallets", + headers={"Authorization": f"Bearer {superuser_token}"}, + ) + assert deleted.status_code == 200 + assert deleted.json()["success"] is True + + active_wallets = await get_wallets(user.id, deleted=False) + assert active_wallets == [] + + +@pytest.mark.anyio +async def test_user_api_create_wallet_validates_currency(): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + + with pytest.raises(ValueError, match="Currency 'INVALID' not allowed."): + await api_users_create_user_wallet(user.id, name="invalid", currency="INVALID") + + wallet = await api_users_create_user_wallet(user.id, name="eur wallet", currency="EUR") + assert wallet.currency == "EUR" diff --git a/tests/api/test_wallet_api.py b/tests/api/test_wallet_api.py new file mode 100644 index 000000000..4087c67b2 --- /dev/null +++ b/tests/api/test_wallet_api.py @@ -0,0 +1,184 @@ +from uuid import uuid4 + +import pytest +from httpx import AsyncClient + +from lnbits.core.crud.wallets import create_wallet, get_wallet +from lnbits.core.models.users import Account +from lnbits.core.services.users import create_user_account + + +def _admin_headers(adminkey: str) -> dict[str, str]: + return {"X-Api-Key": adminkey, "Content-type": "application/json"} + + +@pytest.mark.anyio +async def test_wallet_api_share_invite_reject_accept_and_delete(http_client: AsyncClient): + owner = await create_user_account( + Account( + id=uuid4().hex, + username=f"owner_{uuid4().hex[:8]}", + email=f"owner_{uuid4().hex[:8]}@lnbits.com", + ) + ) + invited = await create_user_account( + Account( + id=uuid4().hex, + username=f"invited_{uuid4().hex[:8]}", + email=f"invited_{uuid4().hex[:8]}@lnbits.com", + ) + ) + source_wallet = owner.wallets[0] + owner_headers = _admin_headers(source_wallet.adminkey) + + invite = await http_client.put( + "/api/v1/wallet/share/invite", + headers=owner_headers, + json={ + "username": invited.username, + "permissions": ["view-payments"], + "status": "invite_sent", + }, + ) + assert invite.status_code == 200 + share_request = invite.json() + assert share_request["request_id"] + + reject = await http_client.delete( + f"/api/v1/wallet/share/invite/{share_request['request_id']}?usr={invited.id}" + ) + assert reject.status_code == 200 + assert reject.json()["success"] is True + + removed_share = await http_client.delete( + f"/api/v1/wallet/share/{share_request['request_id']}", + headers=owner_headers, + ) + assert removed_share.status_code == 200 + assert removed_share.json()["success"] is True + + invite = await http_client.put( + "/api/v1/wallet/share/invite", + headers=owner_headers, + json={ + "username": invited.username, + "permissions": ["view-payments", "receive-payments"], + "status": "invite_sent", + }, + ) + assert invite.status_code == 200 + share_request = invite.json() + + create_shared = await http_client.post( + f"/api/v1/wallet?usr={invited.id}", + json={ + "name": "shared", + "wallet_type": "lightning-shared", + "shared_wallet_id": source_wallet.id, + }, + ) + assert create_shared.status_code == 200 + mirror_wallet = create_shared.json() + assert mirror_wallet["shared_wallet_id"] == source_wallet.id + + approve = await http_client.put( + "/api/v1/wallet/share", + headers=owner_headers, + json={ + "username": invited.username, + "shared_with_wallet_id": mirror_wallet["id"], + "permissions": ["view-payments", "receive-payments"], + "status": "approved", + }, + ) + assert approve.status_code == 200 + assert approve.json()["status"] == "approved" + + delete_share = await http_client.delete( + f"/api/v1/wallet/share/{share_request['request_id']}", + headers=owner_headers, + ) + assert delete_share.status_code == 200 + assert delete_share.json()["success"] is True + assert await get_wallet(mirror_wallet["id"]) is None + + +@pytest.mark.anyio +async def test_wallet_api_paginated_update_reset_and_store_paylinks( + http_client: AsyncClient, +): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + extra_wallet = await create_wallet(user_id=user.id, wallet_name="second") + first_wallet = user.wallets[0] + + page = await http_client.get(f"/api/v1/wallet/paginated?usr={user.id}&limit=10") + assert page.status_code == 200 + assert page.json()["total"] >= 2 + + renamed = await http_client.put( + "/api/v1/wallet/renamed-wallet", + headers=_admin_headers(first_wallet.adminkey), + ) + assert renamed.status_code == 200 + assert renamed.json()["name"] == "renamed-wallet" + + original_admin_key = extra_wallet.adminkey + reset = await http_client.put(f"/api/v1/wallet/reset/{extra_wallet.id}?usr={user.id}") + assert reset.status_code == 200 + assert reset.json()["adminkey"] != original_admin_key + + stored = await http_client.put( + f"/api/v1/wallet/stored_paylinks/{extra_wallet.id}", + headers=_admin_headers(reset.json()["adminkey"]), + json={ + "links": [ + { + "lnurl": "alice@example.com", + "label": "Alice", + } + ] + }, + ) + assert stored.status_code == 200 + assert stored.json()[0]["lnurl"] == "alice@example.com" + + forbidden = await http_client.put( + f"/api/v1/wallet/stored_paylinks/{extra_wallet.id}", + headers=_admin_headers(first_wallet.adminkey), + json={"links": []}, + ) + assert forbidden.status_code == 403 + + updated = await http_client.patch( + "/api/v1/wallet", + headers=_admin_headers(first_wallet.adminkey), + json={"icon": "bolt", "color": "amber", "pinned": True}, + ) + assert updated.status_code == 200 + assert updated.json()["extra"]["icon"] == "bolt" + assert updated.json()["extra"]["color"] == "amber" + assert updated.json()["extra"]["pinned"] is True + + +@pytest.mark.anyio +async def test_wallet_api_shared_wallet_requires_source_id(http_client: AsyncClient): + user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + + response = await http_client.post( + f"/api/v1/wallet?usr={user.id}", + json={"wallet_type": "lightning-shared"}, + ) + assert response.status_code == 400 + assert response.json()["detail"] == "Shared wallet ID is required for shared wallets." diff --git a/tests/api/test_websocket_api.py b/tests/api/test_websocket_api.py new file mode 100644 index 000000000..fdd14ced2 --- /dev/null +++ b/tests/api/test_websocket_api.py @@ -0,0 +1,31 @@ +from unittest.mock import AsyncMock + +from pytest_mock.plugin import MockerFixture + + +def test_websocket_api_connects_and_updates(test_client): + with test_client.websocket_connect("/api/v1/ws/demo-item") as websocket: + response = test_client.post("/api/v1/ws/demo-item", params={"data": "hello"}) + assert response.status_code == 200 + assert response.json() == {"sent": True, "data": "hello"} + assert websocket.receive_text() == "hello" + + response = test_client.get("/api/v1/ws/demo-item/world") + assert response.status_code == 200 + assert response.json() == {"sent": True, "data": "world"} + assert websocket.receive_text() == "world" + + +def test_websocket_api_reports_send_failures(test_client, mocker: MockerFixture): + mocker.patch( + "lnbits.core.views.websocket_api.websocket_manager.send", + AsyncMock(side_effect=RuntimeError("boom")), + ) + + post_response = test_client.post("/api/v1/ws/demo-item", params={"data": "oops"}) + assert post_response.status_code == 200 + assert post_response.json() == {"sent": False, "data": "oops"} + + get_response = test_client.get("/api/v1/ws/demo-item/oops") + assert get_response.status_code == 200 + assert get_response.json() == {"sent": False, "data": "oops"}