zapretgui/tests/test_exit_watch_autorestart.py

"""Exit-watch / foreign winws / auto-restart acceptance tests
(.agent/tasks/exit-watch-autorestart/spec.md).

AC1: exit watcher fires for our dead current process, suppresses on identity
     mismatch and intentional-stop states, starts only on confirmed spawn.
AC3/AC4: main-thread handler outcomes and the auto-restart budget.
AC5: foreign winws classification and the warn-once observer.
AC6: scan guard TTL + blockcheck wiring contract.
AC7: structured post-mortem resolution (transient flag, scan-guard silence).
"""

from __future__ import annotations

import ntpath

import inspect
import threading
import time
from pathlib import Path
from types import SimpleNamespace
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch


PROJECT_SRC = Path(__file__).resolve().parents[1] / "src"
if str(PROJECT_SRC) not in sys.path:
    sys.path.insert(0, str(PROJECT_SRC))


class _LogRecorder:
    def __init__(self) -> None:
        self.records: list[tuple[str, str]] = []

    def __call__(self, message, level="INFO", *args, **kwargs) -> None:
        self.records.append((str(message), str(level)))

    def messages(self, level: str) -> list[str]:
        return [message for message, lvl in self.records if lvl == level]


def _make_base_runner():
    from winws_runtime.runners.zapret1_runner import Winws1StrategyRunner

    runner = object.__new__(Winws1StrategyRunner)
    runner.running_process = None
    runner.current_launch_label = None
    runner._unexpected_process_exit_callback = None
    return runner


class ExitWatcherTests(unittest.TestCase):
    def test_callback_fires_for_current_dead_process_in_running_state(self) -> None:
        from winws_runtime.runners.preset_runner_support import PresetRunnerState

        runner = _make_base_runner()
        process = SimpleNamespace(poll=lambda: 1)
        runner.running_process = process
        runner.get_runner_state_snapshot = lambda: SimpleNamespace(state=PresetRunnerState.RUNNING)
        callback = Mock()
        runner._unexpected_process_exit_callback = callback

        runner._on_watched_process_exit(process)

        callback.assert_called_once_with()

    def test_suppressed_when_process_was_replaced(self) -> None:
        runner = _make_base_runner()
        old_process = SimpleNamespace(poll=lambda: 1)
        runner.running_process = SimpleNamespace(poll=lambda: None)  # new process
        callback = Mock()
        runner._unexpected_process_exit_callback = callback

        runner._on_watched_process_exit(old_process)

        callback.assert_not_called()

    def test_suppressed_during_intentional_stop_states(self) -> None:
        from winws_runtime.runners.preset_runner_support import PresetRunnerState

        for state in (PresetRunnerState.STOPPING, PresetRunnerState.IDLE):
            runner = _make_base_runner()
            process = SimpleNamespace(poll=lambda: 1)
            runner.running_process = process
            runner.get_runner_state_snapshot = lambda s=state: SimpleNamespace(state=s)
            callback = Mock()
            runner._unexpected_process_exit_callback = callback

            runner._on_watched_process_exit(process)

            callback.assert_not_called()

    def test_watcher_thread_invokes_exit_handler_after_wait(self) -> None:
        runner = _make_base_runner()
        fired = threading.Event()
        runner._on_watched_process_exit = lambda _process: fired.set()
        process = SimpleNamespace(wait=lambda: 0)

        runner._start_process_exit_watcher(process)

        self.assertTrue(fired.wait(timeout=2.0))

    def test_zapret2_spawn_success_starts_watcher(self) -> None:
        from winws_runtime.runners.zapret2_runner import Winws2StrategyRunner

        with tempfile.TemporaryDirectory() as tmp:
            runner = object.__new__(Winws2StrategyRunner)
            runner.winws_exe = "winws2.exe"
            runner.work_dir = tmp
            runner.running_process = None
            runner.current_launch_label = None
            runner.current_strategy_args = None
            runner._last_spawn_exit_code = None
            runner._last_spawn_stderr = ""
            runner._last_startup_output_path = ""
            runner._prepare_state_for_spawn_locked = Mock()
            runner._set_runner_state_locked = Mock()
            runner._log_winws2_launch_command = Mock()
            runner._create_startup_info = Mock(return_value=None)
            runner._set_last_error = Mock()
            runner._run_preset_dry_run_locked = Mock(return_value=True)
            runner._start_process_exit_watcher = Mock()

            artifact = SimpleNamespace(
                launch_args=("@config.txt",),
                preset_path="preset.txt",
                normalized_text="--wf-tcp-out=443\n",
            )
            fake_process = SimpleNamespace(pid=1234, returncode=None)

            from winws_runtime.engine.startup import REASON_READY, EngineStartOutcome

            with (
                patch("winws_runtime.runners.zapret2_runner.subprocess.Popen", return_value=fake_process),
                patch(
                    "winws_runtime.runners.zapret2_runner.wait_engine_ready",
                    return_value=EngineStartOutcome(True, REASON_READY, True, 0.03),
                ),
            ):
                self.assertTrue(
                    runner._spawn_process_locked(artifact, "Preset", preset_switch=False)
                )

            runner._start_process_exit_watcher.assert_called_once_with(fake_process)


