-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathchannelindirection_test.py
82 lines (72 loc) · 2.63 KB
/
channelindirection_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from channelindirection import ChannelInDirection
from enumtypes import FeeType
from lnmodel import Htlc
def test_set_get_fee():
cd = ChannelInDirection(num_slots=2)
cd.set_fee(FeeType.UPFRONT, base_fee=1, fee_rate=0.01)
cd.set_fee(FeeType.SUCCESS, base_fee=2, fee_rate=0.02)
assert cd.upfront_fee_function(100) == 2
assert cd.success_fee_function(100) == 4
def test_channel_direction():
cd = ChannelInDirection(num_slots=2)
# Before adding in-flight payments: all slots are free
assert(cd.all_slots_free())
assert(cd.get_num_slots() == 2)
t_0, htlc_0 = 1, Htlc("pid", 100, True)
t_1, htlc_1 = 5, Htlc("pid", 100, True)
t_2, htlc_2 = 6, Htlc("pid", 100, True)
t_3 = 4
# Push HTLC 0
has_slot, htlcs = cd.ensure_free_slots(t_0)
assert(has_slot and not htlcs)
cd.push_htlc(t_0, htlc_0)
# Queue is not full yet
assert(cd.get_num_slots_occupied() == 1)
assert(not cd.all_slots_busy())
# Push HTLC 1
has_slot, htlcs = cd.ensure_free_slots(t_1)
assert(has_slot and not htlcs)
cd.push_htlc(t_1, htlc_1)
# Now the queue is full
assert(cd.get_num_slots_occupied() == 2)
assert(cd.all_slots_busy())
# Push HTLC 2
has_slot, htlcs = cd.ensure_free_slots(t_2)
assert(has_slot and htlcs)
resolution_time, htlc = htlcs[0]
# We got a free slot by popping an outdated in-flight htlc
cd.push_htlc(t_2, htlc_2)
# The queue is full again (popped htlc 1, pushed htlc 2)
assert(cd.get_num_slots_occupied() == 2)
assert(cd.all_slots_busy())
# Push HTLC 3
has_slot, htlcs = cd.ensure_free_slots(t_3)
assert(not has_slot and not htlcs)
# Queue is full, and we can't pop anything: can't store htlc 3
def test_reset_slots():
cd = ChannelInDirection(num_slots=2)
cd.push_htlc(1, Htlc("pid", 100, True))
cd.reset_slots(num_slots=3)
assert(cd.get_num_slots_occupied() == 0)
assert(cd.get_num_slots() == 3)
def test_is_jammed():
cd = ChannelInDirection(num_slots=2)
cd.push_htlc(1, Htlc("pid", 100, True))
cd.push_htlc(1, Htlc("pid", 100, True))
assert(cd.get_earliest_htlc_resolution_time() == 1)
assert(cd.is_jammed(time=0))
cd.reset_slots(num_slots=2)
cd.push_htlc(1, Htlc("pid", 100, True))
cd.push_htlc(0, Htlc("pid", 100, True))
assert(cd.get_earliest_htlc_resolution_time() == 0)
assert(not cd.is_jammed(time=0))
def test_unsuccessful_ensure_free_slot():
cd = ChannelInDirection(num_slots=2)
cd.push_htlc(0, Htlc("pid", 100, True))
cd.push_htlc(0, Htlc("pid", 100, True))
has_slot, htlcs = cd.ensure_free_slots(time=0)
assert(has_slot and len(htlcs) == 1)
cd.push_htlc(1, Htlc("pid", 100, True))
has_slot, htlcs = cd.ensure_free_slots(time=0, num_slots_needed=2)
assert(not has_slot and not htlcs)
assert(cd.get_num_slots_occupied() == 2)