diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7c7f56f --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +.eggs/ +build/ +dist/ + +# Tooling / caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.tox/ +.venv/ +venv/ diff --git a/README.md b/README.md index 56e879a..8a7c97e 100644 --- a/README.md +++ b/README.md @@ -51,10 +51,48 @@ client = AgateClient() device_info = client.query_agate_info(device_id) ``` +## Local connect (device hotspot) + +The aGate runs its own SoftAP (gateway `10.100.1.1`); joining it is what +lets the client talk to the device directly. The client derives the hotspot +credentials from the 20-char SN (`AP_` / ``, or the +`AP_` / `12345678` fallback for malformed SNs), joins, **verifies the +association**, then talks to the device over `https://10.100.1.1:18000` +and the encrypted socket channel. + +```python +from franklinwh_agate_client import ( + AgateClient, NmcliWifiBackend, NoOpWifiBackend, derive_hotspot_credentials, +) + +# Linux desktop/laptop with NetworkManager: +client = AgateClient(wifi_backend=NmcliWifiBackend()) +profile = client.join_hotspot(sn) # joins + verifies; raises HotspotJoinError on failure +local = client.attach_local(sn) # LocalDeviceClient -> https://10.100.1.1:18000 +info = local.obtain_agate_info() # or local.query_bms_info(), ... + +# Container, or a host where you join the hotspot manually: +wm = AgateClient(wifi_backend=NoOpWifiBackend(local_ip="10.100.1.2")).wifi_manager +wm.assume_connected(derive_hotspot_credentials(sn)) +``` + +`WifiManager.join_*` are fail-closed: they poll the backend's `is_connected` +against `settle_timeout` (default 30 s) and raise `HotspotJoinError` if the +association never confirms (`verify=False` for fire-and-forget). On Linux, +`NmcliWifiBackend` uses `nmcli ... device wifi list -f SSID,ACTIVE` for the +association check (a merely *visible* AP does not count) and per-Wi-Fi- +interface `IP4.ADDRESS` lookups for the host IP. + ## Development This library is a work in progress. Additional operations will be added as reverse engineering continues. +Tests (no radio or NetworkManager required): + +``` +python -m unittest discover -s tests +``` + ## License Internal use only. Reverse engineering findings for research purposes. diff --git a/franklinwh_agate_client/__init__.py b/franklinwh_agate_client/__init__.py index 8b24516..639dace 100644 --- a/franklinwh_agate_client/__init__.py +++ b/franklinwh_agate_client/__init__.py @@ -7,8 +7,33 @@ Provides access to Bluetooth and Wi-Fi network I/O stack for AGATE devices. from .client import AgateClient from .bluetooth import BluetoothManager -from .wifi import WifiManager +from .wifi import ( + DEVICE_GATEWAY_IP, + FALLBACK_PASSWORD, + VALID_SN_LENGTH, + HotspotJoinError, + HotspotProfile, + NoOpWifiBackend, + NmcliWifiBackend, + WifiBackend, + WifiManager, + derive_hotspot_credentials, +) from .api import ApiClient __version__ = "0.1.0" -__all__ = ["AgateClient", "BluetoothManager", "WifiManager", "ApiClient"] +__all__ = [ + "AgateClient", + "BluetoothManager", + "ApiClient", + "WifiManager", + "WifiBackend", + "NmcliWifiBackend", + "NoOpWifiBackend", + "HotspotProfile", + "HotspotJoinError", + "derive_hotspot_credentials", + "DEVICE_GATEWAY_IP", + "FALLBACK_PASSWORD", + "VALID_SN_LENGTH", +] diff --git a/franklinwh_agate_client/client.py b/franklinwh_agate_client/client.py index f865ae3..7d4fc03 100644 --- a/franklinwh_agate_client/client.py +++ b/franklinwh_agate_client/client.py @@ -2,25 +2,69 @@ Agate Client - Main interface for AGATE device operations. """ +from __future__ import annotations + +from typing import Any, Optional + +from .bluetooth import BluetoothManager +from .wifi import HotspotProfile, WifiBackend, WifiManager + + class AgateClient: """Main client for FranklinWH AGATE device operations.""" - - def __init__(self, device_id=None): + + def __init__(self, device_id: Optional[str] = None, + wifi_backend: Optional[WifiBackend] = None) -> None: self.device_id = device_id - self.bluetooth_manager = None - self.wifi_manager = None - - def query_agate_info(self, device_id=None): + self.bluetooth_manager: Optional[BluetoothManager] = None + self.wifi_manager = WifiManager(backend=wifi_backend) + + # -- device hotspot (local connect) ------------------------------------- + def join_hotspot(self, sn: str, **kwargs: Any) -> HotspotProfile: + """Join the device SoftAP (derives creds, joins, verifies).""" + return self.wifi_manager.join_device_hotspot(sn, **kwargs) + + def leave_hotspot(self) -> None: + """Disconnect from the device hotspot.""" + self.wifi_manager.leave_device_hotspot() + + def attach_local(self, sn: str, *, verify_join: bool = True, + settle_timeout: float = 30.0, poll_interval: float = 1.0, + **local_kwargs: Any): + """Join the device hotspot, then return a ``LocalDeviceClient`` + aimed at the device gateway (``https://10.100.1.1:18000`` by default). + + ``verify_join`` gates the association check (TLS verification of the + device API is a separate concern — pass ``verify=`` in + ``local_kwargs``). Raises + :class:`~franklinwh_agate_client.wifi.HotspotJoinError` if the + association never verifies. (``LocalDeviceClient`` imports + ``requests``; install it before calling this on a cloud-less host.) + + Extra keyword arguments are passed through to ``LocalDeviceClient`` + (``port``, ``scheme``, ``verify``, ``socket_port``, ``key_radix``, + ``timeout``, ``extra_headers``, ``session``). + """ + profile = self.wifi_manager.join_device_hotspot( + sn, verify=verify_join, settle_timeout=settle_timeout, + poll_interval=poll_interval, + ) + from .device_local import LocalDeviceClient + + return LocalDeviceClient(sn=sn, host=profile.gateway_ip, **local_kwargs) + + # -- cloud / stub operations (pending RE) ------------------------------- + def query_agate_info(self, device_id: Optional[str] = None): """Query AGATE device information.""" # Reverse engineered operation: queryAgateInfo pass - - def query_device_info_http(self, device_id=None): + + def query_device_info_http(self, device_id: Optional[str] = None): """Query device info via HTTP.""" # Reverse engineered operation: queryDeviceInfoHttp pass - - def query_modes_by_http(self, device_id=None): + + def query_modes_by_http(self, device_id: Optional[str] = None): """Query operating modes via HTTP.""" # Reverse engineered operation: queryModesByHttp pass diff --git a/franklinwh_agate_client/wifi.py b/franklinwh_agate_client/wifi.py index 8ae778c..764f7b6 100644 --- a/franklinwh_agate_client/wifi.py +++ b/franklinwh_agate_client/wifi.py @@ -7,25 +7,34 @@ and ``com.alternadom.wifiiot.WifiIotPlugin`` (Java side, ``base.apk``) — see The app joins the **device SoftAP** so that the phone can talk to the aGate over its local HTTP/socket interfaces (gateway ``10.100.1.1``). -Credential derivation (confirmed from the disassembly, 20-char SN rule):: +How this module maps onto that:: - ssid = "AP_" + sn[11:] pwd = sn[8:] - else (invalid length) - ssid = "AP_" + sn pwd = "12345678" + * credential derivation (``derive_hotspot_credentials``) — faithful + reconstruction of ``getDeviceApSsid`` / ``getDeviceApDefaultPwd`` (§10). + * the association itself (``WifiBackend.join`` / ``leave``) — platform + work. :class:`NmcliWifiBackend` drives NetworkManager on a Linux + desktop/laptop; :class:`NoOpWifiBackend` records state only, for + hosts whose own tooling owns the radio (containers, manual joins). + * ``WifiManager`` — the facade that makes the join *functional*: + derive → join → **verify** (poll ``is_connected`` against a deadline, + fail-closed) → expose our IP and the device gateway. -Actual joining is delegated to a :class:`WifiBackend` so this module stays -testable without a radio. On a Linux desktop/laptop use -:class:`NmcliWifiBackend` (NetworkManager); in a container or when joining -manually use :class:`NoOpWifiBackend`. +Note on the app's *other* "wifi" operations: ``connectWifi`` (0x0f) and the +ESP-provisioning/onboarding flow (§4, §9) provision the **device** with a +home router — they are device-side RPCs (see ``LocalDeviceClient`` / +``CommunicationCmd``), not host-radio actions. The compat shims here +(``query_connect_wifi`` / ``wifi_onboarding_add_wifi``) only perform the +host-side join the app performs at the end of onboarding. """ from __future__ import annotations import logging import subprocess +import time from abc import ABC, abstractmethod from dataclasses import dataclass -from typing import Optional +from typing import Callable, List, Optional log = logging.getLogger(__name__) @@ -37,6 +46,11 @@ VALID_SN_LENGTH = 20 FALLBACK_PASSWORD = "12345678" +#: Address prefixes tried (in order) when picking the host address after a +#: join. The aGate gateway is 10.100.1.1 and its DHCP normally hands out +#: 10.100.1.x; the other prefixes are generic private-range fallbacks. +_ADDRESS_PREFIX_PREFERENCE = ("10.100.", "10.", "192.168.", "172.16.") + @dataclass(frozen=True) class HotspotProfile: @@ -47,6 +61,10 @@ class HotspotProfile: gateway_ip: str = DEVICE_GATEWAY_IP +class HotspotJoinError(RuntimeError): + """A join was issued but the association never verified in time.""" + + def derive_hotspot_credentials(sn: str) -> "HotspotProfile": """``getDeviceApSsid`` / ``getDeviceApDefaultPwd``. @@ -69,7 +87,18 @@ def derive_hotspot_credentials(sn: str) -> "HotspotProfile": class WifiBackend(ABC): - """Platform abstraction for joining/leaving a Wi-Fi network.""" + """Platform abstraction for joining/leaving a Wi-Fi network. + + Contract: + + * ``join`` must raise on hard failure (missing tool, rejected SSID…). + * ``leave`` is best-effort: do not raise when there is simply nothing + to disconnect. + * ``is_connected`` must be **fail-closed** (return ``False`` when + uncertain, e.g. the underlying tooling is absent). The manager polls + it to decide whether a join actually stuck, so a wrong ``True`` is + far worse than a late ``False``. + """ @abstractmethod def join(self, profile: HotspotProfile) -> None: @@ -77,22 +106,29 @@ class WifiBackend(ABC): @abstractmethod def leave(self) -> None: - """Disconnect from the device hotspot.""" + """Disconnect from the current Wi-Fi network (best-effort).""" @abstractmethod def get_ip(self) -> Optional[str]: - """Our own IP on the current Wi-Fi network (device-side gateway view).""" + """Our own IPv4 address on the currently associated network.""" def is_connected(self, profile: HotspotProfile) -> bool: - """Best-effort: did the join stick?""" + """Fail-closed: is ``profile`` the network we are associated to?""" return False + def get_signal(self) -> Optional[int]: + """Signal strength (0-100) of the associated AP, or ``None``.""" + return None + class NoOpWifiBackend(WifiBackend): - """Assumes the user has already connected (manually or via another tool). + """Records join/leave state without touching a radio. - Useful in sandboxes/containers without a radio, and when the host's own - tooling handles the association. + For sandboxes/containers and for hosts where the operator joins the + device hotspot manually (or via the host's own tooling). Verification + then trivially passes: the backend records the profile ``join`` was + given. Use :meth:`assume_connected` when the association was made + outside of this process. """ def __init__(self, local_ip: Optional[str] = None) -> None: @@ -100,10 +136,17 @@ class NoOpWifiBackend(WifiBackend): self._profile: Optional[HotspotProfile] = None self.joined: bool = False - def join(self, profile: HotspotProfile) -> None: + def assume_connected(self, profile: HotspotProfile, + local_ip: Optional[str] = None) -> None: + """Declare that the host is already associated with ``profile``.""" + if local_ip is not None: + self._local_ip = local_ip self._profile = profile self.joined = True + def join(self, profile: HotspotProfile) -> None: + self.assume_connected(profile) + def leave(self) -> None: self._profile = None self.joined = False @@ -116,18 +159,33 @@ class NoOpWifiBackend(WifiBackend): class NmcliWifiBackend(WifiBackend): - """Join via ``nmcli`` (NetworkManager) — for Linux desktops/laptops.""" + """Join via ``nmcli`` (NetworkManager) — for Linux desktops/laptops. - def __init__(self, nmcli: str = "nmcli", ipcmd: str = "ip") -> None: + Verified against nmcli/NetworkManager 1.52 source (``devices.c``): + + * association state: ``nmcli -t -f SSID,ACTIVE device wifi list`` — + the ``ACTIVE`` column is ``yes`` *only* for the AP we are associated + to. (There is no ``device wifi show`` subcommand; the command table + is ``list / connect / hotspot / rescan / show-password`` and a bare + ``show`` is accepted only by prefix-matching ``show-password``. Using + a scan list or that accident to mean "associated" is wrong, and + version-dependent — hence ``-f SSID,ACTIVE``.) + * our IP: ``nmcli -g IP4.ADDRESS device show `` per Wi-Fi + interface (never the unfiltered all-device dump, which would happily + report a wired/VPN address). + * signal: the ``SIGNAL`` column of the active AP from the same list. + """ + + def __init__(self, nmcli: str = "nmcli", timeout: float = 30.0) -> None: self._nmcli = nmcli - self._ipcmd = ipcmd + self._timeout = timeout def _run(self, *args: str) -> str: proc = subprocess.run( [self._nmcli, *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, - timeout=30, + timeout=self._timeout, ) if proc.returncode != 0: raise RuntimeError( @@ -136,80 +194,170 @@ class NmcliWifiBackend(WifiBackend): ) return proc.stdout.decode("utf-8", "replace").strip() + def _wifi_ifaces(self) -> List[str]: + """Wi-Fi interface names in an associated/active state.""" + try: + out = self._run("-t", "-f", "DEVICE,TYPE,STATE", "device", "show") + except (RuntimeError, subprocess.TimeoutExpired, OSError): + return [] + ifaces: List[str] = [] + for line in out.splitlines(): + parts = line.split(":") + if len(parts) >= 3 and parts[1] == "wifi" and parts[2] in ("activated", "connected"): + ifaces.append(parts[0]) + return ifaces + def join(self, profile: HotspotProfile) -> None: self._run("device", "wifi", "connect", profile.ssid, "password", profile.password) def leave(self) -> None: - self._run("device", "wifi", "disconnect") + # Best-effort: there may be nothing to disconnect (radio off, no + # association). Tolerate it — teardown must not fail the flow. + try: + self._run("device", "wifi", "disconnect") + except (RuntimeError, subprocess.TimeoutExpired, OSError) as exc: + log.warning("wifi disconnect did not complete cleanly: %s", exc) def get_ip(self) -> Optional[str]: - try: - out = self._run("-g", "IP4.ADDRESS", "device", "show") - except (RuntimeError, subprocess.TimeoutExpired, OSError): - return None - for line in out.splitlines(): - addr = line.strip() - if addr.startswith("10.") or addr.startswith("192.168."): - return addr - return None + candidates: List[str] = [] + for dev in self._wifi_ifaces(): + try: + out = self._run("-g", "IP4.ADDRESS", "device", "show", dev) + except (RuntimeError, subprocess.TimeoutExpired, OSError): + continue + candidates.extend(a.strip() for a in out.splitlines() if a.strip()) + for prefix in _ADDRESS_PREFIX_PREFERENCE: + for addr in candidates: + if addr.startswith(prefix): + return addr + return candidates[0] if candidates else None def is_connected(self, profile: HotspotProfile) -> bool: try: - out = self._run("-g", "SSID", "device", "wifi", "show") + out = self._run("-t", "-f", "SSID,ACTIVE", "device", "wifi", "list") except (RuntimeError, subprocess.TimeoutExpired, OSError): - return False - return profile.ssid in out.splitlines() + return False # fail-closed: cannot confirm association + return f"{profile.ssid}:yes" in out.splitlines() + + def get_signal(self) -> Optional[int]: + try: + out = self._run("-t", "-f", "SSID,SIGNAL,ACTIVE", "device", "wifi", "list") + except (RuntimeError, subprocess.TimeoutExpired, OSError): + return None + for line in out.splitlines(): + # SSIDs may contain ':'; the trailing field is the ACTIVE flag. + parts = line.rsplit(":", 2) + if len(parts) == 3 and parts[2] == "yes": + try: + return int(parts[1]) + except ValueError: + return None + return None class WifiManager: - """Facade over a :class:`WifiBackend` (keeps the original stub API).""" + """Facade over a :class:`WifiBackend` — derive, join, **verify**. + + ``join_*`` do not take the backend's exit code on faith: after + ``join`` they poll ``backend.is_connected(profile)`` against + ``settle_timeout`` (NetworkManager can report success before the + association actually settles, and it can fail a few seconds later). + An unverified join raises :class:`HotspotJoinError` and leaves + ``connected`` ``False``. ``verify=False`` restores fire-and-forget. + """ def __init__(self, backend: Optional[WifiBackend] = None) -> None: self.backend = backend or NoOpWifiBackend() self.connected = False + self.last_profile: Optional[HotspotProfile] = None - # -- new, clean API ---------------------------------------------------- - def join_device_hotspot(self, sn: str) -> HotspotProfile: - """Derive the profile from the SN and join. Returns the profile.""" - profile = derive_hotspot_credentials(sn) - self.backend.join(profile) - self.connected = True + # -- internal ---------------------------------------------------------- + def _join(self, profile: HotspotProfile, *, verify: bool, + settle_timeout: float, poll_interval: float) -> HotspotProfile: + self.last_profile = profile + self.connected = False + self.backend.join(profile) # raises on hard failure + ok = self._verify( + lambda: self.backend.is_connected(profile), + settle_timeout, poll_interval, + ) if verify else True + self.connected = ok + if not ok: + raise HotspotJoinError( + f"join to '{profile.ssid}' was issued but association was " + f"not confirmed within {settle_timeout:g}s" + ) return profile + @staticmethod + def _verify(check: Callable[[], bool], timeout: float, + poll_interval: float) -> bool: + """Poll ``check()`` until it is true or ``timeout`` elapses.""" + deadline = time.monotonic() + timeout + while True: + try: + if check(): + return True + except Exception as exc: # a broken backend must fail closed + log.debug("is_connected check raised: %s", exc) + if time.monotonic() >= deadline: + return False + time.sleep(max(poll_interval, 0.0)) + + # -- device hotspot ------------------------------------------------------ + def join_device_hotspot(self, sn: str, *, verify: bool = True, + settle_timeout: float = 30.0, + poll_interval: float = 1.0) -> HotspotProfile: + """Derive the profile from the SN, join, and verify the association.""" + profile = derive_hotspot_credentials(sn) + return self._join(profile, verify=verify, + settle_timeout=settle_timeout, + poll_interval=poll_interval) + + def join_network(self, ssid: str, password: str, *, verify: bool = True, + settle_timeout: float = 30.0, + poll_interval: float = 1.0) -> HotspotProfile: + """Join an arbitrary network (host side).""" + return self._join(HotspotProfile(ssid=ssid, password=password), + verify=verify, settle_timeout=settle_timeout, + poll_interval=poll_interval) + def leave_device_hotspot(self) -> None: self.backend.leave() self.connected = False + self.last_profile = None + # -- status -------------------------------------------------------------- def local_ip(self) -> Optional[str]: + """Our own IP on the associated network (``None`` if unknown).""" return self.backend.get_ip() + def gateway_ip(self) -> str: + """The aGate's IP on the hotspot (where the local API lives).""" + return self.last_profile.gateway_ip if self.last_profile else DEVICE_GATEWAY_IP + + def get_wifi_signal_strength(self) -> Optional[int]: + return self.backend.get_signal() + # -- original stub names (kept for API compatibility) ------------------- def query_connect_wifi(self, ssid: str, password: Optional[str] = None, - sn: Optional[str] = None): - """``queryConnectWifi`` — join (or rejoin) a network.""" + sn: Optional[str] = None, **join_kwargs: object): + """Join an arbitrary network (host side); with ``sn`` only, join the + device hotspot — same as :meth:`join_device_hotspot`. + + NB: in the app, ``queryConnectWifi`` is the *device* RPC + (``CommunicationCmd.CONNECT_WIFI``, 0x0f) that tells the aGate to + join a router — a provisioning operation, not a host-radio action. + """ if sn is not None and password is None: - profile = derive_hotspot_credentials(sn) - else: - profile = HotspotProfile(ssid=ssid, password=password or "") - self.backend.join(profile) - self.connected = True - return profile + return self.join_device_hotspot(sn, **join_kwargs) + return self.join_network(ssid, password or "", **join_kwargs) - def wifi_onboarding_add_wifi(self, ssid: str, password: str): - """``wifiOnboardingAddWifi`` — register a known network.""" - self.backend.join(HotspotProfile(ssid=ssid, password=password)) - return True - - def get_wifi_signal_strength(self) -> Optional[int]: - if not isinstance(self.backend, NmcliWifiBackend): - return None - try: - out = self.backend._run("-g", "SIGNAL", "device", "wifi", "show") - except (RuntimeError, subprocess.TimeoutExpired, OSError): - return None - for line in out.splitlines(): - digits = "".join(ch for ch in line if ch.isdigit()) - if digits: - return int(digits) - return None + def wifi_onboarding_add_wifi(self, ssid: str, password: str, **join_kwargs: object): + """Host-side step of the app's Wi-Fi onboarding: after the aGate has + been provisioned with the home router (device-side — see + ``CommunicationCmd.SCAN_WIFI_LIST`` / ``CONNECT_WIFI``), the phone + rejoins the home network. This performs that host-side rejoin. + """ + return self.join_network(ssid, password, **join_kwargs) diff --git a/tests/test_wifi.py b/tests/test_wifi.py new file mode 100644 index 0000000..629aa62 --- /dev/null +++ b/tests/test_wifi.py @@ -0,0 +1,267 @@ +"""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()