voculinux/tests/test_auto_pause_monitor.py

"""
Tests for auto-pause process matching, monitor pause/resume, and model unload.

Exercises the shipped matching helpers and AutoPauseMonitor against real
callbacks — not a parallel reimplementation of the match logic.
"""

import os
import sys
import unittest
from unittest.mock import MagicMock, patch

# Heavy deps used transitively when importing recognition manager
mock_pywhispercpp = MagicMock()
mock_pywhispercpp.model = MagicMock()
mock_pywhispercpp.model.Model = MagicMock()

sys.modules.setdefault("vosk", MagicMock())
sys.modules.setdefault("whisper", MagicMock())
sys.modules.setdefault("pywhispercpp", mock_pywhispercpp)
sys.modules.setdefault("pywhispercpp.model", mock_pywhispercpp.model)
sys.modules.setdefault("requests", MagicMock())
sys.modules.setdefault("pyaudio", MagicMock())
sys.modules.setdefault("wave", MagicMock())
sys.modules.setdefault("tqdm", MagicMock())
sys.modules.setdefault("numpy", MagicMock())
sys.modules.setdefault("torch", MagicMock())
sys.modules.setdefault("psutil", MagicMock())

from vocalinux.auto_pause_monitor import (  # noqa: E402
    AutoPauseMonitor,
    any_configured_process_running,
    configured_names_set,
    normalize_process_name,
)
from vocalinux.common_types import RecognitionState  # noqa: E402
from vocalinux.ui.config_manager import DEFAULT_CONFIG  # noqa: E402


class TestNormalizeAndMatch(unittest.TestCase):
    """Pure matching helpers (no I/O)."""

    def test_normalize_basename_and_case(self):
        self.assertEqual(normalize_process_name("Overwatch"), "overwatch")
        self.assertEqual(normalize_process_name("/usr/bin/Steam"), "steam")
        self.assertEqual(normalize_process_name("game.EXE"), "game")
        self.assertEqual(normalize_process_name("  chrome  "), "chrome")
        self.assertEqual(normalize_process_name(""), "")

    def test_configured_names_set_drops_empty(self):
        self.assertEqual(
            configured_names_set(["Overwatch", "", "  ", "steam"]),
            {"overwatch", "steam"},
        )

    def test_match_when_process_running(self):
        self.assertTrue(
            any_configured_process_running(
                ["overwatch", "steam"],
                ["bash", "Overwatch.exe", "Xorg"],
            )
        )

    def test_no_match_when_absent(self):
        self.assertFalse(
            any_configured_process_running(
                ["overwatch"],
                ["bash", "chrome", "Xorg"],
            )
        )

    def test_empty_config_never_matches(self):
        self.assertFalse(any_configured_process_running([], ["overwatch", "steam"]))
        self.assertFalse(any_configured_process_running(["", "  "], ["overwatch"]))

    def test_match_via_exe_basename_path(self):
        self.assertTrue(
            any_configured_process_running(
                ["firefox"],
                ["/usr/lib/firefox/firefox"],
            )
        )


