← Watch feed
tests: move ln graph definitions and prepare_invoice to lnhelpers.py
Public commit record
83/100 · Strong
What the developer wrote
Authored by f321x
tests: move ln graph definitions and prepare_invoice to lnhelpers.py
This makes them reusable from test files other than test_lnpeer and
removes the deferred test_lnpeer import in prepare_chans_and_peers_in_graph.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context✓ Mentions testing or verification
What changed, and why it matters
This commit is purely a test-code refactoring: it moves Lightning Network test helpers (graph definitions and an invoice preparation function) from one test file to another shared helper file. There is no change to production code, no security fix, and no vulnerability introduced.
Recommended action
No action required. This is a benign test refactoring.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The diff shows a pure relocation of test utilities. _GRAPH_DEFINITIONS, high_fee_channel, low_fee_channel, depleted_channel, and prepare_invoice are moved from tests/test_lnpeer.py to tests/lnhelpers.py. test_lnpeer.py now imports them from lnhelpers. The only functional change is removing a deferred import of test_lnpeer inside prepare_chans_and_peers_in_graph so it can use the shared _GRAPH_DEFINITIONS directly. No production logic is modified.
Changed components
tests/lnhelpers.pytests/test_lnpeer.pyInspect captured patch +185 / −185
diff --git a/tests/lnhelpers.py b/tests/lnhelpers.py
index a36f933..532c06e 100644
--- a/tests/lnhelpers.py
+++ b/tests/lnhelpers.py
@@ -1,6 +1,7 @@
import asyncio
import copy
import os
+from decimal import Decimal
from pprint import pformat
from typing import NamedTuple, Tuple, Dict, Mapping, TYPE_CHECKING, Sequence
@@ -11,16 +12,18 @@ from electrum import (
)
from electrum.coinchooser import PRNG
from electrum.network import ProxySettings
-from electrum.bolt11 import BOLT11Addr
+from electrum.bitcoin import COIN, sha256
+from electrum.bolt11 import encode_bolt11_invoice, BOLT11Addr, decode_bolt11_invoice
+from electrum.invoices import PR_UNPAID, Invoice, LN_EXPIRY_NEVER
from electrum.lnpeer import Peer
from electrum.lnutil import (
LnFeatures, PaymentFeeBudget, LOCAL, REMOTE, ChannelType, LocalConfig, RemoteConfig,
- OnlyPubkeyKeypair, secret_to_pubkey,
+ OnlyPubkeyKeypair, secret_to_pubkey, RECEIVED,
)
from electrum.lnchannel import ChannelState, Channel
from electrum.lnrouter import LNPathFinder
from electrum.channel_db import ChannelDB
-from electrum.lnworker import LNWallet, PaySession
+from electrum.lnworker import LNWallet, PaySession, PaymentInfo
from electrum.simple_config import SimpleConfig
from electrum.fee_policy import FeeTimeEstimates, FEE_ETA_TARGETS
from electrum.wallet import Standard_Wallet
@@ -31,6 +34,121 @@ if TYPE_CHECKING:
from . import ElectrumTestCase
+high_fee_channel = {
+ 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ 'local_base_fee_msat': 500_000,
+ 'local_fee_rate_millionths': 500,
+ 'remote_base_fee_msat': 500_000,
+ 'remote_fee_rate_millionths': 500,
+}
+
+low_fee_channel = {
+ 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ 'local_base_fee_msat': 1_000,
+ 'local_fee_rate_millionths': 1,
+ 'remote_base_fee_msat': 1_000,
+ 'remote_fee_rate_millionths': 1,
+}
+
+depleted_channel = {
+ 'local_balance_msat': 330 * 1000, # local pays anchors
+ 'remote_balance_msat': 10 * bitcoin.COIN * 1000,
+ 'local_base_fee_msat': 1_000,
+ 'local_fee_rate_millionths': 1,
+ 'remote_base_fee_msat': 1_000,
+ 'remote_fee_rate_millionths': 1,
+}
+
+_GRAPH_DEFINITIONS = {
+ # A -- B
+ 'single_chan' : {
+ 'alice': {
+ 'channels': {
+ 'bob': [
+ {
+ 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
+ },
+ ],
+ },
+ },
+ 'bob': {
+ },
+ },
+ # A
+ # high fee / \ low fee
+ # B C
+ # high fee \ / low fee
+ # D
+ 'square_graph': {
+ 'alice': {
+ 'channels': {
+ # we should use copies of channel definitions if
+ # we want to independently alter them in a test
+ 'bob': [high_fee_channel.copy()],
+ 'carol': [low_fee_channel.copy()],
+ },
+ },
+ 'bob': {
+ 'channels': {
+ 'dave': [high_fee_channel.copy()],
+ },
+ 'config': {
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_TRAMPOLINE_PAYMENTS: True,
+ },
+ },
+ 'carol': {
+ 'channels': {
+ 'dave': [low_fee_channel.copy()],
+ },
+ 'config': {
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_TRAMPOLINE_PAYMENTS: True,
+ },
+ },
+ 'dave': {
+ },
+ },
+ # A -- B -- C -- D -- E
+ 'line_graph': {
+ 'alice': {
+ 'channels': {
+ 'bob': [low_fee_channel.copy()],
+ },
+ },
+ 'bob': { # Trampoline Forwarder
+ 'channels': {
+ 'carol': [low_fee_channel.copy()],
+ },
+ 'config': {
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
+ },
+ },
+ 'carol': {
+ 'channels': {
+ 'dave': [low_fee_channel.copy()],
+ },
+ 'config': {
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
+ },
+ },
+ 'dave': { # Trampoline Forwarder
+ 'channels': {
+ 'edward': [low_fee_channel.copy()],
+ },
+ 'config': {
+ SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
+ },
+ },
+ 'edward': {
+ },
+ },
+}
+
+
class MockNetwork:
def __init__(self, *, config: SimpleConfig):
self.lnwatcher = None
@@ -225,6 +343,63 @@ def transport_pair(k1, k2, name1, name2):
return t1, t2
+def prepare_invoice(
+ w2: MockLNWallet, # receiver
+ *,
+ amount_msat=100_000_000,
+ include_routing_hints=False,
+ payment_preimage: bytes = None,
+ payment_hash: bytes = None,
+ invoice_features: LnFeatures = None,
+ min_final_cltv_delta: int = None,
+ expiry: int = None,
+) -> Tuple[BOLT11Addr, Invoice]:
+ amount_btc = amount_msat/Decimal(COIN*1000)
+ if payment_preimage is None and not payment_hash:
+ payment_preimage = os.urandom(32)
+ if payment_hash is None:
+ payment_hash = sha256(payment_preimage)
+ if payment_preimage:
+ w2.save_preimage(payment_hash, payment_preimage)
+ if include_routing_hints:
+ routing_hints = w2.calc_routing_hints_for_invoice(amount_msat)
+ else:
+ routing_hints = []
+ trampoline_hints = []
+ if invoice_features is None:
+ invoice_features = w2.features.for_bolt11_invoice()
+ if invoice_features.supports(LnFeatures.PAYMENT_SECRET_OPT):
+ payment_secret = w2.get_payment_secret(payment_hash)
+ else:
+ payment_secret = None
+ if min_final_cltv_delta is None:
+ min_final_cltv_delta = lnutil.MIN_FINAL_CLTV_DELTA_ACCEPTED
+ info = PaymentInfo(
+ payment_hash=payment_hash,
+ amount_msat=amount_msat,
+ direction=RECEIVED,
+ status=PR_UNPAID,
+ min_final_cltv_delta=min_final_cltv_delta,
+ expiry_delay=expiry or LN_EXPIRY_NEVER,
+ invoice_features=invoice_features,
+ )
+ w2.save_payment_info(info)
+ lnaddr1 = BOLT11Addr(
+ paymenthash=payment_hash,
+ amount=amount_btc,
+ tags=[
+ ('c', min_final_cltv_delta),
+ ('d', 'coffee'),
+ ('9', invoice_features),
+ ('x', expiry or 3600),
+ ] + routing_hints,
+ payment_secret=payment_secret,
+ )
+ invoice = encode_bolt11_invoice(lnaddr1, w2.node_keypair.privkey)
+ lnaddr2 = decode_bolt11_invoice(invoice) # unlike lnaddr1, this now has a pubkey set
+ return lnaddr2, Invoice.from_bech32(invoice)
+
+
def prepare_lnwallets(elec_test_case: 'ElectrumTestCase', graph_definition) -> Mapping[str, MockLNWallet]:
workers = {} # type: Dict[str, MockLNWallet]
for a, definition in graph_definition.items():
@@ -239,10 +414,8 @@ def prepare_chans_and_peers_in_graph(
workers: Dict[str, MockLNWallet] = None,
channels: dict[Tuple[str, str], list[Channel]] = None,
) -> Graph:
- from . import test_lnpeer
-
if graph_definition is None:
- graph_definition = test_lnpeer._GRAPH_DEFINITIONS['single_chan']
+ graph_definition = _GRAPH_DEFINITIONS['single_chan']
graph_definition = copy.deepcopy(graph_definition) # paranoia
# create workers
diff --git a/tests/test_lnpeer.py b/tests/test_lnpeer.py
index ca6ee29..a3e70ed 100644
--- a/tests/test_lnpeer.py
+++ b/tests/test_lnpeer.py
@@ -3,7 +3,6 @@ import dataclasses
import shutil
import copy
import tempfile
-from decimal import Decimal
import os
from contextlib import contextmanager
from collections import defaultdict
@@ -29,8 +28,8 @@ from electrum import constants
from electrum import bip32
from electrum.network import Network, ProxySettings
from electrum import simple_config, lnutil
-from electrum.bolt11 import encode_bolt11_invoice, BOLT11Addr, decode_bolt11_invoice
-from electrum.bitcoin import COIN, sha256
+from electrum.bolt11 import encode_bolt11_invoice, BOLT11Addr
+from electrum.bitcoin import sha256
from electrum.transaction import Transaction
from electrum.util import NetworkRetryManager, bfh, OldTaskGroup, EventListener, InvoiceError
from electrum.lnpeer import Peer
@@ -44,134 +43,17 @@ from electrum.lnworker import LNWallet, NoPathFound, SentHtlcInfo, PaySession, L
from electrum.lnmsg import encode_msg, decode_msg
from electrum import lnmsg
from electrum.logging import console_stderr_handler, Logger
-from electrum.lnworker import PaymentInfo
from electrum.lnonion import OnionFailureCode, OnionRoutingFailure, OnionHopsDataSingle, OnionPacket
from electrum.lnutil import LOCAL, REMOTE, UpdateAddHtlc, RecvMPPResolution, RevocationStore
-from electrum.invoices import PR_PAID, PR_UNPAID, Invoice, LN_EXPIRY_NEVER
+from electrum.invoices import PR_PAID, PR_UNPAID, Invoice
from electrum.interface import GracefulDisconnect
-from electrum.simple_config import SimpleConfig
from electrum.fee_policy import FeeTimeEstimates, FEE_ETA_TARGETS
from electrum.mpp_split import split_amount_normal
from electrum.wallet import Abstract_Wallet, Standard_Wallet
from .test_bitcoin import needs_test_with_all_chacha20_implementations
from . import ElectrumTestCase, restore_wallet_from_text__for_unittest, lnhelpers
-from .lnhelpers import Graph, MockLNWallet, create_test_channels
-
-
-high_fee_channel = {
- 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- 'local_base_fee_msat': 500_000,
- 'local_fee_rate_millionths': 500,
- 'remote_base_fee_msat': 500_000,
- 'remote_fee_rate_millionths': 500,
-}
-
-low_fee_channel = {
- 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- 'local_base_fee_msat': 1_000,
- 'local_fee_rate_millionths': 1,
- 'remote_base_fee_msat': 1_000,
- 'remote_fee_rate_millionths': 1,
-}
-
-depleted_channel = {
- 'local_balance_msat': 330 * 1000, # local pays anchors
- 'remote_balance_msat': 10 * bitcoin.COIN * 1000,
- 'local_base_fee_msat': 1_000,
- 'local_fee_rate_millionths': 1,
- 'remote_base_fee_msat': 1_000,
- 'remote_fee_rate_millionths': 1,
-}
-
-_GRAPH_DEFINITIONS = {
- # A -- B
- 'single_chan' : {
- 'alice': {
- 'channels': {
- 'bob': [
- {
- 'local_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- 'remote_balance_msat': 10 * bitcoin.COIN * 1000 // 2,
- },
- ],
- },
- },
- 'bob': {
- },
- },
- # A
- # high fee / \ low fee
- # B C
- # high fee \ / low fee
- # D
- 'square_graph': {
- 'alice': {
- 'channels': {
- # we should use copies of channel definitions if
- # we want to independently alter them in a test
- 'bob': [high_fee_channel.copy()],
- 'carol': [low_fee_channel.copy()],
- },
- },
- 'bob': {
- 'channels': {
- 'dave': [high_fee_channel.copy()],
- },
- 'config': {
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_TRAMPOLINE_PAYMENTS: True,
- },
- },
- 'carol': {
- 'channels': {
- 'dave': [low_fee_channel.copy()],
- },
- 'config': {
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_TRAMPOLINE_PAYMENTS: True,
- },
- },
- 'dave': {
- },
- },
- # A -- B -- C -- D -- E
- 'line_graph': {
- 'alice': {
- 'channels': {
- 'bob': [low_fee_channel.copy()],
- },
- },
- 'bob': { # Trampoline Forwarder
- 'channels': {
- 'carol': [low_fee_channel.copy()],
- },
- 'config': {
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
- },
- },
- 'carol': {
- 'channels': {
- 'dave': [low_fee_channel.copy()],
- },
- 'config': {
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
- },
- },
- 'dave': { # Trampoline Forwarder
- 'channels': {
- 'edward': [low_fee_channel.copy()],
- },
- 'config': {
- SimpleConfig.EXPERIMENTAL_LN_FORWARD_PAYMENTS: True,
- },
- },
- 'edward': {
- },
- },
-}
+from .lnhelpers import Graph, MockLNWallet, create_test_channels, low_fee_channel, depleted_channel
class PaymentDone(Exception): pass
@@ -210,68 +92,13 @@ class TestPeer(ElectrumTestCase):
def setUp(self):
super().setUp()
- self.GRAPH_DEFINITIONS = copy.deepcopy(_GRAPH_DEFINITIONS)
+ self.GRAPH_DEFINITIONS = copy.deepcopy(lnhelpers._GRAPH_DEFINITIONS)
async def asyncTearDown(self):
electrum.trampoline._TRAMPOLINE_NODES_UNITTESTS = {}
await super().asyncTearDown()
- @staticmethod
- def prepare_invoice(
- w2: MockLNWallet, # receiver
- *,
- amount_msat=100_000_000,
- include_routing_hints=False,
- payment_preimage: bytes = None,
- payment_hash: bytes = None,
- invoice_features: LnFeatures = None,
- min_final_cltv_delta: int = None,
- expiry: int = None,
- ) -> Tuple[BOLT11Addr, Invoice]:
- amount_btc = amount_msat/Decimal(COIN*1000)
- if payment_preimage is None and not payment_hash:
- payment_preimage = os.urandom(32)
- if payment_hash is None:
- payment_hash = sha256(payment_preimage)
- if payment_preimage:
- w2.save_preimage(payment_hash, payment_preimage)
- if include_routing_hints:
- routing_hints = w2.calc_routing_hints_for_invoice(amount_msat)
- else:
- routing_hints = []
- trampoline_hints = []
- if invoice_features is None:
- invoice_features = w2.features.for_bolt11_invoice()
- if invoice_features.supports(LnFeatures.PAYMENT_SECRET_OPT):
- payment_secret = w2.get_payment_secret(payment_hash)
- else:
- payment_secret = None
- if min_final_cltv_delta is None:
- min_final_cltv_delta = lnutil.MIN_FINAL_CLTV_DELTA_ACCEPTED
- info = PaymentInfo(
- payment_hash=payment_hash,
- amount_msat=amount_msat,
- direction=RECEIVED,
- status=PR_UNPAID,
- min_final_cltv_delta=min_final_cltv_delta,
- expiry_delay=expiry or LN_EXPIRY_NEVER,
- invoice_features=invoice_features,
- )
- w2.save_payment_info(info)
- lnaddr1 = BOLT11Addr(
- paymenthash=payment_hash,
- amount=amount_btc,
- tags=[
- ('c', min_final_cltv_delta),
- ('d', 'coffee'),
- ('9', invoice_features),
- ('x', expiry or 3600),
- ] + routing_hints,
- payment_secret=payment_secret,
- )
- invoice = encode_bolt11_invoice(lnaddr1, w2.node_keypair.privkey)
- lnaddr2 = decode_bolt11_invoice(invoice) # unlike lnaddr1, this now has a pubkey set
- return lnaddr2, Invoice.from_bech32(invoice)
+ prepare_invoice = staticmethod(lnhelpers.prepare_invoice)
async def _activate_trampoline(self, w: MockLNWallet):
if w.network.channel_db:
Risk score
Our methodology →Why this scored 15/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.