mirror of
https://github.com/LBRYFoundation/LBRY-Vault.git
synced 2025-08-23 17:47:31 +00:00
e56e849505
- replace undoing logic with new HTLCManager class - separate SENT/RECEIVED - move UpdateAddHtlc to lnutil
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
import unittest
|
|
from electrum.lnutil import RECEIVED, LOCAL, REMOTE, SENT, HTLCOwner
|
|
from electrum.lnhtlc import HTLCManager
|
|
from typing import NamedTuple
|
|
|
|
class H(NamedTuple):
|
|
owner : str
|
|
htlc_id : int
|
|
|
|
class TestHTLCManager(unittest.TestCase):
|
|
def test_race(self):
|
|
A = HTLCManager()
|
|
B = HTLCManager()
|
|
ah0, bh0 = H('A', 0), H('B', 0)
|
|
B.recv_htlc(A.send_htlc(ah0))
|
|
self.assertTrue(B.expect_sig[RECEIVED])
|
|
self.assertTrue(A.expect_sig[SENT])
|
|
self.assertFalse(B.expect_sig[SENT])
|
|
self.assertFalse(A.expect_sig[RECEIVED])
|
|
self.assertEqual(B.log[REMOTE]['locked_in'][0][LOCAL], 1)
|
|
A.recv_htlc(B.send_htlc(bh0))
|
|
self.assertTrue(B.expect_sig[RECEIVED])
|
|
self.assertTrue(A.expect_sig[SENT])
|
|
self.assertTrue(A.expect_sig[SENT])
|
|
self.assertTrue(B.expect_sig[RECEIVED])
|
|
self.assertEqual(B.current_htlcs(LOCAL), [])
|
|
self.assertEqual(A.current_htlcs(LOCAL), [])
|
|
self.assertEqual(B.pending_htlcs(LOCAL), [(RECEIVED, ah0)])
|
|
self.assertEqual(A.pending_htlcs(LOCAL), [(RECEIVED, bh0)])
|
|
A.send_ctx()
|
|
B.recv_ctx()
|
|
B.send_ctx()
|
|
A.recv_ctx()
|
|
self.assertEqual(B.pending_htlcs(LOCAL), [(RECEIVED, ah0), (SENT, bh0)][::-1])
|
|
self.assertEqual(A.pending_htlcs(LOCAL), [(RECEIVED, bh0), (SENT, ah0)][::-1])
|
|
B.send_rev()
|
|
A.recv_rev()
|
|
A.send_rev()
|
|
B.recv_rev()
|
|
self.assertEqual(B.current_htlcs(LOCAL), [(RECEIVED, ah0), (SENT, bh0)][::-1])
|
|
self.assertEqual(A.current_htlcs(LOCAL), [(RECEIVED, bh0), (SENT, ah0)][::-1])
|
|
|
|
def test_no_race(self):
|
|
A = HTLCManager()
|
|
B = HTLCManager()
|
|
B.recv_htlc(A.send_htlc(H('A', 0)))
|
|
self.assertEqual(len(B.pending_htlcs(REMOTE)), 1)
|
|
A.send_ctx()
|
|
B.recv_ctx()
|
|
B.send_rev()
|
|
A.recv_rev()
|
|
B.send_ctx()
|
|
A.recv_ctx()
|
|
A.send_rev()
|
|
B.recv_rev()
|
|
self.assertEqual(len(A.current_htlcs(LOCAL)), 1)
|
|
self.assertEqual(len(B.current_htlcs(LOCAL)), 1)
|
|
B.send_settle(0)
|
|
A.recv_settle(0)
|
|
self.assertEqual(A.htlcs_by_direction(REMOTE, RECEIVED), [H('A', 0)])
|
|
self.assertNotEqual(A.current_htlcs(LOCAL), [])
|
|
self.assertNotEqual(B.current_htlcs(REMOTE), [])
|
|
self.assertEqual(A.pending_htlcs(LOCAL), [])
|
|
self.assertEqual(B.pending_htlcs(REMOTE), [])
|
|
B.send_ctx()
|
|
A.recv_ctx()
|
|
A.send_rev()
|
|
B.recv_rev()
|
|
A.send_ctx()
|
|
B.recv_ctx()
|
|
B.send_rev()
|
|
A.recv_rev()
|
|
self.assertEqual(B.current_htlcs(LOCAL), [])
|
|
self.assertEqual(A.current_htlcs(LOCAL), [])
|
|
self.assertEqual(A.current_htlcs(REMOTE), [])
|
|
self.assertEqual(B.current_htlcs(REMOTE), [])
|
|
self.assertEqual(len(A.settled_htlcs(LOCAL)), 1)
|
|
self.assertEqual(len(A.sent_in_ctn(2)), 1)
|
|
self.assertEqual(len(B.received_in_ctn(2)), 1)
|
|
|
|
def test_settle_while_owing(self):
|
|
A = HTLCManager()
|
|
B = HTLCManager()
|
|
B.recv_htlc(A.send_htlc(H('A', 0)))
|
|
A.send_ctx()
|
|
B.recv_ctx()
|
|
B.send_rev()
|
|
A.recv_rev()
|
|
B.send_settle(0)
|
|
A.recv_settle(0)
|
|
self.assertEqual(B.pending_htlcs(REMOTE), [])
|
|
B.send_ctx()
|
|
A.recv_ctx()
|
|
A.send_rev()
|
|
B.recv_rev()
|