feat: add hub-managed Wallet Connections scaffolding
This commit is contained in:
copilot-swe-agent[bot]
committed by
GitHub
GitHub
parent
5860f40e82
commit
9673f6733a
@@ -0,0 +1,225 @@
|
||||
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"}),
|
||||
):
|
||||
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)
|
||||
|
||||
list_resp = await server.api_nwc_wallets()
|
||||
self.assertEqual(len(list_resp["wallets"]), 1)
|
||||
self.assertNotIn("pairing_uri", 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()
|
||||
Reference in New Issue
Block a user