Automated tests (#566)

* return error for wrong key

* payment check use key dependency

* more expressive error

* re-add optional key

* more tests

* more

* more granular

* more testing

* custom event_loop

* tests work

* fix lots of mypy errors

* test_public_api

* both files

* remove unused import

* tests

* tests working

* rm empty file

* minimal test

* set FAKE_WALLET_SECRET="ToTheMoon1"

* set FAKE_WALLET_SECRET="ToTheMoon1"

* trial and error

* trial and error

* test postgres

* test postgres

* test postgres

* test postgres

* test postgres

* test postgres

* test build

* skip mypy

tests/conftest.py

@@ -4,26 +4,124 @@ 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
+from lnbits.core.views.api import api_payments_create_invoice, CreateInvoiceData
+
+from lnbits.core.crud import create_account, create_wallet, get_wallet
+from tests.helpers import credit_wallet, get_random_invoice_data
+
+from lnbits.db import Database
+from lnbits.core.models import User, Wallet, Payment, BalanceCheck
+from typing import Tuple
+
+
 @pytest.fixture(scope="session")
-def app():
-    # yield and pass the app to the test
-    app = create_app()
+def event_loop():
     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())
+    yield loop
     loop.close()
 
 
-@pytest.fixture
+# use session scope to run once before and once after all tests
+@pytest.fixture(scope="session")
+def app(event_loop):
+    app = create_app()
+    # use redefined version of the event loop for scope="session"
+    # loop = asyncio.get_event_loop()
+    loop = event_loop
+    loop.run_until_complete(migrate_databases())
+    yield app
+    # # get the current event loop and gracefully stop any running tasks
+    # loop = event_loop
+    loop.run_until_complete(loop.shutdown_asyncgens())
+    # loop.close()
+
+
+@pytest.fixture(scope="session")
 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()
+
+
+@pytest.fixture(scope="session")
+async def db():
+    yield Database("database")
+
+
+@pytest.fixture(scope="session")
+async def from_user_wallet():
+    user = await create_account()
+    wallet = await create_wallet(user_id=user.id, wallet_name="test_wallet_from")
+    await credit_wallet(
+        wallet_id=wallet.id,
+        amount=99999999,
+    )
+    # print("new from_user_wallet:", wallet)
+    yield user, wallet
+
+
+@pytest.fixture(scope="session")
+async def to_user_wallet():
+    user = await create_account()
+    wallet = await create_wallet(user_id=user.id, wallet_name="test_wallet_to")
+    await credit_wallet(
+        wallet_id=wallet.id,
+        amount=99999999,
+    )
+    # print("new to_user_wallet:", wallet)
+    yield user, wallet
+
+
+@pytest.fixture(scope="session")
+async def inkey_headers_from(from_user_wallet):
+    _, wallet = from_user_wallet
+    yield {
+        "X-Api-Key": wallet.inkey,
+        "Content-type": "application/json",
+    }
+
+
+@pytest.fixture(scope="session")
+async def adminkey_headers_from(from_user_wallet):
+    _, wallet = from_user_wallet
+    yield {
+        "X-Api-Key": wallet.adminkey,
+        "Content-type": "application/json",
+    }
+
+
+@pytest.fixture(scope="session")
+async def inkey_headers_to(to_user_wallet):
+    _, wallet = to_user_wallet
+    yield {
+        "X-Api-Key": wallet.inkey,
+        "Content-type": "application/json",
+    }
+
+
+@pytest.fixture(scope="session")
+async def adminkey_headers_to(to_user_wallet):
+    _, wallet = to_user_wallet
+    yield {
+        "X-Api-Key": wallet.adminkey,
+        "Content-type": "application/json",
+    }
+
+
+@pytest.fixture(scope="session")
+async def invoice(to_user_wallet):
+    _, wallet = to_user_wallet
+    data = await get_random_invoice_data()
+    invoiceData = CreateInvoiceData(**data)
+    # print("--------- New invoice!")
+    # print("wallet:")
+    # print(wallet)
+    stuff_lock = asyncio.Lock()
+    async with stuff_lock:
+        invoice = await api_payments_create_invoice(invoiceData, wallet)
+    await asyncio.sleep(1)
+    # print("invoice")
+    # print(invoice)
+    yield invoice
+    del invoice

