distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for Transport bid integration (banlist, self-send, failure tracking)."""
2
3import asyncio
4from i2p_transport.manager import TransportManager
5from i2p_transport.transport_base import Transport, TransportBid, TransportStyle
6
7
8class MockTransport(Transport):
9 """Transport that always succeeds or always fails."""
10 def __init__(self, succeed=True):
11 self._succeed = succeed
12 self._style = TransportStyle.NTCP2
13
14 @property
15 def style(self):
16 return self._style
17
18 async def start(self):
19 pass
20
21 async def stop(self):
22 pass
23
24 async def bid(self, peer_hash):
25 return TransportBid(latency_ms=10, transport=self)
26
27 async def send(self, peer_hash, data):
28 return self._succeed
29
30 @property
31 def current_address(self):
32 return None
33
34 @property
35 def reachability(self):
36 from i2p_transport.transport_base import ReachabilityStatus
37 return ReachabilityStatus.OK
38
39 @property
40 def is_running(self):
41 return True
42
43
44class MockBanlist:
45 def __init__(self, banned=None):
46 self._banned = set(banned or [])
47
48 def is_banlisted(self, peer_hash):
49 return peer_hash in self._banned
50
51
52def _run(coro):
53 return asyncio.run(coro)
54
55
56def test_reject_self_send():
57 my_hash = b"\x01" * 32
58 tm = TransportManager(router_hash=my_hash)
59 tm.register(MockTransport())
60 assert _run(tm.send(my_hash, b"data")) is False
61
62
63def test_reject_banlisted():
64 peer = b"\x02" * 32
65 banlist = MockBanlist(banned=[peer])
66 tm = TransportManager(banlist=banlist)
67 tm.register(MockTransport())
68 assert _run(tm.send(peer, b"data")) is False
69
70
71def test_success_clears_failure():
72 peer = b"\x03" * 32
73 tm = TransportManager()
74 tm.register(MockTransport(succeed=True))
75 assert _run(tm.send(peer, b"data")) is True
76 assert tm._failed_counts.get(peer) is None
77
78
79def test_failure_tracking():
80 peer = b"\x04" * 32
81 tm = TransportManager()
82 tm.register(MockTransport(succeed=False))
83 _run(tm.send(peer, b"data")) # fail 1
84 assert tm._failed_counts[peer] == 1
85 _run(tm.send(peer, b"data")) # fail 2
86 assert tm._failed_counts[peer] == 2
87 # Third attempt should be rejected (> 1 failures)
88 assert _run(tm.send(peer, b"data")) is False
89
90
91def test_no_banlist_still_works():
92 tm = TransportManager() # no banlist, no router_hash
93 tm.register(MockTransport())
94 assert _run(tm.send(b"\x05" * 32, b"data")) is True
95
96
97def test_clear_failed_count():
98 tm = TransportManager()
99 peer = b"\x06" * 32
100 tm._failed_counts[peer] = 5
101 tm.clear_failed_count(peer)
102 assert peer not in tm._failed_counts