fix(wifi): make Connect work from the setup AP (#571)

* fix(wifi): make Connect work from the setup AP

Joining a network from LEDMatrix-Setup has to take the AP down first, which
drops the phone that sent the request. The connect endpoint answered only
after the attempt finished, so the browser never got a reply and the WiFi
tab's Connect button appeared to do nothing.

- /wifi/connect answers 202 immediately while the AP is active and connects
  in a background thread; the result (never the password) is reported via
  /wifi/status as last_connect_attempt. A second connect while one is
  pending gets 409.
- connect_to_network holds a /tmp flag for the attempt; the monitor daemon
  skips AP management while it is fresh. Previously the daemon's
  disconnected counter, accumulated over the whole AP session, re-enabled
  the AP on its next tick in the middle of the connect.
- The WiFi tab and captive setup page explain the handoff up front, and on
  reopening show why the last attempt failed. The wrong-password message
  now works: the route sets the error_type the captive page checks.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(wifi): serialize connect attempts on both paths

Addresses CodeRabbit review on #571:

- Check for a pending attempt before branching on AP state. A background
  attempt takes the AP down long before it finishes, so a second click
  used to bypass the 409 and start a competing synchronous connect.
- Record pending for the synchronous (non-AP) path too, so two requests
  can't overlap and have the first clear the daemon's in-progress flag
  while the second is still connecting.
- Clear the pending state if the background thread fails to start, rather
  than refusing every later request until restart.
- Say the setup network returns "within a few minutes": a stale flag plus
  the daemon's grace period can take longer than one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-14 12:40:41 -04:00
committed by GitHub
co-authored by Claude Opus 5
parent d51f7ada14
commit 7e580dc005
7 changed files with 484 additions and 31 deletions
+126
View File
@@ -26,6 +26,9 @@ def wifi_manager():
"""Patch WiFiManager where it is defined; yield the instance mock."""
with patch("src.wifi_manager.WiFiManager") as cls:
instance = MagicMock()
# A bare MagicMock is truthy, which would send every connect down
# the setup-AP background path.
instance._is_ap_mode_active.return_value = False
cls.return_value = instance
yield instance
@@ -97,6 +100,129 @@ class TestConnect:
assert "RuntimeError" in body["details"]
class TestConnectThroughSetupAp:
"""With the setup AP up, the phone making the request is connected through
the very network the connect tears down. A synchronous answer can never
arrive, so the route answers 202 first and connects in the background."""
URL = "/api/v3/wifi/connect"
@pytest.fixture(autouse=True)
def ap_active(self, api_v3_module, wifi_manager, monkeypatch):
from web_interface.blueprints.api_v3 import wifi as wifi_routes
wifi_manager._is_ap_mode_active.return_value = True
monkeypatch.setattr(wifi_routes, "_AP_HANDOFF_DELAY_SECONDS", 0)
monkeypatch.setattr(wifi_routes, "_last_connect_attempt", None)
self.spawned = []
monkeypatch.setattr(wifi_routes, "_spawn", self.spawned.append)
self.routes = wifi_routes
def _run_spawned(self):
assert len(self.spawned) == 1
self.spawned.pop()()
def test_answers_202_before_connecting(self, api_v3_client, wifi_manager):
response = api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "hunter22"})
assert response.status_code == 202
assert response.get_json()["status"] == "pending"
wifi_manager.connect_to_network.assert_not_called()
assert self.routes._last_connect_snapshot()["state"] == "pending"
def test_background_result_is_reported_by_status(self, api_v3_client, wifi_manager):
wifi_manager.connect_to_network.return_value = (True, "Connected to HomeNet")
api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "hunter22"})
self._run_spawned()
wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "hunter22")
wifi_manager.config = {}
wifi_manager.get_wifi_status.return_value = MagicMock(
connected=True, ssid="HomeNet", ip_address="10.0.0.5", signal=70,
ap_mode_active=False)
attempt = api_v3_client.get("/api/v3/wifi/status").get_json()["data"]["last_connect_attempt"]
assert attempt["ssid"] == "HomeNet"
assert attempt["state"] == "success"
assert "hunter22" not in str(attempt)
def test_wrong_password_is_flagged(self, api_v3_client, wifi_manager):
wifi_manager.connect_to_network.return_value = (
False, "wrong_password: Secrets were required, but not provided")
api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "hunter22"})
self._run_spawned()
attempt = self.routes._last_connect_snapshot()
assert attempt["state"] == "failed"
assert attempt["error_type"] == "wrong_password"
assert attempt["message"] == "Incorrect password for HomeNet"
def test_manager_exception_is_recorded_as_a_failure(self, api_v3_client, wifi_manager):
wifi_manager.connect_to_network.side_effect = RuntimeError("nmcli vanished")
api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
self._run_spawned()
attempt = self.routes._last_connect_snapshot()
assert attempt["state"] == "failed"
assert "RuntimeError" in attempt["message"]
def test_a_second_connect_while_one_is_pending_is_refused(self, api_v3_client, wifi_manager):
api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
response = api_v3_client.post(self.URL, json={"ssid": "OtherNet"})
assert response.status_code == 409
assert len(self.spawned) == 1
def test_pending_is_honoured_after_the_ap_has_gone_down(self, api_v3_client, wifi_manager):
# The background attempt tears the AP down long before it finishes; a
# second click must not slip through as a synchronous connect.
api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
wifi_manager._is_ap_mode_active.return_value = False
response = api_v3_client.post(self.URL, json={"ssid": "OtherNet"})
assert response.status_code == 409
wifi_manager.connect_to_network.assert_not_called()
def test_a_thread_that_fails_to_start_does_not_leave_pending_behind(
self, api_v3_client, wifi_manager, monkeypatch):
def refuse(_target):
raise RuntimeError("can't start new thread")
monkeypatch.setattr(self.routes, "_spawn", refuse)
assert api_v3_client.post(self.URL, json={"ssid": "HomeNet"}).status_code == 500
assert self.routes._last_connect_snapshot() is None
monkeypatch.setattr(self.routes, "_spawn", self.spawned.append)
assert api_v3_client.post(self.URL, json={"ssid": "HomeNet"}).status_code == 202
class TestConnectSerializedWithoutAp:
"""Without the AP the route connects synchronously, but still one attempt
at a time: a second connect_to_network would remove the in-progress flag
the daemon relies on when the first finishes."""
URL = "/api/v3/wifi/connect"
@pytest.fixture(autouse=True)
def no_prior_attempt(self, api_v3_module, monkeypatch):
from web_interface.blueprints.api_v3 import wifi as wifi_routes
monkeypatch.setattr(wifi_routes, "_last_connect_attempt", None)
self.routes = wifi_routes
def test_a_connect_during_a_synchronous_connect_is_refused(
self, api_v3_client, wifi_manager):
inner = []
def connect(ssid, password):
if not inner:
inner.append(api_v3_client.post(self.URL, json={"ssid": "OtherNet"}))
return True, f"Connected to {ssid}"
wifi_manager.connect_to_network.side_effect = connect
response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
assert response.status_code == 200
assert inner[0].status_code == 409
wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "")
assert self.routes._last_connect_snapshot()["state"] == "success"
def test_an_exception_does_not_leave_pending_behind(self, api_v3_client, wifi_manager):
wifi_manager.connect_to_network.side_effect = RuntimeError("nmcli gone")
assert api_v3_client.post(self.URL, json={"ssid": "HomeNet"}).status_code == 500
assert self.routes._last_connect_snapshot()["state"] == "failed"
class TestDisconnect:
URL = "/api/v3/wifi/disconnect"
+51
View File
@@ -40,6 +40,57 @@ def _wire(wm, connected, ethernet, ap_active):
wm._save_cached_scan = MagicMock()
wm._FORCE_AP_FLAG_PATH = MagicMock()
wm._FORCE_AP_FLAG_PATH.exists.return_value = False
wm._connect_in_progress = MagicMock(return_value=False)
class TestConnectInProgressHoldsOffTheDaemon:
"""Connecting from the setup AP takes the AP down before joining. The
daemon had already counted the whole AP session as "disconnected", so its
next tick re-enabled the AP in the middle of the connect."""
def test_no_ap_enable_while_a_connect_is_running(self, wm):
_wire(wm, connected=False, ethernet=False, ap_active=False)
wm._connect_in_progress.return_value = True
wm._disconnected_checks = 10 # accumulated while the AP was up
changed, *_ = wm.check_and_manage_ap_mode_with_state()
assert changed is False
wm.enable_ap_mode.assert_not_called()
assert wm._disconnected_checks == 0
def test_grace_period_restarts_after_the_connect(self, wm):
_wire(wm, connected=False, ethernet=False, ap_active=False)
wm._connect_in_progress.return_value = True
wm._disconnected_checks = 10
wm.check_and_manage_ap_mode_with_state()
wm._connect_in_progress.return_value = False
wm.check_and_manage_ap_mode_with_state()
wm.enable_ap_mode.assert_not_called()
def test_flag_is_held_for_the_connect_and_removed_after(self, wm, tmp_path):
flag = tmp_path / "connecting"
wm._CONNECT_IN_PROGRESS_FLAG_PATH = flag
seen = []
wm._connect_validated = lambda ssid, pw: seen.append(flag.exists()) or (True, "ok")
assert wm.connect_to_network("HomeNet", "hunter22") == (True, "ok")
assert seen == [True]
assert not flag.exists()
def test_flag_is_removed_when_the_connect_raises(self, wm, tmp_path):
flag = tmp_path / "connecting"
wm._CONNECT_IN_PROGRESS_FLAG_PATH = flag
wm._connect_validated = MagicMock(side_effect=RuntimeError("nmcli gone"))
with pytest.raises(RuntimeError):
wm.connect_to_network("HomeNet", "hunter22")
assert not flag.exists()
def test_a_stale_flag_from_a_dead_process_is_ignored(self, wm, tmp_path):
flag = tmp_path / "connecting"
flag.touch()
wm._CONNECT_IN_PROGRESS_FLAG_PATH = flag
assert WiFiManager._connect_in_progress(wm) is True
old = flag.stat().st_mtime - WiFiManager._CONNECT_FLAG_MAX_AGE_SECONDS - 1
os.utime(flag, (old, old))
assert WiFiManager._connect_in_progress(wm) is False
class TestWithState:
@@ -134,6 +134,7 @@ class TestConnectEndpointSurfacesTheRefusal:
def wifi_manager(self):
with patch("src.wifi_manager.WiFiManager") as cls:
instance = MagicMock()
instance._is_ap_mode_active.return_value = False
cls.return_value = instance
yield instance