Sovran_Systems/Sovran_SystemsOS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Address code review feedback: BOLT11 regex, error messages, PORT env var, gofmt, test helpers

Browse Source 
- nwc_hub_manager.py: Fix BOLT11 regex to accept only valid prefixes (bc/tb/bcrt/tbs); add path to timeout error message; add early-exit to _get_app_pending_txs pagination
- nwc_lnurl_service.py: Allow NWC_LNURL_PORT env var override for port
- nwc-wallets.nix: Remove || true from gofmt so build fails on syntax errors
- test_wallet_connections.py: Extract _call_body() helper; use c.args[] access
This commit is contained in:
copilot-swe-agent[bot]
2026-07-27 03:21:40 +00:00
committed by GitHub
parent 5ebd51e5f9
commit 0e13779773
4 changed files with 44 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/sovran_systemsos_web/nwc_hub_manager.py
+29 -6
View File
@@ -206,7 +206,7 @@ class AlbyHubManager:
time.sleep(2)
raise AlbyHubError(
"dependency_unavailable",
f"Timed out waiting for required file",
f"Timed out waiting for required file: {path}",
)
def _wait_for_hub_api(self, timeout: int = 120) -> None:
@@ -531,10 +531,33 @@ class AlbyHubManager:
return int(budget.get("usedBudget", 0) or 0)
def _get_app_pending_txs(self, app_id: int) -> list[dict]:
txs = self._paginate(
f"/api/apps/{app_id}/transactions?limit={{limit}}&offset={{offset}}"
)
return [t for t in txs if t.get("state", "").lower() in ("pending",)]
"""Return pending transactions for the app.
Paginates only as far as needed: stops after finding the first
pending transaction since the caller rejects *any* pending tx.
"""
path_tmpl = f"/api/apps/{app_id}/transactions?limit={{limit}}&offset={{offset}}"
page_size = 100
token = self.ensure_ready()
offset = 0
pending: list[dict] = []
while True:
path = path_tmpl.format(limit=page_size, offset=offset)
page = self._request("GET", path, token=token)
if isinstance(page, list):
items = page
elif isinstance(page, dict):
items = page.get("transactions") or []
else:
items = []
for t in items:
if t.get("state", "").lower() == "pending":
pending.append(t)
return pending # early exit: one is enough to block
if len(items) < page_size:
break
offset += page_size
return pending
def drain_wallet(self, identifier: str) -> dict:
"""Drain all whole-satoshi funds from an isolated app to the primary wallet.
@@ -689,7 +712,7 @@ class AlbyHubManager:
raise AlbyHubError("invoice_creation_failed", "Hub returned empty invoice.")
# Require a valid BOLT11 prefix (mainnet, testnet, signet, regtest)
if not re.match(r"^ln[a-z]{2,6}[0-9]", invoice, re.IGNORECASE):
if not re.match(r"^ln(bc|tb|bcrt|tbs)[0-9]", invoice, re.IGNORECASE):
raise AlbyHubError(
"invalid_invoice", "Hub returned a non-BOLT11 invoice string."
)
app/sovran_systemsos_web/nwc_lnurl_service.py
+1 -1
View File
@@ -32,7 +32,7 @@ logger = logging.getLogger(__name__)
# ── Configuration ─────────────────────────────────────────────────
LNURL_BIND_HOST = "127.0.0.1"
LNURL_PORT = 8181
LNURL_PORT = int(os.environ.get("NWC_LNURL_PORT", "8181"))
DOMAIN_FILE = "/var/lib/domains/lightning"
NWC_ALIAS_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,31}$")
app/tests/test_wallet_connections.py
+13 -8
View File
@@ -414,6 +414,11 @@ class ManagerCreateTests(unittest.TestCase):
m._request = MagicMock(side_effect=_request)
return m
@staticmethod
def _call_body(call):
"""Return the ``body`` kwarg from a MagicMock call_args."""
return call.kwargs.get("body") or {}
def test_create_returns_pairing_uri_once(self):
m = self._mgr()
result = m.create_wallet("New Wallet", "new", "receive_only", None)
@@ -425,9 +430,9 @@ class ManagerCreateTests(unittest.TestCase):
m.create_wallet("W", "w", "receive_only", None)
create_call = next(
c for c in m._request.call_args_list
if c[0][0] == "POST" and c[0][1] == "/api/apps"
if c.args[0] == "POST" and c.args[1] == "/api/apps"
)
body = create_call[1].get("body") or create_call[0][2] if len(create_call[0]) > 2 else create_call.kwargs.get("body")
body = self._call_body(create_call)
self.assertTrue(body.get("isolated"))
def test_create_receive_only_scopes(self):
@@ -435,9 +440,9 @@ class ManagerCreateTests(unittest.TestCase):
m.create_wallet("W", "w", "receive_only", None)
create_call = next(
c for c in m._request.call_args_list
if c[0][0] == "POST" and "/api/apps" in c[0][1]
if c.args[0] == "POST" and "/api/apps" in c.args[1]
)
body = create_call.kwargs.get("body") or (create_call[0][2] if len(create_call[0]) > 2 else {})
body = self._call_body(create_call)
self.assertNotIn("pay_invoice", body.get("scopes", []))
for scope in mgr.RECEIVE_ONLY_SCOPES:
self.assertIn(scope, body.get("scopes", []))
@@ -447,9 +452,9 @@ class ManagerCreateTests(unittest.TestCase):
m.create_wallet("W", "w", "send_receive_limited", 5000)
create_call = next(
c for c in m._request.call_args_list
if c[0][0] == "POST" and "/api/apps" in c[0][1]
if c.args[0] == "POST" and "/api/apps" in c.args[1]
)
body = create_call.kwargs.get("body") or (create_call[0][2] if len(create_call[0]) > 2 else {})
body = self._call_body(create_call)
self.assertIn("pay_invoice", body.get("scopes", []))
def test_create_includes_managed_metadata(self):
@@ -457,9 +462,9 @@ class ManagerCreateTests(unittest.TestCase):
m.create_wallet("W", "w", "receive_only", None)
create_call = next(
c for c in m._request.call_args_list
if c[0][0] == "POST" and "/api/apps" in c[0][1]
if c.args[0] == "POST" and "/api/apps" in c.args[1]
)
body = create_call.kwargs.get("body") or (create_call[0][2] if len(create_call[0]) > 2 else {})
body = self._call_body(create_call)
meta = body.get("metadata", {})
self.assertEqual(meta.get("app_store_app_id"), "uncle-jim")
self.assertEqual(meta.get("lnurl_alias"), "w")