class TestAutoPauseMonitor(unittest.TestCase):
    """Poller state machine: pause once, resume once, no double-fire."""

    def _make_monitor(self, enabled=True, apps=None, snapshot=None):
        apps = apps if apps is not None else ["overwatch"]
        config = {"enabled": enabled, "apps": list(apps), "interval": 5.0}
        pause_cb = MagicMock()
        resume_cb = MagicMock()
        running = snapshot if snapshot is not None else (lambda: set())

        def get_config():
            return config["enabled"], config["apps"], config["interval"]

        monitor = AutoPauseMonitor(
            get_config=get_config,
            on_pause=pause_cb,
            on_resume=resume_cb,
            process_snapshot=running if callable(running) else (lambda: running),
            use_glib=False,
        )
        return monitor, pause_cb, resume_cb, config

    def test_pause_when_match_appears(self):
        running = {"bash", "overwatch"}
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            snapshot=lambda: running,
        )

        matched = monitor.check_once()

        self.assertTrue(matched)
        self.assertTrue(monitor.paused)
        pause_cb.assert_called_once()
        resume_cb.assert_not_called()

    def test_no_pause_when_disabled(self):
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            enabled=False,
            snapshot=lambda: {"overwatch"},
        )

        matched = monitor.check_once()

        self.assertFalse(matched)
        self.assertFalse(monitor.paused)
        pause_cb.assert_not_called()
        resume_cb.assert_not_called()

    def test_no_pause_when_no_match(self):
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            snapshot=lambda: {"bash", "chrome"},
        )

        matched = monitor.check_once()

        self.assertFalse(matched)
        pause_cb.assert_not_called()

    def test_resume_when_match_clears(self):
        state = {"running": {"overwatch"}}
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            snapshot=lambda: state["running"],
        )

        monitor.check_once()
        pause_cb.assert_called_once()

        state["running"] = {"bash"}
        matched = monitor.check_once()

        self.assertFalse(matched)
        self.assertFalse(monitor.paused)
        resume_cb.assert_called_once()

    def test_no_double_pause(self):
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            snapshot=lambda: {"overwatch"},
        )

        monitor.check_once()
        monitor.check_once()

        pause_cb.assert_called_once()
        resume_cb.assert_not_called()

    def test_no_double_resume(self):
        state = {"running": {"overwatch"}}
        monitor, pause_cb, resume_cb, _ = self._make_monitor(
            snapshot=lambda: state["running"],
        )
        monitor.check_once()
        state["running"] = set()
        monitor.check_once()
        monitor.check_once()

        resume_cb.assert_called_once()

    def test_disabling_while_paused_resumes(self):
        state = {"enabled": True, "running": {"overwatch"}}
        pause_cb = MagicMock()
        resume_cb = MagicMock()

        def get_config():
            return state["enabled"], ["overwatch"], 5.0

        monitor = AutoPauseMonitor(
            get_config=get_config,
            on_pause=pause_cb,
            on_resume=resume_cb,
            process_snapshot=lambda: state["running"],
            use_glib=False,
        )
        monitor.check_once()
        pause_cb.assert_called_once()

        state["enabled"] = False
        monitor.check_once()
        resume_cb.assert_called_once()
        self.assertFalse(monitor.paused)

    def test_pause_callback_exception_does_not_raise(self):
        pause_cb = MagicMock(side_effect=RuntimeError("boom"))
        monitor = AutoPauseMonitor(
            get_config=lambda: (True, ["x"], 5.0),
            on_pause=pause_cb,
            process_snapshot=lambda: {"x"},
            use_glib=False,
        )
        # Should not raise
        monitor.check_once()
        self.assertTrue(monitor.paused)

    def test_start_stop_without_glib(self):
        monitor, _, _, _ = self._make_monitor()
        monitor.start()
        self.assertTrue(monitor.active)
        monitor.stop()
        self.assertFalse(monitor.active)
        monitor.shutdown()
        self.assertFalse(monitor.active)

    def test_start_is_idempotent(self):
        monitor, _, _, _ = self._make_monitor()
        monitor.start()
        monitor.start()
        self.assertTrue(monitor.active)
        monitor.stop()

    def test_resume_callback_exception_does_not_raise(self):
        state = {"running": {"x"}}
        resume_cb = MagicMock(side_effect=RuntimeError("resume boom"))
        monitor = AutoPauseMonitor(
            get_config=lambda: (True, ["x"], 5.0),
            on_pause=MagicMock(),
            on_resume=resume_cb,
            process_snapshot=lambda: state["running"],
            use_glib=False,
        )
        monitor.check_once()
        state["running"] = set()
        monitor.check_once()  # should not raise
        resume_cb.assert_called_once()
        self.assertFalse(monitor.paused)

    def test_read_config_handles_get_config_failure(self):
        monitor = AutoPauseMonitor(
            get_config=MagicMock(side_effect=RuntimeError("config broken")),
            use_glib=False,
        )
        matched = monitor.check_once()
        self.assertFalse(matched)
        self.assertFalse(monitor.paused)

    def test_read_config_coerces_invalid_apps_and_interval(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (True, "not-a-list", "bad"),
            process_snapshot=lambda: {"overwatch"},
            use_glib=False,
        )
        # invalid apps → treated as empty → no match even if overwatch runs
        self.assertFalse(monitor.check_once())

        monitor2 = AutoPauseMonitor(
            get_config=lambda: (True, ["overwatch"], 999),  # clamps to max 60
            process_snapshot=lambda: {"overwatch"},
            use_glib=False,
        )
        enabled, apps, interval = monitor2._read_config()
        self.assertTrue(enabled)
        self.assertEqual(list(apps), ["overwatch"])
        self.assertEqual(interval, 60.0)

        monitor3 = AutoPauseMonitor(
            get_config=lambda: (True, ["x"], 0),  # clamps to min 1
            use_glib=False,
        )
        _, _, interval3 = monitor3._read_config()
        self.assertEqual(interval3, 1.0)

    @patch("vocalinux.auto_pause_monitor.GLib", create=True)
    def test_start_with_glib_schedules_idle_then_timeout(self, _unused):
        """Drive real GLib scheduling helpers via mocked gi.repository.GLib."""
        mock_glib = MagicMock()
        mock_glib.idle_add.return_value = 11
        mock_glib.timeout_add_seconds.return_value = 22

        with patch.dict(
            "sys.modules",
            {"gi": MagicMock(), "gi.repository": MagicMock(GLib=mock_glib)},
        ):
            # Force re-import path used inside methods
            import vocalinux.auto_pause_monitor as apm

            monitor = apm.AutoPauseMonitor(
                get_config=lambda: (False, [], 5.0),
                process_snapshot=lambda: set(),
                use_glib=True,
            )
            monitor.start()
            self.assertTrue(monitor.active)
            mock_glib.idle_add.assert_called()
            # Simulate the idle callback (immediate first poll)
            result = monitor._glib_poll()
            self.assertFalse(result)  # always SOURCE_REMOVE style
            mock_glib.timeout_add_seconds.assert_called()
            monitor.stop()
            mock_glib.source_remove.assert_called()

    def test_glib_poll_when_stopped_returns_false(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (False, [], 5.0),
            use_glib=False,
        )
        monitor._running = False
        self.assertFalse(monitor._glib_poll())

    def test_glib_poll_swallows_check_once_errors(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (True, ["x"], 5.0),
            process_snapshot=MagicMock(side_effect=RuntimeError("psutil down")),
            use_glib=False,
        )
        monitor._running = True
        # use_glib False so reschedule is a no-op after poll
        self.assertFalse(monitor._glib_poll())

    def test_cancel_timeout_noop_when_none(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (False, [], 5.0),
            use_glib=False,
        )
        monitor._timeout_id = None
        monitor._cancel_timeout()  # no raise
        self.assertIsNone(monitor._timeout_id)

    def test_cancel_timeout_handles_source_remove_failure(self):
        mock_glib = MagicMock()
        mock_glib.source_remove.side_effect = RuntimeError("bad id")
        with patch.dict(
            "sys.modules",
            {"gi": MagicMock(), "gi.repository": MagicMock(GLib=mock_glib)},
        ):
            monitor = AutoPauseMonitor(
                get_config=lambda: (False, [], 5.0),
                use_glib=False,
            )
            monitor._timeout_id = 99
            monitor._cancel_timeout()
            self.assertIsNone(monitor._timeout_id)

    def test_schedule_skipped_when_not_running(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (False, [], 5.0),
            use_glib=True,
        )
        monitor._running = False
        monitor._schedule_next_poll(immediate=True)
        self.assertIsNone(monitor._timeout_id)

    def test_schedule_handles_glib_import_failure(self):
        monitor = AutoPauseMonitor(
            get_config=lambda: (False, [], 5.0),
            use_glib=True,
        )
        monitor._running = True
        real_import = __import__

        def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
            if name == "gi" or name.startswith("gi."):
                raise ImportError("no gi")
            return real_import(name, globals, locals, fromlist, level)

        with patch("builtins.__import__", side_effect=fake_import):
            monitor._schedule_next_poll(immediate=False)
        self.assertIsNone(monitor._timeout_id)


