distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for tunnel dispatcher."""
2
3import os
4
5import pytest
6
7
8class TestTunnelDispatcher:
9 def test_construct(self):
10 from i2p_tunnel.dispatcher import TunnelDispatcher
11 d = TunnelDispatcher()
12 assert d.inbound_count() == 0
13 assert d.outbound_count() == 0
14
15 def test_register_inbound(self):
16 from i2p_tunnel.dispatcher import TunnelDispatcher
17 from i2p_data.tunnel import TunnelId
18 d = TunnelDispatcher()
19 d.register_inbound(TunnelId(100), gateway=os.urandom(32))
20 assert d.inbound_count() == 1
21
22 def test_register_outbound(self):
23 from i2p_tunnel.dispatcher import TunnelDispatcher
24 from i2p_data.tunnel import TunnelId
25 d = TunnelDispatcher()
26 d.register_outbound(TunnelId(200), gateway=os.urandom(32))
27 assert d.outbound_count() == 1
28
29 def test_lookup_inbound(self):
30 from i2p_tunnel.dispatcher import TunnelDispatcher
31 from i2p_data.tunnel import TunnelId
32 d = TunnelDispatcher()
33 gw = os.urandom(32)
34 d.register_inbound(TunnelId(100), gateway=gw)
35 info = d.lookup_inbound(TunnelId(100))
36 assert info is not None
37 assert info["gateway"] == gw
38
39 def test_lookup_missing(self):
40 from i2p_tunnel.dispatcher import TunnelDispatcher
41 from i2p_data.tunnel import TunnelId
42 d = TunnelDispatcher()
43 assert d.lookup_inbound(TunnelId(999)) is None
44
45 def test_unregister(self):
46 from i2p_tunnel.dispatcher import TunnelDispatcher
47 from i2p_data.tunnel import TunnelId
48 d = TunnelDispatcher()
49 d.register_inbound(TunnelId(50), gateway=os.urandom(32))
50 d.unregister_inbound(TunnelId(50))
51 assert d.inbound_count() == 0
52
53 def test_dispatch_message(self):
54 from i2p_tunnel.dispatcher import TunnelDispatcher
55 from i2p_data.tunnel import TunnelId
56 d = TunnelDispatcher()
57 delivered = []
58 d.register_inbound(TunnelId(10), gateway=os.urandom(32),
59 handler=lambda data: delivered.append(data))
60 d.dispatch_inbound(TunnelId(10), b"test message")
61 assert len(delivered) == 1
62 assert delivered[0] == b"test message"
63
64 def test_dispatch_unknown_tunnel(self):
65 from i2p_tunnel.dispatcher import TunnelDispatcher
66 from i2p_data.tunnel import TunnelId
67 d = TunnelDispatcher()
68 # Should not raise, just drop
69 d.dispatch_inbound(TunnelId(999), b"dropped")