class UnexpectedExitHandlerTests(unittest.TestCase):
    def _make_events(self, *, phase: str = "running", command_port=None):
        from app.feature_facades.runtime_parts import RuntimeEvents

        runtime_service = SimpleNamespace(
            snapshot=lambda: SimpleNamespace(phase=phase),
            mark_start_failed=Mock(),
        )
        return RuntimeEvents(runtime_service=runtime_service, command_port=command_port)

    def _resolution(self, *, transient: bool):
        from winws_runtime.health.post_mortem import PostMortemResolution

        return PostMortemResolution(
            message="winws2.exe неожиданно завершился: тестовая причина",
            kind="transient_dll_init" if transient else "silent_exit",
            transient=transient,
            exit_code=0xC0000142 if transient else 1,
        )

    def test_no_action_when_resolution_is_none(self) -> None:
        events = self._make_events()
        recorder = _LogRecorder()
        with patch("log.log.log", recorder):
            events.handle_unexpected_process_exit(None)

        events.runtime_service.mark_start_failed.assert_not_called()
        self.assertEqual(recorder.messages("ERROR"), [])

    def test_no_action_when_phase_is_not_active(self) -> None:
        events = self._make_events(phase="stopped")
        events.handle_unexpected_process_exit(self._resolution(transient=False))
        events.runtime_service.mark_start_failed.assert_not_called()

    def test_non_transient_failure_publishes_single_error(self) -> None:
        events = self._make_events()
        recorder = _LogRecorder()
        resolution = self._resolution(transient=False)
        with patch("log.log.log", recorder):
            events.handle_unexpected_process_exit(resolution)

        self.assertEqual(recorder.messages("ERROR"), [resolution.message])
        events.runtime_service.mark_start_failed.assert_called_once_with(resolution.message)

    def test_transient_failure_triggers_auto_restart(self) -> None:
        command_port = SimpleNamespace(start=Mock(return_value=True))
        events = self._make_events(command_port=command_port)
        recorder = _LogRecorder()
        resolution = self._resolution(transient=True)
        with patch("log.log.log", recorder):
            events.handle_unexpected_process_exit(resolution)

        command_port.start.assert_called_once_with()
        self.assertEqual(recorder.messages("ERROR"), [])
        self.assertTrue(
            any("автоматический перезапуск" in msg.lower() for msg in recorder.messages("WARNING"))
        )
        events.runtime_service.mark_start_failed.assert_called_once_with(resolution.message)

    def test_exhausted_budget_disables_auto_restart(self) -> None:
        command_port = SimpleNamespace(start=Mock(return_value=True))
        events = self._make_events(command_port=command_port)
        now = time.monotonic()
        events.auto_restart_history = [now - 1.0, now - 2.0]
        recorder = _LogRecorder()
        resolution = self._resolution(transient=True)
        with patch("log.log.log", recorder):
            events.handle_unexpected_process_exit(resolution)

        command_port.start.assert_not_called()
        errors = recorder.messages("ERROR")
        self.assertEqual(len(errors), 1)
        self.assertIn("Автоперезапуск отключён", errors[0])
        events.runtime_service.mark_start_failed.assert_called_once()

    def test_auto_restart_budget_window_slides(self) -> None:
        events = self._make_events()
        self.assertTrue(events._try_reserve_auto_restart(now=1000.0))
        self.assertTrue(events._try_reserve_auto_restart(now=1001.0))
        self.assertFalse(events._try_reserve_auto_restart(now=1002.0))
        self.assertTrue(events._try_reserve_auto_restart(now=1000.0 + 601.0 + 1.0))


