mirror of
https://github.com/lnbits/lnbits.git
synced 2026-08-14 12:52:47 +02:00
313 lines
9.7 KiB
Python
313 lines
9.7 KiB
Python
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 import update_wallet_balance
|
|
from lnbits.core.services.users import create_user_account
|
|
from lnbits.settings import settings
|
|
|
|
|
|
@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_custom_lightning_address_owner_rules(
|
|
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",
|
|
)
|
|
)
|
|
wallet = user.wallets[0]
|
|
headers = _admin_headers(wallet.adminkey)
|
|
|
|
settings.lnbits_ln_address_mode = "core_first"
|
|
settings.lnbits_allow_custom_wallet_lightning_addresses = False
|
|
disabled = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=headers,
|
|
json={"lightning_address": "custom.name"},
|
|
)
|
|
assert disabled.status_code == 403
|
|
|
|
settings.lnbits_allow_custom_wallet_lightning_addresses = True
|
|
settings.lnbits_wallet_lightning_address_blacklist = ["admin"]
|
|
blacklisted = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=headers,
|
|
json={"lightning_address": "admin"},
|
|
)
|
|
assert blacklisted.status_code == 400
|
|
|
|
invalid = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=headers,
|
|
json={"lightning_address": "custom+tag"},
|
|
)
|
|
assert invalid.status_code == 400
|
|
|
|
existing_wallet = await create_wallet(
|
|
user_id=user.id,
|
|
wallet_name="existing lightning address",
|
|
)
|
|
existing = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=_admin_headers(existing_wallet.adminkey),
|
|
json={"lightning_address": "pay.link"},
|
|
)
|
|
assert existing.status_code == 200
|
|
|
|
conflict = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=headers,
|
|
json={"lightning_address": "pay.link"},
|
|
)
|
|
assert conflict.status_code == 400
|
|
|
|
updated = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=headers,
|
|
json={"lightning_address": "custom.name"},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["lightning_address"] == "custom.name"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wallet_api_custom_lightning_address_charges_fee(
|
|
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",
|
|
)
|
|
)
|
|
fee_user = await create_user_account(
|
|
Account(
|
|
id=uuid4().hex,
|
|
username=f"fees_{uuid4().hex[:8]}",
|
|
email=f"fees_{uuid4().hex[:8]}@lnbits.com",
|
|
)
|
|
)
|
|
wallet = user.wallets[0]
|
|
fee_wallet = fee_user.wallets[0]
|
|
await update_wallet_balance(wallet=wallet, amount=2_000)
|
|
|
|
settings.lnbits_ln_address_mode = "core_first"
|
|
settings.lnbits_allow_custom_wallet_lightning_addresses = True
|
|
settings.lnbits_charge_wallet_lightning_addresses = True
|
|
settings.lnbits_wallet_lightning_address_price_sats = 1_000
|
|
settings.lnbits_service_fee_wallet = fee_wallet.id
|
|
|
|
updated = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=_admin_headers(wallet.adminkey),
|
|
json={"lightning_address": "paid.name"},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["lightning_address"] == "paid.name"
|
|
|
|
charged_wallet = await get_wallet(wallet.id)
|
|
credited_wallet = await get_wallet(fee_wallet.id)
|
|
assert charged_wallet
|
|
assert credited_wallet
|
|
assert charged_wallet.balance == 1_000
|
|
assert credited_wallet.balance == 1_000
|
|
|
|
settings.lnbits_service_fee_wallet = None
|
|
missing_fee_wallet = await http_client.patch(
|
|
"/api/v1/wallet",
|
|
headers=_admin_headers(wallet.adminkey),
|
|
json={"lightning_address": "paid.other"},
|
|
)
|
|
assert missing_fee_wallet.status_code == 400
|
|
assert missing_fee_wallet.json()["detail"] == (
|
|
"Lightning Address fee wallet is not configured."
|
|
)
|
|
|
|
|
|
@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."
|
|
)
|
|
|
|
|
|
def _admin_headers(adminkey: str) -> dict[str, str]:
|
|
return {"X-Api-Key": adminkey, "Content-type": "application/json"}
|