Unit tests for FastAPI branch

Run via `make test`

tests/conftest.py

@@ -1,12 +1,28 @@
+import asyncio
 import pytest
-
+from httpx import AsyncClient
 from lnbits.app import create_app
+from lnbits.commands import migrate_databases
+from lnbits.settings import HOST, PORT
+import tests.mocks
 
+# use session scope to run once before and once after all tests
+@pytest.fixture(scope="session")
+def app():
+    # yield and pass the app to the test
+    app = create_app()
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(migrate_databases())
+    yield app
+    # get the current event loop and gracefully stop any running tasks
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(loop.shutdown_asyncgens())
+    loop.close()
 
 @pytest.fixture
-async def client():
-    app = create_app()
-    app.config["TESTING"] = True
-
-    async with app.test_client() as client:
-        yield client
+async def client(app):
+    client = AsyncClient(app=app, base_url=f'http://{HOST}:{PORT}')
+    # yield and pass the client to the test
+    yield client
+    # close the async client after the test has finished
+    await client.aclose()

tests/core/test_views.py

@@ -1,6 +0,0 @@
-import pytest
-
-
-async def test_homepage(client):
-    r = await client.get("/")
-    assert b"Add a new wallet" in await r.get_data()

tests/core/views/test_generic.py

@@ -0,0 +1,7 @@
+import pytest
+from tests.conftest import client
+
+@pytest.mark.asyncio
+async def test_core_views_generic(client):
+    response = await client.get("/")
+    assert response.status_code == 200

tests/extensions/bleskomat/conftest.py

@@ -0,0 +1,57 @@
+import json
+import pytest
+import secrets
+from lnbits.core.crud import create_account, create_wallet
+from lnbits.extensions.bleskomat.crud import create_bleskomat, create_bleskomat_lnurl
+from lnbits.extensions.bleskomat.models import CreateBleskomat
+from lnbits.extensions.bleskomat.helpers import generate_bleskomat_lnurl_secret, generate_bleskomat_lnurl_signature, prepare_lnurl_params, query_to_signing_payload
+from lnbits.extensions.bleskomat.exchange_rates import exchange_rate_providers
+
+exchange_rate_providers["dummy"] = {
+    "name": "dummy",
+    "domain": None,
+    "api_url": None,
+    "getter": lambda data, replacements: str(1e8),# 1 BTC = 100000000 sats
+}
+
+@pytest.fixture
+async def bleskomat():
+    user = await create_account()
+    wallet = await create_wallet(user_id=user.id, wallet_name="bleskomat_test")
+    data = CreateBleskomat(
+        name="Test Bleskomat",
+        fiat_currency="EUR",
+        exchange_rate_provider="dummy",
+        fee="0"
+    )
+    bleskomat = await create_bleskomat(data=data, wallet_id=wallet.id)
+    return bleskomat
+
+@pytest.fixture
+async def lnurl(bleskomat):
+    query = {
+        "tag": "withdrawRequest",
+        "nonce": secrets.token_hex(10),
+        "tag": "withdrawRequest",
+        "minWithdrawable": "50000",
+        "maxWithdrawable": "50000",
+        "defaultDescription": "test valid sig",
+    }
+    tag = query["tag"]
+    params = prepare_lnurl_params(tag, query)
+    payload = query_to_signing_payload(query)
+    signature = generate_bleskomat_lnurl_signature(
+        payload=payload,
+        api_key_secret=bleskomat.api_key_secret,
+        api_key_encoding=bleskomat.api_key_encoding
+    )
+    secret = generate_bleskomat_lnurl_secret(bleskomat.api_key_id, signature)
+    params = json.JSONEncoder().encode(params)
+    lnurl = await create_bleskomat_lnurl(
+        bleskomat=bleskomat, secret=secret, tag=tag, params=params, uses=1
+    )
+    return {
+        "bleskomat": bleskomat,
+        "lnurl": lnurl,
+        "secret": secret,
+    }

tests/extensions/bleskomat/test_lnurl_api.py

