rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 02:48:38 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5
wrolpi/controller/test/test_admin.py
lrnselfreliance d701846254
Some checks are pending
CodeQL / Analyze (push) Waiting to run
Details
Controller Tests / controller-tests (push) Waiting to run
Details
Jest Tests / jest-tests (push) Waiting to run
Details
Hotspot captive portal: welcome page with the WROLPi's address (#709) 
* Hotspot captive portal: welcome page with the WROLPi's address

Devices that join the hotspot probe a fixed plain-HTTP URL to check for
Internet access.  NetworkManager's shared-mode dnsmasq now resolves those
probe hostnames (and `wrolpi`) to the hotspot address via an include the
Controller writes when the hotspot starts, the probes land on the Controller
(port 80), and are redirected to /portal: "This is your WROLPi, its address is
10.42.0.1, open WROLPi / open the Controller".

Only the probe hostnames are hijacked so the hotspot keeps forwarding
everything else when the WROLPi has an uplink.  The include is corrected and
the connection re-activated once if NetworkManager picked a non-default
subnet.  Writing the include is best-effort; the hotspot starts regardless.

`hotspot.captive_portal` (controller.yaml, default true) toggles the probe
redirects and the include; like the other hotspot settings it applies on the
next hotspot start, so saving while the hotspot is up reports
restart_required and both Controller UIs say so.  The status endpoint gains
the hotspot address and portal URL, shown in both UIs.  repair.sh only
creates the include directory so a disabled setting survives upgrades.
Native only: Docker has no hotspot.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: address review

* React: re-read the hotspot address whenever the HotspotToggle starts or
  stops the hotspot, matching the fallback UI's refresh-on-status.
* A WPA3 start that reverts to WPA2 leaves the hotspot up, so it now gets
  the captive portal address correction too.
* The corrective `nmcli connection up` logs a warning on a non-zero exit;
  dnsmasq would otherwise keep the old address silently.
* HotspotStatusResponse.captive_portal defaults to true like everywhere else.
* Trim design narrative from the module docstring and React comments.
* The include-before-activate test now observes the include at the moment
  of the `nmcli ... hotspot` call instead of inferring from call order.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: never forward any record type for probe hosts

`address=` only synthesizes A/AAAA answers.  iOS also asks for the HTTPS
record type; dnsmasq forwarded that upstream, Apple's answer aliased to
captive.g.aaplimg.com, and the phone fetched the genuine "Success" from
Apple instead of the portal.  Each probe host now also gets `local=`, so
dnsmasq answers every type itself, and the CDN alias is a probe host too.

Observed with dnsmasq query logging on a hotspot with an eth0 uplink.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: let acknowledged clients finish so the OS offers "Done"

A phone's sign-in sheet only shows "Done" once the OS re-probes and gets
the genuine success reply.  The Controller now remembers each client (by
hotspot IP) that has loaded the portal page and answers its later probes
with the reply each OS expects (Apple "Success" HTML, Android 204, Windows
"Microsoft Connect Test", ...).  The set is cleared when the hotspot
starts so a new session shows the page again.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: acknowledge on Continue, not on view

iOS fetches the portal page in the background before opening the sheet,
then re-probes after the sheet's own navigation.  Acknowledging on view
made that re-probe return "Success" as the sheet's content.  Phones only
re-check the network (and offer "Done") after a full-page navigation, so
the welcome page now has a Continue button leading to /portal/continue,
which is the only place a client is acknowledged.  The Continue page says
to tap Done and links to WROLPi and the Controller.

Also hijack clients3.google.com (older Android probe host).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* help: bump wrolpi-help to the captive portal and Speed Test docs

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: comments describe the Continue design; clear acks on stop

Three comments still described acknowledging on view, which c72a7f7b
removed.  Acknowledged clients are now also cleared when the hotspot
stops, so a lease reused in a later session sees the portal page.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-13 20:10:02 -06:00

1830 lines
87 KiB
Python
Raw Blame History