class ForeignWinwsDetectionTests(unittest.TestCase):
    def test_find_foreign_classifies_by_path(self) -> None:
        from winws_runtime.runtime import process_probe

        entries = [(100, "winws2.exe"), (200, "winws.exe"), (300, "winws2.exe")]
        paths = {
            100: r"c:\zapret\exe\winws2.exe",  # canonical
            200: r"c:\other\winws.exe",       # foreign
            300: "",                            # unqueryable → skipped
        }
        with (
            patch.object(process_probe, "_iter_winws_process_entries", return_value=entries),
            patch.object(process_probe, "_query_process_image_path", side_effect=lambda pid: paths[pid]),
            patch.object(
                process_probe,
                "get_expected_winws_paths",
                return_value={
                    "winws2.exe": r"c:\zapret\exe\winws2.exe",
                    "winws.exe": r"c:\zapret\exe\winws.exe",
                },
            ),
        ):
            foreign = process_probe.find_foreign_winws_processes()

        self.assertEqual([record.pid for record in foreign], [200])
        self.assertEqual(foreign[0].exe_path, r"c:\other\winws.exe")

    def test_single_snapshot_scan_matches_separate_probes(self) -> None:
        from winws_runtime.runtime import process_probe

        entries = [(300, "winws2.exe"), (100, "winws2.exe"), (200, "winws.exe"), (400, "winws.exe")]
        paths = {
            100: r"c:\zapret\exe\winws2.exe",  # canonical
            300: r"c:\zapret\exe\winws2.exe",  # canonical
            200: r"c:\other\winws.exe",       # foreign
            400: "",                            # unqueryable → skipped
        }
        expected = {
            "winws2.exe": r"c:\zapret\exe\winws2.exe",
            "winws.exe": r"c:\zapret\exe\winws.exe",
        }
        snapshots = []

        def entries_once():
            snapshots.append(1)
            return list(entries)

        with (
            patch.object(process_probe, "_iter_winws_process_entries", side_effect=entries_once),
            patch.object(process_probe, "_query_process_image_path", side_effect=lambda pid: paths[pid]),
            patch.object(process_probe, "get_expected_winws_paths", return_value=expected),
            patch.object(process_probe.os.path, "exists", return_value=True),
            # Пути в тесте Windows-овские: на Linux basename их бы не разобрал.
            patch.object(process_probe, "_normalize_path", side_effect=lambda path: path),
            patch.object(process_probe.os.path, "basename", ntpath.basename),
        ):
            scan = process_probe.scan_winws_processes()
            self.assertEqual(len(snapshots), 1)
            separate_canonical = process_probe.get_canonical_winws_process_pids()
            separate_foreign = {
                record.pid: record.exe_path for record in process_probe.find_foreign_winws_processes()
            }

        self.assertEqual(scan.canonical_pids, {"winws2.exe": [100, 300]})
        self.assertEqual(scan.foreign_paths, {200: r"c:\other\winws.exe"})
        self.assertEqual(scan.canonical_pids, separate_canonical)
        self.assertEqual(scan.foreign_paths, separate_foreign)

    def test_scan_skips_canonical_when_expected_exe_is_missing(self) -> None:
        from winws_runtime.runtime import process_probe

        with (
            patch.object(process_probe, "_iter_winws_process_entries", return_value=[(100, "winws2.exe")]),
            patch.object(
                process_probe, "_query_process_image_path", return_value=r"c:\zapret\exe\winws2.exe"
            ),
            patch.object(
                process_probe,
                "get_expected_winws_paths",
                return_value={"winws2.exe": r"c:\zapret\exe\winws2.exe"},
            ),
            patch.object(process_probe.os.path, "exists", return_value=False),
        ):
            scan = process_probe.scan_winws_processes()

        self.assertEqual(scan.canonical_pids, {})
        self.assertEqual(scan.foreign_paths, {})

    def _make_objects(self, *, running: bool = True):
        from app.feature_facades.runtime_parts import RuntimeObjects

        runtime_service = SimpleNamespace(
            snapshot=lambda: SimpleNamespace(running=running),
        )
        return RuntimeObjects(runtime_service=runtime_service)

    def test_warns_once_per_foreign_pid_set_while_running(self) -> None:
        objects = self._make_objects(running=True)
        recorder = _LogRecorder()
        foreign = {123: r"c:\other\winws.exe"}
        with (
            patch("log.log.log", recorder),
            patch(
                "winws_runtime.runtime.scan_guard.is_external_winws_scan_active",
                return_value=False,
            ),
        ):
            objects.observe_foreign_processes(foreign)
            objects.observe_foreign_processes(foreign)  # same set → no repeat
            objects.observe_foreign_processes({})       # cleared → reset
            objects.observe_foreign_processes(foreign)  # returns → warn again

        errors = recorder.messages("ERROR")
        self.assertEqual(len(errors), 2)
        self.assertIn("посторонний winws", errors[0])
        self.assertIn("PID 123", errors[0])

    def test_silent_when_dpi_not_running(self) -> None:
        objects = self._make_objects(running=False)
        recorder = _LogRecorder()
        with patch("log.log.log", recorder):
            objects.observe_foreign_processes({123: r"c:\other\winws.exe"})
        self.assertEqual(recorder.messages("ERROR"), [])

    def test_silent_while_external_scan_active(self) -> None:
        objects = self._make_objects(running=True)
        recorder = _LogRecorder()
        with (
            patch("log.log.log", recorder),
            patch(
                "winws_runtime.runtime.scan_guard.is_external_winws_scan_active",
                return_value=True,
            ),
        ):
            objects.observe_foreign_processes({123: r"c:\other\winws.exe"})
        self.assertEqual(recorder.messages("ERROR"), [])


