229 lines
9.0 KiB
Python
229 lines
9.0 KiB
Python
import json
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
|
|
def _install_web_stubs():
|
|
if "fastapi" in sys.modules:
|
|
return
|
|
|
|
class _HTTPException(Exception):
|
|
def __init__(self, status_code=None, detail=None):
|
|
super().__init__(detail)
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
|
|
class _FastAPI:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def mount(self, *args, **kwargs):
|
|
return None
|
|
|
|
def add_middleware(self, *args, **kwargs):
|
|
return None
|
|
|
|
def __getattr__(self, _name):
|
|
def _decorator_factory(*args, **kwargs):
|
|
def _decorator(func):
|
|
return func
|
|
|
|
return _decorator
|
|
|
|
return _decorator_factory
|
|
|
|
class _BaseModel:
|
|
pass
|
|
|
|
class _StaticFiles:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
class _Jinja2Templates:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
class _BaseHTTPMiddleware:
|
|
pass
|
|
|
|
fastapi_module = types.ModuleType("fastapi")
|
|
fastapi_module.FastAPI = _FastAPI
|
|
fastapi_module.HTTPException = _HTTPException
|
|
sys.modules["fastapi"] = fastapi_module
|
|
|
|
responses_module = types.ModuleType("fastapi.responses")
|
|
responses_module.HTMLResponse = object
|
|
responses_module.RedirectResponse = object
|
|
|
|
class _JSONResponse:
|
|
def __init__(self, content=None, status_code=200):
|
|
self.content = content
|
|
self.status_code = status_code
|
|
self.body = json.dumps(content or {}).encode("utf-8")
|
|
|
|
responses_module.JSONResponse = _JSONResponse
|
|
sys.modules["fastapi.responses"] = responses_module
|
|
|
|
staticfiles_module = types.ModuleType("fastapi.staticfiles")
|
|
staticfiles_module.StaticFiles = _StaticFiles
|
|
sys.modules["fastapi.staticfiles"] = staticfiles_module
|
|
|
|
templating_module = types.ModuleType("fastapi.templating")
|
|
templating_module.Jinja2Templates = _Jinja2Templates
|
|
sys.modules["fastapi.templating"] = templating_module
|
|
|
|
requests_module = types.ModuleType("fastapi.requests")
|
|
requests_module.Request = object
|
|
sys.modules["fastapi.requests"] = requests_module
|
|
|
|
pydantic_module = types.ModuleType("pydantic")
|
|
pydantic_module.BaseModel = _BaseModel
|
|
sys.modules["pydantic"] = pydantic_module
|
|
|
|
starlette_base_module = types.ModuleType("starlette.middleware.base")
|
|
starlette_base_module.BaseHTTPMiddleware = _BaseHTTPMiddleware
|
|
sys.modules["starlette.middleware.base"] = starlette_base_module
|
|
|
|
starlette_middleware_module = types.ModuleType("starlette.middleware")
|
|
starlette_middleware_module.base = starlette_base_module
|
|
sys.modules["starlette.middleware"] = starlette_middleware_module
|
|
|
|
starlette_module = types.ModuleType("starlette")
|
|
starlette_module.middleware = starlette_middleware_module
|
|
sys.modules["starlette"] = starlette_module
|
|
|
|
|
|
_install_web_stubs()
|
|
from sovran_systemsos_web import server
|
|
|
|
|
|
class FakeJSONResponse:
|
|
def __init__(self, content=None, status_code=200):
|
|
self.content = content
|
|
self.status_code = status_code
|
|
self.body = json.dumps(content or {}).encode("utf-8")
|
|
|
|
|
|
class WalletConnectionsRegistryTests(unittest.TestCase):
|
|
def test_node_role_feature_allow_list_includes_nwc_wallets(self):
|
|
self.assertIn("nwc-wallets", server.ROLE_FEATURES["node"])
|
|
self.assertNotIn("nwc-wallets", server.ROLE_FEATURES["desktop"])
|
|
|
|
def test_wallet_connections_feature_metadata(self):
|
|
feat = next(f for f in server.FEATURE_REGISTRY if f["id"] == "nwc-wallets")
|
|
self.assertEqual(feat["name"], "Wallet Connections")
|
|
self.assertTrue(feat["needs_domain"])
|
|
self.assertEqual(feat["domain_name"], "lightning")
|
|
self.assertEqual(
|
|
[(p["port"], p["protocol"]) for p in feat["port_requirements"]],
|
|
[("80", "TCP"), ("443", "TCP")],
|
|
)
|
|
|
|
|
|
class WalletConnectionsBehaviorTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_alias_validation_rules(self):
|
|
self.assertTrue(server._nwc_validate_alias("app_1"))
|
|
self.assertTrue(server._nwc_validate_alias("a-1"))
|
|
self.assertFalse(server._nwc_validate_alias("_bad"))
|
|
self.assertFalse(server._nwc_validate_alias("Upper"))
|
|
self.assertFalse(server._nwc_validate_alias("a" * 33))
|
|
|
|
async def test_pairing_uri_returned_only_on_create(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
state_file = Path(td) / "state.json"
|
|
domain_file = Path(td) / "lightning"
|
|
domain_file.write_text("pay.example.com\n")
|
|
with (
|
|
patch.object(server, "JSONResponse", FakeJSONResponse),
|
|
patch.object(server, "NWC_STATE_FILE", str(state_file)),
|
|
patch.object(server, "NWC_DOMAIN_FILE", str(domain_file)),
|
|
patch.object(server, "_nwc_test_address", return_value={"ok": False, "error": "public_endpoint_unreachable"}),
|
|
patch.object(server, "_generate_qr_base64", return_value="data:image/png;base64,abc"),
|
|
):
|
|
req = types.SimpleNamespace(
|
|
name="My Wallet",
|
|
alias="my-wallet",
|
|
access_preset="receive_only",
|
|
spending_limit_sats=None,
|
|
)
|
|
create_resp = await server.api_nwc_create_wallet(req)
|
|
create_body = json.loads(create_resp.body.decode("utf-8"))
|
|
self.assertIn("pairing_uri", create_body)
|
|
self.assertEqual(create_body.get("pairing_qrcode"), "data:image/png;base64,abc")
|
|
|
|
list_resp = await server.api_nwc_wallets()
|
|
self.assertEqual(len(list_resp["wallets"]), 1)
|
|
self.assertNotIn("pairing_uri", list_resp["wallets"][0])
|
|
self.assertNotIn("pairing_qrcode", list_resp["wallets"][0])
|
|
|
|
async def test_create_reports_public_verification_success(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
state_file = Path(td) / "state.json"
|
|
domain_file = Path(td) / "lightning"
|
|
domain_file.write_text("pay.example.com\n")
|
|
with (
|
|
patch.object(server, "JSONResponse", FakeJSONResponse),
|
|
patch.object(server, "NWC_STATE_FILE", str(state_file)),
|
|
patch.object(server, "NWC_DOMAIN_FILE", str(domain_file)),
|
|
patch.object(server, "_nwc_test_address", return_value={"ok": True}),
|
|
):
|
|
req = types.SimpleNamespace(
|
|
name="Wallet Success",
|
|
alias="wallet-success",
|
|
access_preset="receive_only",
|
|
spending_limit_sats=None,
|
|
)
|
|
create_resp = await server.api_nwc_create_wallet(req)
|
|
create_body = json.loads(create_resp.body.decode("utf-8"))
|
|
self.assertTrue(create_body["result"]["wallet_created"])
|
|
self.assertTrue(create_body["result"]["public_endpoint_verification"]["ok"])
|
|
|
|
async def test_lnurl_callback_rejects_appid_mismatch(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
state_file = Path(td) / "state.json"
|
|
domain_file = Path(td) / "lightning"
|
|
domain_file.write_text("pay.example.com\n")
|
|
state_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"wallets": [
|
|
{
|
|
"id": "wallet-1",
|
|
"pubkey": "pubkey-1",
|
|
"name": "Wallet 1",
|
|
"alias": "wallet1",
|
|
"access_preset": "receive_only",
|
|
"spending_limit_sats": None,
|
|
"remaining_budget_sats": None,
|
|
"balance_sats": 0,
|
|
"dust_msat": 0,
|
|
"pending_transactions": 0,
|
|
"min_sendable_msat": 1000,
|
|
"max_sendable_msat": 1000000,
|
|
"created_at": 0,
|
|
}
|
|
]
|
|
}
|
|
)
|
|
)
|
|
with (
|
|
patch.object(server, "JSONResponse", FakeJSONResponse),
|
|
patch.object(server, "NWC_STATE_FILE", str(state_file)),
|
|
patch.object(server, "NWC_DOMAIN_FILE", str(domain_file)),
|
|
patch.object(server, "_nwc_issue_invoice", return_value={"appId": "wrong", "pr": "lnbc1..."}),
|
|
):
|
|
resp = await server.api_lnurl_callback("wallet1", amount="1000")
|
|
body = json.loads(resp.body.decode("utf-8"))
|
|
self.assertEqual(resp.status_code, 502)
|
|
self.assertEqual(body["error"], "invoice_attribution_failed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|