distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for packet scheduler and congestion control."""
2
3import pytest
4
5
6class TestRetransmitTimer:
7 def test_initial_rto(self):
8 from i2p_streaming.scheduler import RetransmitTimer
9 timer = RetransmitTimer()
10 assert timer.rto == 3000 # Initial RTO: 3 seconds
11
12 def test_update_rtt(self):
13 from i2p_streaming.scheduler import RetransmitTimer
14 timer = RetransmitTimer()
15 timer.update_rtt(100) # 100ms RTT
16 assert timer.rto > 0
17 assert timer.rto < 3000 # Should be less than initial
18
19 def test_backoff(self):
20 from i2p_streaming.scheduler import RetransmitTimer
21 timer = RetransmitTimer()
22 rto1 = timer.rto
23 timer.backoff()
24 assert timer.rto == rto1 * 2
25
26 def test_max_backoff(self):
27 from i2p_streaming.scheduler import RetransmitTimer
28 timer = RetransmitTimer()
29 for _ in range(20):
30 timer.backoff()
31 assert timer.rto <= 60000 # Max 60 seconds
32
33 def test_update_rtt_smoothing(self):
34 from i2p_streaming.scheduler import RetransmitTimer
35 timer = RetransmitTimer()
36 timer.update_rtt(100)
37 rto1 = timer.rto
38 timer.update_rtt(200)
39 rto2 = timer.rto
40 # RTO should change but be smoothed
41 assert rto2 != rto1
42
43
44class TestCongestionWindow:
45 def test_initial_window(self):
46 from i2p_streaming.scheduler import CongestionWindow
47 cw = CongestionWindow(initial_size=6)
48 assert cw.size == 6
49
50 def test_slow_start_increase(self):
51 from i2p_streaming.scheduler import CongestionWindow
52 cw = CongestionWindow(initial_size=1, ssthresh=16)
53 cw.on_ack()
54 assert cw.size == 2 # Doubles in slow start
55
56 def test_congestion_avoidance(self):
57 from i2p_streaming.scheduler import CongestionWindow
58 cw = CongestionWindow(initial_size=16, ssthresh=16)
59 cw.on_ack()
60 # In congestion avoidance, increase is additive (~1/size per ACK)
61 assert cw.size > 16
62 assert cw.size <= 17
63
64 def test_on_loss(self):
65 from i2p_streaming.scheduler import CongestionWindow
66 cw = CongestionWindow(initial_size=20, ssthresh=32)
67 cw.on_loss()
68 # AIMD: halve the window
69 assert cw.size == 10
70 assert cw.ssthresh == 10
71
72 def test_minimum_window(self):
73 from i2p_streaming.scheduler import CongestionWindow
74 cw = CongestionWindow(initial_size=1, ssthresh=16)
75 cw.on_loss()
76 assert cw.size >= 1 # Never go below 1
77
78 def test_max_window(self):
79 from i2p_streaming.scheduler import CongestionWindow
80 cw = CongestionWindow(initial_size=1, ssthresh=16, max_size=128)
81 for _ in range(200):
82 cw.on_ack()
83 assert cw.size <= 128
84
85
86class TestPacketScheduler:
87 def test_can_send_within_window(self):
88 from i2p_streaming.scheduler import PacketScheduler
89 sched = PacketScheduler(window_size=10)
90 assert sched.can_send()
91
92 def test_cannot_send_window_full(self):
93 from i2p_streaming.scheduler import PacketScheduler
94 sched = PacketScheduler(window_size=2)
95 sched.on_packet_sent(seq=0)
96 sched.on_packet_sent(seq=1)
97 assert not sched.can_send()
98
99 def test_ack_frees_window(self):
100 from i2p_streaming.scheduler import PacketScheduler
101 sched = PacketScheduler(window_size=1)
102 sched.on_packet_sent(seq=0)
103 assert not sched.can_send()
104 sched.on_ack(seq=0)
105 assert sched.can_send()
106
107 def test_outstanding_count(self):
108 from i2p_streaming.scheduler import PacketScheduler
109 sched = PacketScheduler(window_size=10)
110 sched.on_packet_sent(seq=0)
111 sched.on_packet_sent(seq=1)
112 assert sched.outstanding == 2
113 sched.on_ack(seq=0)
114 assert sched.outstanding == 1