distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for Banlist."""
2
3import time
4
5from i2p_router.banlist import Banlist
6
7
8def test_ban_and_check():
9 bl = Banlist()
10 peer = b"\x01" * 32
11 bl.banlist_router(peer, "test")
12 assert bl.is_banlisted(peer)
13
14
15def test_ban_expires():
16 bl = Banlist()
17 peer = b"\x02" * 32
18 bl.banlist_router(peer, "test", duration_ms=1)
19 time.sleep(0.01)
20 assert not bl.is_banlisted(peer)
21
22
23def test_permanent_ban():
24 bl = Banlist()
25 peer = b"\x03" * 32
26 bl.banlist_router_forever(peer, "bad peer")
27 assert bl.is_banlisted(peer)
28 entry = bl.get_entry(peer)
29 assert entry is not None
30 assert entry.expiration_ms == 0
31
32
33def test_transport_specific_ban():
34 bl = Banlist()
35 peer = b"\x04" * 32
36 bl.banlist_router_transport(peer, "NTCP2", "timeout", duration_ms=60000)
37 assert bl.is_banlisted_transport(peer, "NTCP2")
38 assert not bl.is_banlisted_transport(peer, "SSU2")
39 assert not bl.is_banlisted(peer) # not globally banned
40
41
42def test_unban():
43 bl = Banlist()
44 peer = b"\x05" * 32
45 bl.banlist_router(peer, "test")
46 bl.unbanlist_router(peer)
47 assert not bl.is_banlisted(peer)
48
49
50def test_cleanup_removes_expired():
51 bl = Banlist()
52 bl.banlist_router(b"\x06" * 32, "a", duration_ms=1)
53 bl.banlist_router(b"\x07" * 32, "b", duration_ms=3600000)
54 time.sleep(0.01)
55 removed = bl.cleanup_expired()
56 assert removed == 1
57 assert bl.count == 1
58
59
60def test_global_ban_covers_transport():
61 bl = Banlist()
62 peer = b"\x08" * 32
63 bl.banlist_router(peer, "global ban")
64 assert bl.is_banlisted_transport(peer, "NTCP2")
65 assert bl.is_banlisted_transport(peer, "SSU2")