From e1647cc75b51e3186ed4987b730226f6212d80bb Mon Sep 17 00:00:00 2001 From: Rainer Poisel Date: Thu, 2 Jan 2025 18:32:54 +0100 Subject: [PATCH] refactor: DHCP handling in OpenWrt --- tests/conftest.py | 8 ++++++++ tests/test_util.py | 25 ++++++++++++++++++++++++- util/openwrt.py | 22 ++++++++++++++++++++++ util/strategy/qemu_network.py | 13 +++---------- util/strategy/qemu_stateful.py | 13 ++----------- 5 files changed, 59 insertions(+), 22 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index fc8a425..1ee1901 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/test_util.py b/tests/test_util.py index bb2cea6..53a1c24 100644 --- a/tests/test_util.py +++ b/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] diff --git a/util/openwrt.py b/util/openwrt.py index 5bbf30a..80cba6e 100644 --- a/util/openwrt.py +++ b/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)) diff --git a/util/strategy/qemu_network.py b/util/strategy/qemu_network.py index 1d61eac..0179c02 100644 --- a/util/strategy/qemu_network.py +++ b/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 diff --git a/util/strategy/qemu_stateful.py b/util/strategy/qemu_stateful.py index 00d5e4e..4e850b3 100644 --- a/util/strategy/qemu_stateful.py +++ b/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}")