tests/core/views/test_api.py

@@ -0,0 +1,118 @@
+import pytest
+from lnbits.core.crud import get_wallet
+
+from ...helpers import get_random_invoice_data
+
+# check if the client is working
+@pytest.mark.asyncio
+async def test_core_views_generic(client):
+    response = await client.get("/")
+    assert response.status_code == 200
+
+
+# check GET /api/v1/wallet: wallet info
+@pytest.mark.asyncio
+async def test_get_wallet(client, inkey_headers_to):
+    response = await client.get("/api/v1/wallet", headers=inkey_headers_to)
+    assert response.status_code < 300
+
+
+# check POST /api/v1/payments: invoice creation
+@pytest.mark.asyncio
+async def test_create_invoice(client, inkey_headers_to):
+    data = await get_random_invoice_data()
+    response = await client.post(
+        "/api/v1/payments", json=data, headers=inkey_headers_to
+    )
+    assert response.status_code < 300
+    assert "payment_hash" in response.json()
+    assert len(response.json()["payment_hash"]) == 64
+    assert "payment_request" in response.json()
+    assert "checking_id" in response.json()
+    assert len(response.json()["checking_id"])
+    return response.json()
+
+
+# check POST /api/v1/payments: make payment
+@pytest.mark.asyncio
+async def test_pay_invoice(client, invoice, adminkey_headers_from):
+    data = {"out": True, "bolt11": invoice["payment_request"]}
+    response = await client.post(
+        "/api/v1/payments", json=data, headers=adminkey_headers_from
+    )
+    assert response.status_code < 300
+    assert len(response.json()["payment_hash"]) == 64
+    assert len(response.json()["checking_id"]) > 0
+
+
+# check GET /api/v1/payments/<hash>: payment status
+@pytest.mark.asyncio
+async def test_check_payment_without_key(client, invoice):
+    # check the payment status
+    response = await client.get(f"/api/v1/payments/{invoice['payment_hash']}")
+    assert response.status_code < 300
+    assert response.json()["paid"] == True
+    assert invoice
+    # not key, that's why no "details"
+    assert "details" not in response.json()
+
+
+# check GET /api/v1/payments/<hash>: payment status
+@pytest.mark.asyncio
+async def test_check_payment_with_key(client, invoice, inkey_headers_to):
+    # check the payment status
+    response = await client.get(
+        f"/api/v1/payments/{invoice['payment_hash']}", headers=inkey_headers_to
+    )
+    assert response.status_code < 300
+    assert response.json()["paid"] == True
+    assert invoice
+    # with key, that's why with "details"
+    assert "details" in response.json()
+
+
+# check POST /api/v1/payments: payment with wrong key type
+@pytest.mark.asyncio
+async def test_pay_invoice_wrong_key(client, invoice, adminkey_headers_from):
+    data = {"out": True, "bolt11": invoice["payment_request"]}
+    # try payment with wrong key
+    wrong_adminkey_headers = adminkey_headers_from.copy()
+    wrong_adminkey_headers["X-Api-Key"] = "wrong_key"
+    response = await client.post(
+        "/api/v1/payments", json=data, headers=wrong_adminkey_headers
+    )
+    assert response.status_code >= 300  # should fail
+
+
+# check POST /api/v1/payments: payment with invoice key [should fail]
+@pytest.mark.asyncio
+async def test_pay_invoice_invoicekey(client, invoice, inkey_headers_from):
+    data = {"out": True, "bolt11": invoice["payment_request"]}
+    # try payment with invoice key
+    response = await client.post(
+        "/api/v1/payments", json=data, headers=inkey_headers_from
+    )
+    assert response.status_code >= 300  # should fail
+
+
+# check POST /api/v1/payments: payment with admin key [should pass]
+@pytest.mark.asyncio
+async def test_pay_invoice_adminkey(client, invoice, adminkey_headers_from):
+    data = {"out": True, "bolt11": invoice["payment_request"]}
+    # try payment with admin key
+    response = await client.post(
+        "/api/v1/payments", json=data, headers=adminkey_headers_from
+    )
+    assert response.status_code < 300  # should pass
+
+
+# check POST /api/v1/payments/decode
+@pytest.mark.asyncio
+async def test_decode_invoice(client, invoice):
+    data = {"data": invoice["payment_request"]}
+    response = await client.post(
+        "/api/v1/payments/decode",
+        json=data,
+    )
+    assert response.status_code < 300
+    assert response.json()["payment_hash"] == invoice["payment_hash"]