class ScanGuardTests(unittest.TestCase):
    def test_flag_lifecycle_and_ttl(self) -> None:
        from winws_runtime.runtime import scan_guard

        fake_now = [1000.0]
        with patch.object(scan_guard.time, "monotonic", side_effect=lambda: fake_now[0]):
            scan_guard.mark_external_winws_scan_active(True, ttl_seconds=10.0)
            self.assertTrue(scan_guard.is_external_winws_scan_active())

            fake_now[0] = 1011.0  # TTL expired — a dead scanner cannot mute us forever
            self.assertFalse(scan_guard.is_external_winws_scan_active())

            fake_now[0] = 1012.0
            scan_guard.mark_external_winws_scan_active(True, ttl_seconds=10.0)
            self.assertTrue(scan_guard.is_external_winws_scan_active())
            scan_guard.mark_external_winws_scan_active(False)
            self.assertFalse(scan_guard.is_external_winws_scan_active())

    def test_strategy_search_holds_scan_flag_for_whole_scan(self) -> None:
        from blockcheck.strategy_search.environment import RealEnvironment

        self.assertIn("mark_external_winws_scan_active(True)", inspect.getsource(RealEnvironment.pre_cleanup))
        # Каждая стратегия продлевает флаг: полный подбор дольше его срока.
        self.assertIn("mark_external_winws_scan_active(True)", inspect.getsource(RealEnvironment.start_session))
        post_source = inspect.getsource(RealEnvironment.post_cleanup)
        self.assertIn("finally", post_source)
        self.assertIn("mark_external_winws_scan_active(False)", post_source)


class ResolveUnexpectedExitTests(unittest.TestCase):
    def _fake_runner(self, exit_code: int, output: str = ""):
        return SimpleNamespace(
            winws_exe=r"g:\zapret\exe\winws2.exe",
            build_post_mortem_snapshot=lambda: {
                "exit_code": exit_code,
                "output": output,
                "strategy_name": "Default v1",
            },
        )

    def test_transient_for_dll_init_failure(self) -> None:
        from winws_runtime.health.post_mortem import resolve_unexpected_exit

        with (
            patch(
                "winws_runtime.runners.runner_factory.get_current_runner",
                return_value=self._fake_runner(0xC0000142),
            ),
            patch(
                "winws_runtime.runtime.scan_guard.is_external_winws_scan_active",
                return_value=False,
            ),
        ):
            resolution = resolve_unexpected_exit()

        self.assertIsNotNone(resolution)
        self.assertTrue(resolution.transient)
        self.assertEqual(resolution.exit_code, 0xC0000142)

    def test_not_transient_for_external_kill_and_crash(self) -> None:
        from winws_runtime.health.post_mortem import resolve_unexpected_exit

        for exit_code in (1, 0xC0000005):
            with (
                patch(
                    "winws_runtime.runners.runner_factory.get_current_runner",
                    return_value=self._fake_runner(exit_code),
                ),
                patch(
                    "winws_runtime.runtime.scan_guard.is_external_winws_scan_active",
                    return_value=False,
                ),
            ):
                resolution = resolve_unexpected_exit()

            self.assertIsNotNone(resolution, hex(exit_code))
            self.assertFalse(resolution.transient, hex(exit_code))

    def test_none_while_external_scan_active(self) -> None:
        from winws_runtime.health.post_mortem import resolve_unexpected_exit

        with (
            patch(
                "winws_runtime.runtime.scan_guard.is_external_winws_scan_active",
                return_value=True,
            ),
            patch("winws_runtime.runners.runner_factory.get_current_runner") as get_runner,
        ):
            self.assertIsNone(resolve_unexpected_exit())
        get_runner.assert_not_called()


if __name__ == "__main__":
    unittest.main(verbosity=2)