Add phone hotspot uplink
Prefer a configured phone hotspot for outbound synchronization, preserve secure write-only credentials, and recover through home WiFi or the local Pi hotspot. Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
@@ -0,0 +1,74 @@
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, mock_open, patch
|
||||
|
||||
from modules import app_network
|
||||
|
||||
|
||||
class AppNetworkTest(unittest.TestCase):
|
||||
@patch("modules.app_network.subprocess.run")
|
||||
@patch(
|
||||
"modules.app_network._helper_command",
|
||||
return_value=["sudo", "helper", "configure"],
|
||||
)
|
||||
def test_configuration_secret_is_sent_only_over_stdin(
|
||||
self,
|
||||
_helper_command,
|
||||
run,
|
||||
):
|
||||
run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
||||
|
||||
app_network.configure_phone_hotspot("My Phone", "private-password")
|
||||
|
||||
command = run.call_args.args[0]
|
||||
self.assertNotIn("private-password", command)
|
||||
self.assertEqual(
|
||||
json.loads(run.call_args.kwargs["input"]),
|
||||
{"ssid": "My Phone", "password": "private-password"},
|
||||
)
|
||||
|
||||
@patch("modules.app_network.subprocess.run")
|
||||
def test_connection_request_is_nonblocking(self, run):
|
||||
run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
||||
|
||||
app_network.request_phone_connection()
|
||||
|
||||
self.assertIn("--no-block", run.call_args.args[0])
|
||||
self.assertIn("yaris-uplink-wait.service", run.call_args.args[0])
|
||||
self.assertEqual(run.call_args.kwargs["timeout"], 5)
|
||||
|
||||
@patch("modules.app_network._phone_ssid", return_value="My Phone")
|
||||
@patch(
|
||||
"modules.app_network._nmcli",
|
||||
return_value="phone-hotspot\n100 (connected)\n4 (full)\n172.20.10.2/28",
|
||||
)
|
||||
def test_status_reports_phone_uplink_without_secret(self, _nmcli, _ssid):
|
||||
with patch("builtins.open", mock_open(read_data='{"action":"connected_phone"}')):
|
||||
status = app_network.get_status()
|
||||
|
||||
self.assertEqual(status["mode"], "phone")
|
||||
self.assertEqual(status["phone_ssid"], "My Phone")
|
||||
self.assertTrue(status["internet_reachable"])
|
||||
self.assertTrue(status["large_upload_allowed"])
|
||||
self.assertNotIn("password", status)
|
||||
|
||||
@patch("modules.app_network._phone_ssid", return_value="")
|
||||
@patch(
|
||||
"modules.app_network._nmcli",
|
||||
return_value="Hotspot\n100 (connected)\n2 (portal)\n10.42.0.1/24",
|
||||
)
|
||||
def test_local_hotspot_does_not_claim_internet_or_large_uploads(
|
||||
self,
|
||||
_nmcli,
|
||||
_ssid,
|
||||
):
|
||||
with patch("builtins.open", side_effect=FileNotFoundError):
|
||||
status = app_network.get_status()
|
||||
|
||||
self.assertEqual(status["mode"], "local_hotspot")
|
||||
self.assertFalse(status["internet_reachable"])
|
||||
self.assertFalse(status["large_upload_allowed"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,245 @@
|
||||
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)
|
||||
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")
|
||||
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",
|
||||
side_effect=[False, True],
|
||||
)
|
||||
def test_local_hotspot_is_restored_when_home_is_unavailable(
|
||||
self,
|
||||
activate,
|
||||
write_status,
|
||||
):
|
||||
self.assertTrue(yaris_uplink.restore_fallback())
|
||||
|
||||
self.assertEqual(
|
||||
activate.call_args_list,
|
||||
[
|
||||
call("home-wifi"),
|
||||
call("Hotspot"),
|
||||
],
|
||||
)
|
||||
write_status.assert_called_once_with("restored_local_hotspot")
|
||||
|
||||
|
||||
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)
|
||||
def test_wait_timeout_preserves_current_client_connection(
|
||||
self,
|
||||
_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_current_connection_preserved",
|
||||
"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")
|
||||
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()
|
||||
Reference in New Issue
Block a user