class TestCollectRunningProcessNames(unittest.TestCase):
    """collect_running_process_names uses real helper with mocked psutil."""

    def _psutil_mock(self):
        mock_psutil = MagicMock()
        mock_psutil.NoSuchProcess = type("NoSuchProcess", (Exception,), {})
        mock_psutil.AccessDenied = type("AccessDenied", (Exception,), {})
        mock_psutil.ZombieProcess = type("ZombieProcess", (Exception,), {})
        return mock_psutil

    def test_collects_name_and_exe_basename(self):
        import vocalinux.auto_pause_monitor as apm

        mock_psutil = self._psutil_mock()
        proc_ok = MagicMock()
        proc_ok.info = {"name": "steam", "exe": "/usr/bin/steam"}
        proc_no_name = MagicMock()
        proc_no_name.info = {"name": None, "exe": "/opt/game/Game.exe"}

        class Denied:
            @property
            def info(self):
                raise mock_psutil.AccessDenied()

        mock_psutil.process_iter.return_value = [proc_ok, Denied(), proc_no_name]

        with patch.dict("sys.modules", {"psutil": mock_psutil}):
            names = apm.collect_running_process_names()

        self.assertIn("steam", names)
        self.assertIn("Game.exe", names)

    def test_collect_skips_empty_entries(self):
        import vocalinux.auto_pause_monitor as apm

        mock_psutil = self._psutil_mock()
        proc = MagicMock()
        proc.info = {"name": "", "exe": None}
        mock_psutil.process_iter.return_value = [proc]
        with patch.dict("sys.modules", {"psutil": mock_psutil}):
            names = apm.collect_running_process_names()
        self.assertEqual(names, set())

    def test_collect_swallows_basename_errors(self):
        import vocalinux.auto_pause_monitor as apm

        mock_psutil = self._psutil_mock()
        proc = MagicMock()
        # os.path.basename on non-path-like may raise; helper must continue
        bad_exe = MagicMock()
        bad_exe.__fspath__ = MagicMock(side_effect=TypeError("nope"))
        proc.info = {"name": "ok", "exe": bad_exe}
        mock_psutil.process_iter.return_value = [proc]
        with patch.dict("sys.modules", {"psutil": mock_psutil}):
            with patch("os.path.basename", side_effect=TypeError("bad")):
                names = apm.collect_running_process_names()
        self.assertIn("ok", names)