tests/core/views/test_public_api.py

@@ -0,0 +1,36 @@
+import pytest
+from lnbits.core.crud import get_wallet
+
+# check if the client is working
+@pytest.mark.asyncio
+async def test_core_views_generic(client):
+    response = await client.get("/")
+    assert response.status_code == 200
+
+
+# check GET /public/v1/payment/{payment_hash}: correct hash [should pass]
+@pytest.mark.asyncio
+async def test_api_public_payment_longpolling(client, invoice):
+    response = await client.get(f"/public/v1/payment/{invoice['payment_hash']}")
+    assert response.status_code < 300
+    assert response.json()["status"] == "paid"
+
+
+# check GET /public/v1/payment/{payment_hash}: wrong hash [should fail]
+@pytest.mark.asyncio
+async def test_api_public_payment_longpolling_wrong_hash(client, invoice):
+    response = await client.get(
+        f"/public/v1/payment/{invoice['payment_hash'] + '0'*64}"
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Payment does not exist."
+
+
+# check GET /.well-known/lnurlp/{username}: wrong username [should fail]
+@pytest.mark.asyncio
+async def test_lnaddress_wrong_hash(client):
+    username = "wrong_name"
+    response = await client.get(f"/.well-known/lnurlp/{username}")
+    assert response.status_code == 200
+    assert response.json()["status"] == "ERROR"
+    assert response.json()["reason"] == "Address not found."

tests/helpers.py

@@ -1,5 +1,7 @@
 import hashlib
 import secrets
+import random
+import string
 from lnbits.core.crud import create_payment
 
 
@@ -14,7 +16,18 @@ async def credit_wallet(wallet_id: str, amount: int):
         payment_hash=payment_hash,
         checking_id=payment_hash,
         preimage=preimage,
-        memo="",
+        memo=f"funding_test_{get_random_string(5)}",
         amount=amount,  # msat
         pending=False,  # not pending, so it will increase the wallet's balance
     )
+
+
+def get_random_string(N=10):
+    return "".join(
+        random.SystemRandom().choice(string.ascii_uppercase + string.digits)
+        for _ in range(10)
+    )
+
+
+async def get_random_invoice_data():
+    return {"out": False, "amount": 10, "memo": f"test_memo_{get_random_string(10)}"}

tests/mocks.py

@@ -1,3 +1,4 @@
+import time
 from mock import AsyncMock
 from lnbits import bolt11
 from lnbits.wallets.base import (
@@ -9,20 +10,51 @@ from lnbits.wallets.base import (
 )
 from lnbits.settings import WALLET
 
+from lnbits.wallets.fake import FakeWallet
+
+from .helpers import get_random_string
+
+# primitive event loop for generate_mock_invoice()
+def drive(c):
+    while True:
+        try:
+            c.send(None)
+        except StopIteration as e:
+            return e.value
+
+
+# generates an invoice with FakeWallet
+async def generate_mock_invoice(**x):
+    invoice = await FakeWallet.create_invoice(
+        FakeWallet(), amount=10, memo=f"mock invoice {get_random_string()}"
+    )
+    return invoice
+
+
 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
-    )
-)
+
+WALLET.create_invoice = generate_mock_invoice
+
+# NOTE: This mock fails since it yields the same invoice multiple
+# times which makes the db throw an error due to uniqueness contraints
+# on the checking ID
+
+# # finally we await it
+# invoice = drive(generate_mock_invoice())
+
+# WALLET.create_invoice = AsyncMock(
+#     return_value=InvoiceResponse(
+#         True,  # ok
+#         invoice.checking_id,  # checking_id (i.e. payment_hash)
+#         invoice.payment_request,  # payment_request
+#         "",  # no error
+#     )
+# )
 
 
 def pay_invoice_side_effect(