diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6764c18 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,85 @@ +name: CI + +on: + push: + pull_request: + workflow_dispatch: + +jobs: + build-and-test: + name: Build and test + runs-on: ubuntu-latest + permissions: + contents: read + + steps: + - name: Checkout + uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 + with: + persist-credentials: false + + - name: Build image + run: docker build --tag micropython-linux:test . + + - name: Substrate conformance (esp32 fakes vs contract) + run: | + docker run --rm \ + -v ${{ github.workspace }}:/work \ + -w /work \ + micropython-linux:test /work/tests/substrate-conformance.py + + - name: Checkout micropython-wifimanager + uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 + with: + repository: mitchins/micropython-wifimanager + # Pinned to a specific commit (not a branch) for deterministic CI. + # Must be a revision that includes the API the suite exercises + # (_scan_available_networks/_build_connection_candidates/_connect_candidates/ + # _load_config) — these post-date the v1.0.2 release tag. + ref: 835861c0ed10b170bcf29a171e1eb11177131084 + path: micropython-wifimanager + persist-credentials: false + + - name: WifiManager regression on substrate (MicroPython runtime) + run: | + docker run --rm \ + -v ${{ github.workspace }}:/work \ + -w /work \ + -e WIFIMANAGER_SRC=/work/micropython-wifimanager \ + micropython-linux:test /work/tests/wifimanager-status-regression.py + + - name: Smoke test image + run: docker run --rm micropython-linux:test -c "import sys; print(sys.version)" + + publish: + name: Publish image + if: github.event_name == 'push' && github.ref == format('refs/heads/{0}', github.event.repository.default_branch) + needs: build-and-test + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - name: Checkout + uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 + with: + persist-credentials: false + + - name: Build image + run: docker build --tag micropython-linux:test . + + - name: Log in to GHCR + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Publish image to GHCR + run: | + IMAGE=$(echo "ghcr.io/${{ github.repository }}" | tr '[:upper:]' '[:lower:]') + docker tag micropython-linux:test "$IMAGE:latest" + docker tag micropython-linux:test "$IMAGE:${{ github.sha }}" + docker push "$IMAGE:latest" + docker push "$IMAGE:${{ github.sha }}" diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index c5a072f..0000000 --- a/.travis.yml +++ /dev/null @@ -1,15 +0,0 @@ -sudo: 'required' - -services: - - 'docker' - -script: - - 'docker build -t $DOCKER_USERNAME/micropython-linux:$TRAVIS_BRANCH .' - -after_success: - - docker login -u $DOCKER_USERNAME -p $DOCKER_PASSWORD - - docker push $DOCKER_USERNAME/micropython-linux:$TRAVIS_BRANCH - -# don't notify me when things fail -notifications: - email: false diff --git a/Dockerfile b/Dockerfile index 830f1bf..276010a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,20 +1,29 @@ -FROM debian:stretch-slim +FROM debian:bookworm-slim + +ARG MICROPYTHON_VERSION=v1.28.0 RUN apt-get update && \ - apt-get install -y build-essential libffi-dev git pkg-config python3 && \ - rm -rf /var/lib/apt/lists/* && \ - git clone https://github.com/micropython/micropython.git && \ - cd micropython && \ - cd mpy-cross && \ - make && \ - cd .. && \ - cd ports/unix && \ - make submodules && \ - make && \ - make test && \ - make install && \ - apt-get purge --auto-remove -y build-essential libffi-dev git pkg-config python3 && \ - cd ../../.. && \ - rm -rf micropython + apt-get install -y --no-install-recommends \ + build-essential \ + ca-certificates \ + git \ + libffi8 \ + libffi-dev \ + pkg-config \ + python3 \ + && git clone --depth 1 --branch ${MICROPYTHON_VERSION} https://github.com/micropython/micropython.git \ + && cd micropython \ + && cd ports/unix \ + && make submodules \ + && make \ + && make install \ + && cd / \ + && rm -rf micropython \ + && apt-get purge --auto-remove -y \ + build-essential \ + git \ + pkg-config \ + python3 \ + && rm -rf /var/lib/apt/lists/* -CMD ["/usr/local/bin/micropython"] +ENTRYPOINT ["/usr/local/bin/micropython"] diff --git a/README.md b/README.md index 1a936f4..efef92b 100644 --- a/README.md +++ b/README.md @@ -1,32 +1,93 @@ # docker-micropython-linux -Docker image that comes loaded with the unix (linux in this case) port of micropython. -#### Why use this? Simple: +A pinned **MicroPython runtime in Docker** plus a small **ESP32 test substrate**, for +running deterministic CI tests of MicroPython libraries — without a board, and without +the false confidence of testing under CPython. -* Running automated tests against libraries intended for micropython (you *can* use python3, but there are edge cases). -* Experimenting with micropython without any complexity of building yourself or overhead of buying a board first. -* There wasn't an existing and well-maintained image. +![CI](https://github.com/mitchins/docker-micropython-linux/actions/workflows/ci.yml/badge.svg) -The image is based off the official Debian-slim (stretch) because it's fairly slim (not as slim as alpine, but that can be a headache to build). +## Why this exists -#### Getting Started +MicroPython is **not** CPython. The standard library is a subset and behaves +differently: there is no `os.path`, no `types`, `base64` may be absent, `os.environ` +is `os.getenv`, and module surfaces differ. A library "tested" under `python3` can pass +its suite and still break on device. This repo runs your library's logic under the +**actual MicroPython interpreter** (the unix port), pinned to a known version, in a +throwaway container. -Providing you have access to Docker, you can run the latest version quite easily by: +On top of that it ships a **scriptable fake of the ESP32 SDK** (`network`, `machine`) +so you can exercise Wi-Fi/state/pin logic deterministically — the fake provides the +*shape* (signatures, return types, exceptions, constants, identity captured from the +real firmware); your test provides the *behaviour*. - docker run -it mitchins/micropython-linux - MicroPython v1.9.4-403-g81e320ae on 2018-07-21; linux version - Use Ctrl-D to exit, Ctrl-E for paste mode - >>> ^C +**Scope, honestly:** this targets **logic/framework correctness** (connection policy, +status handling, config parsing, callbacks, pin I/O sequencing) — not RF, timing, or +electrical behaviour. It is a high-signal, low-flake complement to on-device testing, +not a replacement. -Tags are available on the Docker Hub listing +## The image -If you check the Dockerfile, you will see it starts up micropython automatically which is stored in /usr/local/bin/micropython +Pulled from GitHub Container Registry (built and published by CI — see below): -To install something else, you can: + docker run -it ghcr.io/mitchins/docker-micropython-linux:latest + MicroPython v1.28.0 on ...; linux version + >>> - # micropython -m upip install micropython-unittest - Installing to: /root/.micropython/lib/ - Warning: pypi.org SSL certificate is not validated - Installing micropython-unittest 0.3.2 from https://files.pythonhosted.org/packages/9a/42/41057b8da94414a17f7028ee08035c3d945befebddc76d58988067ddaf0f/micropython-unittest-0.3.2.tar.gz +It builds the unix port of MicroPython from source on `debian:bookworm-slim`. Override +the version at build time: -That's about it, the goal is to maintain the latest and stable tagged versions so regressions can be tested as needed, or experimented features accessed. + docker build --build-arg MICROPYTHON_VERSION=v1.28.0 -t micropython-linux . + +The container's entrypoint is `micropython`, so arguments pass straight through: + + docker run --rm ghcr.io/mitchins/docker-micropython-linux:latest -c "import sys; print(sys.version)" + docker run --rm -v "$PWD":/work -w /work ghcr.io/mitchins/docker-micropython-linux:latest /work/script.py + +## The ESP32 substrate + +In [`substrate/`](substrate/): + +| File | Purpose | +|---|---| +| `CONTRACT-esp32-v1.28.md` | The SDK contract, captured from the esp32 port C source (MicroPython v1.28.0) + ESP-IDF v5.5.1 | +| `network.py` | `WLAN` fake — singleton-per-interface, `bytes` SSIDs, real `STAT_*` values, scriptable + spy | +| `machine.py` | `Pin` fake — `int` reads, readable `OUT`, scriptable inputs + spy | +| `aiotest.py` | `run_iterations()` — drives an async loop (e.g. `manage()`) deterministically, no wall-clock waits | + +The fakes are *scriptable*, not emulated. A test programs outcomes and spies on calls; +it never authors a mock: + +```python +import sys; sys.path.insert(0, "substrate") +import network + +sta = network.WLAN(network.STA_IF) +sta.active(True) +sta.program_scan([{"ssid": "Home", "bssid": b"\x0a\x0b\x0c\x0d\x0e\x0f", "rssi": -50}]) + +# ... exercise your library against `network` ... + +assert sta.call_count("connect") == 1 +assert sta.last_call("connect").kwargs["bssid"] == b"\x0a\x0b\x0c\x0d\x0e\x0f" +``` + +See [`tests/`](tests/) for a full worked example: `wifimanager-status-regression.py` +tests [micropython-wifimanager](https://github.com/mitchins/micropython-wifimanager) +against the substrate (including driving its async `manage()` loop), and +`substrate-conformance.py` pins the fakes to the captured contract so they can't drift. + +## CI + +[`.github/workflows/ci.yml`](.github/workflows/ci.yml) on every push/PR: + +1. Build the image +2. **Substrate conformance** — fakes vs `CONTRACT-esp32-v1.28.md` +3. **Library regression** — wifimanager suite in the MicroPython runtime (against a pinned library revision) +4. Smoke test +5. On push: **publish** the image to `ghcr.io/mitchins/docker-micropython-linux` + +## Status / roadmap + +- Pinned to MicroPython **v1.28.0**, ESP32 profile. +- Deferred: ESP8266 profile; broader `machine` (`irq`, ADC, I2C/SPI) — added when a real + consumer needs them. The substrate establishes the pattern; it is single-target today. diff --git a/substrate/CONTRACT-esp32-v1.28.md b/substrate/CONTRACT-esp32-v1.28.md new file mode 100644 index 0000000..81b4843 --- /dev/null +++ b/substrate/CONTRACT-esp32-v1.28.md @@ -0,0 +1,101 @@ +# ESP32 SDK fake — conformance contract + +**Pinned to:** MicroPython **v1.28.0**, **esp32** port. +**Source of truth:** captured from the esp32 port C bindings at the `v1.28.0` tag +(`ports/esp32/network_wlan.c`, `ports/esp32/modnetwork.h`, +`ports/esp32/modnetwork_globals.h`, `ports/esp32/machine_pin.c`) — **not** docs, +not the unix port (which exposes neither `network.WLAN` nor esp `machine.Pin`). + +The fakes in this directory must "walk and quack" like these surfaces: the library +under test must not be able to tell the difference at the call boundary. The fake +supplies *shape* (signatures, return types, exceptions, constants, identity); the +**test supplies behaviour** via the control surface. + +--- + +## `network` module constants + +| Symbol | Value | Provenance / stability | +|---|---|---| +| `STA_IF` | `0` | micropython-owned, stable | +| `AP_IF` | `1` | micropython-owned, stable | +| `STAT_IDLE` | `1000` | `modnetwork.h` enum — **stable** | +| `STAT_CONNECTING` | `1001` | **stable** | +| `STAT_GOT_IP` | `1010` | **stable** | +| `STAT_BEACON_TIMEOUT` | `200` | `WIFI_REASON_BEACON_TIMEOUT` (ESP-IDF v5.5.1) | +| `STAT_NO_AP_FOUND` | `201` | `WIFI_REASON_NO_AP_FOUND` (ESP-IDF v5.5.1) | +| `STAT_WRONG_PASSWORD` | `202` | `WIFI_REASON_AUTH_FAIL` (ESP-IDF v5.5.1) | +| `STAT_ASSOC_FAIL` | `203` | `WIFI_REASON_ASSOC_FAIL` (ESP-IDF v5.5.1) | +| `STAT_CONNECT_FAIL` | `203` | **aliases `STAT_ASSOC_FAIL`** on esp32 (same int, per micropython source) | +| `STAT_HANDSHAKE_TIMEOUT` | `204` | `WIFI_REASON_HANDSHAKE_TIMEOUT` (ESP-IDF v5.5.1) | + +Failure-code source: `wifi_err_reason_t` in +`components/esp_wifi/include/esp_wifi_types_generic.h` at the **esp-idf `v5.5.1`** tag — +the IDF version MicroPython v1.28.0 recommends (`ports/esp32/README.md`). For reference, +`WIFI_REASON_CONNECTION_FAIL = 205` (which MicroPython's `status()` also reports as +wrong-password → `202`). + +**Two-zone confidence.** The happy-path codes (`1000/1001/1010`) are MicroPython's +own and stable across versions. The failure codes above are ESP-IDF enum values pinned +to v5.5.1; they are version-sensitive, so re-capture from the matching IDF header if the +image's MicroPython/IDF pairing changes. On the rp2/cyw43 port these constants have +entirely different values (`CYW43_LINK_*`) — which is why this fake is esp-only and +version-pinned. + +## `network.WLAN` + +- `WLAN(interface=STA_IF)` returns the **same singleton per interface** + (esp32 `make_new` returns static `wlan_sta_obj` / `wlan_ap_obj`). Bad id → `ValueError`. + +| Method | Signature | Returns | Raises | +|---|---|---|---| +| `active([bool])` | optional arg | `bool` | `OSError` on driver error | +| `connect(ssid=None, key=None, *, bssid=None)` | `ssid`/`key` positional; `bssid` **kw-only**, 6 bytes | `None` | `ValueError` (bad bssid len), `OSError` | +| `disconnect()` | — | `None` | `OSError` | +| `status([param])` | no-arg, or `'rssi'` / `'stations'` | no-arg → **int** (STA only; AP → `None`); `'rssi'` → int; `'stations'` → list of 1-tuples of `bytes` | `ValueError` (unknown param), `OSError` | +| `scan()` | — | **list of 6-tuples** | `OSError("STA must be active")` if STA inactive | +| `isconnected()` | — | `bool` | — | +| `ifconfig([tuple])` | get/set | 4-tuple `(ip, mask, gw, dns)` of `str` | — | +| `config(key)` / `config(**kw)` | get/set | get → value; set → `None` | — | + +**`scan()` tuple shape (load-bearing):** +`(ssid: bytes, bssid: bytes, channel: int, RSSI: int, security: int, hidden)` +- `ssid` and `bssid` are **`bytes`**, never `str`. +- on esp32 `hidden` is **always `False`** (never populated; `// XXX hidden?` in source). + +## `machine.Pin` + +- `Pin(id, mode=None, pull=-1, *, value=None, drive=None, hold=None)`. + Bad id → `ValueError("invalid pin")`. + +| Member | Behaviour | +|---|---| +| `value([x])` | no-arg → **int `0`/`1`** (`gpio_get_level`); with arg → `None` (sets) | +| `pin(x)` | call form ≡ `value(x)` | +| `on()` / `off()` / `toggle()` | return `None` | +| `init(...)` / `irq(...)` | exist (`irq` deferred for day-zero) | + +**Mode constants:** `IN`=`GPIO_MODE_INPUT` (`1`), **`OUT`=`GPIO_MODE_INPUT_OUTPUT` (`3`)**, +`OPEN_DRAIN`=`..._OD` (`7`). Because `OUT` is *input-output* on esp32, **reading an +output pin returns the last written level** — the fake replicates this. Pulls: +`PULL_UP`, `PULL_DOWN`. + +--- + +## Confirmed quack-leaks the fake must honour + +1. `scan()`/`status('stations')` return **`bytes`**, not `str`. +2. `WLAN(iface)` is a **singleton per interface** (identity is observable). +3. `scan()` `hidden` element is **always `False`** on esp32. +4. `Pin.value()` returns **`int`**, not `bool`. +5. `Pin.OUT` is **readable** (read-back of written level). +6. `STAT_*` failure codes are **esp/IDF-specific** and `CONNECT_FAIL == ASSOC_FAIL`. + +## Provenance summary + +- `network`/`machine` shapes, constants, return types, exceptions: MicroPython esp32 + port at the **v1.28.0** tag. +- Failure `STAT_*` integers: ESP-IDF **v5.5.1** `wifi_err_reason_t`. + +Re-capture both if the image bumps its MicroPython version (and with it the +recommended IDF version). diff --git a/substrate/aiotest.py b/substrate/aiotest.py new file mode 100644 index 0000000..04dc156 --- /dev/null +++ b/substrate/aiotest.py @@ -0,0 +1,62 @@ +"""Deterministic async driving for substrate-based tests. + +``run_iterations`` runs an infinite-loop coroutine (e.g. ``WifiManager.manage``) for +a bounded number of its ``await sleep`` points, with **no wall-clock delay**, then +returns. It temporarily replaces the supplied asyncio module's ``sleep`` / ``sleep_ms`` +with a fast, counting stop so the loop advances a known number of times and unwinds +cleanly. This lets tests exercise real async control flow without 10-second waits or +flaky timing. +""" + + +class _StopLoop(Exception): + pass + + +def run_iterations(coro_factory, iterations, asyncio_module): + """Run ``coro_factory()`` until its sleep points have fired ``iterations`` times. + + coro_factory: zero-arg callable returning a fresh coroutine (e.g. ``mgr.manage``). + iterations: number of loop turns (sleep points) to allow before stopping. + asyncio_module: the asyncio module the code-under-test actually uses (patched/restored). + + Returns the number of loop turns executed. + """ + if iterations < 1: + raise ValueError("iterations must be >= 1") + + counter = {"n": 0} + saved = {} + for name in ("sleep", "sleep_ms"): + if hasattr(asyncio_module, name): + saved[name] = getattr(asyncio_module, name) + real_sleep = saved.get("sleep") + real_sleep_ms = saved.get("sleep_ms") + + async def fast_sleep(*_args, **_kwargs): + counter["n"] += 1 + if real_sleep is not None: + await real_sleep(0) + elif real_sleep_ms is not None: + await real_sleep_ms(0) + if counter["n"] >= iterations: + raise _StopLoop() + + for name in saved: + setattr(asyncio_module, name, fast_sleep) + + try: + runner = getattr(asyncio_module, "run", None) + try: + if runner is not None: + runner(coro_factory()) + else: + loop = asyncio_module.get_event_loop() + loop.run_until_complete(coro_factory()) + except _StopLoop: + pass + finally: + for name, fn in saved.items(): + setattr(asyncio_module, name, fn) + + return counter["n"] diff --git a/substrate/machine.py b/substrate/machine.py new file mode 100644 index 0000000..16d28bc --- /dev/null +++ b/substrate/machine.py @@ -0,0 +1,100 @@ +"""Contract-accurate fake of the esp32 ``machine`` module (MicroPython v1.28.0). + +Day-zero subset: ``Pin``. See CONTRACT-esp32-v1.28.md. Scriptable fake: GPIO level +is shared per pin id (survives reconstruction, like real hardware), reads return +``int``, and ``OUT`` pins are readable. Use ``set_external()`` to script an input +level and ``call_count()`` to spy on writes. +""" + + +class Pin: + # Mode constants — esp32 values (note OUT is INPUT_OUTPUT, i.e. readable). + IN = 1 + OUT = 3 + OPEN_DRAIN = 7 + PULL_UP = 1 + PULL_DOWN = 2 + IRQ_RISING = 1 + IRQ_FALLING = 2 + + _levels = {} # id -> level (shared GPIO state, like the real chip) + _instances = {} # id -> Pin (a given GPIO maps to one logical pin) + + def __new__(cls, pin_id, *args, **kwargs): + if not isinstance(pin_id, int) or pin_id < 0: + raise ValueError("invalid pin") + inst = cls._instances.get(pin_id) + if inst is None: + inst = object.__new__(cls) + inst.id = pin_id + inst.mode = None + inst.pull = -1 + inst.calls = [] + cls._instances[pin_id] = inst + cls._levels.setdefault(pin_id, 0) + if args or kwargs: + inst._configure(args, kwargs) + return inst + + def __init__(self, pin_id=None, *args, **kwargs): + pass + + def _configure(self, args, kwargs): + mode = args[0] if len(args) >= 1 else kwargs.get("mode") + if mode is not None: + self.mode = mode + if len(args) >= 2: + self.pull = args[1] + elif "pull" in kwargs: + self.pull = kwargs["pull"] + v = kwargs.get("value") + if v is not None: + Pin._levels[self.id] = 1 if v else 0 + + # ---- device-facing API -------------------------------------------------- + def value(self, *args): + self.calls.append(("value", args)) + if args: + Pin._levels[self.id] = 1 if args[0] else 0 + return None + return Pin._levels[self.id] + + def __call__(self, *args): + return self.value(*args) + + def on(self): + self.calls.append(("on", ())) + Pin._levels[self.id] = 1 + return None + + def off(self): + self.calls.append(("off", ())) + Pin._levels[self.id] = 0 + return None + + def toggle(self): + self.calls.append(("toggle", ())) + Pin._levels[self.id] = 0 if Pin._levels[self.id] else 1 + return None + + def init(self, *args, **kwargs): + self._configure(args, kwargs) + return None + + def irq(self, *args, **kwargs): + self.calls.append(("irq", args, kwargs)) + return None + + # ---- test control surface (NOT present on real hardware) ---------------- + def set_external(self, level): + """Script the level this pin reads, as if driven externally.""" + Pin._levels[self.id] = 1 if level else 0 + + def call_count(self, name): + return sum(1 for c in self.calls if c[0] == name) + + +def reset_all(): + """Clear all pin state — call between tests for zero-effort isolation.""" + Pin._instances = {} + Pin._levels = {} diff --git a/substrate/network.py b/substrate/network.py new file mode 100644 index 0000000..fc4ef71 --- /dev/null +++ b/substrate/network.py @@ -0,0 +1,201 @@ +"""Contract-accurate fake of the esp32 ``network`` module (MicroPython v1.28.0). + +See CONTRACT-esp32-v1.28.md for the captured contract. This is a *scriptable fake*, +not an emulator: it guarantees the device-facing shape (signatures, return types, +exceptions, constants, singleton identity) and lets the test drive behaviour through +the control surface (``program_scan`` / ``fail_connect`` / ``connect_after`` / spy). +The library under test imports this as ``network`` and cannot tell the difference. +""" + +# --- contract constants (esp32 v1.28.0) ------------------------------------- +STA_IF = 0 +AP_IF = 1 + +# happy-path: micropython-owned, stable across versions +STAT_IDLE = 1000 +STAT_CONNECTING = 1001 +STAT_GOT_IP = 1010 + +# failure codes: ESP-IDF wifi_err_reason_t, captured from esp-idf v5.5.1 (the IDF +# version MicroPython v1.28.0 recommends). CONNECT_FAIL aliases ASSOC_FAIL on esp32. +STAT_BEACON_TIMEOUT = 200 +STAT_NO_AP_FOUND = 201 +STAT_WRONG_PASSWORD = 202 +STAT_ASSOC_FAIL = 203 +STAT_CONNECT_FAIL = 203 +STAT_HANDSHAKE_TIMEOUT = 204 + +_DISCONNECTED_IFCONFIG = ("0.0.0.0", "0.0.0.0", "0.0.0.0", "0.0.0.0") +_CONNECTED_IFCONFIG = ("192.168.4.2", "255.255.255.0", "192.168.4.1", "8.8.8.8") +_NO_BSSID = b"\x00\x00\x00\x00\x00\x00" + + +def _as_bytes(value): + if value is None: + return None + if isinstance(value, bytes): + return value + return value.encode("utf-8") + + +class _Call: + def __init__(self, name, args, kwargs): + self.name = name + self.args = args + self.kwargs = kwargs + + def __repr__(self): + return "Call(%s, %r, %r)" % (self.name, self.args, self.kwargs) + + +class WLAN: + """Singleton per interface, matching esp32 ``make_new``.""" + + _instances = {} + + def __new__(cls, interface=STA_IF): + if interface not in (STA_IF, AP_IF): + raise ValueError("invalid WLAN interface identifier") + inst = cls._instances.get(interface) + if inst is None: + inst = object.__new__(cls) + inst.interface = interface + inst.reset() + cls._instances[interface] = inst + return inst + + def __init__(self, interface=STA_IF): + pass + + # ---- test control surface (NOT present on real hardware) ---------------- + def reset(self): + self._active = False + self._connected = False + self._scan = [] + self._config = {} + self._connect_after = 0 # status() polls to remain CONNECTING before GOT_IP + self._fail_status = None # if set, connect "fails" and status() returns this + self._connect_ssid = None + self._connect_bssid = None + self.calls = [] + + def program_scan(self, entries): + """Set scan results. Accepts dicts or tuples; normalises to the contract + 6-tuple with ``bytes`` ssid/bssid and ``hidden`` always ``False``.""" + norm = [] + for e in entries: + if isinstance(e, dict): + ssid = _as_bytes(e["ssid"]) + bssid = _as_bytes(e.get("bssid")) or _NO_BSSID + channel = e.get("channel", 0) + rssi = e.get("rssi", -50) + security = e.get("security", 0) + else: + ssid, bssid = _as_bytes(e[0]), _as_bytes(e[1]) + channel, rssi, security = e[2], e[3], e[4] + norm.append((ssid, bssid, channel, rssi, security, False)) + self._scan = norm + + def fail_connect(self, status=STAT_NO_AP_FOUND): + """Make subsequent ``connect()`` calls fail; ``status()`` returns ``status``.""" + self._fail_status = status + + def connect_after(self, polls): + """Stay CONNECTING for ``polls`` ``status()`` reads, then report GOT_IP.""" + self._connect_after = polls + + def call_count(self, name): + return sum(1 for c in self.calls if c.name == name) + + def last_call(self, name): + for c in reversed(self.calls): + if c.name == name: + return c + return None + + def _record(self, name, args, kwargs): + self.calls.append(_Call(name, args, kwargs)) + + # ---- device-facing API (contract-accurate) ------------------------------ + def active(self, *args): + self._record("active", args, {}) + if args: + self._active = bool(args[0]) + return self._active + + def scan(self): + self._record("scan", (), {}) + if not self._active: + raise OSError("STA must be active") + return list(self._scan) + + def connect(self, ssid=None, key=None, *, bssid=None): + if bssid is not None and not isinstance(bssid, (bytes, bytearray)): + raise ValueError("bad bssid type") + if bssid is not None and len(bssid) != 6: + raise ValueError("bad bssid len") + self._record("connect", (ssid, key), {"bssid": bssid}) + self._connect_ssid = ssid + self._connect_bssid = bssid + if self._fail_status is not None or self._connect_after > 0: + self._connected = False + else: + self._connected = True # default: connect succeeds unless scripted otherwise + return None + + def disconnect(self): + self._record("disconnect", (), {}) + self._connected = False + return None + + def isconnected(self): + self._record("isconnected", (), {}) + return self._connected + + def status(self, *args): + self._record("status", args, {}) + if args: + param = args[0] + if param == "rssi": + return self._scan[0][3] if self._scan else -50 + if param == "stations": + return [] + raise ValueError("unknown status param") + if self.interface == AP_IF: + return None + if self._fail_status is not None: + return self._fail_status + if self._connect_after > 0: + self._connect_after -= 1 + if self._connect_after == 0: + self._connected = True + return STAT_CONNECTING + return STAT_GOT_IP if self._connected else STAT_IDLE + + def ifconfig(self, *args): + self._record("ifconfig", args, {}) + if args: + self._config["ifconfig"] = args[0] + return None + if "ifconfig" in self._config: + return self._config["ifconfig"] + return _CONNECTED_IFCONFIG if self._connected else _DISCONNECTED_IFCONFIG + + def config(self, *args, **kwargs): + self._record("config", args, kwargs) + if kwargs: + self._config.update(kwargs) + return None + if len(args) == 1: + if args[0] == "ssid": + if self._connect_ssid is not None: + return self._connect_ssid + return self._config.get("ssid") + return self._config.get(args[0]) + return None + + +def reset_all(): + """Reset every interface to defaults — call between tests for zero-effort isolation.""" + for inst in WLAN._instances.values(): + inst.reset() diff --git a/tests/stubs/logging.py b/tests/stubs/logging.py new file mode 100644 index 0000000..b7f8b33 --- /dev/null +++ b/tests/stubs/logging.py @@ -0,0 +1,124 @@ +"""Logging stub compatible with wifi_manager imports under MicroPython.""" + +import sys + +CRITICAL = 50 +ERROR = 40 +WARNING = 30 +INFO = 20 +DEBUG = 10 +NOTSET = 0 + +_LEVEL_STR = { + CRITICAL: "CRIT", + ERROR: "ERROR", + WARNING: "WARN", + INFO: "INFO", + DEBUG: "DEBUG", +} + +_stream = sys.stderr +_level = INFO +_loggers = {} + + +class Logger: + level = NOTSET + + def __init__(self, name): + self.name = name + + def _level_str(self, level): + return _LEVEL_STR.get(level, "LVL%s" % level) + + def set_level(self, level): + self.level = level + + def is_enabled_for(self, level): + return level >= (self.level or _level) + + def log(self, level, msg, *args): + if level >= (self.level or _level): + _stream.write("[%s][%s]: " % (self._level_str(level), self.name)) + if args: + _stream.write((str(msg) % args) + "\n") + else: + _stream.write(str(msg) + "\n") + + def debug(self, msg, *args): + self.log(DEBUG, msg, *args) + + def info(self, msg, *args): + self.log(INFO, msg, *args) + + def warning(self, msg, *args): + self.log(WARNING, msg, *args) + + def error(self, msg, *args): + self.log(ERROR, msg, *args) + + def critical(self, msg, *args): + self.log(CRITICAL, msg, *args) + + def exc(self, e, msg, *args): + self.log(ERROR, msg, *args) + + def exception(self, msg, *args): + self.exc(sys.exc_info()[1], msg, *args) + + +Logger.setLevel = Logger.set_level +Logger.isEnabledFor = Logger.is_enabled_for + + +def get_logger(name): + if name in _loggers: + return _loggers[name] + logger = Logger(name) + _loggers[name] = logger + return logger + + +def getLogger(name=None): + return get_logger(name or "root") + + +def info(msg, *args): + get_logger("root").info(msg, *args) + + +def debug(msg, *args): + get_logger("root").debug(msg, *args) + + +def warning(msg, *args): + get_logger("root").warning(msg, *args) + + +warn = warning + + +def error(msg, *args): + get_logger("root").error(msg, *args) + + +def critical(msg, *args): + get_logger("root").critical(msg, *args) + + +def exception(msg, *args): + get_logger("root").exception(msg, *args) + + +def basic_config(level=INFO, filename=None, stream=None, format=None): # noqa: A002 + global _level, _stream + _level = level + if stream is not None: + _stream = stream + if filename is not None: + pass + if format is not None: + pass + + +basicConfig = basic_config diff --git a/tests/stubs/webrepl.py b/tests/stubs/webrepl.py new file mode 100644 index 0000000..8a88441 --- /dev/null +++ b/tests/stubs/webrepl.py @@ -0,0 +1,5 @@ +"""WebREPL stub for deterministic regression execution.""" + + +def start(*_args, **_kwargs): + return True diff --git a/tests/substrate-conformance.py b/tests/substrate-conformance.py new file mode 100644 index 0000000..c0ab442 --- /dev/null +++ b/tests/substrate-conformance.py @@ -0,0 +1,214 @@ +#!/usr/bin/env micropython +"""Conformance test for the esp32 substrate fakes against CONTRACT-esp32-v1.28.md. + +This guards the fakes from drifting away from the captured SDK contract. It tests the +substrate in isolation (no library under test), so it can run anywhere the image runs. +If a fake stops "quacking" like the real esp32 v1.28.0 surface, this fails. +""" + +import os +import sys + +_S_IFDIR = 0x4000 + + +def _dirname(path): + return path.rsplit(os.sep, 1)[0] if os.sep in path else "" + + +def _join(a, b): + if not a: + return b + return a.rstrip(os.sep) + os.sep + b + + +def _isdir(path): + try: + return (os.stat(path)[0] & _S_IFDIR) != 0 + except OSError: + return False + + +_SUBSTRATE_DIR = _join(_dirname(_dirname(__file__)), "substrate") +if _isdir(_SUBSTRATE_DIR): + sys.path.insert(0, _SUBSTRATE_DIR) + +import network +import machine + + +def _run_case(name, fn): + try: + fn() + print("PASS:", name) + except Exception: + print("FAIL:", name) + raise + + +def _raises(exc, fn): + try: + fn() + except exc: + return True + except Exception as other: + raise AssertionError("expected %s, got %r" % (exc, other)) + raise AssertionError("expected %s, nothing raised" % exc) + + +# --- network constants ------------------------------------------------------- +def case_network_constants(): + assert network.STA_IF == 0 + assert network.AP_IF == 1 + # happy-path codes: micropython-owned, stable + assert network.STAT_IDLE == 1000 + assert network.STAT_CONNECTING == 1001 + assert network.STAT_GOT_IP == 1010 + # failure codes: ESP-IDF v5.5.1 wifi_err_reason_t (the IDF version MP v1.28.0 uses) + assert network.STAT_BEACON_TIMEOUT == 200 + assert network.STAT_NO_AP_FOUND == 201 + assert network.STAT_WRONG_PASSWORD == 202 # WIFI_REASON_AUTH_FAIL + assert network.STAT_ASSOC_FAIL == 203 + assert network.STAT_HANDSHAKE_TIMEOUT == 204 + # esp32 quirk: CONNECT_FAIL aliases ASSOC_FAIL (same int) + assert network.STAT_CONNECT_FAIL == network.STAT_ASSOC_FAIL == 203 + + +# --- WLAN identity + lifecycle ---------------------------------------------- +def case_wlan_singleton_identity(): + network.reset_all() + assert network.WLAN(network.STA_IF) is network.WLAN(network.STA_IF) + assert network.WLAN(network.AP_IF) is network.WLAN(network.AP_IF) + assert network.WLAN(network.STA_IF) is not network.WLAN(network.AP_IF) + assert _raises(ValueError, lambda: network.WLAN(99)) + + +def case_scan_requires_active_and_shape(): + network.reset_all() + sta = network.WLAN(network.STA_IF) + assert _raises(OSError, sta.scan) # inactive -> OSError("STA must be active") + sta.active(True) + assert sta.active() is True + sta.program_scan([{"ssid": "Net", "bssid": b"\x01\x02\x03\x04\x05\x06", "rssi": -55}]) + results = sta.scan() + assert isinstance(results, list) + row = results[0] + assert len(row) == 6 + assert isinstance(row[0], bytes) and row[0] == b"Net" # ssid is bytes + assert isinstance(row[1], bytes) and len(row[1]) == 6 # bssid is bytes + assert isinstance(row[2], int) and isinstance(row[3], int) and isinstance(row[4], int) + assert row[5] is False # esp32 hidden is always False + + +def case_connect_contract_and_spy(): + network.reset_all() + sta = network.WLAN(network.STA_IF) + assert sta.connect("Net", "pw", bssid=b"\x01\x02\x03\x04\x05\x06") is None + assert sta.isconnected() is True + assert sta.status() == network.STAT_GOT_IP + assert sta.call_count("connect") == 1 + assert sta.last_call("connect").kwargs["bssid"] == b"\x01\x02\x03\x04\x05\x06" + # bssid must be exactly 6 bytes + assert _raises(ValueError, lambda: sta.connect("Net", bssid=b"\x01\x02")) + assert _raises(TypeError, lambda: sta.connect("Net", foo="bar")) + + +def case_status_contract(): + network.reset_all() + sta = network.WLAN(network.STA_IF) + assert sta.status() == network.STAT_IDLE # no-arg STA -> int + assert isinstance(sta.status("rssi"), int) # 'rssi' -> int + assert isinstance(sta.status("stations"), list) # 'stations' -> list + assert _raises(ValueError, lambda: sta.status("bogus")) + assert network.WLAN(network.AP_IF).status() is None # AP no-arg -> None + # scripted failure status + sta.reset() + sta.fail_connect(network.STAT_NO_AP_FOUND) + sta.connect("X") + assert sta.isconnected() is False + assert sta.status() == network.STAT_NO_AP_FOUND + # connect_after: CONNECTING then GOT_IP + sta.reset() + sta.connect_after(2) + sta.connect("X") + assert sta.status() == network.STAT_CONNECTING + assert sta.status() == network.STAT_CONNECTING + assert sta.status() == network.STAT_GOT_IP + + +def case_ifconfig_and_config(): + network.reset_all() + sta = network.WLAN(network.STA_IF) + disc = sta.ifconfig() + assert isinstance(disc, tuple) and len(disc) == 4 + assert disc[0] == "0.0.0.0" + for part in disc: + assert isinstance(part, str) + sta.connect("Office", "pw") + assert sta.ifconfig()[0] != "0.0.0.0" + assert sta.config("ssid") == "Office" + sta.config(essid="ap-name") + assert sta.config("essid") == "ap-name" + custom = ("10.0.0.2", "255.255.255.0", "10.0.0.1", "1.1.1.1") + assert sta.ifconfig(custom) is None + assert sta.ifconfig() == custom + + +# --- machine.Pin ------------------------------------------------------------- +def case_pin_constants_and_io(): + machine.reset_all() + assert machine.Pin.IN == 1 + assert machine.Pin.OUT == 3 # GPIO_MODE_INPUT_OUTPUT (readable) + assert machine.Pin.OPEN_DRAIN == 7 + + p = machine.Pin(5, machine.Pin.OUT) + assert p.value() == 0 and isinstance(p.value(), int) # value() -> int + assert p.value(1) is None # value(x) -> None + assert p.value() == 1 + assert p(0) is None and p() == 0 # call form == value + p.on() + assert machine.Pin(5).value() == 1 # OUT readable via fresh handle (same GPIO) + assert p.toggle() is None and p.value() == 0 + assert p.call_count("on") == 1 + + +def case_pin_external_input_and_isolation(): + machine.reset_all() + button = machine.Pin(0, machine.Pin.IN, machine.Pin.PULL_UP) + assert button.value() == 0 + button.set_external(1) + assert button.value() == 1 + machine.reset_all() + assert machine.Pin(0).value() == 0 # reset_all clears GPIO state + + +def case_pin_invalid_id(): + machine.reset_all() + assert _raises(ValueError, lambda: machine.Pin(-1)) + + +def case_pin_irq_present(): + machine.reset_all() + assert hasattr(machine.Pin(0, machine.Pin.IN), "irq") + + +def main(): + cases = [ + ("network_constants", case_network_constants), + ("wlan_singleton_identity", case_wlan_singleton_identity), + ("scan_requires_active_and_shape", case_scan_requires_active_and_shape), + ("connect_contract_and_spy", case_connect_contract_and_spy), + ("status_contract", case_status_contract), + ("ifconfig_and_config", case_ifconfig_and_config), + ("pin_constants_and_io", case_pin_constants_and_io), + ("pin_external_input_and_isolation", case_pin_external_input_and_isolation), + ("pin_invalid_id", case_pin_invalid_id), + ("pin_irq_present", case_pin_irq_present), + ] + for name, fn in cases: + _run_case(name, fn) + print("substrate conformance complete") + + +if __name__ == "__main__": + main() diff --git a/tests/wifimanager-status-regression.py b/tests/wifimanager-status-regression.py new file mode 100644 index 0000000..8539a30 --- /dev/null +++ b/tests/wifimanager-status-regression.py @@ -0,0 +1,389 @@ +#!/usr/bin/env micropython +"""Deterministic, device-status focused regression checks for micropython-wifimanager. + +These run against the canonical esp32 substrate fakes in ../substrate (network, +machine) plus small webrepl/logging stubs. The test never defines a mock: it programs +the substrate (program_scan / fail_connect / connect_after) and spies on it +(call_count / last_call), then exercises real WifiManager logic with no radios. +""" + +import json +import os +import sys + +try: + import ubinascii as _b64 +except ImportError: + import binascii as _b64 + + +_MISSING_CONFIG_PATH = "/tmp/wifimanager-missing.json" +_REGRESSION_CONFIG_PATH = "/tmp/wifimanager-regression.json" + +_S_IFDIR = 0x4000 + + +def _dirname(path): + if os.sep in path: + return path.rsplit(os.sep, 1)[0] + return "" + + +def _join(*parts): + out = "" + for part in parts: + if not part: + continue + if not out: + out = part + else: + out = out.rstrip(os.sep) + os.sep + part.lstrip(os.sep) + return out + + +def _stat_mode(path): + try: + return os.stat(path)[0] + except OSError: + return None + + +def _isdir(path): + mode = _stat_mode(path) + return mode is not None and (mode & _S_IFDIR) != 0 + + +def _exists(path): + return _stat_mode(path) is not None + + +_SCRIPT_DIR = _dirname(__file__) + + +def _request(method, path, password, body=""): + auth_blob = _b64.b2a_base64(("admin:%s" % password).encode("utf-8")).decode("utf-8").strip() + headers = [ + "%s %s HTTP/1.1" % (method, path), + "Authorization: Basic %s" % auth_blob, + "Content-Length: %d" % len(body), + "", + body, + ] + return "\r\n".join(headers) + + +def _run_case(name, fn): + try: + fn() + print("PASS:", name) + except Exception: + print("FAIL:", name) + raise + + +def _install_substrate(): + """Install the esp32 substrate (network, machine) + stubs (webrepl, logging) as + importable modules, reset to defaults. Substrate dir is inserted last so it wins.""" + repo_root = _dirname(_SCRIPT_DIR) + stubs_dir = _join(_SCRIPT_DIR, "stubs") + substrate_dir = _join(repo_root, "substrate") + for path in (stubs_dir, substrate_dir): + if _isdir(path): + sys.path.insert(0, path) + + import network + import machine + import webrepl # noqa: F401 + import logging # noqa: F401 + sys.modules["network"] = network + sys.modules["machine"] = machine + sys.modules["webrepl"] = webrepl + sys.modules["logging"] = logging + + network.reset_all() + machine.reset_all() + return network, machine + + +def _load_wifi_manager(): + target = os.getenv("WIFIMANAGER_SRC") + if not target: + raise RuntimeError("Set WIFIMANAGER_SRC to a checked-out micropython-wifimanager path") + if not _isdir(target): + raise RuntimeError("WIFIMANAGER_SRC does not exist: %s" % target) + + sys.path.insert(0, target) + network, _machine = _install_substrate() + + from wifi_manager.wifi_manager import WifiManager + + _reset_manager_state(WifiManager) + return WifiManager, network + + +def _reset_manager_state(manager): + manager._connection_callbacks = [] + manager._last_connection_state = None + manager.webrepl_triggered = False + manager._config_server_enabled = False + manager._config_server_password = "micropython" + manager.config_file = _REGRESSION_CONFIG_PATH + manager._ap_start_policy = "never" + + +def _always_true(*_args, **_kwargs): + return True + + +def _always_false(*_args, **_kwargs): + return False + + +def case_connectivity_preference_and_scan_order(): + manager, network = _load_wifi_manager() + manager.preferred_networks = [ + {"ssid": "Flat_AP", "password": "a"}, + {"ssid": "Strong_AP", "password": "b"}, + ] + sta = network.WLAN(network.STA_IF) + sta.active(True) + sta.program_scan([ + {"ssid": "Strong_AP", "bssid": b"\x01\x02\x03\x04\x05\x06", "rssi": -40}, + {"ssid": "Flat_AP", "bssid": b"\x01\x02\x03\x04\x05\x07", "rssi": -90}, + {"ssid": "Guest", "bssid": b"\x01\x02\x03\x04\x05\x08", "rssi": -20}, + ]) + + ordered = manager._scan_available_networks() + assert [item["ssid"] for item in ordered[:3]] == ["Guest", "Strong_AP", "Flat_AP"] + + candidates = manager._build_connection_candidates(ordered) + assert len(candidates) == 2 + assert candidates[0]["ssid"] == "Flat_AP" + assert candidates[1]["ssid"] == "Strong_AP" + assert candidates[1]["bssid"] == b"\x01\x02\x03\x04\x05\x06" + + +def case_status_edge_callbacks_emit_only_on_state_changes(): + manager, network = _load_wifi_manager() + sta = network.WLAN(network.STA_IF) + events = [] + + def callback(event, **kwargs): + events.append((event, kwargs)) + + manager.on_connection_change(callback) + + # Starts disconnected -> first poll emits a single "disconnected" transition. + manager._check_and_notify_connection_state() + assert len(events) == 1 + assert events[0][0] == "disconnected" + assert events[0][1] == {} + + # Script a successful association to "Office". + sta.active(True) + sta.program_scan([{"ssid": "Office", "bssid": b"\x0a\x0b\x0c\x0d\x0e\x0f", "rssi": -45}]) + sta.connect("Office", "pw") + manager._check_and_notify_connection_state() + assert len(events) == 2 + assert events[1][0] == "connected" + assert events[1][1]["ssid"] == "Office" + + # No state change -> no duplicate transition. + manager._check_and_notify_connection_state() + assert len(events) == 2, "should not emit duplicate connected transition" + + +def case_fallback_ap_relies_on_network_status(): + manager, network = _load_wifi_manager() + sta = network.WLAN(network.STA_IF) + manager._ap_start_policy = "fallback" + + # Connected (GOT_IP) -> no AP wanted. + sta.connect("Home", "pw") + assert manager.wants_accesspoint() is False + + # No AP found -> AP wanted. + sta.reset() + sta.fail_connect(network.STAT_NO_AP_FOUND) + sta.connect("Home", "pw") + assert manager.wants_accesspoint() is True + + # Still connecting -> AP wanted. + sta.reset() + sta.connect_after(3) + sta.connect("Home", "pw") + assert manager.wants_accesspoint() is True + + +def case_connect_candidates_uses_requested_bssid_and_notifies(): + manager, network = _load_wifi_manager() + sta = network.WLAN(network.STA_IF) + events = [] + manager.on_connection_change(lambda event, **kwargs: events.append((event, kwargs))) + manager.preferred_networks = [{"ssid": "Flat_AP", "password": "alpha"}] + + sta.active(True) + sta.program_scan([ + {"ssid": "Flat_AP", "bssid": b"\x10\x11\x12\x13\x14\x15", "rssi": -30}, + {"ssid": "Flat_AP", "bssid": b"\x01\x02\x03\x04\x05\x06", "rssi": -70}, + ]) + + available = manager._scan_available_networks() + candidates = manager._build_connection_candidates(available) + assert candidates[0]["bssid"] == b"\x10\x11\x12\x13\x14\x15" + assert candidates[1]["bssid"] == b"\x01\x02\x03\x04\x05\x06" + + connected = manager._connect_candidates(candidates) + assert connected is True + assert len(events) == 1 + assert events[0][0] == "connected" + + # Spy on the real connect() call: the strongest BSSID was requested. + assert sta.call_count("connect") == 1 + assert sta.last_call("connect").kwargs["bssid"] == b"\x10\x11\x12\x13\x14\x15" + + +def case_pin_substrate_readback_and_spy(): + _network, machine = _install_substrate() + + led = machine.Pin(2, machine.Pin.OUT) + led.on() + assert machine.Pin(2).value() == 1 # OUT is readable on esp32; same GPIO via new handle + assert isinstance(led.value(), int) + led.off() + assert led.value() == 0 + + button = machine.Pin(0, machine.Pin.IN, machine.Pin.PULL_UP) + button.set_external(1) + assert button.value() == 1 + + assert led.call_count("on") == 1 + assert led.call_count("off") == 1 + + +def case_async_manage_connects_when_ap_available(): + manager, network = _load_wifi_manager() + from wifi_manager import wifi_manager as wm_module + import aiotest + + # manage() -> setup_network() reloads config each turn, so the networks must come + # from the config file (not a poked attribute, which _load_config would overwrite). + async_config = "/tmp/wifimanager-async.json" + with open(async_config, "w") as handle: + handle.write(json.dumps({ + "schema": 2, + "known_networks": [{"ssid": "Home", "password": "pw"}], + "access_point": {"config": {"essid": "AP"}, "start_policy": "never"}, + "config_server": {"enabled": False}, + })) + manager.config_file = async_config + + events = [] + manager.on_connection_change(lambda event, **kwargs: events.append((event, kwargs))) + + sta = network.WLAN(network.STA_IF) + sta.active(True) + sta.program_scan([{"ssid": "Home", "bssid": b"\x0a\x0b\x0c\x0d\x0e\x0f", "rssi": -50}]) + assert sta.isconnected() is False + + # Drive the real async manage() loop for a few turns with no wall-clock waiting. + turns = aiotest.run_iterations(manager.manage, 4, wm_module.asyncio) + assert turns == 4 + + # The loop saw "not connected", ran setup_network(), and converged to connected. + assert sta.isconnected() is True + assert sta.call_count("connect") >= 1 + assert [e[0] for e in events][:2] == ["disconnected", "connected"] + + +def case_config_masking_is_stable_on_post(): + manager, _ = _load_wifi_manager() + with open(_REGRESSION_CONFIG_PATH, "w") as handle: + handle.write( + json.dumps( + { + "schema": 2, + "known_networks": [ + {"ssid": "Home", "password": "home-pass", "enables_webrepl": True}, + {"ssid": "Office", "password": "office-pass", "enables_webrepl": False}, + ], + "access_point": { + "config": {"essid": "Micropython-AP", "channel": 11, "password": "change-me"}, + "enables_webrepl": False, + "start_policy": "always", + }, + "config_server": {"enabled": False, "password": "secret"}, + } + ) + ) + + original_setup_network = manager.setup_network + manager.config_file = _REGRESSION_CONFIG_PATH + manager._config_server_password = "secret" + manager.setup_network = _always_true + + try: + payload = json.dumps( + { + "schema": 2, + "known_networks": [ + {"ssid": "Home", "password": "***", "enables_webrepl": True}, + {"ssid": "Office", "password": "office-new"}, + ], + "access_point": { + "config": {"essid": "Micropython-AP", "channel": 6, "password": "***"}, + "enables_webrepl": False, + "start_policy": "always", + }, + } + ) + response = manager._handle_config_request(_request("POST", "/config", "secret", body=payload)) + assert "200 OK" in response + + with open(_REGRESSION_CONFIG_PATH, "r") as handle: + written = json.load(handle) + assert written["known_networks"][0]["password"] == "home-pass" + assert written["known_networks"][1]["password"] == "office-new" + assert written["access_point"]["config"]["password"] == "change-me" + finally: + manager.setup_network = original_setup_network + + +def case_missing_config_triggers_recovery_ap(): + manager, _ = _load_wifi_manager() + manager.config_file = _MISSING_CONFIG_PATH + if _exists(_MISSING_CONFIG_PATH): + os.unlink(_MISSING_CONFIG_PATH) + + original_start_config_server = manager.start_config_server + manager.start_config_server = _always_false + + try: + loaded = manager._load_config() + assert loaded is False + assert manager.ap_config["config"]["essid"] == "MicroPython-AP" + assert manager.ap_config["start_policy"] == "always" + finally: + manager.start_config_server = original_start_config_server + + +def main(): + cases = [ + ("connectivity_preference_and_scan_order", case_connectivity_preference_and_scan_order), + ("status_edge_callbacks_emit_only_on_state_changes", case_status_edge_callbacks_emit_only_on_state_changes), + ("fallback_ap_relies_on_network_status", case_fallback_ap_relies_on_network_status), + ("connect_candidates_uses_requested_bssid", case_connect_candidates_uses_requested_bssid_and_notifies), + ("pin_substrate_readback_and_spy", case_pin_substrate_readback_and_spy), + ("async_manage_connects_when_ap_available", case_async_manage_connects_when_ap_available), + ("config_masking_is_stable_on_post", case_config_masking_is_stable_on_post), + ("missing_config_triggers_recovery_ap", case_missing_config_triggers_recovery_ap), + ] + + for name, fn in cases: + _run_case(name, fn) + + print("wifi-manager deterministic examples complete") + + +if __name__ == "__main__": + main()