distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for naming/address resolution."""
2
3import os
4import base64
5
6import pytest
7
8
9class TestLookupResult:
10 def test_success(self):
11 from i2p_client.naming import LookupResult
12 dest_data = os.urandom(387)
13 r = LookupResult(success=True, destination_data=dest_data)
14 assert r.success
15 assert r.destination_data == dest_data
16 assert r.error_code is None
17
18 def test_failure(self):
19 from i2p_client.naming import LookupResult
20 r = LookupResult(success=False, error_code=1)
21 assert not r.success
22 assert r.destination_data is None
23 assert r.error_code == 1
24
25
26class TestHostsTxtService:
27 def _make_hosts_entry(self, hostname, dest_data):
28 """Create a hosts.txt line: hostname=base64(dest_data)"""
29 b64 = base64.b64encode(dest_data).decode("ascii")
30 return f"{hostname}={b64}"
31
32 def test_lookup_found(self):
33 from i2p_client.naming import HostsTxtService
34 dest_data = os.urandom(387)
35 hosts_txt = self._make_hosts_entry("test.i2p", dest_data)
36 svc = HostsTxtService.from_string(hosts_txt)
37 result = svc.lookup("test.i2p")
38 assert result.success
39 assert result.destination_data == dest_data
40
41 def test_lookup_not_found(self):
42 from i2p_client.naming import HostsTxtService
43 svc = HostsTxtService.from_string("")
44 result = svc.lookup("missing.i2p")
45 assert not result.success
46
47 def test_multiple_entries(self):
48 from i2p_client.naming import HostsTxtService
49 d1 = os.urandom(387)
50 d2 = os.urandom(387)
51 lines = "\n".join([
52 self._make_hosts_entry("a.i2p", d1),
53 self._make_hosts_entry("b.i2p", d2),
54 ])
55 svc = HostsTxtService.from_string(lines)
56 assert svc.lookup("a.i2p").destination_data == d1
57 assert svc.lookup("b.i2p").destination_data == d2
58
59 def test_comment_lines_ignored(self):
60 from i2p_client.naming import HostsTxtService
61 d = os.urandom(387)
62 lines = f"# comment\n{self._make_hosts_entry('x.i2p', d)}\n# another comment"
63 svc = HostsTxtService.from_string(lines)
64 assert svc.lookup("x.i2p").success
65
66 def test_blank_lines_ignored(self):
67 from i2p_client.naming import HostsTxtService
68 d = os.urandom(387)
69 lines = f"\n\n{self._make_hosts_entry('y.i2p', d)}\n\n"
70 svc = HostsTxtService.from_string(lines)
71 assert svc.lookup("y.i2p").success
72
73 def test_case_insensitive(self):
74 from i2p_client.naming import HostsTxtService
75 d = os.urandom(387)
76 lines = self._make_hosts_entry("MyHost.I2P", d)
77 svc = HostsTxtService.from_string(lines)
78 assert svc.lookup("myhost.i2p").success
79 assert svc.lookup("MYHOST.I2P").success
80
81 def test_list_all(self):
82 from i2p_client.naming import HostsTxtService
83 d1 = os.urandom(387)
84 d2 = os.urandom(387)
85 lines = "\n".join([
86 self._make_hosts_entry("a.i2p", d1),
87 self._make_hosts_entry("b.i2p", d2),
88 ])
89 svc = HostsTxtService.from_string(lines)
90 all_hosts = svc.list_all()
91 assert len(all_hosts) == 2
92 assert "a.i2p" in all_hosts
93 assert "b.i2p" in all_hosts
94
95
96class TestNamingService:
97 def test_abstract_lookup(self):
98 from i2p_client.naming import NamingService
99 # NamingService is abstract
100 with pytest.raises(TypeError):
101 NamingService()