class TestUnloadModelAndAutoPauseBlock(unittest.TestCase):
    """SpeechRecognitionManager.unload_model and start_recognition guard."""

    def _make_manager(self):
        from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager

        with patch.object(SpeechRecognitionManager, "__init__", lambda self: None):
            mgr = SpeechRecognitionManager()

        mgr.engine = "vosk"
        mgr.state = RecognitionState.IDLE
        mgr.model = MagicMock(name="model")
        mgr.recognizer = MagicMock(name="recognizer")
        mgr._model_initialized = True
        mgr._auto_paused = False
        mgr._idle_unloaded = False
        mgr._model_lock = MagicMock()
        mgr._model_lock.__enter__ = MagicMock(return_value=None)
        mgr._model_lock.__exit__ = MagicMock(return_value=False)
        mgr._http_session = None
        mgr._faster_whisper_engine = None
        mgr._reconnection_attempts = 0
        mgr._update_state = MagicMock()
        mgr._init_vosk = MagicMock()
        mgr._init_whisper = MagicMock()
        mgr._init_whispercpp = MagicMock()
        mgr._init_parakeet = MagicMock()
        mgr._init_faster_whisper = MagicMock()
        mgr._init_remote_api = MagicMock()
        mgr.stop_recognition = MagicMock()
        return mgr

    def test_unload_model_clears_resources_and_sets_flag(self):
        mgr = self._make_manager()

        mgr.unload_model()

        self.assertIsNone(mgr.model)
        self.assertIsNone(mgr.recognizer)
        self.assertFalse(mgr._model_initialized)
        self.assertTrue(mgr.is_auto_paused)
        mgr.stop_recognition.assert_not_called()  # already IDLE

    def test_unload_stops_active_recognition_first(self):
        mgr = self._make_manager()
        mgr.state = RecognitionState.LISTENING

        mgr.unload_model()

        mgr.stop_recognition.assert_called_once()
        self.assertTrue(mgr.is_auto_paused)

    def test_unload_closes_http_session(self):
        mgr = self._make_manager()
        session = MagicMock()
        mgr._http_session = session

        mgr.unload_model()

        session.close.assert_called_once()
        self.assertIsNone(mgr._http_session)

    def test_unload_http_session_close_error_is_swallowed(self):
        mgr = self._make_manager()
        session = MagicMock()
        session.close.side_effect = RuntimeError("already closed")
        mgr._http_session = session

        mgr.unload_model()

        self.assertIsNone(mgr._http_session)
        self.assertTrue(mgr.is_auto_paused)

    def test_unload_continues_if_gc_raises(self):
        mgr = self._make_manager()
        with patch("gc.collect", side_effect=RuntimeError("gc failed")):
            mgr.unload_model()
        self.assertTrue(mgr.is_auto_paused)
        self.assertIsNone(mgr.model)

    def test_start_recognition_blocked_when_auto_paused(self):
        mgr = self._make_manager()
        mgr._auto_paused = True
        # model not ready either; auto-pause should win first
        with (
            patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"),
            patch("vocalinux.speech_recognition.recognition_manager._show_notification"),
        ):
            self.assertFalse(mgr.start_recognition())

        self.assertEqual(mgr.state, RecognitionState.IDLE)
        # Should not have transitioned to listening
        mgr._update_state.assert_not_called()

    def test_unload_idle_keepalive_sets_idle_flag_not_auto_pause(self):
        mgr = self._make_manager()
        mgr.unload_model(reason="idle_keepalive")
        self.assertFalse(mgr.is_auto_paused)
        self.assertTrue(mgr.is_idle_unloaded)
        self.assertIsNone(mgr.model)

    def test_ensure_model_loaded_reloads_after_idle_unload(self):
        mgr = self._make_manager()
        mgr.unload_model(reason="idle_keepalive")

        def init_vosk():
            mgr.model = MagicMock()
            mgr._model_initialized = True

        mgr._init_vosk.side_effect = init_vosk
        self.assertTrue(mgr.ensure_model_loaded())
        self.assertFalse(mgr.is_idle_unloaded)
        self.assertTrue(mgr.model_ready)

    def test_ensure_model_loaded_blocked_when_auto_paused(self):
        mgr = self._make_manager()
        mgr.unload_model(reason="auto_pause")
        self.assertFalse(mgr.ensure_model_loaded())
        mgr._init_vosk.assert_not_called()

    def test_start_recognition_lazy_reloads_after_idle_unload(self):
        mgr = self._make_manager()
        mgr.unload_model(reason="idle_keepalive")

        def init_vosk():
            mgr.model = MagicMock()
            mgr._model_initialized = True

        mgr._init_vosk.side_effect = init_vosk
        mgr.should_record = False
        mgr._recognition_mode = "toggle"
        mgr.audio_buffer = []
        mgr._segment_queue = None
        mgr.audio_thread = None
        mgr.recognition_thread = None

        with (
            patch("vocalinux.speech_recognition.recognition_manager.play_start_sound"),
            patch(
                "vocalinux.speech_recognition.recognition_manager.threading.Thread"
            ) as mock_thread,
        ):
            mock_thread.return_value = MagicMock()
            mgr.start_recognition()

        self.assertFalse(mgr.is_idle_unloaded)
        mgr._update_state.assert_called()

    def test_reinitialize_clears_auto_pause_and_reloads(self):
        mgr = self._make_manager()
        mgr._auto_paused = True
        mgr.model = None
        mgr._model_initialized = False

        # Simulate successful reinit setting model ready
        def init_vosk():
            mgr.model = MagicMock()
            mgr._model_initialized = True

        mgr._init_vosk.side_effect = init_vosk

        mgr.reinitialize_after_resume()

        mgr._init_vosk.assert_called_once()
        self.assertFalse(mgr.is_auto_paused)
        self.assertIsNotNone(mgr.model)

    def test_pause_then_resume_via_monitor_and_engine(self):
        """End-to-end of the shipped path: match → unload → clear → reinit."""
        mgr = self._make_manager()
        state = {"running": {"overwatch"}}

        def on_pause():
            mgr.unload_model()

        def on_resume():
            def init_vosk():
                mgr.model = MagicMock()
                mgr._model_initialized = True

            mgr._init_vosk.side_effect = init_vosk
            mgr.reinitialize_after_resume()

        monitor = AutoPauseMonitor(
            get_config=lambda: (True, ["overwatch"], 5.0),
            on_pause=on_pause,
            on_resume=on_resume,
            process_snapshot=lambda: state["running"],
            use_glib=False,
        )

        monitor.check_once()
        self.assertTrue(mgr.is_auto_paused)
        self.assertIsNone(mgr.model)

        state["running"] = set()
        monitor.check_once()
        self.assertFalse(mgr.is_auto_paused)
        self.assertIsNotNone(mgr.model)
        self.assertTrue(mgr._model_initialized)


