InputRemapper/tests/unit/test_focus_service_client.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# input-remapper - GUI for device specific keyboard mappings
# Copyright (C) 2025 sezanzeb <b8x45ygc9@mozmail.com>
#
# This file is part of input-remapper.
#
# input-remapper is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# input-remapper is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with input-remapper.  If not, see <https://www.gnu.org/licenses/>.

import unittest
from unittest.mock import MagicMock, patch

from inputremapper.gui.focus_service_client import (
    FocusServiceClient,
    ensure_focus_service_running,
)
from tests.lib.test_setup import test_setup


@test_setup
class TestFocusServiceClient(unittest.TestCase):
    def setUp(self) -> None:
        self.bus = MagicMock()
        self.proxy = MagicMock()
        self.bus.get_proxy.return_value = self.proxy
        self.client = FocusServiceClient(bus=self.bus)

    # -- get_status --------------------------------------------------------

    def test_get_status_success(self):
        self.proxy.get_status.return_value = '{"backend": "xorg", "enabled": true}'
        status = self.client.get_status()
        self.assertEqual(status, {"backend": "xorg", "enabled": True})

    def test_get_status_failure_returns_none_and_resets(self):
        self.proxy.get_status.side_effect = Exception("no service")
        self.assertIsNone(self.client.get_status())
        # the cached proxy is dropped so the next call reconnects
        self.assertIsNone(self.client._proxy)

    def test_get_focused_app_success(self):
        self.proxy.get_focused_app.return_value = (
            '{"app_id": "firefox", "title": "Moz"}'
        )
        self.assertEqual(
            self.client.get_focused_app(), {"app_id": "firefox", "title": "Moz"}
        )

    def test_get_focused_app_failure_returns_empty(self):
        self.proxy.get_focused_app.side_effect = Exception("boom")
        self.assertEqual(
            self.client.get_focused_app(), {"app_id": "", "title": ""}
        )

    # -- reload / set_enabled (fire and forget) ----------------------------

    def test_reload_success(self):
        self.client.reload()
        self.proxy.reload.assert_called_once()

    def test_reload_failure_is_swallowed(self):
        self.proxy.reload.side_effect = Exception("boom")
        # must not raise
        self.client.reload()
        self.assertIsNone(self.client._proxy)

    def test_set_enabled(self):
        self.client.set_enabled(True)
        self.proxy.set_enabled.assert_called_once_with(True)

    def test_set_enabled_failure_is_swallowed(self):
        self.proxy.set_enabled.side_effect = Exception("boom")
        self.client.set_enabled(False)
        self.assertIsNone(self.client._proxy)

    # -- focus_changed signal ---------------------------------------------

    def test_connect_focus_changed_success(self):
        received = []
        ok = self.client.connect_focus_changed(received.append)
        self.assertTrue(ok)
        self.proxy.focus_changed.connect.assert_called_once()

        # the wrapped handler parses the json payload before calling back
        handler = self.proxy.focus_changed.connect.call_args[0][0]
        handler('{"app_id": "firefox", "title": "Moz"}')
        self.assertEqual(received, [{"app_id": "firefox", "title": "Moz"}])

    def test_connect_focus_changed_invalid_payload_does_not_raise(self):
        received = []
        self.client.connect_focus_changed(received.append)
        handler = self.proxy.focus_changed.connect.call_args[0][0]
        # invalid json must be swallowed (no callback, no exception)
        handler("not json")
        self.assertEqual(received, [])

    def test_connect_focus_changed_failure(self):
        self.proxy.focus_changed.connect.side_effect = Exception("boom")
        ok = self.client.connect_focus_changed(lambda app: None)
        self.assertFalse(ok)
        self.assertIsNone(self.client._proxy)

    def test_disconnect_without_handler_is_noop(self):
        self.client.disconnect_focus_changed()
        self.proxy.focus_changed.disconnect.assert_not_called()

    def test_disconnect_after_connect(self):
        self.client.connect_focus_changed(lambda app: None)
        self.client.disconnect_focus_changed()
        self.proxy.focus_changed.disconnect.assert_called_once()
        self.assertIsNone(self.client._focus_changed_handler)

    def test_disconnect_failure_clears_handler(self):
        self.client.connect_focus_changed(lambda app: None)
        self.proxy.focus_changed.disconnect.side_effect = Exception("boom")
        self.client.disconnect_focus_changed()
        # handler is cleared even if the disconnect failed
        self.assertIsNone(self.client._focus_changed_handler)


@test_setup
class TestEnsureFocusServiceRunning(unittest.TestCase):
    def test_does_not_start_when_already_running(self):
        with patch(
            "inputremapper.gui.focus_service_client.ProcessUtils."
            "count_python_processes",
            return_value=1,
        ), patch(
            "inputremapper.gui.focus_service_client.subprocess.Popen"
        ) as popen:
            ensure_focus_service_running()
        popen.assert_not_called()

    def test_starts_when_not_running(self):
        with patch(
            "inputremapper.gui.focus_service_client.ProcessUtils."
            "count_python_processes",
            return_value=0,
        ), patch(
            "inputremapper.gui.focus_service_client.subprocess.Popen"
        ) as popen:
            ensure_focus_service_running()
        popen.assert_called_once()
        self.assertEqual(
            popen.call_args[0][0], ["input-remapper-focus-service"]
        )

    def test_start_failure_is_swallowed(self):
        with patch(
            "inputremapper.gui.focus_service_client.ProcessUtils."
            "count_python_processes",
            return_value=0,
        ), patch(
            "inputremapper.gui.focus_service_client.subprocess.Popen",
            side_effect=OSError("boom"),
        ):
            # must not raise
            ensure_focus_service_running()


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