AaronFiore/electrum
1
0
Fork 0
Code Activity

test_lnpeer: add unit test for direct payment to trampoline node

Browse Source
This commit is contained in:
ThomasV
2022-09-20 11:28:51 +02:00
parent 478046289a
commit 9251f482b5
1 changed files with 20 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
electrum/tests/test_lnpeer.py
View File

@@ -715,11 +715,15 @@ class TestPeer(TestCaseForTestnet):
with self.assertRaises(SuccessfulTest): with self.assertRaises(SuccessfulTest):
run(f()) run(f())
@needs_test_with_all_chacha20_implementations def _test_simple_payment(self, trampoline: bool):
def test_payment(self):
"""Alice pays Bob a single HTLC via direct channel.""" """Alice pays Bob a single HTLC via direct channel."""
alice_channel, bob_channel = create_test_channels() alice_channel, bob_channel = create_test_channels()
p1, p2, w1, w2, _q1, _q2 = self.prepare_peers(alice_channel, bob_channel) p1, p2, w1, w2, _q1, _q2 = self.prepare_peers(alice_channel, bob_channel)
async def turn_on_trampoline_alice():
if w1.network.channel_db:
w1.network.channel_db.stop()
await w1.network.channel_db.stopped_event.wait()
w1.network.channel_db = None
async def pay(lnaddr, pay_req): async def pay(lnaddr, pay_req):
self.assertEqual(PR_UNPAID, w2.get_payment_status(lnaddr.paymenthash)) self.assertEqual(PR_UNPAID, w2.get_payment_status(lnaddr.paymenthash))
result, log = await w1.pay_invoice(pay_req) result, log = await w1.pay_invoice(pay_req)
@@ -727,6 +731,8 @@ class TestPeer(TestCaseForTestnet):
self.assertEqual(PR_PAID, w2.get_payment_status(lnaddr.paymenthash)) self.assertEqual(PR_PAID, w2.get_payment_status(lnaddr.paymenthash))
raise PaymentDone() raise PaymentDone()
async def f(): async def f():
if trampoline:
await turn_on_trampoline_alice()
async with OldTaskGroup() as group: async with OldTaskGroup() as group:
await group.spawn(p1._message_loop()) await group.spawn(p1._message_loop())
await group.spawn(p1.htlc_switch()) await group.spawn(p1.htlc_switch())
@@ -737,9 +743,21 @@ class TestPeer(TestCaseForTestnet):
invoice_features = lnaddr.get_features() invoice_features = lnaddr.get_features()
self.assertFalse(invoice_features.supports(LnFeatures.BASIC_MPP_OPT)) self.assertFalse(invoice_features.supports(LnFeatures.BASIC_MPP_OPT))
await group.spawn(pay(lnaddr, pay_req)) await group.spawn(pay(lnaddr, pay_req))
# declare bob as trampoline node
electrum.trampoline._TRAMPOLINE_NODES_UNITTESTS = {
'bob': LNPeerAddr(host="127.0.0.1", port=9735, pubkey=w2.node_keypair.pubkey),
}
with self.assertRaises(PaymentDone): with self.assertRaises(PaymentDone):
run(f()) run(f())
@needs_test_with_all_chacha20_implementations
def test_simple_payment(self):
self._test_simple_payment(trampoline=False)
@needs_test_with_all_chacha20_implementations
def test_simple_payment_trampoline(self):
self._test_simple_payment(trampoline=True)
@needs_test_with_all_chacha20_implementations @needs_test_with_all_chacha20_implementations
def test_payment_race(self): def test_payment_race(self):
"""Alice and Bob pay each other simultaneously. """Alice and Bob pay each other simultaneously.