"""Tests for SSU2 ACK bitfield tracking.""" import pytest class TestSSU2Bitfield: def test_set_and_get(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() bf.set(0) bf.set(5) bf.set(100) assert bf.get(0) is True assert bf.get(5) is True assert bf.get(100) is True def test_get_unset(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() bf.set(10) assert bf.get(0) is False assert bf.get(9) is False assert bf.get(11) is False assert bf.get(255) is False def test_highest(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() assert bf.get_highest() == -1 bf.set(3) assert bf.get_highest() == 3 bf.set(50) assert bf.get_highest() == 50 bf.set(25) # Set a lower one; highest should not change assert bf.get_highest() == 50 def test_shift_offset(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() bf.set(0) bf.set(1) bf.set(5) bf.set(10) bf.shift_offset(5) # Packets 0, 1 should be gone; 5 and 10 should remain assert bf.get(0) is False assert bf.get(1) is False assert bf.get(5) is True assert bf.get(10) is True def test_to_ack_blocks(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() # Contiguous: 0,1,2,3,4 for i in range(5): bf.set(i) blocks = bf.to_ack_blocks() # All contiguous: should be a single (5, 0) ack block assert len(blocks) >= 1 total_acked = sum(a for a, _ in blocks) assert total_acked == 5 def test_from_ack_blocks(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield # through=9, blocks: 3 acked (9,8,7), 2 nacked (6,5), 5 acked (4,3,2,1,0) blocks = [(3, 2), (5, 0)] bf = SSU2Bitfield.from_ack_blocks(through=9, blocks=blocks) assert bf.get(9) is True assert bf.get(8) is True assert bf.get(7) is True assert bf.get(6) is False assert bf.get(5) is False assert bf.get(4) is True assert bf.get(3) is True assert bf.get(2) is True assert bf.get(1) is True assert bf.get(0) is True def test_roundtrip(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() # Set some packet numbers with gaps for pn in [0, 1, 2, 5, 6, 10, 11, 12]: bf.set(pn) through = bf.get_highest() blocks = bf.to_ack_blocks() bf2 = SSU2Bitfield.from_ack_blocks(through, blocks) # Verify all set bits match for pn in range(through + 1): assert bf.get(pn) == bf2.get(pn), f"Mismatch at packet {pn}" def test_sparse_bitfield(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() # Every 10th packet for i in range(0, 100, 10): bf.set(i) assert bf.get_highest() == 90 assert len(bf) == 10 for i in range(100): if i % 10 == 0: assert bf.get(i) is True else: assert bf.get(i) is False def test_contains(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() bf.set(42) assert 42 in bf assert 43 not in bf def test_len(self): from i2p_transport.ssu2_bitfield import SSU2Bitfield bf = SSU2Bitfield() assert len(bf) == 0 bf.set(0) bf.set(10) bf.set(20) assert len(bf) == 3