1from __future__ import annotations
2
3import hashlib
4
5from protocol import Clock, Receiver, Sender, SimulatedChannel, run_until_complete
6
7
8def make_data(size: int = 10 * 1024) -> bytes:
9
10 return bytes((i * 17 + i // 3) % 256 for i in range(size))
11
12
13def run_case(loss_rate: float, *, duplicate_rate: float = 0.10, seed: int = 99):
14 clock = Clock()
15 data_channel = SimulatedChannel(
16 clock,
17 loss_rate=loss_rate,
18 duplicate_rate=duplicate_rate,
19 min_delay=1,
20 max_delay=6,
21 seed=seed,
22 )
23 ack_channel = SimulatedChannel(
24 clock,
25 loss_rate=loss_rate,
26 duplicate_rate=duplicate_rate / 2,
27 min_delay=1,
28 max_delay=4,
29 seed=seed + 1000,
30 )
31 receiver = Receiver(ack_channel, max_seq=256)
32 sent = make_data()
33 sender = Sender(
34 sent,
35 data_channel,
36 receiver,
37 window_size=7,
38 max_seq=256,
39 segment_size=128,
40 timeout=20,
41 )
42 receiver.set_ack_target(sender.receive_ack)
43 run_until_complete(sender, receiver, data_channel, ack_channel, clock)
44 return sent, receiver, sender
45
46
47def sha(data: bytes) -> str:
48 return hashlib.sha256(data).hexdigest()
49
50
51def test_in_order_delivery_under_loss() -> None:
52 sent, receiver, _sender = run_case(0.30, seed=123)
53 assert receiver.data() == sent
54
55
56def test_no_duplicate_delivery() -> None:
57 sent, receiver, _sender = run_case(0.10, duplicate_rate=0.35, seed=321)
58 assert len(receiver.data()) == len(sent)
59 assert receiver.delivered_packets == (len(sent) + 127) // 128
60 assert receiver.data() == sent
61
62
63def test_window_never_exceeded() -> None:
64 _sent, _receiver, sender = run_case(0.30, seed=456)
65 assert sender.max_in_flight <= sender.window_size
66
67
68def test_checksum_matches_for_every_loss_rate() -> None:
69 for loss in (0.0, 0.10, 0.30):
70 sent, receiver, _sender = run_case(loss, seed=789 + int(loss * 100))
71 assert sha(receiver.data()) == sha(sent)
72
73
74def run_all() -> None:
75 test_in_order_delivery_under_loss()
76 test_no_duplicate_delivery()
77 test_window_never_exceeded()
78 test_checksum_matches_for_every_loss_rate()
79
80
81if __name__ == "__main__":
82 run_all()
83 print("all tests passed")
84
Discussion
No comments yet. Start the discussion. Recorded by @patrick-toulme.