"""
Unit tests for controller.lib.admin module.
"""
from unittest import mock
import pytest
import yaml
from controller.lib.config import reload_config_from_drive
from controller.lib import captive_portal
from controller.lib.admin import (
apply_timezone_from_config,
block_bluetooth,
stop_hotspot,
disable_throttle,
unblock_bluetooth,
start_hotspot,
enable_throttle,
get_bluetooth_status,
get_hotspot_status,
get_hotspot_status_dict,
get_throttle_status,
get_timezone_status_dict,
get_device_hotspot_protocols,
get_hotspot_device,
get_hotspot_password,
get_hotspot_protocol,
get_hotspot_ssid,
get_wifi_devices,
migrate_hotspot_settings_from_wrolpi_config,
reboot_system,
restart_all_services,
set_timezone,
shutdown_system,
update_hotspot_settings,
get_desktop_status,
get_desktop_status_dict,
start_desktop,
stop_desktop,
get_vnc_backends,
get_vnc_status,
get_vnc_status_dict,
start_vnc,
stop_vnc,
BluetoothStatus,
DesktopStatus,
VncStatus,
HotspotStatus,
GovernorStatus,
)
class TestGetBluetoothStatus:
"""Tests for get_bluetooth_status function."""
@pytest.mark.asyncio
async def test_returns_bluetooth_status_enum(self):
"""Should return a BluetoothStatus enum value."""
result = await get_bluetooth_status()
assert isinstance(result, BluetoothStatus)
@pytest.mark.asyncio
async def test_returns_unknown_in_docker_mode(self):
"""Should return unknown when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = await get_bluetooth_status()
assert result == BluetoothStatus.unknown
@pytest.mark.asyncio
async def test_returns_unavailable_when_rfkill_not_found(self):
"""Should return unavailable when rfkill is not installed."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
result = await get_bluetooth_status()
assert result == BluetoothStatus.unavailable
@pytest.mark.asyncio
async def test_returns_on_when_unblocked(self):
"""Should return on when Bluetooth is unblocked."""
rfkill_json = '{"rfkilldevices": [{"id": 0, "type": "bluetooth", "device": "hci0", "soft": "unblocked", "hard": "unblocked"}]}'
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (rfkill_json.encode(), b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_bluetooth_status()
assert result == BluetoothStatus.on
@pytest.mark.asyncio
async def test_returns_off_when_soft_blocked(self):
"""Should return off when Bluetooth is soft blocked."""
rfkill_json = '{"rfkilldevices": [{"id": 0, "type": "bluetooth", "device": "hci0", "soft": "blocked", "hard": "unblocked"}]}'
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (rfkill_json.encode(), b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_bluetooth_status()
assert result == BluetoothStatus.off
@pytest.mark.asyncio
async def test_returns_off_when_hard_blocked(self):
"""Should return off when Bluetooth is hard blocked."""
rfkill_json = '{"rfkilldevices": [{"id": 0, "type": "bluetooth", "device": "hci0", "soft": "unblocked", "hard": "blocked"}]}'
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (rfkill_json.encode(), b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_bluetooth_status()
assert result == BluetoothStatus.off
@pytest.mark.asyncio
async def test_returns_unavailable_when_no_bluetooth_device(self):
"""Should return unavailable when no Bluetooth devices exist."""
rfkill_json = '{"rfkilldevices": [{"id": 0, "type": "wlan", "device": "phy0", "soft": "unblocked", "hard": "unblocked"}]}'
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (rfkill_json.encode(), b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_bluetooth_status()
assert result == BluetoothStatus.unavailable
@pytest.mark.asyncio
async def test_handles_timeout(self):
"""Should return unknown on timeout."""
import asyncio
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
mock_proc = mock.AsyncMock()
mock_proc.communicate.side_effect = asyncio.TimeoutError()
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
# wait_for wraps communicate, but we mock at the proc level.
# The timeout is handled by asyncio.wait_for in the function.
result = await get_bluetooth_status()
assert result == BluetoothStatus.unknown
class TestBluetoothBlockUnblock:
"""Tests for unblock_bluetooth / block_bluetooth — they share the same shape."""
@pytest.mark.asyncio
@pytest.mark.parametrize("func,rfkill_arg", [
(unblock_bluetooth, "unblock"),
(block_bluetooth, "block"),
])
async def test_calls_rfkill_with_correct_action(self, func, rfkill_arg):
"""Each function should invoke rfkill with the expected action."""
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (b"", b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc) as mock_exec:
result = await func()
assert result["success"] is True
mock_exec.assert_called_once_with(
"/usr/sbin/rfkill", rfkill_arg, "bluetooth",
stdout=mock.ANY,
stderr=mock.ANY,
)
@pytest.mark.asyncio
@pytest.mark.parametrize("func", [unblock_bluetooth, block_bluetooth])
async def test_handles_rfkill_not_found(self, func):
"""Should handle rfkill not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
result = await func()
assert result["success"] is False
def _mock_systemctl_show_proc(load_state: str, active_state: str) -> mock.AsyncMock:
"""A mocked `systemctl show` process reporting the given unit states."""
mock_proc = mock.AsyncMock()
output = f"LoadState={load_state}\nActiveState={active_state}\n"
mock_proc.communicate.return_value = (output.encode(), b"")
mock_proc.returncode = 0
return mock_proc
class TestGetDesktopStatus:
"""Tests for get_desktop_status function."""
@pytest.mark.asyncio
async def test_returns_unknown_in_docker_mode(self):
"""Should return unknown when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = await get_desktop_status()
assert result == DesktopStatus.unknown
@pytest.mark.asyncio
@pytest.mark.parametrize("load_state,active_state,expected", [
("loaded", "active", DesktopStatus.on),
("loaded", "inactive", DesktopStatus.off),
("loaded", "failed", DesktopStatus.off),
("not-found", "inactive", DesktopStatus.unavailable),
])
async def test_parses_systemctl_show(self, load_state, active_state, expected):
"""Should map the display manager's LoadState/ActiveState to a DesktopStatus."""
mock_proc = _mock_systemctl_show_proc(load_state, active_state)
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_desktop_status()
assert result == expected
@pytest.mark.asyncio
async def test_returns_unavailable_when_systemctl_not_found(self):
"""Should return unavailable when systemctl is not installed."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
result = await get_desktop_status()
assert result == DesktopStatus.unavailable
@pytest.mark.asyncio
async def test_returns_unknown_when_systemctl_fails(self):
"""Should return unknown when systemctl returns an error."""
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (b"", b"some error")
mock_proc.returncode = 1
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await get_desktop_status()
assert result == DesktopStatus.unknown
class TestGetDesktopStatusDict:
"""Tests for get_desktop_status_dict function."""
@pytest.mark.asyncio
@pytest.mark.parametrize("status,running,available", [
(DesktopStatus.on, True, True),
(DesktopStatus.off, False, True),
(DesktopStatus.unavailable, False, False),
(DesktopStatus.unknown, False, False),
])
async def test_maps_status_to_flags(self, status, running, available):
"""Should map each DesktopStatus to running/available flags."""
with mock.patch("controller.lib.admin.get_desktop_status", return_value=status):
result = await get_desktop_status_dict()
assert result["running"] is running
assert result["available"] is available
if available:
assert result["reason"] is None
else:
assert result["reason"]
class TestDesktopStartStop:
"""Tests for start_desktop / stop_desktop — they share the same shape."""
@pytest.mark.asyncio
@pytest.mark.parametrize("func,systemctl_action", [
(start_desktop, "start"),
(stop_desktop, "stop"),
])
async def test_calls_systemctl_with_correct_action(self, func, systemctl_action):
"""Each function should invoke systemctl with the expected action; never enable/disable."""
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (b"", b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc) as mock_exec:
result = await func()
assert result["success"] is True
assert mock_exec.call_args_list[0] == mock.call(
"systemctl", systemctl_action, "display-manager.service",
stdout=mock.ANY,
stderr=mock.ANY,
)
# Fail open: never enable/disable the unit or change the default target.
forbidden = {"enable", "disable", "mask", "unmask", "set-default", "isolate"}
for call in mock_exec.call_args_list:
assert not forbidden.intersection(call.args), call.args
@pytest.mark.asyncio
@pytest.mark.parametrize("func", [start_desktop, stop_desktop])
async def test_reports_systemctl_error(self, func):
"""Should report systemctl's stderr when the action fails."""
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (b"", b"Unit display-manager.service not loaded.")
mock_proc.returncode = 5
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await func()
assert result["success"] is False
assert "not loaded" in result["error"]
@pytest.mark.asyncio
@pytest.mark.parametrize("func", [start_desktop, stop_desktop])
async def test_handles_systemctl_not_found(self, func):
"""Should handle systemctl not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
result = await func()
assert result["success"] is False
class TestStopDesktopDropsToConsole:
"""Stopping the desktop must leave the physical display on a usable terminal.
The display manager owns its own VT (7 on Raspberry Pi OS); stopping it leaves
that VT blank with only a cursor, so the display is switched to the console VT.
"""
@staticmethod
def _proc(returncode=0, stdout=b"", stderr=b""):
proc = mock.AsyncMock()
proc.communicate.return_value = (stdout, stderr)
proc.returncode = returncode
return proc
@pytest.mark.asyncio
async def test_activates_and_repaints_console(self):
"""Should check for a getty, activate the console VT, then bounce to repaint it.
The console VT is normally already in the foreground when the display manager
stops, but the monitor is left black until something forces a redraw.
"""
procs = [self._proc(), self._proc(0, b"active\n"), self._proc(), self._proc(), self._proc()]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs) as mock_exec:
with mock.patch("builtins.open", mock.mock_open()) as mock_console:
result = await stop_desktop()
assert result["success"] is True
assert result["switched_to_console"] is True
commands = [call.args for call in mock_exec.call_args_list]
assert commands[0] == ("systemctl", "stop", "display-manager.service")
assert commands[1] == ("systemctl", "is-active", "getty@tty1.service")
# Land on the console, bounce away, and come back so it is repainted.
assert commands[2:] == [("/usr/bin/chvt", "1"), ("/usr/bin/chvt", "2"), ("/usr/bin/chvt", "1")]
mock_console.assert_called_once_with("/dev/tty1", "w")
written = mock_console().write.call_args.args[0]
assert "stopped the desktop" in written
assert "systemctl start display-manager" in written
@pytest.mark.asyncio
async def test_console_write_failure_is_tolerated(self):
"""A console that cannot be written to must not fail the stop."""
procs = [self._proc(), self._proc(0, b"active\n"), self._proc(), self._proc(), self._proc()]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs):
with mock.patch("builtins.open", side_effect=OSError("No such device")):
result = await stop_desktop()
assert result["success"] is True
assert result["switched_to_console"] is True
@pytest.mark.asyncio
async def test_reports_failure_when_stranded_on_repaint_vt(self):
"""If the bounce cannot return to the console, say so rather than claim success."""
procs = [self._proc(), self._proc(0, b"active\n"), self._proc(), self._proc(), self._proc(1)]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs):
result = await stop_desktop()
assert result["success"] is True
assert result["switched_to_console"] is False
@pytest.mark.asyncio
async def test_does_not_switch_when_no_getty(self):
"""Without a getty on the console VT, switching would be just as blank — skip it."""
procs = [self._proc(), self._proc(3, b"inactive\n")]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs) as mock_exec:
result = await stop_desktop()
assert result["success"] is True
assert result["switched_to_console"] is False
assert len(mock_exec.call_args_list) == 2 # no chvt
@pytest.mark.asyncio
async def test_chvt_failure_does_not_fail_the_stop(self):
"""A missing or failing chvt is reported, but the desktop is still stopped."""
procs = [self._proc(), self._proc(0, b"active\n"), FileNotFoundError()]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs):
result = await stop_desktop()
assert result["success"] is True
assert result["switched_to_console"] is False
@pytest.mark.asyncio
async def test_no_switch_when_stop_failed(self):
"""A failed stop leaves the desktop up, so the display must not be switched."""
procs = [self._proc(5, b"", b"Unit not loaded.")]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=procs) as mock_exec:
result = await stop_desktop()
assert result["success"] is False
assert "switched_to_console" not in result
assert len(mock_exec.call_args_list) == 1
@pytest.mark.asyncio
async def test_start_never_switches_vt(self):
"""Starting the desktop must not touch the VT; the display manager claims one itself."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=self._proc()) as mock_exec:
result = await start_desktop()
assert result["success"] is True
assert "switched_to_console" not in result
assert len(mock_exec.call_args_list) == 1
class TestGetHotspotStatus:
"""Tests for get_hotspot_status function."""
def test_returns_hotspot_status_enum(self):
"""Should return a HotspotStatus enum value."""
result = get_hotspot_status()
assert isinstance(result, HotspotStatus)
def test_returns_unknown_in_docker_mode(self):
"""Should return unknown when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = get_hotspot_status()
assert result == HotspotStatus.unknown
def test_handles_nmcli_not_found(self):
"""Should handle nmcli not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_current_ssid", return_value=None):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = get_hotspot_status()
assert result == HotspotStatus.unknown
def test_handles_timeout(self):
"""Should handle subprocess timeout."""
import subprocess
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_current_ssid", return_value=None):
with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("nmcli", 5)):
result = get_hotspot_status()
assert result == HotspotStatus.unknown
class TestGetHotspotStatusWithRealOutput:
"""Tests for get_hotspot_status using real nmcli output from RPi (10.0.0.9)."""
# Real nmcli output when WiFi radio is off (wlan0 software-disabled).
NMCLI_RADIO_OFF = (
"eth0: connected to Wired connection 1\n"
"\t\"eth0\"\n"
"\tethernet (macb), 2C:CF:67:08:87:88, hw, mtu 1500\n"
"\n"
"lo: connected (externally) to lo\n"
"\t\"lo\"\n"
"\tloopback (unknown), 00:00:00:00:00:00, sw, mtu 65536\n"
"\n"
"wlan0: unavailable\n"
"\t\"Broadcom Wi-Fi\"\n"
"\twifi (brcmfmac), 2C:CF:67:08:87:89, sw disabled, hw, mtu 1500\n"
)
# Real nmcli output when WiFi radio is on but no hotspot (wlan0 disconnected).
NMCLI_RADIO_ON_NO_HOTSPOT = (
"eth0: connected to Wired connection 1\n"
"\t\"eth0\"\n"
"\tethernet (macb), 2C:CF:67:08:87:88, hw, mtu 1500\n"
"\n"
"lo: connected (externally) to lo\n"
"\t\"lo\"\n"
"\tloopback (unknown), 00:00:00:00:00:00, sw, mtu 65536\n"
"\n"
"wlan0: disconnected\n"
"\t\"Broadcom Wi-Fi\"\n"
"\twifi (brcmfmac), 2C:CF:67:08:87:89, hw, mtu 1500\n"
)
# Real nmcli output when hotspot is active (wlan0 connected to Hotspot).
NMCLI_HOTSPOT_ACTIVE = (
"wlan0: connected to Hotspot\n"
"\t\"Broadcom Wi-Fi\"\n"
"\twifi (brcmfmac), 2C:CF:67:08:87:89, hw, mtu 1500\n"
"\n"
"eth0: connected to Wired connection 1\n"
"\t\"eth0\"\n"
"\tethernet (macb), 2C:CF:67:08:87:88, hw, mtu 1500\n"
"\n"
"lo: connected (externally) to lo\n"
"\t\"lo\"\n"
"\tloopback (unknown), 00:00:00:00:00:00, sw, mtu 65536\n"
)
def test_radio_off_returns_off(self):
"""nmcli 'wlan0: unavailable' (radio sw disabled) should return HotspotStatus.off."""
mock_result = mock.Mock()
mock_result.returncode = 0
mock_result.stdout = self.NMCLI_RADIO_OFF
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_current_ssid", return_value=None):
with mock.patch("subprocess.run", return_value=mock_result):
result = get_hotspot_status()
assert result == HotspotStatus.off
def test_radio_on_no_hotspot_returns_disconnected(self):
"""nmcli 'wlan0: disconnected' (radio on, no hotspot) should return disconnected."""
mock_result = mock.Mock()
mock_result.returncode = 0
mock_result.stdout = self.NMCLI_RADIO_ON_NO_HOTSPOT
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_current_ssid", return_value=None):
with mock.patch("subprocess.run", return_value=mock_result):
result = get_hotspot_status()
assert result == HotspotStatus.disconnected
def test_hotspot_active_returns_connected(self):
"""nmcli 'wlan0: connected to Hotspot' should return HotspotStatus.connected."""
mock_result = mock.Mock()
mock_result.returncode = 0
mock_result.stdout = self.NMCLI_HOTSPOT_ACTIVE
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_current_ssid", return_value=None):
with mock.patch("subprocess.run", return_value=mock_result):
result = get_hotspot_status()
assert result == HotspotStatus.connected
class TestGetHotspotStatusDict:
"""Tests for get_hotspot_status_dict function."""
def test_available_when_radio_off(self):
"""Hotspot should be available when WiFi radio is off,
because start_hotspot() turns the radio on first."""
with mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.off):
result = get_hotspot_status_dict()
assert result["available"] is True
assert result["enabled"] is False
def test_unavailable_when_unknown(self):
"""Hotspot should be unavailable when status is unknown (Docker, no nmcli)."""
with mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.unknown):
result = get_hotspot_status_dict()
assert result["available"] is False
def test_enabled_when_connected(self):
"""Hotspot should be enabled and available when connected."""
with mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.connected):
result = get_hotspot_status_dict()
assert result["available"] is True
assert result["enabled"] is True
def test_ip_and_portal_url_when_connected(self, reset_runtime_config):
with mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.connected), \
mock.patch("controller.lib.captive_portal.get_hotspot_ip", return_value='10.42.0.1'), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False):
result = get_hotspot_status_dict()
assert result["ip"] == '10.42.0.1'
assert result["captive_portal"] is True
assert result["portal_url"] == 'http://10.42.0.1/portal'
def test_no_ip_when_off(self, reset_runtime_config):
with mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.off), \
mock.patch("controller.lib.captive_portal.get_hotspot_ip") as get_ip:
result = get_hotspot_status_dict()
assert result["ip"] is None
assert result["portal_url"] is None
get_ip.assert_not_called()
class TestHotspotSettingsMigration:
"""Legacy hotspot settings in wrolpi.yaml are copied into controller.yaml on startup."""
@staticmethod
def _write_wrolpi_yaml(test_config_directory, **values):
lines = ''.join(f'{key}: {value}\n' for key, value in values.items())
(test_config_directory / 'wrolpi.yaml').write_text(lines)
def test_migrates_wrolpi_yaml_values(self, test_directory, test_config_directory, mock_config_path,
reset_runtime_config):
"""wrolpi.yaml hotspot settings should be copied to controller.yaml."""
mock_config_path.write_text("{}\n")
self._write_wrolpi_yaml(test_config_directory, hotspot_device='wlp2s0', hotspot_ssid='RafaelPi',
hotspot_password='password123')
with mock.patch("controller.lib.admin.get_media_directory", return_value=test_directory):
migrate_hotspot_settings_from_wrolpi_config()
assert get_hotspot_device() == 'wlp2s0'
assert get_hotspot_ssid() == 'RafaelPi'
assert get_hotspot_password() == 'password123'
# The migrated values are persisted in controller.yaml.
saved = yaml.safe_load(mock_config_path.read_text())
assert saved['hotspot'] == {'device': 'wlp2s0', 'ssid': 'RafaelPi', 'password': 'password123'}
def test_does_not_overwrite_explicit_controller_values(self, test_directory, test_config_directory,
mock_config_path, reset_runtime_config):
"""Explicit controller.yaml values win over legacy wrolpi.yaml values."""
mock_config_path.write_text("hotspot:\n device: wlan1\n")
reload_config_from_drive()
self._write_wrolpi_yaml(test_config_directory, hotspot_device='wlp2s0', hotspot_ssid='RafaelPi')
with mock.patch("controller.lib.admin.get_media_directory", return_value=test_directory):
migrate_hotspot_settings_from_wrolpi_config()
# The explicit device is kept, but the SSID is still migrated.
assert get_hotspot_device() == 'wlan1'
assert get_hotspot_ssid() == 'RafaelPi'
def test_noop_without_wrolpi_yaml(self, test_directory, test_config_directory, mock_config_path,
reset_runtime_config):
"""Nothing happens when wrolpi.yaml does not exist."""
mock_config_path.write_text("{}\n")
with mock.patch("controller.lib.admin.get_media_directory", return_value=test_directory):
migrate_hotspot_settings_from_wrolpi_config()
assert get_hotspot_device() == 'wlan0'
assert yaml.safe_load(mock_config_path.read_text()) == {}
def test_noop_when_drive_not_mounted(self, test_directory, test_config_directory, mock_config_path,
reset_runtime_config):
"""Nothing happens when controller.yaml does not exist (drive not mounted)."""
self._write_wrolpi_yaml(test_config_directory, hotspot_device='wlp2s0')
with mock.patch("controller.lib.admin.get_media_directory", return_value=test_directory):
migrate_hotspot_settings_from_wrolpi_config()
assert get_hotspot_device() == 'wlan0'
class TestUpdateHotspotSettings:
"""Tests for update_hotspot_settings function."""
def test_updates_and_saves(self, mock_config_path, reset_runtime_config):
"""New settings should be applied to the runtime config and saved to controller.yaml."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = update_hotspot_settings(device='wlp2s0', ssid='RafaelPi', password='password123')
assert result["success"] is True
assert get_hotspot_device() == 'wlp2s0'
saved = yaml.safe_load(mock_config_path.read_text())
assert saved['hotspot'] == {'device': 'wlp2s0', 'ssid': 'RafaelPi', 'password': 'password123'}
def test_partial_update(self, mock_config_path, reset_runtime_config):
"""Only provided values are changed."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = update_hotspot_settings(device='wlp2s0')
assert result["success"] is True
assert result["device"] == 'wlp2s0'
assert result["ssid"] == 'WROLPi'
assert result["password"] == 'wrolpi hotspot'
def test_updates_protocol(self, mock_config_path, reset_runtime_config):
"""The protocol is saved to controller.yaml and returned."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = update_hotspot_settings(protocol='wpa3')
assert result["success"] is True
assert result["protocol"] == 'wpa3'
assert get_hotspot_protocol() == 'wpa3'
saved = yaml.safe_load(mock_config_path.read_text())
assert saved['hotspot']['protocol'] == 'wpa3'
def test_updates_captive_portal(self, mock_config_path, reset_runtime_config):
"""The captive portal toggle is saved to controller.yaml and returned."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False):
result = update_hotspot_settings(captive_portal=False)
assert result["success"] is True
assert result["captive_portal"] is False
saved = yaml.safe_load(mock_config_path.read_text())
assert saved['hotspot']['captive_portal'] is False
def test_restart_required_only_while_hotspot_up(self, mock_config_path, reset_runtime_config):
"""A change applies on the next hotspot start, so say so when the hotspot is running."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.connected):
assert update_hotspot_settings(captive_portal=False)["restart_required"] is True
# Nothing changed: nothing to apply.
assert update_hotspot_settings()["restart_required"] is False
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.admin.get_hotspot_status", return_value=HotspotStatus.off):
assert update_hotspot_settings(captive_portal=True)["restart_required"] is False
def test_rejects_invalid_protocol(self, mock_config_path, reset_runtime_config):
"""Only wpa2 and wpa3 are valid protocols."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = update_hotspot_settings(protocol='wep')
assert result["success"] is False
assert "protocol" in result["error"].lower()
assert get_hotspot_protocol() == 'wpa2'
def test_rejects_short_password(self, mock_config_path, reset_runtime_config):
"""nmcli requires a WPA password of 8 to 63 characters."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = update_hotspot_settings(password='short')
assert result["success"] is False
assert "8" in result["error"]
assert get_hotspot_password() == 'wrolpi hotspot'
def test_rejects_in_docker_mode(self, reset_runtime_config):
"""Hotspot settings cannot be changed in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = update_hotspot_settings(device='wlp2s0')
assert result["success"] is False
assert "Docker" in result["error"]
def test_noop_does_not_save(self, tmp_path, reset_runtime_config):
"""An empty update succeeds without writing controller.yaml."""
missing_path = tmp_path / 'missing' / 'controller.yaml'
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
# save_config() would raise RuntimeError here; success proves it was skipped.
with mock.patch("controller.lib.config.CONFIG_PATH_ON_DRIVE", missing_path):
result = update_hotspot_settings()
assert result["success"] is True
assert not missing_path.exists()
def test_error_when_drive_not_mounted(self, tmp_path, reset_runtime_config):
"""Saving fails cleanly when the primary drive is not mounted."""
missing_path = tmp_path / 'missing' / 'controller.yaml'
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.config.CONFIG_PATH_ON_DRIVE", missing_path):
result = update_hotspot_settings(device='wlp2s0')
assert result["success"] is False
assert "not mounted" in result["error"]
# `iw dev wlan0 info` output; only the "wiphy" line matters.
IW_DEV_INFO = """Interface wlan0
\tifindex 3
\twdev 0x1
\taddr d8:3a:dd:11:22:33
\ttype managed
\twiphy 0
\ttxpower 31.00 dBm
"""
# Raspberry Pi brcmfmac (verified on real Pi 4/Pi 5 hardware): AP mode, but no
# AP/VLAN (fullmac driver) and no AP-mode SAE, so WPA3 hotspot is not possible.
# "SAE with AUTHENTICATE command" is station-mode WPA3 (joining, not hosting).
IW_PHY_BRCMFMAC = """Wiphy phy0
\twiphy index: 0
\tmax # scan SSIDs: 10
\tSupported Ciphers:
\t\t* WEP40 (00-0f-ac:1)
\t\t* WEP104 (00-0f-ac:5)
\t\t* TKIP (00-0f-ac:2)
\t\t* CCMP-128 (00-0f-ac:4)
\t\t* CMAC (00-0f-ac:6)
\tAvailable Antennas: TX 0 RX 0
\tSupported interface modes:
\t\t * IBSS
\t\t * managed
\t\t * AP
\t\t * P2P-client
\t\t * P2P-GO
\t\t * P2P-device
\tDevice supports SAE with AUTHENTICATE command
\tSupported extended features:
\t\t* [ CQM_RSSI_LIST ]: multiple CQM_RSSI_THOLD records
\t\t* [ DFS_OFFLOAD ]: DFS offload
"""
# mac80211 driver (Intel/Atheros style): AP/VLAN plus the CMAC cipher means
# wpa_supplicant can run SAE in software for the AP.
IW_PHY_MAC80211 = """Wiphy phy0
\tmax # scan SSIDs: 4
\tSupported Ciphers:
\t\t* WEP40 (00-0f-ac:1)
\t\t* TKIP (00-0f-ac:2)
\t\t* CCMP-128 (00-0f-ac:4)
\t\t* CMAC (00-0f-ac:6)
\tSupported interface modes:
\t\t * IBSS
\t\t * managed
\t\t * AP
\t\t * AP/VLAN
\t\t * monitor
\t\t * mesh point
\tSupported extended features:
\t\t* [ RRM ]: RRM
\t\t* [ FILS_STA ]: FILS shared key authentication support
"""
# Fullmac driver with AP-mode SAE offload advertised by the firmware.
# Older iw releases print "Extended features:" instead of "Supported extended features:".
IW_PHY_SAE_OFFLOAD_AP = """Wiphy phy0
\tSupported Ciphers:
\t\t* CCMP-128 (00-0f-ac:4)
\tSupported interface modes:
\t\t * managed
\t\t * AP
\tExtended features:
\t\t* [ SAE_OFFLOAD ]: SAE offload support
\t\t* [ SAE_OFFLOAD_AP ]: SAE offload support (AP mode)
"""
# Station-only hardware which cannot host a hotspot at all.
IW_PHY_NO_AP = """Wiphy phy0
\tSupported Ciphers:
\t\t* CCMP-128 (00-0f-ac:4)
\tSupported interface modes:
\t\t * managed
\t\t * monitor
"""
class TestGetDeviceHotspotProtocols:
"""Tests for get_device_hotspot_protocols; WPA3 requires SAE support in AP mode."""
@staticmethod
def _protocols(phy_info: str) -> list:
dev_info = mock.Mock(returncode=0, stdout=IW_DEV_INFO, stderr="")
phy = mock.Mock(returncode=0, stdout=phy_info, stderr="")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=[dev_info, phy]) as mock_run:
result = get_device_hotspot_protocols("wlan0")
# The wiphy index from `iw dev` is used to query the phy.
assert mock_run.call_args_list[1][0][0] == ["/usr/sbin/iw", "phy", "phy0", "info"]
return result
def test_rpi_brcmfmac_is_wpa2_only(self):
"""The Raspberry Pi's fullmac driver cannot host a WPA3 (SAE) AP."""
assert self._protocols(IW_PHY_BRCMFMAC) == ["wpa2"]
def test_mac80211_driver_supports_wpa3(self):
"""mac80211 drivers (AP/VLAN listed) with the CMAC cipher can run software SAE."""
assert self._protocols(IW_PHY_MAC80211) == ["wpa2", "wpa3"]
def test_sae_offload_ap_supports_wpa3(self):
"""Firmware advertising SAE_OFFLOAD_AP can host a WPA3 AP."""
assert self._protocols(IW_PHY_SAE_OFFLOAD_AP) == ["wpa2", "wpa3"]
def test_station_only_hardware_has_no_protocols(self):
"""A device without AP mode cannot host a hotspot at all."""
assert self._protocols(IW_PHY_NO_AP) == []
def test_returns_empty_in_docker_mode(self):
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
assert get_device_hotspot_protocols("wlan0") == []
def test_unknown_device_has_no_protocols(self):
"""`iw dev` fails for a device which does not exist."""
dev_info = mock.Mock(returncode=237, stdout="", stderr="command failed: No such device")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=[dev_info]):
assert get_device_hotspot_protocols("wlan9") == []
def test_assumes_wpa2_when_iw_missing(self):
"""Without `iw` we cannot probe; assume the WPA2 hotspot nmcli has always created."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
assert get_device_hotspot_protocols("wlan0") == ["wpa2"]
class TestGetWifiDevices:
"""Tests for get_wifi_devices function."""
# Real terse nmcli output from an x86 PC: `nmcli -t -f DEVICE,TYPE device`
NMCLI_TERSE_DEVICES = (
"eno1:ethernet\n"
"wlp2s0:wifi\n"
"p2p-dev-wlp2s0:wifi-p2p\n"
"lo:loopback\n"
)
def test_returns_wifi_devices_only(self):
"""Should return only wifi-type devices, excluding wifi-p2p and ethernet."""
mock_result = mock.Mock(returncode=0, stdout=self.NMCLI_TERSE_DEVICES)
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", return_value=mock_result):
assert get_wifi_devices() == ["wlp2s0"]
def test_returns_empty_in_docker_mode(self):
"""Should return an empty list in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
assert get_wifi_devices() == []
def test_returns_empty_when_nmcli_missing(self):
"""Should return an empty list when nmcli is not installed."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
assert get_wifi_devices() == []
def test_returns_empty_when_nmcli_fails(self):
"""Should return an empty list when nmcli returns an error."""
mock_result = mock.Mock(returncode=8, stdout="")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", return_value=mock_result):
assert get_wifi_devices() == []
class TestDockerModeErrors:
"""Admin functions should return an error result when running in Docker mode."""
@pytest.mark.parametrize("func,is_async,args", [
(unblock_bluetooth, True, ()),
(block_bluetooth, True, ()),
(start_hotspot, False, ()),
(stop_hotspot, False, ()),
(start_desktop, True, ()),
(stop_desktop, True, ()),
(start_vnc, True, ()),
(stop_vnc, True, ()),
(enable_throttle, False, ()),
(disable_throttle, False, ()),
(shutdown_system, False, ()),
(reboot_system, False, ()),
(restart_all_services, True, ()),
(set_timezone, False, ("America/Denver",)),
])
@pytest.mark.asyncio
async def test_returns_error_in_docker_mode(self, func, is_async, args):
"""In Docker mode each function should refuse with a Docker-mentioning error."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = await func(*args) if is_async else func(*args)
assert result["success"] is False
# set_timezone uses "error" string verbatim; others use the same shape.
error = result.get("error", "")
assert "Docker" in error
HOTSPOT_IP_JSON = '[{"ifname": "wlan0", "addr_info": [{"family": "inet", "local": "10.42.0.1"}]}]'
class TestStartHotspot:
"""Tests for start_hotspot function."""
def test_handles_nmcli_not_found(self):
"""Should handle nmcli not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = start_hotspot()
assert result["success"] is False
assert "nmcli" in result.get("error", "").lower()
def test_uses_config_values(self):
"""Should use get_config_value for device, ssid, and password."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
def mock_get_config(key, default=None):
config_map = {
'hotspot.device': 'wlan1',
'hotspot.ssid': 'TestSSID',
'hotspot.password': 'testpassword123',
}
return config_map.get(key, default)
with mock.patch("controller.lib.admin.get_config_value", side_effect=mock_get_config):
radio_on = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan1:disconnected\n", stderr="")
ok = mock.Mock(returncode=0, stdout="", stderr="")
# The final call reads the hotspot's address for the captive portal.
ip_addr = mock.Mock(returncode=0, stdout=HOTSPOT_IP_JSON, stderr="")
with mock.patch("subprocess.run",
side_effect=[radio_on, device_ready, ok, ok, ok, ip_addr]) as mock_subprocess:
with mock.patch("time.sleep"):
start_hotspot()
# Find the hotspot creation call (nmcli device wifi hotspot)
calls = mock_subprocess.call_args_list
hotspot_call = [c for c in calls if 'hotspot' in c[0][0]]
assert hotspot_call, "Hotspot creation call not found"
cmd_args = hotspot_call[0][0][0]
assert "wlan1" in cmd_args
assert "TestSSID" in cmd_args
assert "testpassword123" in cmd_args
@staticmethod
def _start_with_nmcli(ip_json=HOTSPOT_IP_JSON):
"""Run start_hotspot with nmcli mocked; returns the subprocess mock and the include path."""
radio_on = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
ok = mock.Mock(returncode=0, stdout="", stderr="")
ip_addr = mock.Mock(returncode=0, stdout=ip_json, stderr="")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False), \
mock.patch("subprocess.run", side_effect=[radio_on, device_ready, ok, ok, ok, ip_addr, ok]) as run, \
mock.patch("time.sleep"):
result = start_hotspot()
return result, run
def test_writes_captive_portal_config_before_creating_hotspot(self, reset_runtime_config):
"""dnsmasq reads the include at activation, so it must exist before `nmcli ... hotspot`."""
include = captive_portal.INCLUDE_FILE
seen = {}
results = iter([
mock.Mock(returncode=0, stdout="", stderr=""), # radio on
mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr=""), # device ready
mock.Mock(returncode=0, stdout="", stderr=""), # nmcli device wifi hotspot
mock.Mock(returncode=0, stdout="", stderr=""), # modify
mock.Mock(returncode=0, stdout="", stderr=""), # up
mock.Mock(returncode=0, stdout=HOTSPOT_IP_JSON, stderr=""), # ip addr
])
def run(cmd, **kwargs):
if 'hotspot' in cmd:
# Observe the include at the moment NetworkManager (and its dnsmasq) start.
seen['include_at_create'] = include.read_text() if include.exists() else None
return next(results)
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False), \
mock.patch("subprocess.run", side_effect=run), \
mock.patch("time.sleep"):
result = start_hotspot()
assert result["success"] is True
assert seen['include_at_create'] is not None
assert 'address=/captive.apple.com/10.42.0.1' in seen['include_at_create']
def test_disabled_removes_captive_portal_config(self, reset_runtime_config):
include = captive_portal.INCLUDE_FILE
include.write_text('address=/captive.apple.com/10.42.0.1\n')
with mock.patch("controller.lib.captive_portal.get_config_value", return_value=False):
result, run = self._start_with_nmcli()
assert result["success"] is True
assert not include.exists()
# No `ip` read and no extra re-activation when the portal is off.
assert not [c for c in run.call_args_list if c[0][0][0] == 'ip']
def test_rewrites_config_when_networkmanager_picked_another_subnet(self, reset_runtime_config):
"""10.42.1.1 is used when another shared connection holds 10.42.0.1."""
other = '[{"ifname": "wlan0", "addr_info": [{"family": "inet", "local": "10.42.1.1"}]}]'
result, run = self._start_with_nmcli(ip_json=other)
assert result["success"] is True
assert 'address=/captive.apple.com/10.42.1.1' in captive_portal.INCLUDE_FILE.read_text()
# Re-activated once more so dnsmasq re-reads the corrected include.
ups = [c for c in run.call_args_list if c[0][0][:3] == ['nmcli', 'connection', 'up']]
assert len(ups) == 2
def test_reactivation_failure_is_logged(self, reset_runtime_config, caplog):
"""dnsmasq keeps the old address if the corrective `up` fails; say so."""
other = '[{"ifname": "wlan0", "addr_info": [{"family": "inet", "local": "10.42.1.1"}]}]'
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
ip_addr = mock.Mock(returncode=0, stdout=other, stderr="")
failed_up = mock.Mock(returncode=4, stdout="", stderr="Error: Connection activation failed")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False), \
mock.patch("subprocess.run", side_effect=[ok, device_ready, ok, ok, ok, ip_addr, failed_up]), \
mock.patch("time.sleep"):
result = start_hotspot()
assert result["success"] is True
assert "Could not re-activate hotspot for captive portal" in caplog.text
assert "Connection activation failed" in caplog.text
def test_wpa3_reverted_hotspot_still_gets_captive_portal_ip(self, reset_runtime_config):
"""A WPA3 failure that reverts to WPA2 leaves the hotspot up; it still needs the right address."""
other = '[{"ifname": "wlan0", "addr_info": [{"family": "inet", "local": "10.42.1.1"}]}]'
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
sae_failed = mock.Mock(returncode=1, stdout="", stderr="Error: sae unsupported")
ip_addr = mock.Mock(returncode=0, stdout=other, stderr="")
def mock_get_config(key, default=None):
return {'hotspot.protocol': 'wpa3'}.get(key, default)
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("controller.lib.captive_portal.is_docker_mode", return_value=False), \
mock.patch("controller.lib.admin.get_config_value", side_effect=mock_get_config), \
mock.patch("subprocess.run", side_effect=[
ok, device_ready, ok, # radio, ready, hotspot
sae_failed, # modify to sae fails
ok, ok, # revert modify, revert up
ip_addr, ok, # captive portal: read address, re-activate
]) as run, \
mock.patch("time.sleep"):
result = start_hotspot()
assert result["success"] is False
assert "reverted to WPA2" in result["error"]
assert 'address=/captive.apple.com/10.42.1.1' in captive_portal.INCLUDE_FILE.read_text()
assert run.call_args_list[-1][0][0] == ["nmcli", "connection", "up", "Hotspot"]
def test_stop_forgets_acknowledged_clients(self):
"""A later joiner reusing the lease must see the portal page, not a silent success."""
captive_portal.acknowledge_client('10.42.0.210')
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False), \
mock.patch("subprocess.run", return_value=mock.Mock(returncode=0, stdout="", stderr="")):
assert stop_hotspot()["success"] is True
assert not captive_portal.is_client_acknowledged('10.42.0.210')
def test_start_forgets_acknowledged_clients(self, reset_runtime_config):
"""A new hotspot session shows every device the portal page again."""
captive_portal.acknowledge_client('10.42.0.210')
result, _ = self._start_with_nmcli()
assert result["success"] is True
assert not captive_portal.is_client_acknowledged('10.42.0.210')
def test_no_reactivation_when_ip_matches(self, reset_runtime_config):
result, run = self._start_with_nmcli()
ups = [c for c in run.call_args_list if c[0][0][:3] == ['nmcli', 'connection', 'up']]
assert len(ups) == 1
@staticmethod
def _run_enable(protocol: str, results: list) -> tuple:
"""Run start_hotspot() with the given hotspot.protocol and subprocess results."""
def mock_get_config(key, default=None):
return {'hotspot.protocol': protocol}.get(key, default)
# After a successful start the hotspot's address is read once for the captive portal.
ip_addr = mock.Mock(returncode=0, stdout=HOTSPOT_IP_JSON, stderr="")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_config_value", side_effect=mock_get_config):
with mock.patch("subprocess.run", side_effect=results + [ip_addr]) as mock_subprocess:
with mock.patch("time.sleep"):
result = start_hotspot()
return result, mock_subprocess.call_args_list
def test_wpa2_sets_psk_key_mgmt(self):
"""The Hotspot connection is explicitly set to WPA2 so a reused profile
does not keep a previously configured protocol."""
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
result, calls = self._run_enable('wpa2', [ok, device_ready, ok, ok, ok])
assert result["success"] is True
modify_call = [c[0][0] for c in calls if 'modify' in c[0][0]]
assert modify_call and "wpa-psk" in modify_call[0]
def test_wpa3_sets_sae_key_mgmt(self):
"""WPA3 modifies the Hotspot connection to SAE with required PMF, then re-activates."""
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
result, calls = self._run_enable('wpa3', [ok, device_ready, ok, ok, ok])
assert result["success"] is True
modify_call = [c[0][0] for c in calls if 'modify' in c[0][0]]
assert modify_call and "sae" in modify_call[0]
assert "802-11-wireless-security.pmf" in modify_call[0]
up_call = [c[0][0] for c in calls if 'up' in c[0][0]]
assert up_call, "Hotspot connection was not re-activated"
def test_wpa3_failure_reverts_to_wpa2(self):
"""If the driver rejects WPA3, the hotspot is reverted to WPA2 and an error returned."""
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
sae_failed = mock.Mock(returncode=1, stdout="", stderr="Error: SAE not supported")
# radio, poll, hotspot, modify(sae) fails, revert modify, revert up.
result, calls = self._run_enable('wpa3', [ok, device_ready, ok, sae_failed, ok, ok])
assert result["success"] is False
assert "wpa3" in result["error"].lower()
modify_calls = [c[0][0] for c in calls if 'modify' in c[0][0]]
assert len(modify_calls) == 2 and "wpa-psk" in modify_calls[1]
def test_wpa3_failed_revert_is_reported(self):
"""If reverting to WPA2 also fails, the error must not claim the hotspot was reverted."""
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
sae_failed = mock.Mock(returncode=1, stdout="", stderr="Error: SAE not supported")
revert_failed = mock.Mock(returncode=1, stdout="", stderr="Error: connection not found")
# radio, poll, hotspot, modify(sae) fails, revert modify fails.
result, _ = self._run_enable('wpa3', [ok, device_ready, ok, sae_failed, revert_failed])
assert result["success"] is False
assert "reverted" not in result["error"].lower()
assert "wpa3" in result["error"].lower()
def test_wpa2_modify_failure_still_succeeds(self):
"""A failure setting WPA2 explicitly is not fatal; nmcli already created a WPA2 hotspot."""
ok = mock.Mock(returncode=0, stdout="", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
failed = mock.Mock(returncode=1, stdout="", stderr="Error: unknown property")
result, _ = self._run_enable('wpa2', [ok, device_ready, ok, failed])
assert result["success"] is True
def test_waits_for_device_ready_after_radio_on(self):
"""Should poll device status after turning radio on, waiting for device
to transition from 'unavailable' before creating hotspot.
Real RPi behavior: radio on takes ~2s for device to become 'disconnected'."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
# Simulate: radio on succeeds, device initially unavailable, then disconnected, then hotspot succeeds
radio_on_result = mock.Mock(returncode=0, stdout="", stderr="")
device_unavailable = mock.Mock(returncode=0, stdout="wlan0:unavailable\n", stderr="")
device_ready = mock.Mock(returncode=0, stdout="wlan0:disconnected\n", stderr="")
hotspot_result = mock.Mock(returncode=0, stdout="", stderr="")
with mock.patch("subprocess.run", side_effect=[
radio_on_result, # nmcli radio wifi on
device_unavailable, # nmcli -t device status (not ready)
device_ready, # nmcli -t device status (ready)
hotspot_result, # nmcli device wifi hotspot
hotspot_result, # nmcli connection modify Hotspot
hotspot_result, # nmcli connection up Hotspot
mock.Mock(returncode=0, stdout=HOTSPOT_IP_JSON, stderr=""), # ip -4 -j addr show
]):
with mock.patch("time.sleep"): # Don't actually sleep in tests
result = start_hotspot()
assert result["success"] is True
def test_fails_if_device_never_becomes_ready(self):
"""Should fail with a clear error if device stays unavailable after radio on."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
radio_on_result = mock.Mock(returncode=0, stdout="", stderr="")
device_unavailable = mock.Mock(returncode=0, stdout="wlan0:unavailable\n", stderr="")
# Device never becomes ready (all polls return unavailable)
with mock.patch("subprocess.run", side_effect=[
radio_on_result,
] + [device_unavailable] * 20):
with mock.patch("time.sleep"):
result = start_hotspot()
assert result["success"] is False
assert "not ready" in result.get("error", "").lower() or "unavailable" in result.get("error", "").lower()
class TestStopHotspot:
"""Tests for stop_hotspot function."""
def test_handles_nmcli_not_found(self):
"""Should handle nmcli not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = stop_hotspot()
assert result["success"] is False
assert "nmcli" in result.get("error", "").lower()
class TestGetThrottleStatus:
"""Tests for get_throttle_status function."""
def test_returns_governor_status_enum(self):
"""Should return a GovernorStatus enum value."""
result = get_throttle_status()
assert isinstance(result, GovernorStatus)
def test_returns_unknown_in_docker_mode(self):
"""Should return unknown when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = get_throttle_status()
assert result == GovernorStatus.unknown
def test_returns_unknown_when_cpufreq_not_available(self, tmp_path):
"""Should return unknown when cpufreq doesn't exist."""
from pathlib import Path
nonexistent_path = tmp_path / "nonexistent"
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch.object(Path, "exists", return_value=False):
result = get_throttle_status()
assert result == GovernorStatus.unknown
def test_reads_governor_from_sysfs(self, tmp_path):
"""Should read governor from sysfs when available."""
from pathlib import Path
governor_path = tmp_path / "scaling_governor"
governor_path.write_text("ondemand")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
# This test verifies the function structure - actual sysfs testing
# would require a real Raspberry Pi environment
result = get_throttle_status()
assert isinstance(result, GovernorStatus)
class TestShutdownSystem:
"""Tests for shutdown_system function."""
def test_handles_shutdown_not_found(self):
"""Should handle shutdown command not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("shutil.which", return_value=None):
result = shutdown_system()
assert result["success"] is False
assert "not found" in result.get("error", "").lower()
def test_calls_shutdown_command(self):
"""Should call shutdown command with correct args after delay."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("shutil.which", return_value="/sbin/shutdown"):
mock_popen = mock.Mock()
with mock.patch("subprocess.Popen", mock_popen):
# Use delay=0 for testing to avoid waiting
result = shutdown_system(delay=0)
assert result["success"] is True
assert "shutting down" in result["message"].lower()
# Give thread time to execute
import time
time.sleep(0.1)
mock_popen.assert_called_once_with(["shutdown", "-h", "now"])
class TestRebootSystem:
"""Tests for reboot_system function."""
def test_handles_reboot_not_found(self):
"""Should handle reboot command not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("shutil.which", return_value=None):
result = reboot_system()
assert result["success"] is False
assert "not found" in result.get("error", "").lower()
def test_calls_reboot_command(self):
"""Should call reboot command after delay."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("shutil.which", return_value="/sbin/reboot"):
mock_popen = mock.Mock()
with mock.patch("subprocess.Popen", mock_popen):
# Use delay=0 for testing to avoid waiting
result = reboot_system(delay=0)
assert result["success"] is True
assert "rebooting" in result["message"].lower()
# Give thread time to execute
import time
time.sleep(0.1)
mock_popen.assert_called_once_with(["reboot"])
class TestRestartAllServices:
"""Tests for restart_all_services function."""
@pytest.mark.asyncio
async def test_restarts_services(self):
"""Should attempt to restart all services."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
# Mock subprocess.run to succeed
mock_result = mock.Mock()
mock_result.returncode = 0
mock_result.stderr = ""
# Mock _restart_self to avoid actual restart
with mock.patch("subprocess.run", return_value=mock_result):
with mock.patch("controller.lib.admin._restart_self"):
with mock.patch("asyncio.get_event_loop") as mock_loop:
mock_loop.return_value.call_later = mock.Mock()
result = await restart_all_services()
assert result["success"] is True
assert "services" in result
# Should include controller as pending
assert result["services"]["wrolpi-controller"]["pending"] is True
class TestGetTimezoneStatus:
"""Tests for get_timezone_status_dict function."""
def test_returns_unavailable_in_docker_mode(self):
"""Should return unavailable when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
result = get_timezone_status_dict()
assert result["available"] is False
assert result["timezone"] is None
assert "Docker" in result["reason"]
def test_returns_timezone_from_timedatectl(self):
"""Should return the current system timezone."""
mock_result = mock.Mock()
mock_result.returncode = 0
mock_result.stdout = "America/Denver\n"
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", return_value=mock_result):
result = get_timezone_status_dict()
assert result["available"] is True
assert result["timezone"] == "America/Denver"
assert result["reason"] is None
def test_handles_timedatectl_not_found(self):
"""Should handle timedatectl not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = get_timezone_status_dict()
assert result["available"] is False
assert "not found" in result["reason"]
def test_handles_timeout(self):
"""Should handle subprocess timeout."""
import subprocess
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("timedatectl", 5)):
result = get_timezone_status_dict()
assert result["available"] is False
assert "timed out" in result["reason"]
class TestSetTimezone:
"""Tests for set_timezone function."""
def test_sets_timezone_successfully(self):
"""Should set timezone via timedatectl."""
mock_result = mock.Mock()
mock_result.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", return_value=mock_result) as mock_run:
result = set_timezone("America/Denver")
assert result["success"] is True
assert result["timezone"] == "America/Denver"
mock_run.assert_called_once_with(
["timedatectl", "set-timezone", "America/Denver"],
capture_output=True,
text=True,
timeout=10,
)
def test_handles_invalid_timezone(self):
"""Should handle timedatectl rejecting an invalid timezone."""
mock_result = mock.Mock()
mock_result.returncode = 1
mock_result.stderr = "Failed to set time zone: Invalid time zone"
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", return_value=mock_result):
result = set_timezone("Invalid/Timezone")
assert result["success"] is False
assert "Invalid" in result["error"]
def test_handles_empty_timezone(self):
"""Should reject empty timezone strings."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
result = set_timezone("")
assert result["success"] is False
assert "empty" in result["error"].lower()
def test_handles_timedatectl_not_found(self):
"""Should handle timedatectl not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = set_timezone("America/Denver")
assert result["success"] is False
assert "not found" in result["error"]
class TestApplyTimezoneFromConfig:
"""Tests for apply_timezone_from_config function."""
def test_skips_in_docker_mode(self):
"""Should do nothing in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
with mock.patch("controller.lib.admin.set_timezone") as mock_set:
apply_timezone_from_config()
mock_set.assert_not_called()
def test_applies_timezone_from_config(self, tmp_path):
"""Should read timezone from wrolpi.yaml and apply it."""
config_dir = tmp_path / 'config'
config_dir.mkdir()
config_file = config_dir / 'wrolpi.yaml'
config_file.write_text("timezone: America/New_York\n")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_media_directory", return_value=tmp_path):
with mock.patch("controller.lib.admin.set_timezone", return_value={"success": True}) as mock_set:
apply_timezone_from_config()
mock_set.assert_called_once_with("America/New_York")
def test_skips_when_no_timezone_in_config(self, tmp_path):
"""Should do nothing when timezone is not set in config."""
config_dir = tmp_path / 'config'
config_dir.mkdir()
config_file = config_dir / 'wrolpi.yaml'
config_file.write_text("download_wait: 60\n")
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_media_directory", return_value=tmp_path):
with mock.patch("controller.lib.admin.set_timezone") as mock_set:
apply_timezone_from_config()
mock_set.assert_not_called()
def test_skips_when_config_file_missing(self, tmp_path):
"""Should do nothing when wrolpi.yaml doesn't exist."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_media_directory", return_value=tmp_path):
with mock.patch("controller.lib.admin.set_timezone") as mock_set:
apply_timezone_from_config()
mock_set.assert_not_called()
def _show_proc(output: str, returncode: int = 0) -> mock.AsyncMock:
"""A mocked `systemctl show` process printing the given properties."""
proc = mock.AsyncMock()
proc.communicate.return_value = (output.encode(), b"")
proc.returncode = returncode
return proc
def _ok_proc() -> mock.AsyncMock:
proc = mock.AsyncMock()
proc.communicate.return_value = (b"", b"")
proc.returncode = 0
return proc
def _multi_show_proc(*blocks: str, returncode: int = 0) -> mock.AsyncMock:
"""A mocked multi-unit `systemctl show`: one blank-line-separated block per unit."""
proc = mock.AsyncMock()
proc.communicate.return_value = ("\n\n".join(blocks).encode(), b"")
proc.returncode = returncode
return proc
def _backends_proc(wayvnc="loaded", wayvnc_active="inactive",
realvnc="loaded", realvnc_active="inactive") -> mock.AsyncMock:
return _multi_show_proc(
f"LoadState={wayvnc}\nActiveState={wayvnc_active}",
f"LoadState={realvnc}\nActiveState={realvnc_active}",
)
class TestGetVncBackends:
"""Both stacks can be installed, and either can be running."""
@pytest.mark.asyncio
async def test_probes_both_units_in_one_call(self):
"""One `systemctl show` for both units; these probes run on every /api/stats."""
with mock.patch("asyncio.create_subprocess_exec", return_value=_backends_proc()) as mock_exec:
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = True
backends, reason = await get_vnc_backends()
mock_exec.assert_called_once()
assert "wayvnc.service" in mock_exec.call_args.args
assert "vncserver-x11-serviced.service" in mock_exec.call_args.args
assert [b["name"] for b in backends] == ["wayvnc", "realvnc"]
assert reason is None
@pytest.mark.asyncio
async def test_wayvnc_is_preferred_for_starting(self):
"""wayvnc comes first, so it is what start_vnc() uses."""
with mock.patch("asyncio.create_subprocess_exec", return_value=_backends_proc()):
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = True
backends, _ = await get_vnc_backends()
assert backends[0]["unit"] == "wayvnc.service"
@pytest.mark.asyncio
async def test_reports_each_backend_active_state(self):
"""A running RealVNC must be reported even though wayvnc is preferred."""
proc = _backends_proc(realvnc_active="active")
with mock.patch("asyncio.create_subprocess_exec", return_value=proc):
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = True
backends, _ = await get_vnc_backends()
assert backends[0] == {"unit": "wayvnc.service", "name": "wayvnc", "active": False}
assert backends[1] == {"unit": "vncserver-x11-serviced.service", "name": "realvnc", "active": True}
@pytest.mark.asyncio
async def test_skips_unconfigured_wayvnc(self):
"""An unconfigured wayvnc is not usable; RealVNC is."""
with mock.patch("asyncio.create_subprocess_exec", return_value=_backends_proc()):
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = False
backends, reason = await get_vnc_backends()
assert [b["name"] for b in backends] == ["realvnc"]
assert reason is None
@pytest.mark.asyncio
async def test_reports_unconfigured_wayvnc_when_alone(self):
"""wayvnc installed but unconfigured, with no RealVNC, explains itself."""
proc = _backends_proc(realvnc="not-found")
with mock.patch("asyncio.create_subprocess_exec", return_value=proc):
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = False
backends, reason = await get_vnc_backends()
assert backends == []
assert "not configured" in reason
@pytest.mark.asyncio
async def test_reports_nothing_installed(self):
"""Neither backend installed is reported plainly."""
proc = _backends_proc(wayvnc="not-found", realvnc="not-found")
with mock.patch("asyncio.create_subprocess_exec", return_value=proc):
backends, reason = await get_vnc_backends()
assert backends == []
assert reason == "No VNC server installed"
@pytest.mark.asyncio
async def test_survives_missing_systemctl(self):
"""A host without systemd simply has no VNC backend."""
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
backends, reason = await get_vnc_backends()
assert backends == []
assert reason
@pytest.mark.asyncio
async def test_does_not_probe_the_compositor(self):
"""Backend choice must not depend on a running compositor.
raspi-config uses `pgrep labwc` to decide Wayland-vs-X11, which reports X11
whenever the desktop is stopped and would pick the wrong backend.
"""
with mock.patch("asyncio.create_subprocess_exec", return_value=_backends_proc()) as mock_exec:
with mock.patch("controller.lib.admin.WAYVNC_CONFIG") as config:
config.is_file.return_value = True
await get_vnc_backends()
for call in mock_exec.call_args_list:
assert "pgrep" not in call.args[0]
class TestGetVncStatus:
"""Tests for get_vnc_status."""
@pytest.mark.asyncio
async def test_returns_unknown_in_docker_mode(self):
"""Should return unknown when in Docker mode."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
assert await get_vnc_status() == VncStatus.unknown
@pytest.mark.asyncio
@pytest.mark.parametrize("backends,expected", [
([{"unit": "u", "name": "wayvnc", "active": True}], VncStatus.on),
([{"unit": "u", "name": "wayvnc", "active": False}], VncStatus.off),
# A running RealVNC counts even though wayvnc is preferred and stopped.
([{"unit": "a", "name": "wayvnc", "active": False},
{"unit": "b", "name": "realvnc", "active": True}], VncStatus.on),
([], VncStatus.unavailable),
])
async def test_on_when_any_backend_runs(self, backends, expected):
"""Any running backend means VNC is on."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
assert await get_vnc_status() == expected
class TestGetVncStatusDict:
"""The status dict gates starting on the desktop being up."""
@staticmethod
def _backends(wayvnc_active=False, realvnc_active=None):
backends = [{"unit": "wayvnc.service", "name": "wayvnc", "active": wayvnc_active}]
if realvnc_active is not None:
backends.append({"unit": "realvnc.service", "name": "realvnc", "active": realvnc_active})
return backends
@pytest.mark.asyncio
@pytest.mark.parametrize("vnc_active,desktop_up,running,can_start", [
(True, True, True, True),
(False, True, False, True),
(False, False, False, False), # desktop stopped: cannot start
(True, False, True, False), # running, but could not be started now
])
async def test_flags(self, vnc_active, desktop_up, running, can_start):
"""running/available/can_start reflect VNC and desktop state."""
desktop = DesktopStatus.on if desktop_up else DesktopStatus.off
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=(self._backends(wayvnc_active=vnc_active), None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=desktop):
result = await get_vnc_status_dict()
assert result["running"] is running
assert result["available"] is True
assert result["can_start"] is can_start
assert result["desktop_running"] is desktop_up
assert result["port"] == 5900
@pytest.mark.asyncio
async def test_names_the_running_backend(self):
"""The UI must not be told wayvnc while RealVNC is the one serving."""
backends = self._backends(wayvnc_active=False, realvnc_active=True)
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.on):
result = await get_vnc_status_dict()
assert result["running"] is True
assert result["backend"] == "realvnc"
@pytest.mark.asyncio
async def test_unavailable_reports_reason(self):
"""No backend installed is reported as unavailable, with the reason."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=([], "No VNC server installed")):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.on):
result = await get_vnc_status_dict()
assert result["available"] is False
assert result["can_start"] is False
assert result["reason"] == "No VNC server installed"
assert result["backend"] is None
@pytest.mark.asyncio
async def test_explains_a_stopped_desktop(self):
"""The reason should name the desktop when that is what blocks VNC."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=(self._backends(), None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.off):
result = await get_vnc_status_dict()
assert "Desktop must be running" in result["reason"]
class TestVncStartStop:
"""Starting requires a desktop; stopping never does, and stops every backend."""
WAYVNC = {"unit": "wayvnc.service", "name": "wayvnc", "active": False}
REALVNC = {"unit": "vncserver-x11-serviced.service", "name": "realvnc", "active": False}
@pytest.mark.asyncio
async def test_start_uses_the_preferred_backend(self):
"""With a desktop up, start should start wayvnc and nothing else."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=([dict(self.WAYVNC), dict(self.REALVNC)], None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.on):
with mock.patch("asyncio.create_subprocess_exec", return_value=_ok_proc()) as mock_exec:
result = await start_vnc()
assert result["success"] is True
mock_exec.assert_called_once_with(
"systemctl", "start", "wayvnc.service", stdout=mock.ANY, stderr=mock.ANY)
@pytest.mark.asyncio
async def test_start_refused_when_desktop_stopped(self):
"""The gate lives here, not only in the UI, so curl callers get it too."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=([dict(self.WAYVNC)], None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.off):
with mock.patch("asyncio.create_subprocess_exec") as mock_exec:
result = await start_vnc()
assert result["success"] is False
assert result["precondition_failed"] is True
assert "Desktop must be running" in result["error"]
mock_exec.assert_not_called()
@pytest.mark.asyncio
async def test_stop_stops_every_running_backend(self):
"""Otherwise the other stack keeps serving while the UI reports VNC off."""
backends = [dict(self.WAYVNC, active=True), dict(self.REALVNC, active=True)]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
with mock.patch("asyncio.create_subprocess_exec", return_value=_ok_proc()) as mock_exec:
result = await stop_vnc()
assert result["success"] is True
stopped = [call.args[2] for call in mock_exec.call_args_list]
assert stopped == ["wayvnc.service", "vncserver-x11-serviced.service"]
@pytest.mark.asyncio
async def test_stop_targets_only_the_running_backend(self):
"""A stopped backend needs no stop; the running one must be stopped."""
backends = [dict(self.WAYVNC, active=False), dict(self.REALVNC, active=True)]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
with mock.patch("asyncio.create_subprocess_exec", return_value=_ok_proc()) as mock_exec:
result = await stop_vnc()
assert result["success"] is True
stopped = [call.args[2] for call in mock_exec.call_args_list]
assert stopped == ["vncserver-x11-serviced.service"]
@pytest.mark.asyncio
async def test_stop_reports_a_backend_that_would_not_stop(self):
"""A VNC server left running must not be reported as stopped."""
backends = [dict(self.WAYVNC, active=True)]
failing = mock.AsyncMock()
failing.communicate.return_value = (b"", b"Job for wayvnc.service failed")
failing.returncode = 1
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
with mock.patch("asyncio.create_subprocess_exec", return_value=failing):
result = await stop_vnc()
assert result["success"] is False
assert "wayvnc.service" in result["error"]
@pytest.mark.asyncio
async def test_stop_allowed_with_desktop_stopped(self):
"""A running VNC must always be stoppable, or it could be stranded on."""
backends = [dict(self.WAYVNC, active=True)]
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends", return_value=(backends, None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.off):
with mock.patch("asyncio.create_subprocess_exec", return_value=_ok_proc()) as mock_exec:
result = await stop_vnc()
assert result["success"] is True
mock_exec.assert_called_once_with(
"systemctl", "stop", "wayvnc.service", stdout=mock.ANY, stderr=mock.ANY)
@pytest.mark.asyncio
@pytest.mark.parametrize("func", [start_vnc, stop_vnc])
async def test_never_enables_or_disables(self, func):
"""Runtime-only: VNC must not be persisted across reboots."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=([dict(self.WAYVNC, active=True)], None)):
with mock.patch("controller.lib.admin.get_desktop_status", return_value=DesktopStatus.on):
with mock.patch("asyncio.create_subprocess_exec", return_value=_ok_proc()) as mock_exec:
await func()
forbidden = {"enable", "disable", "mask", "unmask", "set-default", "isolate"}
for call in mock_exec.call_args_list:
assert not forbidden.intersection(call.args), call.args
@pytest.mark.asyncio
@pytest.mark.parametrize("func", [start_vnc, stop_vnc])
async def test_reports_missing_backend(self, func):
"""Without a backend both actions explain themselves instead of failing obscurely."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("controller.lib.admin.get_vnc_backends",
return_value=([], "No VNC server installed")):
result = await func()
assert result["success"] is False
assert result["error"] == "No VNC server installed"
Reference in a new issue View git blame Copy permalink