kfj001
4c0b9de3df
wifi.py presented itself as a join/leave abstraction but provided no real
behavior: the default backend was a no-op, the facade never verified the
association, and the nmcli queries were wrong. This makes the module do what
it claims, verified end-to-end against the recovered protocol and nmcli 1.52.
wifi.py
- WifiManager: join now verifies the association by polling
backend.is_connected against a settle_timeout (fail-closed). An
unverified join raises HotspotJoinError and leaves connected=False;
verify=False restores fire-and-forget. Tracks last_profile and adds
gateway_ip().
- NmcliWifiBackend: the association check now uses the ACTIVE column
(`device wifi list -f SSID,ACTIVE`). The old `device wifi show` was a
prefix-match accident onto `show-password` and reported in-range APs, not
the associated one. get_ip() queries per-Wi-Fi-interface IP4.ADDRESS
(preferring the 10.100 aGate subnet) instead of the unfiltered all-device
dump that would return a wired/VPN address. Signal is read from the active
AP; the dead ipcmd param is removed; leave() is best-effort.
- WifiBackend: documented the fail-closed contract and added get_signal()
so the manager no longer reaches into a private _run via isinstance.
- NoOpWifiBackend: added assume_connected() for manual/external joins.
client.py
- AgateClient now owns a WifiManager and exposes join_hotspot(),
leave_hotspot(), and attach_local() (join + verify + LocalDeviceClient
aimed at the device gateway). LocalDeviceClient is imported lazily so the
package still works without `requests`.
__init__.py
- Export the backends, HotspotProfile, HotspotJoinError, the SN-derivation
helper, and the module constants.
tests/test_wifi.py (new)
- 20 stdlib-unittest cases: SN derivation, verify pass/fail, fail-closed
behavior, the in-range-but-not-associated regression, IP preference,
tolerant leave, and the client wiring. No radio or pytest required.
Also:
- README: documented the local-connect (device hotspot) flow.
- .gitignore: ignore Python build/cache artifacts.
268 lines
9.3 KiB
Python
268 lines
9.3 KiB
Python
"""Tests for the Wi-Fi join/leave abstraction (no radio, no NetworkManager).
|
|
|
|
Run with either::
|
|
|
|
python -m unittest discover -s tests
|
|
python -m pytest tests # if pytest is installed
|
|
|
|
The ``NmcliWifiBackend`` tests inject a fake ``_run`` so they exercise the
|
|
parsing / fail-closed logic without a real ``nmcli`` binary.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
|
|
|
|
from franklinwh_agate_client.wifi import (
|
|
DEVICE_GATEWAY_IP,
|
|
HotspotJoinError,
|
|
HotspotProfile,
|
|
NmcliWifiBackend,
|
|
NoOpWifiBackend,
|
|
WifiBackend,
|
|
WifiManager,
|
|
derive_hotspot_credentials,
|
|
)
|
|
|
|
|
|
class _FakeBackend(WifiBackend):
|
|
"""An in-memory backend whose association outcome we control."""
|
|
|
|
def __init__(self, associated: bool = True, ip: str = "10.100.1.2") -> None:
|
|
self.associated = associated
|
|
self._ip = ip
|
|
self._profile: HotspotProfile | None = None
|
|
self.joined_calls = 0
|
|
self.leave_calls = 0
|
|
|
|
def join(self, profile: HotspotProfile) -> None:
|
|
self.joined_calls += 1
|
|
self._profile = profile
|
|
|
|
def leave(self) -> None:
|
|
self.leave_calls += 1
|
|
self._profile = None
|
|
|
|
def get_ip(self):
|
|
return self._ip if self._profile else None
|
|
|
|
def is_connected(self, profile: HotspotProfile) -> bool:
|
|
return self.associated and self._profile is not None and self._profile == profile
|
|
|
|
def get_signal(self):
|
|
return 73 if self._profile else None
|
|
|
|
|
|
class _BoomBackend(_FakeBackend):
|
|
"""join() raises — models a hard association failure."""
|
|
|
|
def join(self, profile: HotspotProfile) -> None:
|
|
self.joined_calls += 1
|
|
raise RuntimeError("nmcli: device not found")
|
|
|
|
|
|
# A 20-char SN so the "valid" derivation branch is exercised.
|
|
SN20 = "SN0123456789ABCDEFGH"
|
|
assert len(SN20) == 20
|
|
|
|
|
|
class TestDeriveHotspotCredentials(unittest.TestCase):
|
|
def test_valid_20char_sn(self):
|
|
p = derive_hotspot_credentials(SN20)
|
|
self.assertEqual(p.ssid, "AP_" + SN20[11:])
|
|
self.assertEqual(p.password, SN20[8:])
|
|
self.assertEqual(p.gateway_ip, DEVICE_GATEWAY_IP)
|
|
|
|
def test_invalid_length_uses_fallback(self):
|
|
p = derive_hotspot_credentials("short")
|
|
self.assertEqual(p.ssid, "AP_short")
|
|
self.assertEqual(p.password, "12345678")
|
|
self.assertEqual(p.gateway_ip, DEVICE_GATEWAY_IP)
|
|
|
|
|
|
class TestWifiManagerWithFakeBackend(unittest.TestCase):
|
|
def _manager(self, **kw) -> WifiManager:
|
|
return WifiManager(backend=_FakeBackend(**kw))
|
|
|
|
def test_join_verified_success(self):
|
|
m = self._manager(associated=True)
|
|
profile = m.join_device_hotspot(SN20, settle_timeout=5, poll_interval=0.01)
|
|
self.assertTrue(m.connected)
|
|
self.assertEqual(profile.ssid, "AP_" + SN20[11:])
|
|
self.assertEqual(m.last_profile, profile)
|
|
self.assertEqual(m.gateway_ip(), DEVICE_GATEWAY_IP)
|
|
self.assertEqual(m.local_ip(), "10.100.1.2")
|
|
|
|
def test_join_unverified_raises_and_stays_disconnected(self):
|
|
m = self._manager(associated=False)
|
|
with self.assertRaises(HotspotJoinError):
|
|
m.join_device_hotspot(SN20, settle_timeout=0.1, poll_interval=0.01)
|
|
self.assertFalse(m.connected)
|
|
|
|
def test_hard_join_failure_propagates(self):
|
|
m = WifiManager(backend=_BoomBackend(associated=True))
|
|
with self.assertRaises(RuntimeError):
|
|
m.join_device_hotspot(SN20)
|
|
self.assertFalse(m.connected)
|
|
|
|
def test_verify_false_skips_check(self):
|
|
m = self._manager(associated=False)
|
|
profile = m.join_device_hotspot(SN20, verify=False)
|
|
self.assertTrue(m.connected)
|
|
self.assertEqual(profile.ssid, "AP_" + SN20[11:])
|
|
|
|
def test_leave_resets_state(self):
|
|
m = self._manager(associated=True)
|
|
m.join_device_hotspot(SN20, settle_timeout=5, poll_interval=0.01)
|
|
m.leave_device_hotspot()
|
|
self.assertFalse(m.connected)
|
|
self.assertIsNone(m.last_profile)
|
|
self.assertEqual(m.gateway_ip(), DEVICE_GATEWAY_IP) # fallback
|
|
self.assertEqual(m.local_ip(), None)
|
|
|
|
def test_signal_delegates_to_backend(self):
|
|
m = self._manager(associated=True)
|
|
m.join_device_hotspot(SN20, settle_timeout=5, poll_interval=0.01)
|
|
self.assertEqual(m.get_wifi_signal_strength(), 73)
|
|
|
|
def test_default_backend_is_noop(self):
|
|
m = WifiManager()
|
|
self.assertIsInstance(m.backend, NoOpWifiBackend)
|
|
m.join_device_hotspot(SN20, settle_timeout=1, poll_interval=0.01)
|
|
self.assertTrue(m.connected)
|
|
|
|
|
|
class TestNoOpBackend(unittest.TestCase):
|
|
def test_assume_connected_roundtrip(self):
|
|
b = NoOpWifiBackend(local_ip="10.100.1.9")
|
|
p = HotspotProfile(ssid="AP_x", password="pw")
|
|
b.assume_connected(p)
|
|
self.assertTrue(b.is_connected(p))
|
|
self.assertEqual(b.get_ip(), "10.100.1.9")
|
|
b.leave()
|
|
self.assertFalse(b.is_connected(p))
|
|
|
|
|
|
class _RunStub:
|
|
"""Replace ``NmcliWifiBackend._run`` with canned output / errors.
|
|
|
|
``responses`` maps a leading-argument tuple
|
|
(e.g. ``("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list")``) to a
|
|
string result or an Exception to raise.
|
|
"""
|
|
|
|
def __init__(self, responses: dict, backend: NmcliWifiBackend) -> None:
|
|
self.responses = responses
|
|
self.backend = backend
|
|
self.calls: list[tuple] = []
|
|
|
|
def __call__(self, *args):
|
|
self.calls.append(args)
|
|
for key, val in self.responses.items():
|
|
if args[: len(key)] == key:
|
|
if isinstance(val, Exception):
|
|
raise val
|
|
return val
|
|
raise AssertionError(f"unexpected nmcli invocation: {args}")
|
|
|
|
|
|
def _patch_run(backend: NmcliWifiBackend, responses: dict) -> _RunStub:
|
|
stub = _RunStub(responses, backend)
|
|
backend._run = stub # type: ignore[method-assign]
|
|
return stub
|
|
|
|
|
|
class TestNmcliParsing(unittest.TestCase):
|
|
def setUp(self):
|
|
self.backend = NmcliWifiBackend()
|
|
self.profile = HotspotProfile(ssid="AP_SN123", password="pw")
|
|
|
|
def test_is_connected_true_when_active_yes(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list"):
|
|
"AP_SN123:yes\nOtherAP:no\n",
|
|
})
|
|
self.assertTrue(self.backend.is_connected(self.profile))
|
|
|
|
def test_is_connected_false_when_in_range_but_not_active(self):
|
|
# Regression: a visible (in-range) AP is NOT an association.
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list"):
|
|
"AP_SN123:no\nOtherAP:yes\n",
|
|
})
|
|
self.assertFalse(self.backend.is_connected(self.profile))
|
|
|
|
def test_is_connected_fail_closed_on_error(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list"):
|
|
RuntimeError("nmcli: no NetworkManager"),
|
|
})
|
|
self.assertFalse(self.backend.is_connected(self.profile))
|
|
|
|
def test_get_ip_prefers_10_100_subnet(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "DEVICE,TYPE,STATE", "device", "show"):
|
|
"wlan0:wifi:activated\neth0:ethernet:activated\n",
|
|
("-g", "IP4.ADDRESS", "device", "show", "wlan0"):
|
|
"192.168.1.50\n10.100.1.50\n",
|
|
})
|
|
self.assertEqual(self.backend.get_ip(), "10.100.1.50")
|
|
|
|
def test_get_ip_ignores_non_wifi_and_unlisted(self):
|
|
# eth0 is not wifi, so it must never be queried for an address.
|
|
stub = _patch_run(self.backend, {
|
|
("-t", "-f", "DEVICE,TYPE,STATE", "device", "show"):
|
|
"wlan0:wifi:activated\neth0:ethernet:activated\n",
|
|
("-g", "IP4.ADDRESS", "device", "show", "wlan0"): "10.100.1.50\n",
|
|
})
|
|
self.assertEqual(self.backend.get_ip(), "10.100.1.50")
|
|
self.assertNotIn(
|
|
("-g", "IP4.ADDRESS", "device", "show", "eth0"), stub.calls
|
|
)
|
|
|
|
def test_get_ip_none_when_no_wifi_activated(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "DEVICE,TYPE,STATE", "device", "show"):
|
|
"eth0:ethernet:activated\n",
|
|
})
|
|
self.assertIsNone(self.backend.get_ip())
|
|
|
|
def test_get_signal_reads_active_ap(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "SSID,SIGNAL,ACTIVE", "device", "wifi", "list"):
|
|
"AP_SN123:73:yes\nOtherAP:20:no\n",
|
|
})
|
|
self.assertEqual(self.backend.get_signal(), 73)
|
|
|
|
def test_get_signal_none_when_no_active(self):
|
|
_patch_run(self.backend, {
|
|
("-t", "-f", "SSID,SIGNAL,ACTIVE", "device", "wifi", "list"):
|
|
"OtherAP:20:no\n",
|
|
})
|
|
self.assertIsNone(self.backend.get_signal())
|
|
|
|
def test_leave_tolerates_nothing_to_disconnect(self):
|
|
stub = _patch_run(self.backend, {
|
|
("device", "wifi", "disconnect"): RuntimeError("There is no Wi-Fi device"),
|
|
})
|
|
self.backend.leave() # must not raise
|
|
self.assertEqual(len(stub.calls), 1)
|
|
|
|
def test_join_passes_ssid_and_password(self):
|
|
stub = _patch_run(self.backend, {
|
|
("device", "wifi", "connect"): "",
|
|
})
|
|
self.backend.join(self.profile)
|
|
self.assertEqual(
|
|
stub.calls[0],
|
|
("device", "wifi", "connect", "AP_SN123", "password", "pw"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|