distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_apps.addressbook — hostname resolution and subscription management."""
2
3import pytest
4
5from i2p_apps.addressbook.book import Addressbook, AddressbookEntry
6from i2p_apps.addressbook.storage import FileStorage
7from i2p_apps.addressbook.subscription import SubscriptionManager
8
9
10# ---------------------------------------------------------------------------
11# Addressbook tests
12# ---------------------------------------------------------------------------
13
14class TestAddressbook:
15 """Tests for the Addressbook hostname→destination database."""
16
17 def test_lookup_unknown_host_returns_none(self):
18 book = Addressbook()
19 assert book.lookup("unknown.i2p") is None
20
21 def test_add_entry_and_lookup(self):
22 book = Addressbook()
23 book.add_entry("example.i2p", "AAAA1234BASE64DEST")
24 assert book.lookup("example.i2p") == "AAAA1234BASE64DEST"
25
26 def test_remove_entry(self):
27 book = Addressbook()
28 book.add_entry("example.i2p", "AAAA1234BASE64DEST")
29 assert book.remove_entry("example.i2p") is True
30 assert book.lookup("example.i2p") is None
31 # Removing a non-existent entry returns False
32 assert book.remove_entry("example.i2p") is False
33
34 def test_list_all_returns_all_entries(self):
35 book = Addressbook()
36 book.add_entry("alpha.i2p", "DEST_A")
37 book.add_entry("beta.i2p", "DEST_B")
38 book.add_entry("gamma.i2p", "DEST_C")
39 all_entries = book.list_all()
40 assert len(all_entries) == 3
41 assert "alpha.i2p" in all_entries
42 assert "beta.i2p" in all_entries
43 assert "gamma.i2p" in all_entries
44
45 def test_case_insensitive_lookups(self):
46 book = Addressbook()
47 book.add_entry("Example.I2P", "DEST_CASE")
48 assert book.lookup("example.i2p") == "DEST_CASE"
49 assert book.lookup("EXAMPLE.I2P") == "DEST_CASE"
50 assert book.lookup("Example.I2P") == "DEST_CASE"
51
52 def test_duplicate_add_updates_existing(self):
53 book = Addressbook()
54 book.add_entry("example.i2p", "OLD_DEST")
55 book.add_entry("example.i2p", "NEW_DEST")
56 assert book.lookup("example.i2p") == "NEW_DEST"
57 assert book.entry_count == 1
58
59 def test_has_entry(self):
60 book = Addressbook()
61 assert book.has_entry("example.i2p") is False
62 book.add_entry("example.i2p", "DEST")
63 assert book.has_entry("example.i2p") is True
64 assert book.has_entry("EXAMPLE.I2P") is True
65
66 def test_entry_count(self):
67 book = Addressbook()
68 assert book.entry_count == 0
69 book.add_entry("a.i2p", "DEST_A")
70 book.add_entry("b.i2p", "DEST_B")
71 assert book.entry_count == 2
72
73 def test_add_entry_stores_source(self):
74 book = Addressbook()
75 book.add_entry("example.i2p", "DEST", source="subscription")
76 entries = book.list_all()
77 assert entries["example.i2p"].source == "subscription"
78
79
80# ---------------------------------------------------------------------------
81# FileStorage tests
82# ---------------------------------------------------------------------------
83
84class TestFileStorage:
85 """Tests for the FileStorage hosts.txt persistence."""
86
87 def test_save_and_load_roundtrip(self, tmp_path):
88 path = str(tmp_path / "hosts.txt")
89 storage = FileStorage(path)
90
91 book = Addressbook()
92 book.add_entry("alpha.i2p", "DEST_ALPHA", source="local")
93 book.add_entry("beta.i2p", "DEST_BETA", source="subscription")
94
95 storage.save(book.list_all())
96 loaded = storage.load()
97
98 assert len(loaded) == 2
99 assert loaded["alpha.i2p"].destination == "DEST_ALPHA"
100 assert loaded["beta.i2p"].destination == "DEST_BETA"
101
102 def test_empty_file_produces_empty_dict(self, tmp_path):
103 path = str(tmp_path / "hosts.txt")
104 # Create an empty file
105 with open(path, "w") as f:
106 f.write("")
107 storage = FileStorage(path)
108 loaded = storage.load()
109 assert loaded == {}
110
111 def test_missing_file_produces_empty_dict(self, tmp_path):
112 path = str(tmp_path / "nonexistent_hosts.txt")
113 storage = FileStorage(path)
114 loaded = storage.load()
115 assert loaded == {}
116
117 def test_hosts_txt_format(self, tmp_path):
118 """hosts.txt format: hostname=destination per line, # for comments."""
119 path = str(tmp_path / "hosts.txt")
120 storage = FileStorage(path)
121
122 book = Addressbook()
123 book.add_entry("example.i2p", "BASE64DEST")
124
125 storage.save(book.list_all())
126
127 with open(path, "r") as f:
128 content = f.read()
129
130 # Should contain hostname=destination
131 assert "example.i2p=BASE64DEST" in content
132
133 def test_load_with_comments_and_blanks(self, tmp_path):
134 path = str(tmp_path / "hosts.txt")
135 with open(path, "w") as f:
136 f.write("# This is a comment\n")
137 f.write("\n")
138 f.write("alpha.i2p=DEST_A\n")
139 f.write("# Another comment\n")
140 f.write("beta.i2p=DEST_B\n")
141 f.write("\n")
142 storage = FileStorage(path)
143 loaded = storage.load()
144 assert len(loaded) == 2
145 assert loaded["alpha.i2p"].destination == "DEST_A"
146 assert loaded["beta.i2p"].destination == "DEST_B"
147
148 def test_path_property(self, tmp_path):
149 path = str(tmp_path / "hosts.txt")
150 storage = FileStorage(path)
151 assert storage.path == path
152
153
154# ---------------------------------------------------------------------------
155# SubscriptionManager tests
156# ---------------------------------------------------------------------------
157
158class TestSubscriptionManager:
159 """Tests for the SubscriptionManager subscription list handling."""
160
161 def test_add_subscription_url(self):
162 mgr = SubscriptionManager()
163 mgr.add_subscription("http://example.i2p/hosts.txt")
164 assert "http://example.i2p/hosts.txt" in mgr.get_subscriptions()
165
166 def test_remove_subscription(self):
167 mgr = SubscriptionManager()
168 mgr.add_subscription("http://example.i2p/hosts.txt")
169 assert mgr.remove_subscription("http://example.i2p/hosts.txt") is True
170 assert len(mgr.get_subscriptions()) == 0
171 assert mgr.remove_subscription("http://nonexistent.i2p/hosts.txt") is False
172
173 def test_parse_subscription_response(self):
174 mgr = SubscriptionManager()
175 text = (
176 "# Subscription list\n"
177 "\n"
178 "alpha.i2p=DEST_ALPHA_BASE64\n"
179 "beta.i2p=DEST_BETA_BASE64\n"
180 "# comment line\n"
181 "gamma.i2p=DEST_GAMMA_BASE64\n"
182 )
183 result = mgr.parse_subscription_response(text)
184 assert len(result) == 3
185 assert result["alpha.i2p"] == "DEST_ALPHA_BASE64"
186 assert result["beta.i2p"] == "DEST_BETA_BASE64"
187 assert result["gamma.i2p"] == "DEST_GAMMA_BASE64"
188
189 def test_parse_subscription_response_empty(self):
190 mgr = SubscriptionManager()
191 result = mgr.parse_subscription_response("")
192 assert result == {}
193
194 def test_merge_new_entries_into_addressbook(self):
195 mgr = SubscriptionManager()
196 book = Addressbook()
197
198 new_entries = {
199 "alpha.i2p": "DEST_A",
200 "beta.i2p": "DEST_B",
201 }
202 count = mgr.merge_into_addressbook(book, new_entries)
203 assert count == 2
204 assert book.lookup("alpha.i2p") == "DEST_A"
205 assert book.lookup("beta.i2p") == "DEST_B"
206
207 def test_existing_entries_not_overwritten_by_subscription(self):
208 mgr = SubscriptionManager()
209 book = Addressbook()
210 book.add_entry("alpha.i2p", "ORIGINAL_DEST")
211
212 new_entries = {
213 "alpha.i2p": "NEW_DEST_FROM_SUB",
214 "beta.i2p": "DEST_B",
215 }
216 count = mgr.merge_into_addressbook(book, new_entries, overwrite=False)
217 # Only beta.i2p is new
218 assert count == 1
219 # alpha.i2p should keep its original destination
220 assert book.lookup("alpha.i2p") == "ORIGINAL_DEST"
221 assert book.lookup("beta.i2p") == "DEST_B"
222
223 def test_merge_with_overwrite(self):
224 mgr = SubscriptionManager()
225 book = Addressbook()
226 book.add_entry("alpha.i2p", "ORIGINAL_DEST")
227
228 new_entries = {
229 "alpha.i2p": "OVERWRITTEN_DEST",
230 }
231 count = mgr.merge_into_addressbook(book, new_entries, overwrite=True)
232 assert count == 1
233 assert book.lookup("alpha.i2p") == "OVERWRITTEN_DEST"
234
235 def test_merge_entries_have_subscription_source(self):
236 mgr = SubscriptionManager()
237 book = Addressbook()
238 new_entries = {"example.i2p": "DEST"}
239 mgr.merge_into_addressbook(book, new_entries)
240 entries = book.list_all()
241 assert entries["example.i2p"].source == "subscription"