"""Tests for i2p_apps.addressbook — hostname resolution and subscription management.""" import pytest from i2p_apps.addressbook.book import Addressbook, AddressbookEntry from i2p_apps.addressbook.storage import FileStorage from i2p_apps.addressbook.subscription import SubscriptionManager # --------------------------------------------------------------------------- # Addressbook tests # --------------------------------------------------------------------------- class TestAddressbook: """Tests for the Addressbook hostname→destination database.""" def test_lookup_unknown_host_returns_none(self): book = Addressbook() assert book.lookup("unknown.i2p") is None def test_add_entry_and_lookup(self): book = Addressbook() book.add_entry("example.i2p", "AAAA1234BASE64DEST") assert book.lookup("example.i2p") == "AAAA1234BASE64DEST" def test_remove_entry(self): book = Addressbook() book.add_entry("example.i2p", "AAAA1234BASE64DEST") assert book.remove_entry("example.i2p") is True assert book.lookup("example.i2p") is None # Removing a non-existent entry returns False assert book.remove_entry("example.i2p") is False def test_list_all_returns_all_entries(self): book = Addressbook() book.add_entry("alpha.i2p", "DEST_A") book.add_entry("beta.i2p", "DEST_B") book.add_entry("gamma.i2p", "DEST_C") all_entries = book.list_all() assert len(all_entries) == 3 assert "alpha.i2p" in all_entries assert "beta.i2p" in all_entries assert "gamma.i2p" in all_entries def test_case_insensitive_lookups(self): book = Addressbook() book.add_entry("Example.I2P", "DEST_CASE") assert book.lookup("example.i2p") == "DEST_CASE" assert book.lookup("EXAMPLE.I2P") == "DEST_CASE" assert book.lookup("Example.I2P") == "DEST_CASE" def test_duplicate_add_updates_existing(self): book = Addressbook() book.add_entry("example.i2p", "OLD_DEST") book.add_entry("example.i2p", "NEW_DEST") assert book.lookup("example.i2p") == "NEW_DEST" assert book.entry_count == 1 def test_has_entry(self): book = Addressbook() assert book.has_entry("example.i2p") is False book.add_entry("example.i2p", "DEST") assert book.has_entry("example.i2p") is True assert book.has_entry("EXAMPLE.I2P") is True def test_entry_count(self): book = Addressbook() assert book.entry_count == 0 book.add_entry("a.i2p", "DEST_A") book.add_entry("b.i2p", "DEST_B") assert book.entry_count == 2 def test_add_entry_stores_source(self): book = Addressbook() book.add_entry("example.i2p", "DEST", source="subscription") entries = book.list_all() assert entries["example.i2p"].source == "subscription" # --------------------------------------------------------------------------- # FileStorage tests # --------------------------------------------------------------------------- class TestFileStorage: """Tests for the FileStorage hosts.txt persistence.""" def test_save_and_load_roundtrip(self, tmp_path): path = str(tmp_path / "hosts.txt") storage = FileStorage(path) book = Addressbook() book.add_entry("alpha.i2p", "DEST_ALPHA", source="local") book.add_entry("beta.i2p", "DEST_BETA", source="subscription") storage.save(book.list_all()) loaded = storage.load() assert len(loaded) == 2 assert loaded["alpha.i2p"].destination == "DEST_ALPHA" assert loaded["beta.i2p"].destination == "DEST_BETA" def test_empty_file_produces_empty_dict(self, tmp_path): path = str(tmp_path / "hosts.txt") # Create an empty file with open(path, "w") as f: f.write("") storage = FileStorage(path) loaded = storage.load() assert loaded == {} def test_missing_file_produces_empty_dict(self, tmp_path): path = str(tmp_path / "nonexistent_hosts.txt") storage = FileStorage(path) loaded = storage.load() assert loaded == {} def test_hosts_txt_format(self, tmp_path): """hosts.txt format: hostname=destination per line, # for comments.""" path = str(tmp_path / "hosts.txt") storage = FileStorage(path) book = Addressbook() book.add_entry("example.i2p", "BASE64DEST") storage.save(book.list_all()) with open(path, "r") as f: content = f.read() # Should contain hostname=destination assert "example.i2p=BASE64DEST" in content def test_load_with_comments_and_blanks(self, tmp_path): path = str(tmp_path / "hosts.txt") with open(path, "w") as f: f.write("# This is a comment\n") f.write("\n") f.write("alpha.i2p=DEST_A\n") f.write("# Another comment\n") f.write("beta.i2p=DEST_B\n") f.write("\n") storage = FileStorage(path) loaded = storage.load() assert len(loaded) == 2 assert loaded["alpha.i2p"].destination == "DEST_A" assert loaded["beta.i2p"].destination == "DEST_B" def test_path_property(self, tmp_path): path = str(tmp_path / "hosts.txt") storage = FileStorage(path) assert storage.path == path # --------------------------------------------------------------------------- # SubscriptionManager tests # --------------------------------------------------------------------------- class TestSubscriptionManager: """Tests for the SubscriptionManager subscription list handling.""" def test_add_subscription_url(self): mgr = SubscriptionManager() mgr.add_subscription("http://example.i2p/hosts.txt") assert "http://example.i2p/hosts.txt" in mgr.get_subscriptions() def test_remove_subscription(self): mgr = SubscriptionManager() mgr.add_subscription("http://example.i2p/hosts.txt") assert mgr.remove_subscription("http://example.i2p/hosts.txt") is True assert len(mgr.get_subscriptions()) == 0 assert mgr.remove_subscription("http://nonexistent.i2p/hosts.txt") is False def test_parse_subscription_response(self): mgr = SubscriptionManager() text = ( "# Subscription list\n" "\n" "alpha.i2p=DEST_ALPHA_BASE64\n" "beta.i2p=DEST_BETA_BASE64\n" "# comment line\n" "gamma.i2p=DEST_GAMMA_BASE64\n" ) result = mgr.parse_subscription_response(text) assert len(result) == 3 assert result["alpha.i2p"] == "DEST_ALPHA_BASE64" assert result["beta.i2p"] == "DEST_BETA_BASE64" assert result["gamma.i2p"] == "DEST_GAMMA_BASE64" def test_parse_subscription_response_empty(self): mgr = SubscriptionManager() result = mgr.parse_subscription_response("") assert result == {} def test_merge_new_entries_into_addressbook(self): mgr = SubscriptionManager() book = Addressbook() new_entries = { "alpha.i2p": "DEST_A", "beta.i2p": "DEST_B", } count = mgr.merge_into_addressbook(book, new_entries) assert count == 2 assert book.lookup("alpha.i2p") == "DEST_A" assert book.lookup("beta.i2p") == "DEST_B" def test_existing_entries_not_overwritten_by_subscription(self): mgr = SubscriptionManager() book = Addressbook() book.add_entry("alpha.i2p", "ORIGINAL_DEST") new_entries = { "alpha.i2p": "NEW_DEST_FROM_SUB", "beta.i2p": "DEST_B", } count = mgr.merge_into_addressbook(book, new_entries, overwrite=False) # Only beta.i2p is new assert count == 1 # alpha.i2p should keep its original destination assert book.lookup("alpha.i2p") == "ORIGINAL_DEST" assert book.lookup("beta.i2p") == "DEST_B" def test_merge_with_overwrite(self): mgr = SubscriptionManager() book = Addressbook() book.add_entry("alpha.i2p", "ORIGINAL_DEST") new_entries = { "alpha.i2p": "OVERWRITTEN_DEST", } count = mgr.merge_into_addressbook(book, new_entries, overwrite=True) assert count == 1 assert book.lookup("alpha.i2p") == "OVERWRITTEN_DEST" def test_merge_entries_have_subscription_source(self): mgr = SubscriptionManager() book = Addressbook() new_entries = {"example.i2p": "DEST"} mgr.merge_into_addressbook(book, new_entries) entries = book.list_all() assert entries["example.i2p"].source == "subscription"