distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for InboundMessageDistributor."""
2
3from i2p_router.inbound_message_distributor import (
4 InboundMessageDistributor, DeliveryMode, I2NPType,
5)
6
7
8def test_filter_allows_garlic_on_client():
9 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
10 assert dist.is_allowed_message_type(I2NPType.GARLIC) is True
11
12
13def test_filter_blocks_tunnel_data_on_client():
14 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
15 assert dist.is_allowed_message_type(I2NPType.TUNNEL_DATA) is False
16
17
18def test_filter_blocks_data_msg_on_client():
19 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
20 assert dist.is_allowed_message_type(I2NPType.DATA) is False
21
22
23def test_filter_allows_delivery_status():
24 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
25 assert dist.is_allowed_message_type(I2NPType.DELIVERY_STATUS) is True
26
27
28def test_distribute_garlic_to_handler():
29 received = []
30 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
31 dist.set_garlic_handler(lambda data: received.append(data))
32 assert dist.distribute(I2NPType.GARLIC, b"garlic_data") is True
33 assert received == [b"garlic_data"]
34
35
36def test_distribute_drops_disallowed():
37 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
38 assert dist.distribute(I2NPType.TUNNEL_DATA, b"data") is False
39
40
41def test_handle_clove_destination_delivery():
42 delivered = []
43 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
44 dist.set_client_manager(lambda dh, data: delivered.append((dh, data)))
45 assert dist.handle_clove(
46 DeliveryMode.DESTINATION, b"payload",
47 dest_hash=b"\x01" * 32,
48 ) is True
49 assert len(delivered) == 1
50
51
52def test_handle_clove_destination_wrong_hash():
53 dist = InboundMessageDistributor(client_hash=b"\x01" * 32)
54 assert dist.handle_clove(
55 DeliveryMode.DESTINATION, b"payload",
56 dest_hash=b"\x02" * 32,
57 ) is False
58
59
60def test_handle_clove_tunnel_delivery():
61 dist = InboundMessageDistributor()
62 assert dist.handle_clove(
63 DeliveryMode.TUNNEL, b"data",
64 router_hash=b"\x03" * 32, tunnel_id=42,
65 ) is True