From 8e0a540c29e78476a7554470cddd11df47cdb94f Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 04:55:00 +0000 Subject: [PATCH 1/3] windows: take the adapter names netsh accepts, and find a built agent OpenIPC/firmware#2381 recovered a bricked Hi3516CV300 as far as "Phase 2: Flash via TFTP" and stopped there, on four separate faults. `list_interfaces()` took its names from `socket.if_nameindex()`, which answers on Windows with "ethernet_0". netsh addresses adapters by their friendly name, has never heard of that, and says so as "Failed to configure the DHCP service. The interface may be disconnected." -- which sends the reader after a cable. The reporter got past it by renaming their adapter to `ethernet_0`. Ask netsh for the list instead, parsed by column position because netsh is localised, so the names we hand out are names we can hand back. `add_ip` treated an address that was already up as fatal, so setting it up by hand first did not help either. Bindability now decides, before and after the command: it is the property the caller actually wants, and unlike command output it is not translated. An address we did not add is no longer removed on the way out -- taking away the operator's own static IP is not ours to do. netsh also returns before the stack can bind the address, so the TFTP bind raced it and lost with EADDRNOTAVAIL. The failure unwound the context manager, which removed the address, so the log showed defib taking away what it had just assigned. `add_ip` now waits for the address to come up. Separately, `get_agent_binary` searched exactly one path: the git checkout four levels above the module. In an installed package that resolves inside site-packages and can never exist, so the whole `defib agent` family was unreachable for anyone who installed defib the documented way -- and the refusal, "No agent binary for 'hi3516cv300'", reads as an unsupported chip on a chip that is supported. It now searches DEFIB_AGENT_DIR, a packaged binaries/ directory, the cache directory and the checkout, tries both the chip name and the mapped build name so a `make SOC=gk7205v300` output is found, and when it comes up empty says what to build and how. hi3516ev200 has had its own stanza in agent/Makefile all along but was missing from the chip map, so `defib agent` refused a chip the agent builds. A test now reads the SOC list out of the Makefile and fails if the map falls behind it again. --- agent/README.md | 18 +++ src/defib/agent/client.py | 122 ++++++++++++--- src/defib/cli/app.py | 18 ++- src/defib/network/ip_manager.py | 163 ++++++++++++++++++-- src/defib/tui/screens/flash_doctor.py | 6 +- tests/test_agent_binary_lookup.py | 109 ++++++++++++++ tests/test_ip_manager_windows.py | 208 ++++++++++++++++++++++++++ 7 files changed, 602 insertions(+), 42 deletions(-) create mode 100644 tests/test_agent_binary_lookup.py create mode 100644 tests/test_ip_manager_windows.py diff --git a/agent/README.md b/agent/README.md index a999f68..0f2846e 100644 --- a/agent/README.md +++ b/agent/README.md @@ -83,6 +83,24 @@ Requires `arm-none-eabi-gcc` (Arch: `pacman -S arm-none-eabi-gcc arm-none-eabi-n Addresses from [qemu-hisilicon](https://github.com/OpenIPC/LoTool) hardware definitions. +### Where defib looks for a built binary + +No prebuilt agent ships in the pip/uv package -- the binary is bare-metal ARM +compiled per SoC, so it has to be built here first. `defib agent` searches, in +order: + +1. `$DEFIB_AGENT_DIR` +2. `defib/agent/binaries/` inside the installed package +3. the agent cache directory (`defib agent` prints the path when it finds + nothing) -- next to the downloaded-firmware cache +4. this `agent/` directory, which is what a git checkout gets for free + +Both `agent-.bin` and `agent-.bin` are tried in each, so a +`make SOC=gk7205v300` build is found even though that chip shares the +gk7205v200 memory map. If you installed defib rather than cloning it, build +here and copy the `.bin` into the cache directory, or point `DEFIB_AGENT_DIR` +at this one. + `hi3520dv200` is a V1-era DVR/NVR SoC: Cortex-A9 (single core), `0x2xxxxxxx` peripheral map, and a HISFC350 SPI flash controller (NOT the FMC100 used by all other supported SoCs). The HISFC350 driver lives in diff --git a/src/defib/agent/client.py b/src/defib/agent/client.py index 798fa08..0e1a6b6 100644 --- a/src/defib/agent/client.py +++ b/src/defib/agent/client.py @@ -7,6 +7,7 @@ from __future__ import annotations import logging +import os import struct import zlib from dataclasses import dataclass, field @@ -142,6 +143,28 @@ def bad_block(self) -> list[SectorResult]: FALLBACK_BAUD = 115200 # Always works +# Which agent build a chip uses. Several chips share one because they share a +# memory map, so the build name is not the chip name. Module scope so the "no +# binary" help text lists exactly what the lookup accepts. +_CHIP_TO_AGENT = { + "hi3516ev300": "hi3516ev300", + "hi3516ev200": "hi3516ev200", + "gk7205v200": "gk7205v200", + "gk7205v300": "gk7205v200", + "gk7202v300": "gk7205v200", + "hi3516cv300": "hi3516cv300", + "hi3516cv500": "hi3516cv500", + "hi3516av300": "hi3516cv500", # cv500-family, same memory map + "hi3516dv300": "hi3516cv500", # cv500-family, same memory map + "hi3519v101": "hi3519v101", + "hi3516av200": "hi3519v101", # 3519v101 family, same memory map + "hi3516cv610": "hi3516cv610", + "hi3516cv608": "hi3516cv610", # cv6xx-family, same memory map + "hi3518ev200": "hi3518ev200", + "hi3520dv200": "hi3520dv200", # V1-era, HISFC350 SPI controller +} + + def get_agent_binary(chip: str) -> Path | None: """Get the path to the pre-compiled agent binary for a chip. @@ -149,37 +172,90 @@ def get_agent_binary(chip: str) -> Path | None: the variant only affects DDR init, not the agent binary, so we strip it before the lookup. """ - chip = chip.split(":", 1)[0] - chip_to_agent = { - "hi3516ev300": "hi3516ev300", - "gk7205v200": "gk7205v200", - "gk7205v300": "gk7205v200", - "gk7202v300": "gk7205v200", - "hi3516cv300": "hi3516cv300", - "hi3516cv500": "hi3516cv500", - "hi3516av300": "hi3516cv500", # cv500-family, same memory map - "hi3516dv300": "hi3516cv500", # cv500-family, same memory map - "hi3519v101": "hi3519v101", - "hi3516av200": "hi3519v101", # 3519v101 family, same memory map - "hi3516cv610": "hi3516cv610", - "hi3516cv608": "hi3516cv610", # cv6xx-family, same memory map - "hi3518ev200": "hi3518ev200", - "hi3520dv200": "hi3520dv200", # V1-era, HISFC350 SPI controller - } - - agent_name = chip_to_agent.get(chip.lower()) + agent_name = agent_binary_for(chip) if not agent_name: return None - candidates = [ - Path(__file__).parent.parent.parent.parent / "agent" / f"agent-{agent_name}.bin", - ] - for path in candidates: + for path in _agent_search_path(agent_name, chip): if path.exists(): return path return None +def agent_binary_for(chip: str) -> str | None: + """The agent build this chip uses, or None if there is no agent for it. + + Several chips share one binary because they share a memory map, so the + build name is not the chip name. + """ + return _CHIP_TO_AGENT.get(chip.split(":", 1)[0].lower()) + + +def _agent_search_path(agent_name: str, chip: str = "") -> list[Path]: + """Where an agent binary may be found, nearest first. + + The only entry this used to have was the git checkout, four levels up from + this module. In an installed package that resolves inside site-packages and + can never exist, so `defib agent` was unreachable for anyone who installed + defib the documented way -- the reporter in OpenIPC/firmware#2381 hit it on + a uv tool install. The binary still has to be compiled per SoC, so the + honest fix is to look where a compiled one plausibly is and to say how to + build one when it is not there. + + Both names are tried in every directory. `make` names its output after the + SOC it was given, and several chips map onto another chip's build, so + `make SOC=gk7205v300` leaves an agent-gk7205v300.bin that a search for the + mapped agent-gk7205v200.bin would walk straight past. + """ + names = [n for n in (chip.split(":", 1)[0].lower(), agent_name) if n] + filenames = list(dict.fromkeys(f"agent-{n}.bin" for n in names)) + + directories = [] + override = os.environ.get("DEFIB_AGENT_DIR") + if override: + directories.append(Path(override)) + # Shipped inside the package, for a wheel that carries prebuilt agents. + directories.append(Path(__file__).parent / "binaries") + # Built by hand and dropped in the cache next to downloaded firmware. + directories.append(get_agent_cache_dir()) + # A git checkout with `make SOC=` already run in agent/. + directories.append(Path(__file__).parent.parent.parent.parent / "agent") + + return [d / f for d in directories for f in filenames] + + +def get_agent_cache_dir() -> Path: + """Where a locally built agent binary can be dropped to be found.""" + from defib.firmware import get_cache_dir + + return get_cache_dir().parent / "agent" + + +def agent_binary_help(chip: str) -> str: + """Say why there is no agent binary, and what to do about it.""" + agent_name = agent_binary_for(chip) + if not agent_name: + supported = ", ".join(sorted(set(_CHIP_TO_AGENT))) + return ( + f"There is no flash agent for '{chip}'. The agent supports: " + f"{supported}. Use `defib install` or `defib restore`, which drive " + f"U-Boot over TFTP instead and work on every supported chip." + ) + looked_in = "\n ".join(str(p) for p in _agent_search_path(agent_name, chip)) + return ( + f"No agent binary for '{chip}' (it uses the {agent_name} build).\n" + f"The agent is bare-metal C compiled per SoC, and no prebuilt binary " + f"ships in the package -- build it from a defib checkout:\n" + f" git clone https://github.com/OpenIPC/defib && cd defib/agent\n" + f" make SOC={agent_name} # needs arm-none-eabi-gcc\n" + f"then either run defib from that checkout, copy agent-{agent_name}.bin " + f"into {get_agent_cache_dir()}, or point DEFIB_AGENT_DIR at it.\n" + f"Looked in:\n {looked_in}\n" + f"Or skip the agent: `defib install` and `defib restore` drive U-Boot " + f"over TFTP and need nothing compiled." + ) + + class FlashAgentClient: """Client for the bare-metal flash agent. diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 2c2c4cf..d9e8fda 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -1036,7 +1036,9 @@ async def _agent_upload_async( from rich.console import Console - from defib.agent.client import FlashAgentClient, get_agent_binary + from defib.agent.client import ( + FlashAgentClient, agent_binary_help, get_agent_binary, + ) from defib.firmware import get_cached_path from defib.profiles.loader import load_profile from defib.protocol.hisilicon_cv6xx import HiSiliconCV6xx @@ -1052,7 +1054,10 @@ async def _agent_upload_async( # Find agent binary agent_path = get_agent_binary(chip) if not agent_path: - msg = f"No agent binary for '{chip}'" + # A bare "No agent binary for ''" reads as an unsupported chip, + # which sent the reporter in OpenIPC/firmware#2381 looking for another + # tool when the answer was that nothing had compiled it yet. + msg = agent_binary_help(chip) if output == "json": print(json_mod.dumps({"event": "error", "message": msg})) else: @@ -1348,7 +1353,9 @@ async def _agent_flash_async( from rich.console import Console - from defib.agent.client import FlashAgentClient, get_agent_binary + from defib.agent.client import ( + FlashAgentClient, agent_binary_help, get_agent_binary, + ) from defib.firmware import get_cached_path from defib.profiles.loader import load_profile from defib.protocol.hisilicon_standard import HiSiliconStandard @@ -1375,7 +1382,10 @@ async def _agent_flash_async( # --- Find agent binary --- agent_path = get_agent_binary(chip) if not agent_path: - msg = f"No agent binary for '{chip}'" + # A bare "No agent binary for ''" reads as an unsupported chip, + # which sent the reporter in OpenIPC/firmware#2381 looking for another + # tool when the answer was that nothing had compiled it yet. + msg = agent_binary_help(chip) if output == "json": print(json_mod.dumps({"event": "error", "message": msg})) else: diff --git a/src/defib/network/ip_manager.py b/src/defib/network/ip_manager.py index bdd862f..a29fcb1 100644 --- a/src/defib/network/ip_manager.py +++ b/src/defib/network/ip_manager.py @@ -8,6 +8,13 @@ - Linux: ip addr add/del - macOS: ifconfig alias - Windows: netsh interface ip add/delete address + +Two Windows facts shape most of what follows, both reported in +OpenIPC/firmware#2381. `netsh` addresses adapters by their *friendly* name +("Ethernet"), which is not what ``socket.if_nameindex()`` returns there +("ethernet_0"), so the names have to come from netsh itself. And netsh returns +before the stack can actually bind the address, so an immediate bind fails with +EADDRNOTAVAIL on an address that is on its way up. """ from __future__ import annotations @@ -40,19 +47,65 @@ async def _run_command(cmd: list[str]) -> tuple[int, str, str]: ) -async def add_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> None: +def _bindable(ip: str) -> bool: + """Can a socket bind this address right now? + + This is the property every caller actually wants -- the TFTP server binds + the address it just assigned -- and unlike parsing command output it is + language-independent, which matters because netsh and `ip` are localised. + """ + import socket + + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + try: + sock.bind((ip, 0)) + except OSError: + return False + return True + + +async def _wait_until_bindable(ip: str, timeout: float = 10.0) -> bool: + """Poll until the address is usable, or the deadline passes. + + Windows returns from `netsh interface ip add address` before the address + is plumbed in, so the bind that follows raced it and lost -- and because + the failure unwound the context manager, the log showed defib removing the + address it had just added, which reads like the tool sabotaging itself + (OpenIPC/firmware#2381). + """ + deadline = asyncio.get_running_loop().time() + timeout + while True: + if _bindable(ip): + return True + if asyncio.get_running_loop().time() >= deadline: + return False + await asyncio.sleep(0.1) + + +async def add_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> bool: """Add a temporary static IP address to a network interface. + Returns True if this call assigned the address, False if it was already + there. The caller needs the difference: an address we did not add is not + ours to take away again. + Args: interface: Network interface name (e.g., "eth0", "en0", "Ethernet"). ip: IP address to assign (e.g., "192.168.1.10"). netmask: Subnet mask (default "255.255.255.0"). Raises: - IPManagerError: If the command fails. + IPManagerError: If the address cannot be made usable. """ prefix = _netmask_to_prefix(netmask) + # Already configured -- by a previous run, by the operator setting it up by + # hand, or by a NIC that keeps it. All three are the state we wanted, so + # succeed and remember that the teardown must leave it alone. + if _bindable(ip): + logger.info("%s is already usable; leaving it as it is", ip) + return False + if sys.platform == "linux": cmd = ["ip", "addr", "add", f"{ip}/{prefix}", "dev", interface] elif sys.platform == "darwin": @@ -66,10 +119,38 @@ async def add_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> Non returncode, stdout, stderr = await _run_command(cmd) if returncode != 0: + detail = stderr.strip() or stdout.strip() + # The command can fail and the address still be there: another process + # won the race, or the platform reports "already exists" as an error. + # Bindability is the answer, not the exit status. + if _bindable(ip): + logger.info("%s reported an error but %s is usable: %s", cmd[0], ip, detail) + return False raise IPManagerError( - f"Failed to add IP {ip} to {interface}: {stderr.strip() or stdout.strip()}" + f"Failed to add IP {ip} to {interface}: {detail}\n" + f"{_interface_name_advice(interface)}" + ) + + if not await _wait_until_bindable(ip): + raise IPManagerError( + f"{ip} was assigned to {interface} but never became usable. " + f"Check that the adapter is connected and that {ip} does not " + f"collide with an address already on this host." ) logger.info("Successfully added %s to %s", ip, interface) + return True + + +def _interface_name_advice(interface: str) -> str: + """Point at the name mismatch that is nearly always the Windows cause.""" + if sys.platform != "win32": + return f"Check that {interface} exists -- `ip -brief link` lists what does." + names = ", ".join(list_interfaces()) or "none found" + return ( + f"On Windows netsh wants the adapter's friendly name, which is not the " + f"name Python reports for it. Adapters on this host: {names}. " + f"Pass one with --nic." + ) async def remove_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> None: @@ -122,11 +203,14 @@ async def temporary_ip( Yields: The assigned IP address. """ - await add_ip(interface, ip, netmask) + added = await add_ip(interface, ip, netmask) try: yield ip finally: - await remove_ip(interface, ip, netmask) + if added: + await remove_ip(interface, ip, netmask) + else: + logger.info("Leaving %s on %s: this run did not add it", ip, interface) def _netmask_to_prefix(netmask: str) -> int: @@ -138,21 +222,74 @@ def _netmask_to_prefix(netmask: str) -> int: return binary.count("1") +def _windows_interfaces() -> list[str]: + """Adapter names in the form netsh will accept. + + `socket.if_nameindex()` answers on Windows, which is what made this look + like it worked: it returns "ethernet_0", netsh has never heard of that, and + the failure it gives back is "Failed to configure the DHCP service. The + interface may be disconnected." -- which sends the reader after a cable. + The reporter in OpenIPC/firmware#2381 got there by renaming their adapter + to `ethernet_0` to match. Ask netsh instead, so the names we hand out are + the names we hand back. + + Parsed positionally rather than by header text, because netsh is localised: + + Admin State State Type Interface Name + ----------------------------------------------------------------- + Enabled Connected Dedicated Ethernet + + Three single-token columns, then a name that may contain spaces. + """ + import subprocess + + try: + proc = subprocess.run( + ["netsh", "interface", "show", "interface"], + capture_output=True, text=True, timeout=10, + ) + except (OSError, subprocess.SubprocessError) as exc: + logger.warning("Could not list adapters with netsh: %s", exc) + return [] + + names: list[str] = [] + seen_separator = False + for line in proc.stdout.splitlines(): + if line.strip().startswith("---"): + # Everything above the rule is the header, in whatever language. + seen_separator = True + continue + if not seen_separator: + continue + fields = line.split(None, 3) + if len(fields) != 4: + continue + name = fields[3].strip() + if name and name not in names: + names.append(name) + return names + + def list_interfaces() -> list[str]: """List available network interfaces. - Returns interface names that can be used with add_ip/remove_ip. + Returns interface names that can be used with add_ip/remove_ip -- which is + the whole point, and is why Windows does not go through if_nameindex. """ import socket interfaces: list[str] = [] - try: - if hasattr(socket, "if_nameindex"): - for _, name in socket.if_nameindex(): - if name != "lo": - interfaces.append(name) - except OSError: - pass + + if sys.platform == "win32": + interfaces = _windows_interfaces() + else: + try: + if hasattr(socket, "if_nameindex"): + for _, name in socket.if_nameindex(): + if name != "lo": + interfaces.append(name) + except OSError: + pass if not interfaces: # Fallback: common defaults diff --git a/src/defib/tui/screens/flash_doctor.py b/src/defib/tui/screens/flash_doctor.py index 471b02e..b125979 100644 --- a/src/defib/tui/screens/flash_doctor.py +++ b/src/defib/tui/screens/flash_doctor.py @@ -474,7 +474,9 @@ async def _do_upload_and_connect(self) -> None: """Try connecting to running agent first, upload if needed.""" import asyncio as aio - from defib.agent.client import FlashAgentClient, get_agent_binary + from defib.agent.client import ( + FlashAgentClient, agent_binary_help, get_agent_binary, + ) from defib.transport.serial import SerialTransport chip = self._chip @@ -507,7 +509,7 @@ async def _do_upload_and_connect(self) -> None: agent_path = get_agent_binary(chip) if not agent_path: - self._log(f"[red]No agent binary for '{chip}'[/]") + self._log(f"[red]{agent_binary_help(chip)}[/]") return agent_data = agent_path.read_bytes() diff --git a/tests/test_agent_binary_lookup.py b/tests/test_agent_binary_lookup.py new file mode 100644 index 0000000..eab38c9 --- /dev/null +++ b/tests/test_agent_binary_lookup.py @@ -0,0 +1,109 @@ +"""The flash agent has to be findable, or say why it is not. + +OpenIPC/firmware#2381: a reporter with a bricked Hi3516CV300 was told to run +`defib agent flash` and got "No agent binary for 'hi3516cv300'" -- which reads +as an unsupported chip. hi3516cv300 is supported; the agent is bare-metal C +compiled per SoC and no prebuilt binary ships in the package. Worse, the only +path ever searched was the git checkout four levels above the module, which in +an installed package resolves inside site-packages and can never exist, so the +whole `defib agent` family was unreachable for anyone who installed defib the +documented way. +""" +from __future__ import annotations + +from pathlib import Path + +from defib.agent.client import ( + _CHIP_TO_AGENT, + _agent_search_path, + agent_binary_for, + agent_binary_help, + get_agent_binary, +) + + +class TestSearchPath: + def test_an_installed_package_has_somewhere_to_look(self): + """More than the checkout, which an installed defib does not have.""" + paths = _agent_search_path("hi3516cv300") + assert len(paths) > 1 + + def test_the_env_override_wins(self, monkeypatch, tmp_path): + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + assert _agent_search_path("hi3516cv300")[0] == tmp_path / "agent-hi3516cv300.bin" + + def test_a_binary_in_the_env_dir_is_found(self, monkeypatch, tmp_path): + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + (tmp_path / "agent-hi3516cv300.bin").write_bytes(b"\x00" * 16) + assert get_agent_binary("hi3516cv300") == tmp_path / "agent-hi3516cv300.bin" + + def test_a_variant_suffix_still_resolves(self, monkeypatch, tmp_path): + """`hi3516av300:emmc` differs only in DDR init, not in the agent.""" + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + (tmp_path / "agent-hi3516cv500.bin").write_bytes(b"\x00" * 16) + assert get_agent_binary("hi3516av300:emmc") is not None + + def test_every_path_is_absolute(self): + assert all(Path(p).is_absolute() for p in _agent_search_path("hi3516cv300")) + + def test_a_build_named_after_the_chip_is_found_too(self, monkeypatch, tmp_path): + """`make SOC=gk7205v300` writes agent-gk7205v300.bin, not the mapped name.""" + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + (tmp_path / "agent-gk7205v300.bin").write_bytes(b"\x00" * 16) + assert get_agent_binary("gk7205v300") == tmp_path / "agent-gk7205v300.bin" + + def test_the_mapped_build_is_still_found_when_that_is_what_exists(self, monkeypatch, tmp_path): + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + (tmp_path / "agent-gk7205v200.bin").write_bytes(b"\x00" * 16) + assert get_agent_binary("gk7205v300") == tmp_path / "agent-gk7205v200.bin" + + def test_every_chip_the_agent_makefile_builds_is_reachable(self): + """The map is what `defib agent` will accept; the Makefile is the truth.""" + makefile = ( + Path(__file__).parent.parent / "agent" / "Makefile" + ).read_text() + built = { + line.split("$(SOC),")[1].split(")")[0] + for line in makefile.splitlines() + if "ifeq ($(SOC)," in line or "else ifeq ($(SOC)," in line + } + assert built, "could not read the SOC list out of agent/Makefile" + assert built <= set(_CHIP_TO_AGENT), ( + f"agent/Makefile builds {sorted(built - set(_CHIP_TO_AGENT))}, " + f"which defib agent will refuse" + ) + + +class TestTheMessage: + def test_a_supported_chip_is_not_reported_as_unsupported(self, monkeypatch, tmp_path): + """This is the sentence that misdirected the reporter.""" + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + msg = agent_binary_help("hi3516cv300") + assert "no prebuilt binary" in msg + assert "make SOC=hi3516cv300" in msg + + def test_it_names_the_shared_build_when_they_differ(self, monkeypatch, tmp_path): + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + msg = agent_binary_help("hi3516dv300") + assert "hi3516cv500 build" in msg + assert "make SOC=hi3516cv500" in msg + + def test_it_says_where_it_looked(self, monkeypatch, tmp_path): + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + assert str(tmp_path) in agent_binary_help("hi3516cv300") + + def test_it_offers_the_route_that_needs_no_compiler(self): + """install/restore drive U-Boot over TFTP and want nothing built.""" + msg = agent_binary_help("hi3516cv300") + assert "defib install" in msg and "defib restore" in msg + + def test_a_chip_with_no_agent_says_so_and_lists_what_has_one(self): + msg = agent_binary_help("hi3516cv200") + assert "no flash agent for 'hi3516cv200'" in msg + assert "hi3516cv300" in msg + assert "defib install" in msg + + def test_agent_binary_for_agrees_with_the_map(self): + for chip, build in _CHIP_TO_AGENT.items(): + assert agent_binary_for(chip) == build + assert agent_binary_for("nonesuch") is None diff --git a/tests/test_ip_manager_windows.py b/tests/test_ip_manager_windows.py new file mode 100644 index 0000000..247d075 --- /dev/null +++ b/tests/test_ip_manager_windows.py @@ -0,0 +1,208 @@ +"""Temporary-IP management must not fight the host it is running on. + +OpenIPC/firmware#2381: a reporter recovering a bricked Hi3516CV300 on Windows +got `defib install` as far as "Phase 2: Flash via TFTP" and no further. Three +separate faults, all in this module: + + * the adapter name came from ``socket.if_nameindex()`` ("ethernet_0"), which + netsh does not accept -- it answered "Failed to configure the DHCP service. + The interface may be disconnected." and they went looking for a cable; + * assigning an address that was already there was fatal ("The object already + exists"), so setting it up by hand first did not help either; + * netsh returns before the address can be bound, so the TFTP bind that came + next raced it, lost, and unwound the context manager -- which removed the + address, making the log read as though defib had taken away what it had + just assigned. + +These tests pin the three fixes and the promise that we only tear down what we +put up. +""" +from __future__ import annotations + +import subprocess +import sys + +import pytest + +from defib.network import ip_manager +from defib.network.ip_manager import ( + IPManagerError, + _windows_interfaces, + add_ip, + list_interfaces, + temporary_ip, +) + + +NETSH_SHOW_INTERFACE = """ +Admin State State Type Interface Name +------------------------------------------------------------------------- +Enabled Connected Dedicated Ethernet +Enabled Disconnected Dedicated Wi-Fi +Enabled Connected Dedicated Ethernet 2 +""" + + +class TestWindowsInterfaceNames: + def test_names_come_from_netsh_not_if_nameindex(self, monkeypatch): + """The names must be the ones netsh will take back.""" + monkeypatch.setattr( + subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess( + a[0], 0, NETSH_SHOW_INTERFACE, "", + ), + ) + assert _windows_interfaces() == ["Ethernet", "Wi-Fi", "Ethernet 2"] + + def test_a_name_with_a_space_survives(self, monkeypatch): + """"Ethernet 2" is a real default name; splitting on whitespace loses it.""" + monkeypatch.setattr( + subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess( + a[0], 0, NETSH_SHOW_INTERFACE, "", + ), + ) + assert "Ethernet 2" in _windows_interfaces() + + def test_the_header_is_not_an_adapter(self, monkeypatch): + """netsh is localised, so the header is skipped by position, not by text.""" + monkeypatch.setattr( + subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess( + a[0], 0, NETSH_SHOW_INTERFACE, "", + ), + ) + assert "Interface Name" not in _windows_interfaces() + + def test_no_netsh_falls_back_rather_than_raising(self, monkeypatch): + monkeypatch.setattr( + subprocess, "run", + lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("netsh")), + ) + assert _windows_interfaces() == [] + + def test_windows_does_not_go_through_if_nameindex(self, monkeypatch): + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + monkeypatch.setattr(ip_manager, "_windows_interfaces", lambda: ["Ethernet"]) + assert list_interfaces() == ["Ethernet"] + + +class TestAddressAlreadyThere: + @pytest.mark.asyncio + async def test_an_address_already_up_is_success_not_failure(self, monkeypatch): + """"The object already exists" is the state we wanted.""" + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: True) + + async def fail(cmd): + raise AssertionError("should not have run a command") + + monkeypatch.setattr(ip_manager, "_run_command", fail) + assert await add_ip("Ethernet", "192.168.1.10") is False + + @pytest.mark.asyncio + async def test_a_command_that_fails_but_leaves_it_usable_is_success(self, monkeypatch): + """Exit status is not the authority; bindability is.""" + states = iter([False, True]) + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: next(states)) + + async def already_exists(cmd): + return 1, "", "The object already exists." + + monkeypatch.setattr(ip_manager, "_run_command", already_exists) + assert await add_ip("Ethernet", "192.168.1.10") is False + + @pytest.mark.asyncio + async def test_a_genuine_failure_still_raises_and_names_the_adapters(self, monkeypatch): + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + monkeypatch.setattr(ip_manager, "_windows_interfaces", lambda: ["Ethernet"]) + + async def disconnected(cmd): + return 1, "", "The interface may be disconnected." + + monkeypatch.setattr(ip_manager, "_run_command", disconnected) + with pytest.raises(IPManagerError) as exc: + await add_ip("ethernet_0", "192.168.1.10") + # The message has to point at the name mismatch, which is the cause. + assert "friendly name" in str(exc.value) + assert "Ethernet" in str(exc.value) + assert "--nic" in str(exc.value) + + +class TestBindRace: + @pytest.mark.asyncio + async def test_add_waits_for_the_address_to_become_usable(self, monkeypatch): + """netsh returns early; the caller binds immediately. Wait it out.""" + calls = {"n": 0} + + def bindable(ip): + calls["n"] += 1 + return calls["n"] > 3 # not there, not there, not there, there + + monkeypatch.setattr(ip_manager, "_bindable", bindable) + + async def ok(cmd): + return 0, "", "" + + monkeypatch.setattr(ip_manager, "_run_command", ok) + assert await add_ip("Ethernet", "192.168.1.10") is True + assert calls["n"] > 1, "did not poll at all" + + @pytest.mark.asyncio + async def test_an_address_that_never_comes_up_says_so(self, monkeypatch): + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + + async def ok(cmd): + return 0, "", "" + + monkeypatch.setattr(ip_manager, "_run_command", ok) + + async def instant(ip, timeout=10.0): + return False + + monkeypatch.setattr(ip_manager, "_wait_until_bindable", instant) + with pytest.raises(IPManagerError, match="never became usable"): + await add_ip("Ethernet", "192.168.1.10") + + +class TestTeardownOnlyUndoesWhatWeDid: + @pytest.mark.asyncio + async def test_an_address_we_added_is_removed(self, monkeypatch): + removed = [] + monkeypatch.setattr(ip_manager, "add_ip", _returning(True)) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + async with temporary_ip("Ethernet", "192.168.1.10"): + pass + assert removed == [("Ethernet", "192.168.1.10")] + + @pytest.mark.asyncio + async def test_an_address_that_was_already_there_is_left_alone(self, monkeypatch): + """Taking away the operator's own static IP is not ours to do.""" + removed = [] + monkeypatch.setattr(ip_manager, "add_ip", _returning(False)) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + async with temporary_ip("Ethernet", "192.168.1.10"): + pass + assert removed == [] + + @pytest.mark.asyncio + async def test_a_failure_inside_the_block_still_removes_ours(self, monkeypatch): + removed = [] + monkeypatch.setattr(ip_manager, "add_ip", _returning(True)) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + with pytest.raises(RuntimeError): + async with temporary_ip("Ethernet", "192.168.1.10"): + raise RuntimeError("TFTP bind failed") + assert removed == [("Ethernet", "192.168.1.10")] + + +def _returning(value): + async def _add(interface, ip, netmask="255.255.255.0"): + return value + return _add + + +def _recording(sink): + async def _remove(interface, ip, netmask="255.255.255.0"): + sink.append((interface, ip)) + return _remove From de5b498d7fde15044eec7d80f213e48d0c2a63a0 Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 04:58:03 +0000 Subject: [PATCH 2/3] tests: drop an unused import, and check netsh for real on Windows Ruff caught the `sys` import the mocked tests did not need, since they patch `ip_manager.sys` rather than their own. Putting it back for a skipif: the CI matrix runs on windows-latest, so the one claim the interface fix rests on -- that netsh can be asked for names netsh will accept -- can be checked against the real command there instead of only against a captured table. --- tests/test_ip_manager_windows.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/test_ip_manager_windows.py b/tests/test_ip_manager_windows.py index 247d075..ccb3bb3 100644 --- a/tests/test_ip_manager_windows.py +++ b/tests/test_ip_manager_windows.py @@ -86,6 +86,20 @@ def test_windows_does_not_go_through_if_nameindex(self, monkeypatch): monkeypatch.setattr(ip_manager, "_windows_interfaces", lambda: ["Ethernet"]) assert list_interfaces() == ["Ethernet"] + @pytest.mark.skipif( + not sys.platform.startswith("win"), reason="needs a real netsh", + ) + def test_real_netsh_yields_at_least_one_adapter(self): + """The mocks above pin the parser; this pins it against the real thing. + + CI runs this matrix on windows-latest, so the one claim the fix rests + on -- that netsh can be asked for names netsh will accept -- is checked + on a real Windows host rather than only against a captured table. + """ + assert _windows_interfaces(), ( + "netsh listed no adapters; the parser or the command has drifted" + ) + class TestAddressAlreadyThere: @pytest.mark.asyncio From cfa74f5c0ba155f07f1604b82e5b256f61e46158 Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 05:32:15 +0000 Subject: [PATCH 3/3] review: keep netsh off the event loop, and never strand a host address Three findings from the review of this branch, all of them mine. `_windows_interfaces()` ran `subprocess.run` with a ten-second timeout, and `list_interfaces()` is reached from the async install and network paths and from add_ip's own failure advice. Blocking the loop there stops every other task, including the serial link to a camera sitting in its bootrom window, which does not wait for us. The netsh table parser is now separate from the call that produces it, so `list_interfaces_async()` can reuse it over `asyncio.create_subprocess_exec` while the synchronous `list-interfaces` command keeps the synchronous one. `add_ip` raised when the bindability wait timed out, having already assigned the address -- and `temporary_ip` cannot clean up, because it does not learn that we own the address until add_ip returns. So a delayed address left the host reconfigured, and the next run read it as the operator's and left it alone for good. Every exit after a successful assignment now takes the address back down, cancellation included, and a rollback that itself fails does not replace the error that caused it. `_agent_search_path` called `get_agent_cache_dir()` eagerly, and that went through `get_cache_dir()`, which creates the directory. An unwritable cache location therefore raised out of `get_agent_binary` even when DEFIB_AGENT_DIR held the binary, and out of the help text whose whole job is to say that nothing was found. Building a search path is now free of side effects; `get_cache_dir` gained `create=False` for callers that are only looking. --- src/defib/agent/client.py | 10 +- src/defib/cli/app.py | 8 +- src/defib/firmware.py | 12 ++- src/defib/network/ip_manager.py | 99 +++++++++++++++---- tests/test_agent_binary_lookup.py | 43 ++++++++ tests/test_ip_manager_windows.py | 159 ++++++++++++++++++++++++++++++ 6 files changed, 302 insertions(+), 29 deletions(-) diff --git a/src/defib/agent/client.py b/src/defib/agent/client.py index 0e1a6b6..8d1b92d 100644 --- a/src/defib/agent/client.py +++ b/src/defib/agent/client.py @@ -225,10 +225,16 @@ def _agent_search_path(agent_name: str, chip: str = "") -> list[Path]: def get_agent_cache_dir() -> Path: - """Where a locally built agent binary can be dropped to be found.""" + """Where a locally built agent binary can be dropped to be found. + + Names the directory without making it. Building the search path must stay + free of side effects: an unwritable cache location used to raise out of + `get_agent_binary` even when DEFIB_AGENT_DIR held the binary, and out of + the help text whose whole job is to explain that nothing was found. + """ from defib.firmware import get_cache_dir - return get_cache_dir().parent / "agent" + return get_cache_dir(create=False).parent / "agent" def agent_binary_help(chip: str) -> str: diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index d9e8fda..5ab3cdf 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -829,7 +829,7 @@ async def _network_async( from rich.console import Console - from defib.network.ip_manager import list_interfaces, temporary_ip + from defib.network.ip_manager import list_interfaces_async, temporary_ip from defib.network.tftp_server import start_tftp_server console = Console() @@ -851,7 +851,7 @@ async def _network_async( # Determine network interface if not nic: - interfaces = list_interfaces() + interfaces = await list_interfaces_async() if interfaces: nic = interfaces[0] if output == "human": @@ -2227,7 +2227,7 @@ async def _install_async( has_firmware, pad_to_size, ) - from defib.network.ip_manager import list_interfaces, temporary_ip + from defib.network.ip_manager import list_interfaces_async, temporary_ip from defib.network.tftp_server import start_tftp_server from defib.recovery.events import LogEvent, ProgressEvent from defib.recovery.session import RecoverySession @@ -2600,7 +2600,7 @@ async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: if not use_pod_tftp: # Host TFTP needs a NIC + host_ip; pod path needs neither. if not nic: - interfaces = list_interfaces() + interfaces = await list_interfaces_async() if interfaces: nic = interfaces[0] else: diff --git a/src/defib/firmware.py b/src/defib/firmware.py index bdbd7e1..a967c0b 100644 --- a/src/defib/firmware.py +++ b/src/defib/firmware.py @@ -68,8 +68,13 @@ } -def get_cache_dir() -> Path: - """Get platform-appropriate cache directory for downloaded firmware.""" +def get_cache_dir(create: bool = True) -> Path: + """Get platform-appropriate cache directory for downloaded firmware. + + `create=False` just names the directory. Callers that are only searching + should use it: an unwritable cache location must not stop a lookup that + would have found the file somewhere else entirely. + """ if sys.platform == "darwin": base = Path.home() / "Library" / "Caches" elif sys.platform == "win32": @@ -77,7 +82,8 @@ def get_cache_dir() -> Path: else: base = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) cache_dir = base / "defib" / "firmware" - cache_dir.mkdir(parents=True, exist_ok=True) + if create: + cache_dir.mkdir(parents=True, exist_ok=True) return cache_dir diff --git a/src/defib/network/ip_manager.py b/src/defib/network/ip_manager.py index a29fcb1..e65de7a 100644 --- a/src/defib/network/ip_manager.py +++ b/src/defib/network/ip_manager.py @@ -32,6 +32,11 @@ class IPManagerError(Exception): """Failed to manage IP address.""" +# netsh is the only source of adapter names netsh will accept. +_NETSH_LIST_CMD = ["netsh", "interface", "show", "interface"] +_NETSH_TIMEOUT = 10 + + async def _run_command(cmd: list[str]) -> tuple[int, str, str]: """Run a shell command asynchronously and return (returncode, stdout, stderr).""" proc = await asyncio.create_subprocess_exec( @@ -126,12 +131,20 @@ async def add_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> boo if _bindable(ip): logger.info("%s reported an error but %s is usable: %s", cmd[0], ip, detail) return False - raise IPManagerError( - f"Failed to add IP {ip} to {interface}: {detail}\n" - f"{_interface_name_advice(interface)}" - ) + advice = await _interface_name_advice(interface) + raise IPManagerError(f"Failed to add IP {ip} to {interface}: {detail}\n{advice}") - if not await _wait_until_bindable(ip): + # From here the address is ours, so every way out of this function has to + # take it back down again -- `temporary_ip` cannot, because it does not + # learn that we own it until we return. Leaving it up would also make the + # next run read it as the operator's and never clean it up. + try: + ready = await _wait_until_bindable(ip) + except BaseException: + await _undo_add(interface, ip, netmask) + raise + if not ready: + await _undo_add(interface, ip, netmask) raise IPManagerError( f"{ip} was assigned to {interface} but never became usable. " f"Check that the adapter is connected and that {ip} does not " @@ -141,11 +154,19 @@ async def add_ip(interface: str, ip: str, netmask: str = "255.255.255.0") -> boo return True -def _interface_name_advice(interface: str) -> str: +async def _undo_add(interface: str, ip: str, netmask: str) -> None: + """Best-effort rollback that must never replace the error that caused it.""" + try: + await remove_ip(interface, ip, netmask) + except Exception: # noqa: BLE001 - rollback failure must not mask the cause + logger.warning("Could not take %s back off %s", ip, interface, exc_info=True) + + +async def _interface_name_advice(interface: str) -> str: """Point at the name mismatch that is nearly always the Windows cause.""" if sys.platform != "win32": return f"Check that {interface} exists -- `ip -brief link` lists what does." - names = ", ".join(list_interfaces()) or "none found" + names = ", ".join(await list_interfaces_async()) or "none found" return ( f"On Windows netsh wants the adapter's friendly name, which is not the " f"name Python reports for it. Adapters on this host: {names}. " @@ -245,16 +266,42 @@ def _windows_interfaces() -> list[str]: try: proc = subprocess.run( - ["netsh", "interface", "show", "interface"], - capture_output=True, text=True, timeout=10, + _NETSH_LIST_CMD, + capture_output=True, text=True, timeout=_NETSH_TIMEOUT, ) except (OSError, subprocess.SubprocessError) as exc: logger.warning("Could not list adapters with netsh: %s", exc) return [] + return _parse_netsh_interfaces(proc.stdout) + + +async def _windows_interfaces_async() -> list[str]: + """As `_windows_interfaces`, without standing on the event loop. + + `list_interfaces()` is reached from the async install and network paths, + and from add_ip's own failure advice. A synchronous `subprocess.run` there + stops every other task until netsh returns -- which is the whole recovery, + including the serial link to a camera sitting in its bootrom window. + """ + try: + returncode, stdout, stderr = await asyncio.wait_for( + _run_command(_NETSH_LIST_CMD), timeout=_NETSH_TIMEOUT, + ) + except (OSError, asyncio.TimeoutError) as exc: + logger.warning("Could not list adapters with netsh: %s", exc) + return [] + if returncode != 0: + logger.warning("netsh could not list adapters: %s", stderr.strip()) + return [] + return _parse_netsh_interfaces(stdout) + + +def _parse_netsh_interfaces(stdout: str) -> list[str]: + """Pull the adapter names out of a `netsh interface show interface` table.""" names: list[str] = [] seen_separator = False - for line in proc.stdout.splitlines(): + for line in stdout.splitlines(): if line.strip().startswith("---"): # Everything above the rule is the header, in whatever language. seen_separator = True @@ -270,11 +317,32 @@ def _windows_interfaces() -> list[str]: return names +async def list_interfaces_async() -> list[str]: + """`list_interfaces` for callers that are already in an event loop.""" + if sys.platform == "win32": + return await _windows_interfaces_async() or _fallback_interfaces() + return list_interfaces() + + +def _fallback_interfaces() -> list[str]: + """Last resort when the platform will not enumerate.""" + if sys.platform == "linux": + return ["eth0", "enp0s3"] + if sys.platform == "darwin": + return ["en0", "en1"] + if sys.platform == "win32": + return ["Ethernet", "Wi-Fi"] + return [] + + def list_interfaces() -> list[str]: """List available network interfaces. Returns interface names that can be used with add_ip/remove_ip -- which is the whole point, and is why Windows does not go through if_nameindex. + + Synchronous, so it belongs to the synchronous CLI commands; + `list_interfaces_async` is the one to reach for inside a coroutine. """ import socket @@ -291,13 +359,4 @@ def list_interfaces() -> list[str]: except OSError: pass - if not interfaces: - # Fallback: common defaults - if sys.platform == "linux": - interfaces = ["eth0", "enp0s3"] - elif sys.platform == "darwin": - interfaces = ["en0", "en1"] - elif sys.platform == "win32": - interfaces = ["Ethernet", "Wi-Fi"] - - return interfaces + return interfaces or _fallback_interfaces() diff --git a/tests/test_agent_binary_lookup.py b/tests/test_agent_binary_lookup.py index eab38c9..251a20d 100644 --- a/tests/test_agent_binary_lookup.py +++ b/tests/test_agent_binary_lookup.py @@ -107,3 +107,46 @@ def test_agent_binary_for_agrees_with_the_map(self): for chip, build in _CHIP_TO_AGENT.items(): assert agent_binary_for(chip) == build assert agent_binary_for("nonesuch") is None + + +class TestLookupHasNoSideEffects: + """Qodo review on OpenIPC/defib#135. + + Building the search path used to create the cache directory, so an + unwritable cache location raised out of `get_agent_binary` even when + DEFIB_AGENT_DIR held the binary -- and out of `agent_binary_help`, whose + entire job is to explain that nothing was found. + """ + + def test_searching_does_not_create_the_cache_directory(self, monkeypatch, tmp_path): + cache = tmp_path / "nonexistent-cache" + monkeypatch.setattr( + "defib.firmware.get_cache_dir", + lambda create=True: cache / "firmware", + ) + _agent_search_path("hi3516cv300") + assert not cache.exists() + + def test_a_binary_is_still_found_when_the_cache_cannot_be_made( + self, monkeypatch, tmp_path, + ): + def refuses(create=True): + if create: + raise OSError("read-only file system") + return tmp_path / "cache" / "firmware" + + monkeypatch.setattr("defib.firmware.get_cache_dir", refuses) + monkeypatch.setenv("DEFIB_AGENT_DIR", str(tmp_path)) + (tmp_path / "agent-hi3516cv300.bin").write_bytes(b"\x00" * 16) + assert get_agent_binary("hi3516cv300") == tmp_path / "agent-hi3516cv300.bin" + + def test_the_help_text_still_renders_when_the_cache_cannot_be_made( + self, monkeypatch, tmp_path, + ): + def refuses(create=True): + if create: + raise OSError("read-only file system") + return tmp_path / "cache" / "firmware" + + monkeypatch.setattr("defib.firmware.get_cache_dir", refuses) + assert "make SOC=hi3516cv300" in agent_binary_help("hi3516cv300") diff --git a/tests/test_ip_manager_windows.py b/tests/test_ip_manager_windows.py index ccb3bb3..9cf26fb 100644 --- a/tests/test_ip_manager_windows.py +++ b/tests/test_ip_manager_windows.py @@ -19,6 +19,7 @@ """ from __future__ import annotations +import asyncio import subprocess import sys @@ -27,9 +28,11 @@ from defib.network import ip_manager from defib.network.ip_manager import ( IPManagerError, + _parse_netsh_interfaces, _windows_interfaces, add_ip, list_interfaces, + list_interfaces_async, temporary_ip, ) @@ -220,3 +223,159 @@ def _recording(sink): async def _remove(interface, ip, netmask="255.255.255.0"): sink.append((interface, ip)) return _remove + + +class TestTheAddressIsNeverStranded: + """Qodo review on OpenIPC/defib#135. + + `add_ip` returns ownership to `temporary_ip`, so anything that raises + between assigning the address and returning happens while nobody is in a + position to clean up. A stranded address is not just litter: the next run + sees it already up and reads it as the operator's, so it is never removed. + """ + + @pytest.mark.asyncio + async def test_a_wait_that_times_out_takes_the_address_back_down(self, monkeypatch): + removed = [] + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + + async def ok(cmd): + return 0, "", "" + + async def never(ip, timeout=10.0): + return False + + monkeypatch.setattr(ip_manager, "_run_command", ok) + monkeypatch.setattr(ip_manager, "_wait_until_bindable", never) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + + with pytest.raises(IPManagerError, match="never became usable"): + await add_ip("Ethernet", "192.168.1.10") + assert removed == [("Ethernet", "192.168.1.10")] + + @pytest.mark.asyncio + async def test_a_cancelled_wait_takes_the_address_back_down(self, monkeypatch): + """Ctrl-C during a recovery must not leave the host reconfigured.""" + removed = [] + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + + async def ok(cmd): + return 0, "", "" + + async def cancelled(ip, timeout=10.0): + raise asyncio.CancelledError() + + monkeypatch.setattr(ip_manager, "_run_command", ok) + monkeypatch.setattr(ip_manager, "_wait_until_bindable", cancelled) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + + with pytest.raises(asyncio.CancelledError): + await add_ip("Ethernet", "192.168.1.10") + assert removed == [("Ethernet", "192.168.1.10")] + + @pytest.mark.asyncio + async def test_a_rollback_that_fails_does_not_hide_the_real_error(self, monkeypatch): + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + + async def ok(cmd): + return 0, "", "" + + async def never(ip, timeout=10.0): + return False + + async def blows_up(interface, ip, netmask="255.255.255.0"): + raise OSError("netsh went away") + + monkeypatch.setattr(ip_manager, "_run_command", ok) + monkeypatch.setattr(ip_manager, "_wait_until_bindable", never) + monkeypatch.setattr(ip_manager, "remove_ip", blows_up) + + with pytest.raises(IPManagerError, match="never became usable"): + await add_ip("Ethernet", "192.168.1.10") + + @pytest.mark.asyncio + async def test_an_address_that_was_already_there_is_not_rolled_back(self, monkeypatch): + removed = [] + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: True) + monkeypatch.setattr(ip_manager, "remove_ip", _recording(removed)) + assert await add_ip("Ethernet", "192.168.1.10") is False + assert removed == [] + + +class TestNetshStaysOffTheEventLoop: + """Qodo review on OpenIPC/defib#135. + + A synchronous `subprocess.run` reached from a coroutine stops every other + task until it returns -- during a recovery that includes the serial link to + a camera sitting in its bootrom window, which does not wait for us. + """ + + @pytest.mark.asyncio + async def test_the_async_path_does_not_call_subprocess_run(self, monkeypatch): + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + + def forbidden(*a, **k): + raise AssertionError("blocking subprocess.run on the event loop") + + monkeypatch.setattr(subprocess, "run", forbidden) + + async def fake(cmd): + return 0, NETSH_SHOW_INTERFACE, "" + + monkeypatch.setattr(ip_manager, "_run_command", fake) + assert await list_interfaces_async() == ["Ethernet", "Wi-Fi", "Ethernet 2"] + + @pytest.mark.asyncio + async def test_a_netsh_that_hangs_does_not_hang_the_recovery(self, monkeypatch): + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + monkeypatch.setattr(ip_manager, "_NETSH_TIMEOUT", 0.05) + + async def hangs(cmd): + await asyncio.sleep(30) + return 0, "", "" + + monkeypatch.setattr(ip_manager, "_run_command", hangs) + # Falls back rather than stalling, and does so promptly. + assert await list_interfaces_async() == ["Ethernet", "Wi-Fi"] + + @pytest.mark.asyncio + async def test_a_failing_netsh_falls_back(self, monkeypatch): + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + + async def fails(cmd): + return 1, "", "The requested operation requires elevation." + + monkeypatch.setattr(ip_manager, "_run_command", fails) + assert await list_interfaces_async() == ["Ethernet", "Wi-Fi"] + + @pytest.mark.asyncio + async def test_the_failure_advice_is_async_too(self, monkeypatch): + """add_ip's own error message reaches the enumerator; that path counts.""" + monkeypatch.setattr(ip_manager.sys, "platform", "win32") + monkeypatch.setattr(ip_manager, "_bindable", lambda ip: False) + + def forbidden(*a, **k): + raise AssertionError("blocking subprocess.run on the event loop") + + monkeypatch.setattr(subprocess, "run", forbidden) + + async def dispatch(cmd): + if cmd[:2] == ["netsh", "interface"] and "show" in cmd: + return 0, NETSH_SHOW_INTERFACE, "" + return 1, "", "The interface may be disconnected." + + monkeypatch.setattr(ip_manager, "_run_command", dispatch) + with pytest.raises(IPManagerError) as exc: + await add_ip("ethernet_0", "192.168.1.10") + assert "Ethernet" in str(exc.value) + + +class TestTheParserIsSeparableFromTheCommand: + def test_parsing_needs_no_subprocess_at_all(self): + """Splitting the two is what let the async path reuse the parser.""" + assert _parse_netsh_interfaces(NETSH_SHOW_INTERFACE) == [ + "Ethernet", "Wi-Fi", "Ethernet 2", + ] + + def test_an_empty_table_is_not_an_adapter(self): + assert _parse_netsh_interfaces("") == []