@@ -0,0 +1,120 @@
+import pytest
+import secrets
+from lnbits.core.crud import get_wallet
+from lnbits.settings import HOST, PORT
+from lnbits.extensions.bleskomat.crud import get_bleskomat_lnurl
+from lnbits.extensions.bleskomat.helpers import generate_bleskomat_lnurl_signature, query_to_signing_payload
+from tests.conftest import client
+from tests.helpers import credit_wallet
+from tests.extensions.bleskomat.conftest import bleskomat, lnurl
+from tests.mocks import WALLET
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_missing_secret(client):
+    response = await client.get("/bleskomat/u")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ERROR", "reason": "Missing secret"}
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_invalid_secret(client):
+    response = await client.get("/bleskomat/u?k1=invalid-secret")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ERROR", "reason": "Invalid secret"}
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_unknown_api_key(client):
+    query = {
+        "id": "does-not-exist",
+        "nonce": secrets.token_hex(10),
+        "tag": "withdrawRequest",
+        "minWithdrawable": "1",
+        "maxWithdrawable": "1",
+        "defaultDescription": "",
+        "f": "EUR",
+    }
+    payload = query_to_signing_payload(query)
+    signature = "xxx"# not checked, so doesn't matter
+    response = await client.get(f'/bleskomat/u?{payload}&signature={signature}')
+    assert response.status_code == 200
+    assert response.json() == {"status": "ERROR", "reason": "Unknown API key"}
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_invalid_signature(client, bleskomat):
+    query = {
+        "id": bleskomat.api_key_id,
+        "nonce": secrets.token_hex(10),
+        "tag": "withdrawRequest",
+        "minWithdrawable": "1",
+        "maxWithdrawable": "1",
+        "defaultDescription": "",
+        "f": "EUR",
+    }
+    payload = query_to_signing_payload(query)
+    signature = "invalid"
+    response = await client.get(f'/bleskomat/u?{payload}&signature={signature}')
+    assert response.status_code == 200
+    assert response.json() == {"status": "ERROR", "reason": "Invalid API key signature"}
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_valid_signature(client, bleskomat):
+    query = {
+        "id": bleskomat.api_key_id,
+        "nonce": secrets.token_hex(10),
+        "tag": "withdrawRequest",
+        "minWithdrawable": "1",
+        "maxWithdrawable": "1",
+        "defaultDescription": "test valid sig",
+        "f": "EUR",# tests use the dummy exchange rate provider
+    }
+    payload = query_to_signing_payload(query)
+    signature = generate_bleskomat_lnurl_signature(
+        payload=payload,
+        api_key_secret=bleskomat.api_key_secret,
+        api_key_encoding=bleskomat.api_key_encoding
+    )
+    response = await client.get(f'/bleskomat/u?{payload}&signature={signature}')
+    assert response.status_code == 200
+    data = response.json()
+    assert data["tag"] == "withdrawRequest"
+    assert data["minWithdrawable"] == 1000
+    assert data["maxWithdrawable"] == 1000
+    assert data["defaultDescription"] == "test valid sig"
+    assert data["callback"] == f'http://{HOST}:{PORT}/bleskomat/u'
+    k1 = data["k1"]
+    lnurl = await get_bleskomat_lnurl(secret=k1)
+    assert lnurl
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_action_insufficient_balance(client, lnurl):
+    bleskomat = lnurl["bleskomat"]
+    secret = lnurl["secret"]
+    pr = "lntb500n1pseq44upp5xqd38rgad72lnlh4gl339njlrsl3ykep82j6gj4g02dkule7k54qdqqcqzpgxqyz5vqsp5h0zgewuxdxcl2rnlumh6g520t4fr05rgudakpxm789xgjekha75s9qyyssq5vhwsy9knhfeqg0wn6hcnppwmum8fs3g3jxkgw45havgfl6evchjsz3s8e8kr6eyacz02szdhs7v5lg0m7wehd5rpf6yg8480cddjlqpae52xu"
+    WALLET.pay_invoice.reset_mock()
+    response = await client.get(f'/bleskomat/u?k1={secret}&pr={pr}')
+    assert response.status_code == 200
+    assert response.json() == {"status": "ERROR", "reason": "Failed to pay invoice: Insufficient balance."}
+    wallet = await get_wallet(bleskomat.wallet)
+    assert wallet.balance_msat == 0
+    bleskomat_lnurl = await get_bleskomat_lnurl(secret)
+    assert bleskomat_lnurl.has_uses_remaining() == True
+    WALLET.pay_invoice.assert_not_called()
+
+@pytest.mark.asyncio
+async def test_bleskomat_lnurl_api_action_success(client, lnurl):
+    bleskomat = lnurl["bleskomat"]
+    secret = lnurl["secret"]
+    pr = "lntb500n1pseq44upp5xqd38rgad72lnlh4gl339njlrsl3ykep82j6gj4g02dkule7k54qdqqcqzpgxqyz5vqsp5h0zgewuxdxcl2rnlumh6g520t4fr05rgudakpxm789xgjekha75s9qyyssq5vhwsy9knhfeqg0wn6hcnppwmum8fs3g3jxkgw45havgfl6evchjsz3s8e8kr6eyacz02szdhs7v5lg0m7wehd5rpf6yg8480cddjlqpae52xu"
+    await credit_wallet(
+        wallet_id=bleskomat.wallet,
+        amount=100000,
+    )
+    wallet = await get_wallet(bleskomat.wallet)
+    assert wallet.balance_msat == 100000
+    WALLET.pay_invoice.reset_mock()
+    response = await client.get(f'/bleskomat/u?k1={secret}&pr={pr}')
+    assert response.json() == {"status": "OK"}
+    wallet = await get_wallet(bleskomat.wallet)
+    assert wallet.balance_msat == 50000
+    bleskomat_lnurl = await get_bleskomat_lnurl(secret)
+    assert bleskomat_lnurl.has_uses_remaining() == False
+    WALLET.pay_invoice.assert_called_once_with(pr)

