distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SSU2 ACK bitfield tracking."""
2
3import pytest
4
5
6class TestSSU2Bitfield:
7 def test_set_and_get(self):
8 from i2p_transport.ssu2_bitfield import SSU2Bitfield
9
10 bf = SSU2Bitfield()
11 bf.set(0)
12 bf.set(5)
13 bf.set(100)
14 assert bf.get(0) is True
15 assert bf.get(5) is True
16 assert bf.get(100) is True
17
18 def test_get_unset(self):
19 from i2p_transport.ssu2_bitfield import SSU2Bitfield
20
21 bf = SSU2Bitfield()
22 bf.set(10)
23 assert bf.get(0) is False
24 assert bf.get(9) is False
25 assert bf.get(11) is False
26 assert bf.get(255) is False
27
28 def test_highest(self):
29 from i2p_transport.ssu2_bitfield import SSU2Bitfield
30
31 bf = SSU2Bitfield()
32 assert bf.get_highest() == -1
33
34 bf.set(3)
35 assert bf.get_highest() == 3
36
37 bf.set(50)
38 assert bf.get_highest() == 50
39
40 bf.set(25) # Set a lower one; highest should not change
41 assert bf.get_highest() == 50
42
43 def test_shift_offset(self):
44 from i2p_transport.ssu2_bitfield import SSU2Bitfield
45
46 bf = SSU2Bitfield()
47 bf.set(0)
48 bf.set(1)
49 bf.set(5)
50 bf.set(10)
51
52 bf.shift_offset(5)
53 # Packets 0, 1 should be gone; 5 and 10 should remain
54 assert bf.get(0) is False
55 assert bf.get(1) is False
56 assert bf.get(5) is True
57 assert bf.get(10) is True
58
59 def test_to_ack_blocks(self):
60 from i2p_transport.ssu2_bitfield import SSU2Bitfield
61
62 bf = SSU2Bitfield()
63 # Contiguous: 0,1,2,3,4
64 for i in range(5):
65 bf.set(i)
66
67 blocks = bf.to_ack_blocks()
68 # All contiguous: should be a single (5, 0) ack block
69 assert len(blocks) >= 1
70 total_acked = sum(a for a, _ in blocks)
71 assert total_acked == 5
72
73 def test_from_ack_blocks(self):
74 from i2p_transport.ssu2_bitfield import SSU2Bitfield
75
76 # through=9, blocks: 3 acked (9,8,7), 2 nacked (6,5), 5 acked (4,3,2,1,0)
77 blocks = [(3, 2), (5, 0)]
78 bf = SSU2Bitfield.from_ack_blocks(through=9, blocks=blocks)
79
80 assert bf.get(9) is True
81 assert bf.get(8) is True
82 assert bf.get(7) is True
83 assert bf.get(6) is False
84 assert bf.get(5) is False
85 assert bf.get(4) is True
86 assert bf.get(3) is True
87 assert bf.get(2) is True
88 assert bf.get(1) is True
89 assert bf.get(0) is True
90
91 def test_roundtrip(self):
92 from i2p_transport.ssu2_bitfield import SSU2Bitfield
93
94 bf = SSU2Bitfield()
95 # Set some packet numbers with gaps
96 for pn in [0, 1, 2, 5, 6, 10, 11, 12]:
97 bf.set(pn)
98
99 through = bf.get_highest()
100 blocks = bf.to_ack_blocks()
101 bf2 = SSU2Bitfield.from_ack_blocks(through, blocks)
102
103 # Verify all set bits match
104 for pn in range(through + 1):
105 assert bf.get(pn) == bf2.get(pn), f"Mismatch at packet {pn}"
106
107 def test_sparse_bitfield(self):
108 from i2p_transport.ssu2_bitfield import SSU2Bitfield
109
110 bf = SSU2Bitfield()
111 # Every 10th packet
112 for i in range(0, 100, 10):
113 bf.set(i)
114
115 assert bf.get_highest() == 90
116 assert len(bf) == 10
117
118 for i in range(100):
119 if i % 10 == 0:
120 assert bf.get(i) is True
121 else:
122 assert bf.get(i) is False
123
124 def test_contains(self):
125 from i2p_transport.ssu2_bitfield import SSU2Bitfield
126
127 bf = SSU2Bitfield()
128 bf.set(42)
129 assert 42 in bf
130 assert 43 not in bf
131
132 def test_len(self):
133 from i2p_transport.ssu2_bitfield import SSU2Bitfield
134
135 bf = SSU2Bitfield()
136 assert len(bf) == 0
137 bf.set(0)
138 bf.set(10)
139 bf.set(20)
140 assert len(bf) == 3