fceab51704
All checks were successful
Zapret source guards / validate (push) Successful in 10s
- отчёт проверки пишется через временный файл; сбой записи попадает в журнал - новый запуск сбрасывает путь прошлого отчёта, чтобы обращение не взяло чужой - Стоп в последней контрольной пробе больше не записывает стратегию как рабочую - совет про нерабочий адрес в hosts не пропадает, когда сломан и другой адрес - сбой открытия журнала запуска пишется в журнал, а не молчит Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
435 lines
20 KiB
Python
435 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import unittest
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from tempfile import TemporaryDirectory
|
||
from unittest.mock import patch
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
from diagnostics import history
|
||
from settings import schema
|
||
from settings.normalize import normalize_check_history, normalize_settings
|
||
from settings.schema import build_default_settings
|
||
|
||
WHEN = datetime(2026, 10, 7, 20, 15, 30)
|
||
|
||
|
||
def _service(label, level, *, control=False, key=""):
|
||
return {"key": key or label.lower(), "label": label, "level": level, "control": control}
|
||
|
||
|
||
def _report(*services, problems=(), scope="all") -> dict:
|
||
return {
|
||
"scope": scope,
|
||
"services": list(services),
|
||
"problems": [{"level": level, "text": text} for level, text in problems],
|
||
}
|
||
|
||
|
||
class EntryTests(unittest.TestCase):
|
||
def test_entry_keeps_summary_and_state_of_every_service(self) -> None:
|
||
report = _report(
|
||
_service("Discord", "ok"),
|
||
_service("YouTube", "fail"),
|
||
_service("Google", "ok", control=True),
|
||
problems=(("fail", "YouTube не открывается"), ("warn", "DNS подменяет ответы")),
|
||
)
|
||
entry = history.blockcheck_entry(report, log_file="C:\\logs\\run.log", when=WHEN)
|
||
|
||
self.assertEqual(entry["kind"], history.KIND_BLOCKCHECK)
|
||
self.assertEqual(entry["time"], "2026-10-07T20:15:30")
|
||
self.assertEqual(entry["title"], "Все сайты")
|
||
self.assertEqual(entry["level"], "fail")
|
||
self.assertEqual(entry["headline"], "YouTube не открывается")
|
||
self.assertEqual(entry["problems"], ["YouTube не открывается", "DNS подменяет ответы"])
|
||
# Контрольные сайты в сравнение не идут: они нужны только чтобы понять, есть ли интернет.
|
||
self.assertEqual(entry["states"], {"Discord": "ok", "YouTube": "fail"})
|
||
self.assertEqual(entry["log_file"], "C:\\logs\\run.log")
|
||
|
||
def test_clean_run_and_scope_title(self) -> None:
|
||
entry = history.blockcheck_entry(_report(_service("Discord", "ok"), scope="main"), when=WHEN)
|
||
|
||
self.assertEqual((entry["level"], entry["headline"], entry["title"]), ("ok", "Всё открывается", "Discord и YouTube"))
|
||
|
||
def test_full_check_has_its_own_title_and_is_not_compared_with_all_sites(self) -> None:
|
||
full = history.blockcheck_entry(_report(_service("YouTube", "fail"), scope="full"), when=WHEN)
|
||
earlier = [history.blockcheck_entry(_report(_service("YouTube", "ok"), scope="all"), when=WHEN)]
|
||
|
||
self.assertEqual(full["title"], "Глубокая диагностика")
|
||
self.assertIsNone(history.previous_run(earlier, full))
|
||
|
||
def test_deep_diagnostics_is_compared_with_the_full_checks_of_before_the_rename(self) -> None:
|
||
full = history.blockcheck_entry(_report(_service("YouTube", "fail"), scope="full"), when=WHEN)
|
||
old = {**history.blockcheck_entry(_report(_service("YouTube", "ok"), scope="full"), when=WHEN), "title": "Полная проверка"}
|
||
|
||
self.assertIs(history.previous_run([old], full), old)
|
||
|
||
def test_overall_level_is_the_worst_problem_level(self) -> None:
|
||
def level(*problems):
|
||
return history.blockcheck_entry(_report(problems=problems), when=WHEN)["level"]
|
||
|
||
self.assertEqual(level(("warn", "a"), ("unknown", "b")), "warn")
|
||
self.assertEqual(level(("unknown", "b")), "unknown")
|
||
self.assertEqual(level(("warn", "a"), ("fail", "c")), "fail")
|
||
|
||
|
||
class DnsServersAdapterTests(unittest.TestCase):
|
||
def test_report_becomes_plain_dictionary_for_the_engine(self) -> None:
|
||
from types import SimpleNamespace
|
||
|
||
from blockcheck import commands
|
||
|
||
from dns.server_check import Finding
|
||
|
||
report = SimpleNamespace(
|
||
findings=(
|
||
Finding(
|
||
"info",
|
||
"dead",
|
||
"Не отвечают совсем: X (1.1.1.1).",
|
||
title="Не отвечают совсем",
|
||
servers=(("X", "1.1.1.1"),),
|
||
),
|
||
Finding("warn", "dot_blocked", "DoT закрыт у: Y."),
|
||
)
|
||
)
|
||
stops: list = []
|
||
|
||
def run_server_check(*, on_progress=None, should_stop=None):
|
||
stops.append(should_stop)
|
||
return report
|
||
|
||
marker = lambda: False # noqa: E731
|
||
with (
|
||
patch("dns.commands.run_server_check", run_server_check),
|
||
patch("dns.server_check_plans.build_text_report", lambda _report: "таблица"),
|
||
):
|
||
result = commands.check_dns_servers(should_stop=marker)
|
||
|
||
self.assertEqual(result["level"], "warn")
|
||
# Находка идёт фразой и готовыми частями; если частей нет — только фразой.
|
||
self.assertEqual(
|
||
result["findings"][0],
|
||
{
|
||
"level": "info",
|
||
"text": "Не отвечают совсем: X (1.1.1.1).",
|
||
"title": "Не отвечают совсем",
|
||
"servers": [["X", "1.1.1.1"]],
|
||
"note": "",
|
||
},
|
||
)
|
||
self.assertEqual(result["findings"][1], {"level": "warn", "text": "DoT закрыт у: Y."})
|
||
self.assertEqual(result["text"], "таблица")
|
||
self.assertIs(stops[0], marker)
|
||
|
||
def test_level_is_worst_finding_or_unknown_without_findings(self) -> None:
|
||
from types import SimpleNamespace
|
||
|
||
from blockcheck import commands
|
||
from dns.server_check import Finding
|
||
|
||
def level(*levels):
|
||
report = SimpleNamespace(findings=tuple(Finding(item, "code", "x") for item in levels))
|
||
with (
|
||
patch("dns.commands.run_server_check", lambda **_k: report),
|
||
patch("dns.server_check_plans.build_text_report", lambda _report: ""),
|
||
):
|
||
return commands.check_dns_servers()["level"]
|
||
|
||
self.assertEqual(level("ok", "info"), "ok")
|
||
self.assertEqual(level("warn", "fail"), "fail")
|
||
self.assertEqual(level(), "unknown")
|
||
|
||
|
||
class ChangesTests(unittest.TestCase):
|
||
@staticmethod
|
||
def _entry(**states):
|
||
return {"kind": "blockcheck", "title": "Все сайты", "time": "2026-10-06T10:00:00", "states": states}
|
||
|
||
def test_stopped_and_recovered_services_are_named(self) -> None:
|
||
before = self._entry(Discord="ok", YouTube="warn", Telegram="fail", X="fail")
|
||
after = self._entry(Discord="fail", YouTube="fail", Telegram="ok", X="fail")
|
||
|
||
self.assertEqual(
|
||
history.describe_changes(before, after),
|
||
["перестали открываться: Discord, YouTube", "снова открываются: Telegram"],
|
||
)
|
||
|
||
def test_unknown_or_missing_state_is_not_a_change(self) -> None:
|
||
before = self._entry(Discord="ok", YouTube="unknown")
|
||
after = self._entry(Discord="unknown", YouTube="fail", Новый="fail")
|
||
|
||
self.assertEqual(history.describe_changes(before, after), [])
|
||
|
||
def test_no_previous_run_means_nothing_to_compare(self) -> None:
|
||
self.assertEqual(history.describe_changes(None, self._entry(Discord="fail")), [])
|
||
|
||
def test_previous_run_is_the_latest_of_same_kind_and_scope(self) -> None:
|
||
runs = [
|
||
{"kind": "blockcheck", "title": "Все сайты", "time": "1"},
|
||
{"kind": "blockcheck", "title": "Discord и YouTube", "time": "2"},
|
||
{"kind": "blockcheck", "title": "Все сайты", "time": "3"},
|
||
{"kind": "dns_servers", "title": "Все сайты", "time": "4"},
|
||
]
|
||
|
||
self.assertEqual(history.previous_run(runs, {"kind": "blockcheck", "title": "Все сайты"})["time"], "3")
|
||
self.assertEqual(history.previous_run(runs, {"kind": "blockcheck", "title": "Discord и YouTube"})["time"], "2")
|
||
self.assertIsNone(history.previous_run(runs, {"kind": "blockcheck", "title": "Другое"}))
|
||
|
||
def test_time_is_shown_short(self) -> None:
|
||
self.assertEqual(history.format_time("2026-10-07T20:15:30"), "07.10 20:15")
|
||
self.assertEqual(history.format_time("когда-то"), "когда-то")
|
||
|
||
|
||
class JsonReportTests(unittest.TestCase):
|
||
def test_document_names_its_format_and_keeps_russian_text(self) -> None:
|
||
report = _report(_service("Discord", "ok"), problems=(("warn", "DNS подменяет ответы"),))
|
||
entry = history.blockcheck_entry(report, when=WHEN)
|
||
text = history.report_json(report, entry, app_version="21.1.7.50")
|
||
document = json.loads(text)
|
||
|
||
self.assertEqual(document["format"], "zapretgui.blockcheck/1")
|
||
self.assertEqual(document["app_version"], "21.1.7.50")
|
||
self.assertEqual(document["time"], "2026-10-07T20:15:30")
|
||
self.assertEqual(document["report"]["problems"][0]["text"], "DNS подменяет ответы")
|
||
self.assertIn("DNS подменяет ответы", text)
|
||
|
||
|
||
class NormalizeTests(unittest.TestCase):
|
||
def test_default_settings_have_empty_history(self) -> None:
|
||
self.assertEqual(build_default_settings()["blockcheck"]["check_history"], [])
|
||
self.assertEqual(normalize_settings({})["blockcheck"]["check_history"], [])
|
||
|
||
def test_only_known_fields_survive_and_garbage_is_dropped(self) -> None:
|
||
runs = normalize_check_history(
|
||
[
|
||
{
|
||
"kind": "blockcheck",
|
||
"time": "2026-10-07T20:15:30",
|
||
"title": "Все сайты",
|
||
"level": "катастрофа",
|
||
"headline": "x" * 1000,
|
||
"problems": ["a", "", 5, "b"] + ["лишнее"] * 30,
|
||
"states": {"Discord": "ok", "YouTube": "сломан", "": "ok"},
|
||
"log_file": "C:\\logs\\run.log",
|
||
"password": "секрет",
|
||
},
|
||
{"kind": "", "time": "2026-10-07T20:15:30"},
|
||
{"kind": "blockcheck"},
|
||
"мусор",
|
||
]
|
||
)
|
||
|
||
self.assertEqual(len(runs), 1)
|
||
run = runs[0]
|
||
self.assertEqual(run["level"], "unknown")
|
||
self.assertEqual(len(run["headline"]), schema.CHECK_HISTORY_TEXT_LIMIT)
|
||
self.assertEqual(run["problems"][:3], ["a", "5", "b"])
|
||
self.assertEqual(len(run["problems"]), schema.CHECK_HISTORY_PROBLEMS_LIMIT)
|
||
self.assertEqual(run["states"], {"Discord": "ok"})
|
||
self.assertNotIn("password", run)
|
||
self.assertEqual(normalize_check_history("не список"), [])
|
||
|
||
def test_only_the_latest_runs_are_kept(self) -> None:
|
||
many = [{"kind": "blockcheck", "time": f"t{index}"} for index in range(schema.CHECK_HISTORY_LIMIT + 7)]
|
||
runs = normalize_check_history(many)
|
||
|
||
self.assertEqual(len(runs), schema.CHECK_HISTORY_LIMIT)
|
||
self.assertEqual(runs[0]["time"], "t7")
|
||
self.assertEqual(runs[-1]["time"], f"t{schema.CHECK_HISTORY_LIMIT + 6}")
|
||
|
||
|
||
class StoreAndRememberTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self._folder = TemporaryDirectory()
|
||
self.addCleanup(self._folder.cleanup)
|
||
self.root = Path(self._folder.name)
|
||
patcher = patch("settings.store.MAIN_DIRECTORY", str(self.root))
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
from settings import store
|
||
|
||
self.store = store
|
||
self.addCleanup(store.close_settings_database)
|
||
store.close_settings_database()
|
||
|
||
def test_history_is_stored_next_to_other_blockcheck_settings(self) -> None:
|
||
self.store.set_blockcheck_settings({"user_domains": ["example.com"]})
|
||
runs = self.store.add_check_history_run({"kind": "blockcheck", "time": "t1", "level": "ok"})
|
||
self.store.add_check_history_run({"kind": "blockcheck", "time": "t2", "level": "fail"})
|
||
|
||
self.assertEqual([run["time"] for run in runs], ["t1"])
|
||
self.assertEqual([run["time"] for run in self.store.get_check_history()], ["t1", "t2"])
|
||
# Запись истории не затёрла соседнюю настройку.
|
||
self.assertEqual(self.store.get_blockcheck_settings()["user_domains"], ["example.com"])
|
||
|
||
def test_remembering_a_run_reports_what_changed_and_writes_json(self) -> None:
|
||
from blockcheck import commands
|
||
|
||
log_file = str(self.root / "blockcheck_run_1.log")
|
||
first = commands.remember_blockcheck_run(_report(_service("Discord", "ok"), _service("YouTube", "ok")), log_file)
|
||
self.assertEqual((first["changes"], first["previous_time"]), ([], ""))
|
||
|
||
second_report = _report(
|
||
_service("Discord", "ok"), _service("YouTube", "fail"), problems=(("fail", "YouTube не открывается"),)
|
||
)
|
||
second = commands.remember_blockcheck_run(second_report, log_file)
|
||
|
||
self.assertEqual(second["changes"], ["перестали открываться: YouTube"])
|
||
self.assertEqual([run["level"] for run in second["history"]], ["ok", "fail"])
|
||
self.assertTrue(second["previous_time"])
|
||
self.assertEqual(len(self.store.get_check_history()), 2)
|
||
document = json.loads(Path(second["json_file"]).read_text("utf-8"))
|
||
self.assertEqual(Path(second["json_file"]).name, "blockcheck_run_1.json")
|
||
self.assertEqual(document["format"], history.REPORT_FORMAT)
|
||
self.assertEqual(document["level"], "fail")
|
||
# Отчёт, открытый из «Прошлых проверок», показывает ту же полосу изменений.
|
||
self.assertEqual(document["report"]["changes"], ["перестали открываться: YouTube"])
|
||
self.assertEqual(document["report"]["previous_time"], second["previous_time"])
|
||
|
||
def test_run_of_another_scope_is_not_compared(self) -> None:
|
||
from blockcheck import commands
|
||
|
||
commands.remember_blockcheck_run(_report(_service("YouTube", "ok"), scope="all"), None)
|
||
note = commands.remember_blockcheck_run(_report(_service("YouTube", "fail"), scope="main"), None)
|
||
|
||
self.assertEqual((note["changes"], note["json_file"]), ([], ""))
|
||
|
||
|
||
class WorkerTests(unittest.TestCase):
|
||
def _worker(self, remember):
|
||
from blockcheck.worker import BlockcheckWorker
|
||
|
||
return BlockcheckWorker(
|
||
report_path=lambda *_a: "",
|
||
save_report=lambda *_a: "",
|
||
remember_run=remember,
|
||
)
|
||
|
||
def test_changes_are_added_to_report_and_printed(self) -> None:
|
||
def remember(report, _log):
|
||
# Как настоящая запись: изменения ложатся в отчёт до сохранения файла.
|
||
report["changes"] = ["перестали открываться: YouTube"]
|
||
return {
|
||
"changes": ["перестали открываться: YouTube"],
|
||
"previous_time": "2026-10-06T10:00:00",
|
||
"json_file": "x.json",
|
||
"lines": ["🕘 С прошлой проверки (06.10 10:00) перестали открываться: YouTube."],
|
||
}
|
||
|
||
worker = self._worker(remember)
|
||
lines: list[str] = []
|
||
worker.log_message.connect(lines.append)
|
||
report = {"scope": "all"}
|
||
worker._remember(report)
|
||
|
||
self.assertEqual(report["changes"], ["перестали открываться: YouTube"])
|
||
self.assertEqual(report["json_file"], "x.json")
|
||
self.assertEqual(report["history"], [])
|
||
self.assertIn("🕘 С прошлой проверки (06.10 10:00) перестали открываться: YouTube.", lines)
|
||
|
||
def test_failed_history_write_does_not_cost_the_result(self) -> None:
|
||
def broken(_report, _log):
|
||
raise OSError("диск переполнен")
|
||
|
||
report = {"scope": "all", "services": []}
|
||
self._worker(broken)._remember(report)
|
||
|
||
self.assertNotIn("changes", report)
|
||
|
||
def test_no_changes_means_no_extra_lines(self) -> None:
|
||
worker = self._worker(lambda report, log: {"changes": [], "previous_time": "", "json_file": ""})
|
||
lines: list[str] = []
|
||
worker.log_message.connect(lines.append)
|
||
worker._remember({"scope": "all"})
|
||
|
||
self.assertEqual(lines, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|
||
|
||
class SingleReportFileTests(unittest.TestCase):
|
||
"""У проверки один файл — .json: в нём и данные, и текст отчёта, и прерванный прогон тоже."""
|
||
|
||
def _run(self, engine_result, *, raises: Exception | None = None):
|
||
from unittest.mock import patch
|
||
|
||
from blockcheck.worker import BlockcheckWorker
|
||
|
||
saved: list[tuple[dict, str]] = []
|
||
remembered: list[tuple[dict, str]] = []
|
||
|
||
def remember(report, path):
|
||
remembered.append((dict(report), path))
|
||
return {"changes": [], "previous_time": "", "json_file": path}
|
||
|
||
def fake_engine(scope, *, emit, partial, **_kwargs):
|
||
emit("строка первая")
|
||
partial({"scope": scope, "services": ["discord"], "partial": True})
|
||
emit("строка вторая")
|
||
if raises is not None:
|
||
raise raises
|
||
return engine_result
|
||
|
||
worker = BlockcheckWorker(
|
||
scope="full",
|
||
report_path=lambda scope: f"run_{scope}.json",
|
||
save_report=lambda report, path: saved.append((report, path)) or path,
|
||
remember_run=remember,
|
||
)
|
||
announced: list = []
|
||
worker.run_log_started.connect(announced.append)
|
||
with patch("diagnostics.engine.run_blockcheck", fake_engine):
|
||
worker.run()
|
||
return saved, remembered, announced
|
||
|
||
def test_finished_run_keeps_its_text_inside_the_report(self) -> None:
|
||
saved, remembered, announced = self._run({"scope": "full", "services": []})
|
||
|
||
[(report, path)] = remembered
|
||
self.assertEqual(path, "run_full.json")
|
||
self.assertEqual(report["text"], ["строка первая", "строка вторая"])
|
||
self.assertEqual((saved, announced), ([], ["run_full.json"]))
|
||
|
||
def test_stopped_and_crashed_runs_are_saved_with_what_was_learned(self) -> None:
|
||
for kwargs, mark in (({"engine_result": {"stopped": True}}, "stopped"), ({"engine_result": None, "raises": RuntimeError("сеть")}, "failed")):
|
||
saved, remembered, announced = self._run(**kwargs)
|
||
[(report, path)] = saved
|
||
self.assertTrue(report[mark])
|
||
self.assertEqual(report["services"], ["discord"])
|
||
self.assertEqual(report["text"][:2], ["строка первая", "строка вторая"])
|
||
self.assertEqual((remembered, announced), ([], [path]))
|
||
|
||
def test_failed_write_keeps_the_old_report_whole_and_leaves_no_temp_file(self) -> None:
|
||
import tempfile
|
||
|
||
from blockcheck import commands
|
||
|
||
with tempfile.TemporaryDirectory() as folder:
|
||
path = str(Path(folder) / "blockcheck_run_1_full.json")
|
||
self.assertEqual(commands.save_blockcheck_report({"scope": "full", "text": ["старый"]}, path), path)
|
||
|
||
with patch("blockcheck.commands.os.replace", side_effect=OSError("disk full")):
|
||
self.assertEqual(commands.save_blockcheck_report({"scope": "full", "text": ["новый"]}, path), "")
|
||
|
||
# Оборванная запись не портит отчёт, лежавший на этом месте, и не оставляет мусора.
|
||
self.assertEqual(commands.load_past_blockcheck_report(path)["text"], ["старый"])
|
||
self.assertEqual(sorted(item.name for item in Path(folder).iterdir()), ["blockcheck_run_1_full.json"])
|
||
|
||
def test_saved_file_is_read_back_as_a_past_report(self) -> None:
|
||
import tempfile
|
||
|
||
from blockcheck import commands
|
||
|
||
with tempfile.TemporaryDirectory() as folder:
|
||
path = str(Path(folder) / "blockcheck_run_1_full.json")
|
||
self.assertEqual(commands.save_blockcheck_report({"scope": "full", "text": ["раз", "два"]}, path), path)
|
||
self.assertEqual(commands.load_past_blockcheck_report(path)["text"], ["раз", "два"])
|
||
self.assertEqual(sorted(item.name for item in Path(folder).iterdir()), ["blockcheck_run_1_full.json"])
|
||
self.assertEqual(commands.save_blockcheck_report({}, None), "")
|