class TestAutoPauseConfigDefaults(unittest.TestCase):
    """DEFAULT_CONFIG schema for auto_pause section."""

    def test_default_config_has_auto_pause_section(self):
        self.assertIn("auto_pause", DEFAULT_CONFIG)
        section = DEFAULT_CONFIG["auto_pause"]
        self.assertFalse(section["enabled"])
        self.assertEqual(section["apps"], [])
        self.assertIn("poll_interval_seconds", section)

    def test_default_config_has_model_keepalive_section(self):
        self.assertIn("model_keepalive", DEFAULT_CONFIG)
        section = DEFAULT_CONFIG["model_keepalive"]
        self.assertFalse(section["enabled"])
        self.assertEqual(section["idle_timeout_seconds"], 300)

    def test_config_manager_round_trip_auto_pause(self):
        import json
        import tempfile
        from unittest.mock import patch

        from vocalinux.ui.config_manager import ConfigManager

        with tempfile.TemporaryDirectory() as tmp:
            config_dir = os.path.join(tmp, "vocalinux")
            config_file = os.path.join(config_dir, "config.json")
            os.makedirs(config_dir, exist_ok=True)

            with (
                patch("vocalinux.ui.config_manager.CONFIG_DIR", config_dir),
                patch("vocalinux.ui.config_manager.CONFIG_FILE", config_file),
            ):
                cm = ConfigManager()
                self.assertFalse(cm.get_bool("auto_pause", "enabled", True))
                self.assertEqual(cm.get("auto_pause", "apps", None), [])

                cm.set("auto_pause", "enabled", True)
                cm.set("auto_pause", "apps", ["Overwatch", "steam"])
                self.assertTrue(cm.save_config())

                cm2 = ConfigManager()
                self.assertTrue(cm2.get_bool("auto_pause", "enabled", False))
                self.assertEqual(cm2.get("auto_pause", "apps"), ["Overwatch", "steam"])

                with open(config_file) as f:
                    raw = json.load(f)
                self.assertIn("auto_pause", raw)
                self.assertEqual(raw["auto_pause"]["apps"], ["Overwatch", "steam"])


# Tray wiring tests live in test_suspend_handler.py (TestTrayAutoPauseWiring)
# so this module never imports tray_indicator and cannot pin a stale gi/GLib
# mock for later suite files.


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