tests/helpers.py

@@ -0,0 +1,19 @@
+import hashlib
+import secrets
+from lnbits.core.crud import create_payment
+
+async def credit_wallet(wallet_id: str, amount: int):
+    preimage = secrets.token_hex(32)
+    m = hashlib.sha256()
+    m.update(f"{preimage}".encode())
+    payment_hash = m.hexdigest()
+    await create_payment(
+        wallet_id=wallet_id,
+        payment_request="",
+        payment_hash=payment_hash,
+        checking_id=payment_hash,
+        preimage=preimage,
+        memo="",
+        amount=amount,# msat
+        pending=False,# not pending, so it will increase the wallet's balance
+    )

tests/mocks.py

@@ -0,0 +1,36 @@
+from mock import AsyncMock
+from lnbits import bolt11
+from lnbits.wallets.base import (
+    StatusResponse,
+    InvoiceResponse,
+    PaymentResponse,
+    PaymentStatus,
+    Wallet,
+)
+from lnbits.settings import WALLET
+
+WALLET.status = AsyncMock(return_value=StatusResponse(
+    "",# no error
+    1000000,# msats
+))
+WALLET.create_invoice = AsyncMock(return_value=InvoiceResponse(
+    True,# ok
+    "6621aafbdd7709ca6eea6203f362d64bd7cb2911baa91311a176b3ecaf2274bd",# checking_id (i.e. payment_hash)
+    "lntb1u1psezhgspp5vcs6477awuyu5mh2vgplxckkf0tuk2g3h253xydpw6e7etezwj7sdqqcqzpgxqyz5vqsp5dxpw8zs77hw5pla4wz4mfujllyxtlpu443auur2uxqdrs8q2h56q9qyyssq65zk30ylmygvv5y4tuwalnf3ttnqjn57ef6rmcqg0s53akem560jh8ptemjcmytn3lrlatw4hv9smg88exv3v4f4lqnp96s0psdrhxsp6pp75q",# payment_request
+    "",# no error
+))
+def pay_invoice_side_effect(payment_request: str):
+    invoice = bolt11.decode(payment_request)
+    return PaymentResponse(
+        True,# ok
+        invoice.payment_hash,# checking_id (i.e. payment_hash)
+        0,# fee_msat
+        "",# no error
+    )
+WALLET.pay_invoice = AsyncMock(side_effect=pay_invoice_side_effect)
+WALLET.get_invoice_status = AsyncMock(return_value=PaymentStatus(
+    True,# paid
+))
+WALLET.get_payment_status = AsyncMock(return_value=PaymentStatus(
+    True,# paid
+))