WifiManager: add test for state machine (#37396)

* test wifi state machine * clean up and another few tests * no unittest :(( * clean up * clean up * try to repro on device * try to repro on device * nice, the flicker is covered by test_user_initiated_skips_dbus_lookup! * add todo soon to be all fixed * documentaiton * test the thread races too * _fire -> fire * duplication * new state * fix some tests * format * combine similar tests * use process_callbacks * clean up * collapse two tests * rm nl * previous messy test * delete old * asked another to ask questions
2026-06-27 17:42:04 +08:00 · 2026-02-25 01:59:19 -08:00
parent 1792a60053
commit d9b5a1e30b
3 changed files with 695 additions and 0 deletions
@@ -23,9 +23,11 @@ class NMDeviceStateReason(IntEnum):
  # https://networkmanager.dev/docs/api/1.46/nm-dbus-types.html#NMDeviceStateReason
  NONE = 0
  UNKNOWN = 1
+  IP_CONFIG_UNAVAILABLE = 5
  NO_SECRETS = 7
  SUPPLICANT_DISCONNECT = 8
  CONNECTION_REMOVED = 38
+  USER_REQUESTED = 39
  SSID_NOT_FOUND = 53
  NEW_ACTIVATION = 60

@@ -0,0 +1,678 @@
+"""Tests for WifiManager._handle_state_change.
+
+Tests the state machine in isolation by constructing a WifiManager with mocked
+DBus, then calling _handle_state_change directly with NM state transitions.
+
+Many tests assert *desired* behavior that the current code doesn't implement yet.
+These are marked with pytest.mark.xfail and document the intended fix.
+"""
+import pytest
+from pytest_mock import MockerFixture
+
+from openpilot.system.ui.lib.networkmanager import NMDeviceState, NMDeviceStateReason
+from openpilot.system.ui.lib.wifi_manager import WifiManager, WifiState, ConnectStatus
+
+
+def _make_wm(mocker: MockerFixture, connections=None):
+  """Create a WifiManager with only the fields _handle_state_change touches."""
+  mocker.patch.object(WifiManager, '_initialize')
+  wm = WifiManager.__new__(WifiManager)
+  wm._exit = True  # prevent stop() from doing anything in __del__
+  wm._conn_monitor = mocker.MagicMock()
+  wm._connections = dict(connections or {})
+  wm._wifi_state = WifiState()
+  wm._callback_queue = []
+  wm._need_auth = []
+  wm._activated = []
+  wm._update_networks = mocker.MagicMock()
+  wm._get_active_wifi_connection = mocker.MagicMock(return_value=(None, None))
+  return wm
+
+
+def fire(wm: WifiManager, new_state: int, prev_state: int = NMDeviceState.UNKNOWN,
+         reason: int = NMDeviceStateReason.NONE) -> None:
+  """Feed a state change into the handler."""
+  wm._handle_state_change(new_state, prev_state, reason)
+
+
+def fire_wpa_connect(wm: WifiManager) -> None:
+  """WPA handshake then IP negotiation through ACTIVATED, as seen on device."""
+  fire(wm, NMDeviceState.NEED_AUTH)
+  fire(wm, NMDeviceState.PREPARE, prev_state=NMDeviceState.NEED_AUTH)
+  fire(wm, NMDeviceState.CONFIG)
+  fire(wm, NMDeviceState.IP_CONFIG)
+  fire(wm, NMDeviceState.IP_CHECK)
+  fire(wm, NMDeviceState.SECONDARIES)
+  fire(wm, NMDeviceState.ACTIVATED)
+
+
+# ---------------------------------------------------------------------------
+# Basic transitions
+# ---------------------------------------------------------------------------
+
+class TestDisconnected:
+  def test_generic_disconnect_clears_state(self, mocker):
+    wm = _make_wm(mocker)
+    wm._wifi_state = WifiState(ssid="Net", status=ConnectStatus.CONNECTED)
+
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.UNKNOWN)
+
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+    wm._update_networks.assert_not_called()
+
+  def test_new_activation_is_noop(self, mocker):
+    """NEW_ACTIVATION means NM is about to connect to another network — don't clear."""
+    wm = _make_wm(mocker)
+    wm._wifi_state = WifiState(ssid="OldNet", status=ConnectStatus.CONNECTED)
+
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.NEW_ACTIVATION)
+
+    assert wm._wifi_state.ssid == "OldNet"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+
+  @pytest.mark.xfail(reason="TODO: CONNECTION_REMOVED should only clear if ssid not in _connections")
+  def test_connection_removed_keeps_other_connecting(self, mocker):
+    """Forget A while connecting to B: CONNECTION_REMOVED for A must not clear B."""
+    wm = _make_wm(mocker, connections={"B": "/path/B"})
+    wm._set_connecting("B")
+
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.CONNECTION_REMOVED)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+  def test_connection_removed_clears_when_forgotten(self, mocker):
+    """Forget A: A is no longer in _connections, so state should clear."""
+    wm = _make_wm(mocker, connections={})
+    wm._wifi_state = WifiState(ssid="A", status=ConnectStatus.CONNECTED)
+
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.CONNECTION_REMOVED)
+
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+
+
+class TestDeactivating:
+  @pytest.mark.xfail(reason="TODO: DEACTIVATING should be a no-op")
+  def test_deactivating_is_noop(self, mocker):
+    """DEACTIVATING should be a no-op — DISCONNECTED follows with correct state.
+
+    Fix: remove the entire DEACTIVATING elif block — do nothing for any reason.
+    """
+    wm = _make_wm(mocker)
+    wm._wifi_state = WifiState(ssid="Net", status=ConnectStatus.CONNECTED)
+
+    fire(wm, NMDeviceState.DEACTIVATING, reason=NMDeviceStateReason.CONNECTION_REMOVED)
+
+    assert wm._wifi_state.ssid == "Net"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+
+
+class TestPrepareConfig:
+  @pytest.mark.xfail(reason="TODO: should skip DBus lookup when ssid already set")
+  def test_user_initiated_skips_dbus_lookup(self, mocker):
+    """User called _set_connecting('B') — PREPARE must not overwrite via DBus.
+
+    Reproduced on device: rapidly tap A then B. PREPARE's DBus lookup returns A's
+    stale conn_path, overwriting ssid to A for 1-2 frames. UI shows the "connecting"
+    indicator briefly jump to the wrong network row then back.
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A", "B": "/path/B"})
+    wm._set_connecting("B")
+    wm._get_active_wifi_connection.return_value = ("/path/A", {})
+
+    fire(wm, NMDeviceState.PREPARE)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+    wm._get_active_wifi_connection.assert_not_called()
+
+  @pytest.mark.parametrize("state", [NMDeviceState.PREPARE, NMDeviceState.CONFIG])
+  def test_auto_connect_looks_up_ssid(self, mocker, state):
+    """Auto-connection (ssid=None): PREPARE and CONFIG must look up ssid from NM."""
+    wm = _make_wm(mocker, connections={"AutoNet": "/path/auto"})
+    wm._get_active_wifi_connection.return_value = ("/path/auto", {})
+
+    fire(wm, state)
+
+    assert wm._wifi_state.ssid == "AutoNet"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+  def test_auto_connect_dbus_fails(self, mocker):
+    """Auto-connection but DBus returns None: ssid stays None, status CONNECTING."""
+    wm = _make_wm(mocker)
+
+    fire(wm, NMDeviceState.PREPARE)
+
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+  def test_auto_connect_conn_path_not_in_connections(self, mocker):
+    """DBus returns a conn_path that doesn't match any known connection."""
+    wm = _make_wm(mocker, connections={"Other": "/path/other"})
+    wm._get_active_wifi_connection.return_value = ("/path/unknown", {})
+
+    fire(wm, NMDeviceState.PREPARE)
+
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+
+class TestNeedAuth:
+  def test_wrong_password_fires_callback(self, mocker):
+    """NEED_AUTH+SUPPLICANT_DISCONNECT from CONFIG = real wrong password."""
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._set_connecting("SecNet")
+
+    fire(wm, NMDeviceState.NEED_AUTH, prev_state=NMDeviceState.CONFIG,
+         reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+    assert len(wm._callback_queue) == 1
+    wm.process_callbacks()
+    cb.assert_called_once_with("SecNet")
+
+  def test_failed_no_secrets_fires_callback(self, mocker):
+    """FAILED+NO_SECRETS = wrong password (weak/gone network)."""
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._set_connecting("WeakNet")
+
+    fire(wm, NMDeviceState.FAILED, reason=NMDeviceStateReason.NO_SECRETS)
+
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+    assert len(wm._callback_queue) == 1
+    wm.process_callbacks()
+    cb.assert_called_once_with("WeakNet")
+
+  def test_need_auth_then_failed_no_double_fire(self, mocker):
+    """Real device sends NEED_AUTH(SUPPLICANT_DISCONNECT) then FAILED(NO_SECRETS) back-to-back.
+
+    The first clears ssid, so the second must not fire a duplicate callback.
+    Real device sequence: NEED_AUTH(CONFIG, SUPPLICANT_DISCONNECT) → FAILED(NEED_AUTH, NO_SECRETS)
+    """
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._set_connecting("BadPass")
+
+    fire(wm, NMDeviceState.NEED_AUTH, prev_state=NMDeviceState.CONFIG,
+         reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+    assert len(wm._callback_queue) == 1
+
+    fire(wm, NMDeviceState.FAILED, prev_state=NMDeviceState.NEED_AUTH,
+         reason=NMDeviceStateReason.NO_SECRETS)
+    assert len(wm._callback_queue) == 1  # no duplicate
+
+    wm.process_callbacks()
+    cb.assert_called_once_with("BadPass")
+
+  def test_no_ssid_no_callback(self, mocker):
+    """If ssid is None when NEED_AUTH fires, no callback enqueued."""
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+
+    fire(wm, NMDeviceState.NEED_AUTH, reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+
+    assert len(wm._callback_queue) == 0
+
+  @pytest.mark.xfail(reason="TODO: interrupted auth (prev=DISCONNECTED) should be ignored")
+  def test_interrupted_auth_ignored(self, mocker):
+    """Switching A->B: NEED_AUTH from A (prev=DISCONNECTED) must not fire callback.
+
+    Reproduced on device: rapidly switching between two saved networks can trigger a
+    rare false "wrong password" dialog for the previous network, even though both have
+    correct passwords. The stale NEED_AUTH has prev_state=DISCONNECTED (not CONFIG).
+
+    Fix: the handler's second param is currently `_` (unused). Rename to `prev_state`
+    and only fire the NEED_AUTH callback when prev_state indicates a real auth failure
+    (e.g. prev_state == CONFIG), not a stale signal from a prior connection.
+    """
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._set_connecting("A")
+    wm._set_connecting("B")
+
+    fire(wm, NMDeviceState.NEED_AUTH, prev_state=NMDeviceState.DISCONNECTED,
+         reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+    assert len(wm._callback_queue) == 0
+
+  def test_need_auth_targets_previous_ssid_via_prev_ssid(self, mocker):
+    """Switch A→B, late NEED_AUTH arrives: prev_ssid mechanism fires callback for A.
+
+    This tests current prev_ssid behavior which we plan to remove.
+    Migration: (1) add prev_state guard to NEED_AUTH (see test_interrupted_auth_ignored),
+    (2) remove prev_ssid from WifiState and handler, (3) delete this test — it becomes
+    redundant with test_interrupted_auth_ignored once prev_state is the guard mechanism.
+    """
+    wm = _make_wm(mocker)
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._set_connecting("A")
+    wm._set_connecting("B")
+
+    fire(wm, NMDeviceState.NEED_AUTH, reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+
+    assert len(wm._callback_queue) == 1
+    wm.process_callbacks()
+    cb.assert_called_once_with("A")
+
+
+class TestPassthroughStates:
+  """NEED_AUTH (generic), IP_CONFIG, IP_CHECK, SECONDARIES, FAILED (generic) are no-ops."""
+
+  @pytest.mark.parametrize("state", [
+    NMDeviceState.NEED_AUTH,
+    NMDeviceState.IP_CONFIG,
+    NMDeviceState.IP_CHECK,
+    NMDeviceState.SECONDARIES,
+    NMDeviceState.FAILED,
+  ])
+  def test_passthrough_is_noop(self, mocker, state):
+    wm = _make_wm(mocker)
+    wm._set_connecting("Net")
+
+    fire(wm, state, reason=NMDeviceStateReason.NONE)
+
+    assert wm._wifi_state.ssid == "Net"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+    assert len(wm._callback_queue) == 0
+
+
+class TestActivated:
+  def test_sets_connected(self, mocker):
+    """ACTIVATED sets status to CONNECTED and fires callback."""
+    wm = _make_wm(mocker, connections={"MyNet": "/path/mynet"})
+    cb = mocker.MagicMock()
+    wm.add_callbacks(activated=cb)
+    wm._set_connecting("MyNet")
+    wm._get_active_wifi_connection.return_value = ("/path/mynet", {})
+
+    fire(wm, NMDeviceState.ACTIVATED)
+
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "MyNet"
+    assert len(wm._callback_queue) == 1
+    wm.process_callbacks()
+    cb.assert_called_once()
+
+  def test_conn_path_none_still_connected(self, mocker):
+    """ACTIVATED but DBus returns None: status CONNECTED, ssid unchanged."""
+    wm = _make_wm(mocker)
+    wm._set_connecting("MyNet")
+
+    fire(wm, NMDeviceState.ACTIVATED)
+
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "MyNet"
+
+  def test_activated_side_effects(self, mocker):
+    """ACTIVATED persists the volatile connection to disk and triggers _update_networks."""
+    wm = _make_wm(mocker, connections={"Net": "/path/net"})
+    wm._set_connecting("Net")
+    wm._get_active_wifi_connection.return_value = ("/path/net", {})
+
+    fire(wm, NMDeviceState.ACTIVATED)
+
+    wm._conn_monitor.send_and_get_reply.assert_called_once()
+    wm._update_networks.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Thread races: _set_connecting on main thread vs _handle_state_change on monitor thread.
+# Uses side_effect on the DBus mock to simulate _set_connecting running mid-handler.
+# ---------------------------------------------------------------------------
+# The deterministic fixes (skip DBus lookup when ssid already set, prev_state guard
+# on NEED_AUTH) also shrink these race windows to near-zero. If races are still
+# visible after, make WifiState frozen (replace() + single atomic assignment) and/or
+# add a narrow lock around _wifi_state reads/writes (not around DBus calls).
+# The NEED_AUTH prev_ssid mutation race is eliminated by removing prev_ssid entirely.
+
+class TestThreadRaces:
+  @pytest.mark.xfail(reason="TODO: PREPARE overwrites _set_connecting via stale DBus lookup")
+  def test_prepare_race_user_tap_during_dbus(self, mocker):
+    """User taps B while PREPARE's DBus call is in flight for auto-connect.
+
+    Monitor thread reads wifi_state (ssid=None), starts DBus call.
+    Main thread: _set_connecting("B"). Monitor thread writes back stale ssid from DBus.
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A", "B": "/path/B"})
+
+    def user_taps_b_during_dbus(*args, **kwargs):
+      wm._set_connecting("B")
+      return ("/path/A", {})
+
+    wm._get_active_wifi_connection.side_effect = user_taps_b_during_dbus
+
+    fire(wm, NMDeviceState.PREPARE)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+  @pytest.mark.xfail(reason="TODO: ACTIVATED overwrites _set_connecting with stale CONNECTED state")
+  def test_activated_race_user_tap_during_dbus(self, mocker):
+    """User taps B right as A finishes connecting (ACTIVATED handler running).
+
+    Monitor thread reads wifi_state (A, CONNECTING), starts DBus call.
+    Main thread: _set_connecting("B"). Monitor thread writes (A, CONNECTED), losing B.
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A", "B": "/path/B"})
+    wm._set_connecting("A")
+
+    def user_taps_b_during_dbus(*args, **kwargs):
+      wm._set_connecting("B")
+      return ("/path/A", {})
+
+    wm._get_active_wifi_connection.side_effect = user_taps_b_during_dbus
+
+    fire(wm, NMDeviceState.ACTIVATED)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+
+# ---------------------------------------------------------------------------
+# Full sequences (NM signal order from real devices)
+# ---------------------------------------------------------------------------
+
+class TestFullSequences:
+  def test_normal_connect(self, mocker):
+    """User connects to saved network: full happy path.
+
+    Real device sequence (switching from another connected network):
+      DEACTIVATING(ACTIVATED, NEW_ACTIVATION) → DISCONNECTED(DEACTIVATING, NEW_ACTIVATION)
+      PREPARE → CONFIG → NEED_AUTH(CONFIG, NONE) → PREPARE(NEED_AUTH, NONE) → CONFIG
+      → IP_CONFIG → IP_CHECK → SECONDARIES → ACTIVATED
+    """
+    wm = _make_wm(mocker, connections={"Home": "/path/home"})
+    wm._get_active_wifi_connection.return_value = ("/path/home", {})
+
+    wm._set_connecting("Home")
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire(wm, NMDeviceState.NEED_AUTH)  # WPA handshake (reason=NONE)
+    fire(wm, NMDeviceState.PREPARE, prev_state=NMDeviceState.NEED_AUTH)
+    fire(wm, NMDeviceState.CONFIG)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.IP_CONFIG)
+    fire(wm, NMDeviceState.IP_CHECK)
+    fire(wm, NMDeviceState.SECONDARIES)
+    fire(wm, NMDeviceState.ACTIVATED)
+
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "Home"
+
+  def test_wrong_password_then_retry(self, mocker):
+    """Wrong password → NEED_AUTH → FAILED → NM auto-reconnects to saved network.
+
+    Real device sequence:
+      PREPARE → CONFIG → NEED_AUTH(CONFIG, NONE)                 ← WPA handshake
+      → PREPARE(NEED_AUTH, NONE) → CONFIG
+      → NEED_AUTH(CONFIG, SUPPLICANT_DISCONNECT)                 ← wrong password
+      → FAILED(NEED_AUTH, NO_SECRETS)                            ← NM gives up
+      → DISCONNECTED(FAILED, NONE)
+      → PREPARE → CONFIG → NEED_AUTH(CONFIG, NONE) → PREPARE    ← auto-reconnect
+      → CONFIG → IP_CONFIG → ... → ACTIVATED
+    """
+    wm = _make_wm(mocker, connections={"Sec": "/path/sec"})
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+
+    wm._set_connecting("Sec")
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire(wm, NMDeviceState.NEED_AUTH)  # WPA handshake (reason=NONE)
+    fire(wm, NMDeviceState.PREPARE, prev_state=NMDeviceState.NEED_AUTH)
+    fire(wm, NMDeviceState.CONFIG)
+
+    fire(wm, NMDeviceState.NEED_AUTH, prev_state=NMDeviceState.CONFIG,
+         reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+    assert len(wm._callback_queue) == 1
+
+    # FAILED(NO_SECRETS) follows but ssid is already cleared — no double-fire
+    fire(wm, NMDeviceState.FAILED, reason=NMDeviceStateReason.NO_SECRETS)
+    assert len(wm._callback_queue) == 1
+
+    fire(wm, NMDeviceState.DISCONNECTED, prev_state=NMDeviceState.FAILED)
+
+    # Retry
+    wm._callback_queue.clear()
+    wm._set_connecting("Sec")
+    wm._get_active_wifi_connection.return_value = ("/path/sec", {})
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+
+  def test_switch_saved_networks(self, mocker):
+    """Switch from A to B (both saved): NM signal sequence from real device.
+
+    Real device sequence:
+      DEACTIVATING(ACTIVATED, NEW_ACTIVATION) → DISCONNECTED(DEACTIVATING, NEW_ACTIVATION)
+      → PREPARE → CONFIG → NEED_AUTH(CONFIG, NONE) → PREPARE(NEED_AUTH, NONE) → CONFIG
+      → IP_CONFIG → IP_CHECK → SECONDARIES → ACTIVATED
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A", "B": "/path/B"})
+    wm._wifi_state = WifiState(ssid="A", status=ConnectStatus.CONNECTED)
+    wm._get_active_wifi_connection.return_value = ("/path/B", {})
+
+    wm._set_connecting("B")
+
+    fire(wm, NMDeviceState.DEACTIVATING, prev_state=NMDeviceState.ACTIVATED,
+         reason=NMDeviceStateReason.NEW_ACTIVATION)
+    fire(wm, NMDeviceState.DISCONNECTED, prev_state=NMDeviceState.DEACTIVATING,
+         reason=NMDeviceStateReason.NEW_ACTIVATION)
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "B"
+
+  @pytest.mark.xfail(reason="TODO: interrupted auth from switching should not fire need_auth")
+  def test_rapid_switch_no_false_wrong_password(self, mocker):
+    """Switch A→B quickly: A's interrupted NEED_AUTH must NOT show wrong password.
+
+    NOTE: The late NEED_AUTH(DISCONNECTED, SUPPLICANT_DISCONNECT) is common when rapidly
+    switching between networks with wrong/new passwords. Less common when switching between
+    saved networks with correct passwords. Not guaranteed — some switches skip it and go
+    straight from DISCONNECTED to PREPARE. The prev_state is consistently DISCONNECTED
+    for stale signals, so the prev_state guard reliably distinguishes them.
+
+    Worst-case signal sequence this protects against:
+      DEACTIVATING(NEW_ACTIVATION) → DISCONNECTED(NEW_ACTIVATION)
+      → NEED_AUTH(DISCONNECTED, SUPPLICANT_DISCONNECT)  ← A's stale auth failure
+      → PREPARE → CONFIG → ... → ACTIVATED  ← B connects
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A", "B": "/path/B"})
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+    wm._wifi_state = WifiState(ssid="A", status=ConnectStatus.CONNECTED)
+    wm._get_active_wifi_connection.return_value = ("/path/B", {})
+
+    wm._set_connecting("B")
+
+    fire(wm, NMDeviceState.DEACTIVATING, prev_state=NMDeviceState.ACTIVATED,
+         reason=NMDeviceStateReason.NEW_ACTIVATION)
+    fire(wm, NMDeviceState.DISCONNECTED, prev_state=NMDeviceState.DEACTIVATING,
+         reason=NMDeviceStateReason.NEW_ACTIVATION)
+    fire(wm, NMDeviceState.NEED_AUTH, prev_state=NMDeviceState.DISCONNECTED,
+         reason=NMDeviceStateReason.SUPPLICANT_DISCONNECT)
+
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+    assert len(wm._callback_queue) == 0
+
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+
+  @pytest.mark.xfail(reason="TODO: forget A while connecting to B should not clear B")
+  def test_forget_A_connect_B(self, mocker):
+    """Forget A while connecting to B: full signal sequence.
+
+    Real device sequence:
+      DEACTIVATING(ACTIVATED, CONNECTION_REMOVED) → DISCONNECTED(DEACTIVATING, CONNECTION_REMOVED)
+      → PREPARE → CONFIG → NEED_AUTH(CONFIG, NONE) → PREPARE(NEED_AUTH, NONE) → CONFIG
+      → IP_CONFIG → IP_CHECK → SECONDARIES → ACTIVATED
+
+    Signal order:
+      1. User: _set_connecting("B"), forget("A") removes A from _connections
+      2. NewConnection for B arrives → _connections["B"] = ...
+      3. DEACTIVATING(CONNECTION_REMOVED) — should be no-op
+      4. DISCONNECTED(CONNECTION_REMOVED) — B is in _connections, must not clear
+      5. PREPARE → CONFIG → NEED_AUTH → PREPARE → CONFIG → ... → ACTIVATED
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A"})
+    wm._wifi_state = WifiState(ssid="A", status=ConnectStatus.CONNECTED)
+
+    wm._set_connecting("B")
+    del wm._connections["A"]
+    wm._connections["B"] = "/path/B"
+
+    fire(wm, NMDeviceState.DEACTIVATING, prev_state=NMDeviceState.ACTIVATED,
+         reason=NMDeviceStateReason.CONNECTION_REMOVED)
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.DISCONNECTED, prev_state=NMDeviceState.DEACTIVATING,
+         reason=NMDeviceStateReason.CONNECTION_REMOVED)
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    wm._get_active_wifi_connection.return_value = ("/path/B", {})
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "B"
+
+  @pytest.mark.xfail(reason="TODO: forget A while connecting to B should not clear B")
+  def test_forget_A_connect_B_late_new_connection(self, mocker):
+    """Forget A, connect B: NewConnection for B arrives AFTER DISCONNECTED.
+
+    This is the worst-case race: B isn't in _connections when DISCONNECTED fires,
+    so the guard can't protect it and state clears. PREPARE must recover by doing
+    the DBus lookup (ssid is None at that point).
+
+    Signal order:
+      1. User: _set_connecting("B"), forget("A") removes A from _connections
+      2. DEACTIVATING(CONNECTION_REMOVED) — B NOT in _connections, should be no-op
+      3. DISCONNECTED(CONNECTION_REMOVED) — B STILL NOT in _connections, clears state
+      4. NewConnection for B arrives late → _connections["B"] = ...
+      5. PREPARE (ssid=None, so DBus lookup recovers) → CONFIG → ACTIVATED
+    """
+    wm = _make_wm(mocker, connections={"A": "/path/A"})
+    wm._wifi_state = WifiState(ssid="A", status=ConnectStatus.CONNECTED)
+
+    wm._set_connecting("B")
+    del wm._connections["A"]
+
+    fire(wm, NMDeviceState.DEACTIVATING, prev_state=NMDeviceState.ACTIVATED,
+         reason=NMDeviceStateReason.CONNECTION_REMOVED)
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.DISCONNECTED, prev_state=NMDeviceState.DEACTIVATING,
+         reason=NMDeviceStateReason.CONNECTION_REMOVED)
+    # B not in _connections yet, so state clears — this is the known edge case
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+
+    # NewConnection arrives late
+    wm._connections["B"] = "/path/B"
+    wm._get_active_wifi_connection.return_value = ("/path/B", {})
+
+    # PREPARE recovers: ssid is None so it looks up from DBus
+    fire(wm, NMDeviceState.PREPARE)
+    assert wm._wifi_state.ssid == "B"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "B"
+
+  def test_auto_connect(self, mocker):
+    """NM auto-connects (no user action, ssid starts None)."""
+    wm = _make_wm(mocker, connections={"AutoNet": "/path/auto"})
+    wm._get_active_wifi_connection.return_value = ("/path/auto", {})
+
+    fire(wm, NMDeviceState.PREPARE)
+    assert wm._wifi_state.ssid == "AutoNet"
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING
+
+    fire(wm, NMDeviceState.CONFIG)
+    fire_wpa_connect(wm)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTED
+    assert wm._wifi_state.ssid == "AutoNet"
+
+  @pytest.mark.xfail(reason="TODO: FAILED(SSID_NOT_FOUND) should emit error for UI")
+  def test_ssid_not_found(self, mocker):
+    """Network drops off after connection starts.
+
+    NM docs: SSID_NOT_FOUND (53) = "The WiFi network could not be found"
+    Expected sequence: PREPARE → CONFIG → FAILED(SSID_NOT_FOUND) → DISCONNECTED
+
+    NOTE: SSID_NOT_FOUND is rare. On-device testing with a disappearing hotspot typically
+    produces FAILED(NO_SECRETS) instead. May be driver-specific or require the network
+    to vanish from scan results mid-connection.
+
+    The UI error callback mechanism is intentionally deferred — for now just clear state.
+    """
+    wm = _make_wm(mocker, connections={"GoneNet": "/path/gone"})
+    cb = mocker.MagicMock()
+    wm.add_callbacks(need_auth=cb)
+
+    wm._set_connecting("GoneNet")
+    fire(wm, NMDeviceState.PREPARE)
+    fire(wm, NMDeviceState.CONFIG)
+    fire(wm, NMDeviceState.FAILED, reason=NMDeviceStateReason.SSID_NOT_FOUND)
+
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+    assert wm._wifi_state.ssid is None
+
+  def test_failed_then_disconnected_clears_state(self, mocker):
+    """After FAILED, NM always transitions to DISCONNECTED to clean up.
+
+    NM docs: FAILED (120) = "failed to connect, cleaning up the connection request"
+    Full sequence: ... → FAILED(reason) → DISCONNECTED(NONE)
+    """
+    wm = _make_wm(mocker)
+    wm._set_connecting("Net")
+
+    fire(wm, NMDeviceState.FAILED, reason=NMDeviceStateReason.NONE)
+    assert wm._wifi_state.status == ConnectStatus.CONNECTING  # FAILED(NONE) is a no-op
+
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.NONE)
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
+
+  def test_user_requested_disconnect(self, mocker):
+    """User explicitly disconnects from the network.
+
+    NM docs: USER_REQUESTED (39) = "Device disconnected by user or client"
+    Expected sequence: DEACTIVATING(USER_REQUESTED) → DISCONNECTED(USER_REQUESTED)
+    """
+    wm = _make_wm(mocker)
+    wm._wifi_state = WifiState(ssid="MyNet", status=ConnectStatus.CONNECTED)
+
+    fire(wm, NMDeviceState.DEACTIVATING, reason=NMDeviceStateReason.USER_REQUESTED)
+    fire(wm, NMDeviceState.DISCONNECTED, reason=NMDeviceStateReason.USER_REQUESTED)
+
+    assert wm._wifi_state.ssid is None
+    assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
@@ -389,6 +389,21 @@ class WifiManager:
    #    Fix: only do DBus lookup when wifi_state.ssid is None (auto-connections);
    #    user-initiated connections already have ssid set via _set_connecting.

+    # TODO: Thread safety — _wifi_state is read/written by both the monitor thread (this
+    # handler) and the main thread (_set_connecting via connect/activate). The GIL makes
+    # individual assignments atomic, but read-then-write patterns can lose main thread writes:
+    #   PREPARE/CONFIG: reads _wifi_state, does slow DBus call, writes back — if
+    #   _set_connecting runs in between, the handler overwrites it with stale state.
+    #   This is both a deterministic bug (stale DBus data) and a thread race. The
+    #   deterministic fix (skip DBus lookup when ssid is set) also shrinks the race
+    #   window to near-zero since the read/write happen back-to-back under the GIL.
+    #   ACTIVATED: same read-then-write pattern with a DBus call in between.
+    #   NEED_AUTH: mutates _wifi_state.prev_ssid in place, which can corrupt a new
+    #   WifiState created by _set_connecting on the main thread.
+    # The deterministic fixes (skip DBus lookup when ssid set, prev_state guard) shrink
+    # the race windows significantly. If still visible, add a narrow lock around
+    # _wifi_state reads/writes (not around DBus calls, to avoid blocking the UI thread).
+
    # TODO: Handle (FAILED, SSID_NOT_FOUND) and emit for ui to show error
    #  Happens when network drops off after starting connection