441 lines
18 KiB
Python
441 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
|
||
class HostsFileManagerTests(unittest.TestCase):
|
||
def test_hosts_manager_has_no_legacy_bootstrap(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
manager = hosts_module.HostsManager()
|
||
self.assertFalse(hasattr(manager, "apply_hosts_bootstrap_if_needed"))
|
||
|
||
def test_execute_hosts_operation_does_not_run_legacy_bootstrap(self) -> None:
|
||
from hosts import commands as hosts_commands
|
||
|
||
calls: list[str] = []
|
||
|
||
class FakeHostsManager:
|
||
def apply_hosts_bootstrap_if_needed(self) -> None:
|
||
raise AssertionError("legacy bootstrap must not run")
|
||
|
||
def apply_service_dns_selections(self, service_dns) -> bool:
|
||
calls.append(f"apply:{service_dns.get('ChatGPT')}")
|
||
return True
|
||
|
||
result = hosts_commands.execute_hosts_operation(
|
||
FakeHostsManager(),
|
||
"apply_selection",
|
||
{"ChatGPT": "fin_dns"},
|
||
)
|
||
|
||
self.assertTrue(result.success)
|
||
self.assertEqual(calls, ["apply:fin_dns"])
|
||
|
||
def test_get_hosts_state_uses_read_only_access_check(self) -> None:
|
||
from hosts import commands as hosts_commands
|
||
|
||
class FakeHostsManager:
|
||
def is_hosts_file_readable(self) -> bool:
|
||
return True
|
||
|
||
def is_hosts_file_accessible(self) -> bool:
|
||
raise AssertionError("status refresh must not probe hosts write access")
|
||
|
||
def get_active_domains_map(self) -> dict[str, str]:
|
||
return {"chatgpt.com": "1.1.1.1"}
|
||
|
||
def is_adobe_domains_active(self) -> bool:
|
||
return False
|
||
|
||
state = hosts_commands.get_hosts_state(FakeHostsManager())
|
||
|
||
self.assertTrue(state.accessible)
|
||
self.assertEqual(state.active_domains, frozenset({"chatgpt.com"}))
|
||
|
||
def test_apply_domain_rows_skips_write_when_content_is_unchanged(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 chatgpt.com",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"",
|
||
]
|
||
)
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", return_value=True) as write_hosts,
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "2.2.2.2")]))
|
||
|
||
write_hosts.assert_not_called()
|
||
self.assertEqual(manager.last_status, "Файл hosts уже актуален: 1 запись")
|
||
|
||
def test_apply_domain_rows_skips_ipv6_when_unavailable(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=""),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=False, create=True),
|
||
):
|
||
self.assertTrue(
|
||
manager.apply_domain_ip_rows(
|
||
[
|
||
("instagram.com", "157.240.245.174"),
|
||
("instagram.com", "2a03:2880:f330:25:face:b00c:0:4420"),
|
||
]
|
||
)
|
||
)
|
||
|
||
self.assertEqual(len(written), 1)
|
||
self.assertIn("157.240.245.174 instagram.com", written[0])
|
||
self.assertNotIn("2a03:2880:f330:25:face:b00c:0:4420 instagram.com", written[0])
|
||
|
||
def test_apply_domain_rows_does_not_clear_hosts_when_all_rows_are_filtered_ipv6(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 old.example",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=False, create=True),
|
||
):
|
||
self.assertFalse(
|
||
manager.apply_domain_ip_rows(
|
||
[("ipv6-only.example", "2a03:2880:f330:25:face:b00c:0:4420")]
|
||
)
|
||
)
|
||
|
||
self.assertEqual(written, [])
|
||
self.assertEqual(manager.last_status, "Нет подходящих hosts-записей для применения")
|
||
|
||
def test_apply_domain_rows_replaces_only_zapretgui_managed_block(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 manual.example",
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"1.1.1.1 old.example",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"10.0.0.2 another.example",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("new.example", "2.2.2.2")]))
|
||
|
||
self.assertEqual(len(written), 1)
|
||
self.assertIn("10.0.0.1 manual.example", written[0])
|
||
self.assertIn("10.0.0.2 another.example", written[0])
|
||
self.assertIn("# >>> zapretgui:hosts managed begin >>>", written[0])
|
||
self.assertIn("2.2.2.2 new.example", written[0])
|
||
self.assertNotIn("1.1.1.1 old.example", written[0])
|
||
|
||
def test_apply_domain_rows_places_managed_block_before_manual_hosts_entries(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 manual.example",
|
||
"10.0.0.2 another.example",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "2.2.2.2")]))
|
||
|
||
self.assertEqual(len(written), 1)
|
||
self.assertIn("# user header", written[0])
|
||
self.assertIn("10.0.0.1 manual.example", written[0])
|
||
self.assertIn("10.0.0.2 another.example", written[0])
|
||
self.assertIn("2.2.2.2 chatgpt.com", written[0])
|
||
self.assertLess(written[0].index("# user header"), written[0].index("# >>> zapretgui:hosts managed begin >>>"))
|
||
self.assertLess(written[0].index("2.2.2.2 chatgpt.com"), written[0].index("10.0.0.1 manual.example"))
|
||
self.assertLess(written[0].index("2.2.2.2 chatgpt.com"), written[0].index("10.0.0.2 another.example"))
|
||
|
||
def test_apply_domain_rows_updates_top_domain_entry_without_adding_duplicate(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 chatgpt.com",
|
||
"10.0.0.2 another.example",
|
||
"10.0.0.3 chatgpt.com",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "2.2.2.2")]))
|
||
|
||
self.assertEqual(len(written), 1)
|
||
chatgpt_lines = [
|
||
line
|
||
for line in written[0].splitlines()
|
||
if line.strip()
|
||
and not line.lstrip().startswith("#")
|
||
and "chatgpt.com" in line.split()[1:]
|
||
]
|
||
self.assertEqual(chatgpt_lines, ["2.2.2.2 chatgpt.com", "10.0.0.3 chatgpt.com"])
|
||
self.assertIn("10.0.0.2 another.example", written[0])
|
||
self.assertEqual(manager.last_status, "Файл hosts обновлён: применено 1 запись")
|
||
|
||
def test_apply_domain_rows_does_not_shift_block_down_on_repeated_updates(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 chatgpt.com",
|
||
"10.0.0.2 another.example",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
current_content = {"text": original}
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
def write_hosts(text: str) -> bool:
|
||
written.append(text)
|
||
current_content["text"] = text
|
||
return True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", side_effect=lambda: current_content["text"]),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=write_hosts),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "2.2.2.2")]))
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "3.3.3.3")]))
|
||
|
||
self.assertEqual(len(written), 2)
|
||
first_lines = written[0].splitlines()
|
||
second_lines = written[1].splitlines()
|
||
first_begin = first_lines.index("# >>> zapretgui:hosts managed begin >>>")
|
||
second_begin = second_lines.index("# >>> zapretgui:hosts managed begin >>>")
|
||
self.assertEqual(second_begin, first_begin)
|
||
self.assertNotIn("\n\n\n# >>> zapretgui:hosts managed begin >>>", written[1])
|
||
|
||
def test_apply_domain_rows_keeps_other_domains_from_same_hosts_line(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 chatgpt.com manual.example # keep",
|
||
"10.0.0.3 chatgpt.com",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
patch.object(hosts_module, "is_ipv6_available", return_value=True),
|
||
):
|
||
self.assertTrue(manager.apply_domain_ip_rows([("chatgpt.com", "2.2.2.2")]))
|
||
|
||
self.assertEqual(len(written), 1)
|
||
self.assertIn("2.2.2.2 chatgpt.com", written[0])
|
||
self.assertIn("10.0.0.1 manual.example # keep", written[0])
|
||
self.assertNotIn("10.0.0.1 chatgpt.com manual.example", written[0])
|
||
|
||
def test_apply_service_selection_with_unknown_rows_does_not_clear_existing_block(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "get_service_domain_ip_rows", return_value=[]),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
):
|
||
self.assertFalse(manager.apply_service_dns_selections({"Missing": "zapret_dns"}))
|
||
|
||
self.assertEqual(written, [])
|
||
self.assertEqual(manager.last_status, "Не найдено записей hosts для выбранных сервисов")
|
||
|
||
def test_active_domains_are_read_from_zapretgui_managed_block(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
content = "\n".join(
|
||
[
|
||
"9.9.9.9 outside.example",
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 managed.example",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"",
|
||
]
|
||
)
|
||
manager = hosts_module.HostsManager()
|
||
with patch.object(hosts_module, "safe_read_hosts_file", return_value=content):
|
||
self.assertEqual(manager.get_active_domains_map(), {"managed.example": "2.2.2.2"})
|
||
|
||
def test_active_domains_keep_top_managed_domain_entry(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
content = "\n".join(
|
||
[
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 ChatGPT.com",
|
||
"3.3.3.3 chatgpt.com",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"",
|
||
]
|
||
)
|
||
manager = hosts_module.HostsManager()
|
||
with patch.object(hosts_module, "safe_read_hosts_file", return_value=content):
|
||
self.assertEqual(manager.get_active_domains_map(), {"chatgpt.com": "2.2.2.2"})
|
||
|
||
def test_active_domain_ip_map_keeps_all_managed_domain_ips(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
content = "\n".join(
|
||
[
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 ChatGPT.com",
|
||
"3.3.3.3 chatgpt.com",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"",
|
||
]
|
||
)
|
||
manager = hosts_module.HostsManager()
|
||
with patch.object(hosts_module, "safe_read_hosts_file", return_value=content):
|
||
self.assertEqual(
|
||
manager.get_active_domain_ip_map(),
|
||
{"chatgpt.com": ["2.2.2.2", "3.3.3.3"]},
|
||
)
|
||
|
||
def test_services_catalog_command_uses_full_active_domain_ip_map(self) -> None:
|
||
from hosts import commands as hosts_commands
|
||
|
||
class FakeHostsManager:
|
||
def get_active_domain_ip_map(self) -> dict[str, list[str]]:
|
||
return {"chatgpt.com": ["2.2.2.2", "3.3.3.3"]}
|
||
|
||
def get_active_domains_map(self) -> dict[str, str]:
|
||
raise AssertionError("нельзя терять дополнительные IP одного домена")
|
||
|
||
with patch("hosts.page_plans.build_services_catalog_plan") as build_plan:
|
||
build_plan.side_effect = lambda **kwargs: kwargs["active_domains_map"]
|
||
|
||
result = hosts_commands.build_services_catalog_plan(
|
||
hosts_runtime=FakeHostsManager(),
|
||
current_selection={},
|
||
direct_title="Direct",
|
||
ai_title="AI",
|
||
other_title="Other",
|
||
)
|
||
|
||
self.assertEqual(result, {"chatgpt.com": ["2.2.2.2", "3.3.3.3"]})
|
||
|
||
def test_clear_hosts_file_removes_only_zapretgui_managed_block(self) -> None:
|
||
from hosts import hosts as hosts_module
|
||
|
||
original = "\n".join(
|
||
[
|
||
"# user header",
|
||
"10.0.0.1 manual.example",
|
||
"# >>> zapretgui:hosts managed begin >>>",
|
||
"# Generated by ZapretGUI. Do not edit this block manually.",
|
||
"2.2.2.2 managed.example",
|
||
"# <<< zapretgui:hosts managed end <<<",
|
||
"10.0.0.2 another.example",
|
||
"",
|
||
]
|
||
)
|
||
written: list[str] = []
|
||
manager = hosts_module.HostsManager()
|
||
manager.is_hosts_file_accessible = lambda: True
|
||
|
||
with (
|
||
patch.object(hosts_module, "safe_read_hosts_file", return_value=original),
|
||
patch.object(hosts_module, "safe_write_hosts_file", side_effect=lambda text: written.append(text) or True),
|
||
):
|
||
self.assertTrue(manager.clear_hosts_file())
|
||
|
||
self.assertEqual(len(written), 1)
|
||
self.assertIn("10.0.0.1 manual.example", written[0])
|
||
self.assertIn("10.0.0.2 another.example", written[0])
|
||
self.assertNotIn("2.2.2.2 managed.example", written[0])
|
||
self.assertNotIn("zapretgui:hosts managed begin", written[0])
|
||
|
||
def test_ipv6_detection_uses_winapi_on_windows(self) -> None:
|
||
from hosts import ipv6_detection
|
||
|
||
ipv6_detection.reset_ipv6_detection_cache()
|
||
with (
|
||
patch.object(ipv6_detection.os, "name", "nt"),
|
||
patch.object(ipv6_detection, "_is_ipv6_available_winapi", return_value=True) as winapi_probe,
|
||
patch.object(ipv6_detection, "_is_ipv6_available_socket_probe", return_value=False) as socket_probe,
|
||
):
|
||
self.assertTrue(ipv6_detection.is_ipv6_available())
|
||
|
||
winapi_probe.assert_called_once()
|
||
socket_probe.assert_not_called()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|