distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for bandwidth estimator."""
2
3import time
4
5import pytest
6
7
8class TestBandwidthEstimator:
9 def test_initial_rates_are_zero(self):
10 from i2p_streaming.bandwidth import BandwidthEstimator
11 bw = BandwidthEstimator()
12 assert bw.send_bps() == 0.0
13 assert bw.recv_bps() == 0.0
14
15 def test_record_sent_increases_send_rate(self):
16 from i2p_streaming.bandwidth import BandwidthEstimator
17 bw = BandwidthEstimator(window_ms=1000)
18 bw.record_sent(1000)
19 assert bw.send_bps() > 0.0
20
21 def test_record_received_increases_recv_rate(self):
22 from i2p_streaming.bandwidth import BandwidthEstimator
23 bw = BandwidthEstimator(window_ms=1000)
24 bw.record_received(1000)
25 assert bw.recv_bps() > 0.0
26
27 def test_send_and_recv_are_independent(self):
28 from i2p_streaming.bandwidth import BandwidthEstimator
29 bw = BandwidthEstimator()
30 bw.record_sent(5000)
31 assert bw.send_bps() > 0.0
32 assert bw.recv_bps() == 0.0
33
34 def test_multiple_records_accumulate(self):
35 from i2p_streaming.bandwidth import BandwidthEstimator
36 bw = BandwidthEstimator(window_ms=5000)
37 bw.record_sent(100)
38 first = bw.send_bps()
39 bw.record_sent(100)
40 second = bw.send_bps()
41 assert second >= first
42
43 def test_ema_smoothing(self):
44 """EMA should smooth out spikes over time."""
45 from i2p_streaming.bandwidth import BandwidthEstimator
46 bw = BandwidthEstimator(window_ms=1000)
47 # Record a large spike
48 bw.record_sent(10000)
49 spike_rate = bw.send_bps()
50 # Wait a bit then record small amounts — rate should decrease
51 time.sleep(0.05)
52 for i in range(10):
53 time.sleep(0.01)
54 bw.record_sent(10)
55 after_small = bw.send_bps()
56 # After many small records with real elapsed time, rate should drop
57 assert after_small < spike_rate
58
59 def test_custom_window_size(self):
60 from i2p_streaming.bandwidth import BandwidthEstimator
61 bw = BandwidthEstimator(window_ms=500)
62 bw.record_sent(1000)
63 assert bw.send_bps() > 0.0
64
65 def test_zero_bytes_recorded(self):
66 from i2p_streaming.bandwidth import BandwidthEstimator
67 bw = BandwidthEstimator()
68 bw.record_sent(0)
69 # Should not error, rate should be 0
70 assert bw.send_bps() == 0.0