yaris-obd/tests/test_yaris_uplink.py

import os
import stat
import tempfile
import unittest
from unittest.mock import MagicMock, call, patch

from deploy import yaris_uplink


class UplinkCredentialTest(unittest.TestCase):
    def test_rejects_invalid_credentials(self):
        with self.assertRaises(yaris_uplink.UplinkError):
            yaris_uplink.validate_credentials(
                {"ssid": "", "password": "long-enough"}
            )
        with self.assertRaises(yaris_uplink.UplinkError):
            yaris_uplink.validate_credentials(
                {"ssid": "Phone", "password": "short"}
            )

    def test_keyfile_preserves_edge_spaces_and_backslashes(self):
        self.assertEqual(
            yaris_uplink.keyfile_escape(r" two\spaces "),
            r"\stwo\\spaces\s",
        )

    @patch("deploy.yaris_uplink.write_status")
    @patch(
        "deploy.yaris_uplink.active_connection",
        return_value="home-wifi",
    )
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    @patch("deploy.yaris_uplink.run_nmcli")
    def test_configure_writes_root_only_keyfile_without_secret_in_nmcli(
        self,
        run_nmcli,
        _connection_exists,
        _active_connection,
        _write_status,
    ):
        with tempfile.TemporaryDirectory() as directory:
            keyfile_path = os.path.join(directory, "phone.nmconnection")
            payload = b'{"ssid":"My Phone","password":"private-password"}'
            with (
                patch.object(yaris_uplink, "PHONE_KEYFILE", keyfile_path),
                patch.object(
                    yaris_uplink.sys,
                    "stdin",
                    MagicMock(buffer=MagicMock(read=MagicMock(return_value=payload))),
                ),
            ):
                yaris_uplink.configure()

            with open(keyfile_path, encoding="utf-8") as handle:
                keyfile = handle.read()
            mode = stat.S_IMODE(os.stat(keyfile_path).st_mode)

        self.assertEqual(mode, 0o600)
        self.assertIn("ssid=My Phone", keyfile)
        self.assertIn("psk=private-password", keyfile)
        self.assertIn("dns=8.8.8.8;1.1.1.1;", keyfile)
        self.assertIn("ignore-auto-dns=true", keyfile)
        for call in run_nmcli.call_args_list:
            self.assertNotIn("private-password", call.args)


class UplinkSelectionTest(unittest.TestCase):
    @patch("deploy.yaris_uplink.time.sleep")
    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.run_nmcli")
    @patch("deploy.yaris_uplink.internet_reachable", return_value=True)
    @patch("deploy.yaris_uplink.phone_is_visible", return_value=True)
    @patch("deploy.yaris_uplink.activate", return_value=True)
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    @patch("deploy.yaris_uplink.active_connection", return_value="Hotspot-local")
    def test_phone_is_preferred_over_home_from_local_hotspot(
        self,
        _active,
        _exists,
        activate,
        _visible,
        _internet,
        _run_nmcli,
        write_status,
        _sleep,
    ):
        self.assertTrue(yaris_uplink.check_uplink())

        activate.assert_called_once_with("phone-hotspot")
        write_status.assert_called_once_with("connected_phone")

    @patch("deploy.yaris_uplink.time.sleep")
    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.run_nmcli")
    @patch("deploy.yaris_uplink.phone_is_visible", return_value=False)
    @patch("deploy.yaris_uplink.activate")
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    @patch("deploy.yaris_uplink.active_connection", return_value="home-wifi")
    def test_missing_phone_does_not_disrupt_home_wifi(
        self,
        _active,
        _exists,
        activate,
        _visible,
        _run_nmcli,
        write_status,
        _sleep,
    ):
        self.assertTrue(yaris_uplink.check_uplink())

        activate.assert_not_called()
        write_status.assert_called_once_with(
            "phone_not_in_range_current_connection_preserved"
        )

    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.activate", return_value=True)
    @patch("deploy.yaris_uplink.internet_reachable", return_value=False)
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    @patch(
        "deploy.yaris_uplink.active_connection",
        return_value="phone-hotspot",
    )
    def test_home_is_fallback_when_phone_has_no_internet(
        self,
        _active,
        _exists,
        _internet,
        activate,
        write_status,
    ):
        self.assertTrue(yaris_uplink.check_uplink())

        activate.assert_called_once_with("home-wifi")
        write_status.assert_called_once_with(
            "phone_without_internet_connected_home"
        )

    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.activate", return_value=False)
    def test_home_unavailable_reports_no_connection(
        self,
        activate,
        write_status,
    ):
        self.assertTrue(yaris_uplink.restore_fallback())

        activate.assert_called_once_with("home-wifi")
        write_status.assert_called_once_with(
            "no_connection", "No home WiFi or phone hotspot available."
        )


