refactor: DHCP handling in OpenWrt

tests/conftest.py

@@ -1,6 +1,9 @@
+from ipaddress import IPv4Address
+
 import pytest
 from labgrid import Target
 from labgrid.driver import ShellDriver, SSHDriver
+from openwrt import enable_dhcp
 from strategy import QEMUNetworkStrategy, Status
 
 
@@ -15,3 +18,8 @@ def shell_command(target: Target, strategy: QEMUNetworkStrategy) -> ShellDriver:
 def ssh_command(target: Target, strategy: QEMUNetworkStrategy) -> SSHDriver:
     strategy.transition("ssh")
     return target.get_driver("SSHDriver")
+
+
+@pytest.fixture(scope="module")
+def dhcp_ip(shell_command: ShellDriver) -> list[IPv4Address]:
+    return enable_dhcp(shell_command)

tests/test_util.py

@@ -52,8 +52,31 @@ def test_openwrt_get_gateway_ip_ok(ip_test: IPTest) -> None:
     assert openwrt.get_gateway_ip(runner) == ip_test.ip_addresses
 
 
+@dataclass
+class IPNameTest:
+    ip_output: str
+    if_name: str
+
+
+@pytest.mark.parametrize(
+    "ip_name_test",
+    [
+        IPNameTest(
+            """default via 192.168.187.2 dev br-lan  src 192.168.187.100
+192.168.187.0/24 dev br-lan scope link  src 192.168.187.100
+""",
+            "br-lan",
+        ),
+    ],
+)
+def test_openwrt_get_default_interface_device_name(ip_name_test: IPNameTest) -> None:
+    runner = MagicMock()
+    runner.run_check.side_effect = [ip_name_test.ip_output.split("\n")]
+    assert openwrt.get_default_interface_device_name(runner) == ip_name_test.if_name
+
+
 @patch("network.shell_run")
-def test_primary_host_ip_ok(mock_shell_run: MagicMock) -> None:
+def test_openwrt_primary_host_ip_ok(mock_shell_run: MagicMock) -> None:
     ip_r_s_output = json.dumps([{"dev": "eth0"}])
     ip_a_s_output = json.dumps([{"addr_info": [{"local": "192.168.1.1"}]}])
     mock_shell_run.side_effect = [ip_r_s_output, ip_a_s_output]

util/openwrt.py

@@ -1,6 +1,10 @@
 import re
+from functools import partial
 from ipaddress import IPv4Address
 
+import service
+import uci
+from func import wait_for
 from process import Runner, run
 
 IPV4_ADDR_REGEX = re.compile(r"inet\s+(\d+\.\d+\.\d+\.\d+)")
@@ -19,3 +23,21 @@ def get_gateway_ip(runner: Runner) -> list[IPv4Address]:
     ip_output = run(runner, "ip -4 r s default")
     matches = IPV4_GATEWAY_IP_REGEX.findall(ip_output)
     return [IPv4Address(match) for match in matches]
+
+
+IPV4_PRIMARY_IF_NAME_REGEX = re.compile(r"default\s+via\s+\d+\.\d+\.\d+\.\d+\s+dev\s+([a-z-]+)")
+
+
+def get_default_interface_device_name(runner: Runner) -> str:
+    ip_output = run(runner, "ip -4 r s default")
+    matches = IPV4_PRIMARY_IF_NAME_REGEX.findall(ip_output)
+    return matches[0]
+
+
+def enable_dhcp(runner: Runner) -> list[IPv4Address]:
+    if uci.get(runner, "network.lan.proto") != "dhcp":
+        uci.set(runner, "network.lan.proto", "dhcp")
+        uci.commit(runner, "network")
+        service.restart(runner, "network", wait=1)
+    assert wait_for(partial(get_gateway_ip, runner), "gateway IP has been assigned", delay=1)
+    return get_ip_addr(runner, get_default_interface_device_name(runner))

util/strategy/qemu_network.py

@@ -16,22 +16,19 @@
 import os
 import subprocess
 import urllib.parse
-from functools import partial
 from pathlib import Path
 
 import attr
 import httpx
-import service
-import uci
 from driver import CustomQEMUDriver, QEMUParams
-from func import retry_exc, wait_for
+from func import retry_exc
 from labgrid import step, target_factory
 from labgrid.driver import ShellDriver, SSHDriver
 from labgrid.driver.exception import ExecutionError
 from labgrid.step import Step
 from labgrid.strategy import Strategy, StrategyError
 from labgrid.util import get_free_port
-from openwrt import get_gateway_ip
+from openwrt import enable_dhcp
 
 from .status import Status
 
@@ -167,16 +164,12 @@ def transition(self, state: Status | str, *, step: Step) -> None:
             self.target.activate(self.shell)
 
             assert self.shell
-            if uci.get(self.shell, "network.lan.proto") != "dhcp":
-                uci.set(self.shell, "network.lan.proto", "dhcp")
-                uci.commit(self.shell, "network")
-                service.restart(self.shell, "network", wait=1)
-            assert wait_for(partial(get_gateway_ip, self.shell), "gateway IP has been assigned", delay=1)
 
         elif state == Status.ssh:
             self.transition(Status.shell)
 
             assert self.shell
+            enable_dhcp(self.shell)
             self.update_network_service()
 
         self.status = state

util/strategy/qemu_stateful.py

@@ -1,17 +1,12 @@
-from functools import partial
-
 import attr
-import service
-import uci
 from driver import StatefulQEMUDriver
-from func import retry_exc, wait_for
+from func import retry_exc
 from labgrid import step, target_factory
 from labgrid.driver import ShellDriver, SSHDriver
 from labgrid.driver.exception import ExecutionError
 from labgrid.step import Step
 from labgrid.strategy import Strategy, StrategyError
 from labgrid.util import get_free_port
-from openwrt import get_gateway_ip
 
 from .status import Status
 
@@ -88,16 +83,12 @@ def transition(self, state: Status | str, *, step: Step) -> None:
             self.target.activate(self.shell)
 
             assert self.shell
-            if uci.get(self.shell, "network.lan.proto") != "dhcp":
-                uci.set(self.shell, "network.lan.proto", "dhcp")
-                uci.commit(self.shell, "network")
-                service.restart(self.shell, "network", wait=1)
-            assert wait_for(partial(get_gateway_ip, self.shell), "gateway IP has been assigned", delay=1)
 
         elif state == Status.ssh:
             self.transition(Status.shell)
 
             assert self.shell
+            enable_dhcp(self.shell)
             self.update_network_service()
         else:
             raise StrategyError(f"no transition found from {self.status} to {status}")