class UplinkWaitTest(unittest.TestCase):
    @patch("deploy.yaris_uplink.check_uplink", return_value=True)
    @patch("deploy.yaris_uplink.phone_is_visible", return_value=True)
    @patch("deploy.yaris_uplink.run_nmcli")
    @patch("deploy.yaris_uplink.time.sleep")
    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.active_connection", return_value="home-wifi")
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    def test_wait_connects_after_phone_appears(
        self,
        _exists,
        _active,
        write_status,
        _sleep,
        _run_nmcli,
        _visible,
        check_uplink,
    ):
        self.assertTrue(yaris_uplink.wait_for_phone())

        write_status.assert_called_once_with("waiting_for_phone")
        check_uplink.assert_called_once_with()

    @patch(
        "deploy.yaris_uplink.time.monotonic",
        side_effect=[100, 221],
    )
    @patch("deploy.yaris_uplink.run_nmcli")
    @patch("deploy.yaris_uplink.time.sleep")
    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.active_connection", return_value="home-wifi")
    @patch("deploy.yaris_uplink.connection_exists", return_value=True)
    @patch("deploy.yaris_uplink._connect_home_or_idle", return_value=True)
    def test_wait_timeout_restores_home_fallback(
        self,
        _home,
        _exists,
        _active,
        write_status,
        _sleep,
        _run_nmcli,
        _monotonic,
    ):
        self.assertTrue(yaris_uplink.wait_for_phone())

        write_status.assert_any_call("waiting_for_phone")
        write_status.assert_called_with(
            "phone_wait_timed_out_fallback_restored",
            "Phone hotspot did not appear within two minutes.",
        )


class UplinkTransitionTest(unittest.TestCase):
    @patch("deploy.yaris_uplink.write_status")
    @patch("deploy.yaris_uplink.active_connection", return_value="Hotspot-local")
    def test_records_networkmanager_fallback(self, _active, write_status):
        yaris_uplink.record_current_connection()

        write_status.assert_called_once_with("network_local_hotspot_active")

    @patch("deploy.yaris_uplink.time.time", return_value=123)
    @patch("deploy.yaris_uplink.ipv4_address", return_value="10.21.0.112")
    @patch(
        "deploy.yaris_uplink.active_connection",
        return_value="phone-hotspot",
    )
    def test_status_preserves_last_phone_address(
        self,
        _active,
        _address,
        _time,
    ):
        with tempfile.TemporaryDirectory() as directory:
            status_path = os.path.join(directory, "status.json")
            with patch.object(yaris_uplink, "STATUS_PATH", status_path):
                yaris_uplink.write_status("network_phone_connected")
                with patch(
                    "deploy.yaris_uplink.active_connection",
                    return_value="home-wifi",
                ):
                    yaris_uplink.write_status("network_home_connected")
                with open(status_path, encoding="utf-8") as handle:
                    status = yaris_uplink.json.load(handle)

        self.assertEqual(status["last_phone_ip"], "10.21.0.112")


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