nsfarm/setup: implement common WAN configuration setups

nsfarm/setup/__init__.py

-from . import alpine, openwrt, updater, utils
+from . import alpine, openwrt, updater, uplink, utils
 
 __all__ = [
     "alpine",
     "openwrt",
     "updater",
+    "uplink",
     "utils",
 ]

nsfarm/setup/uplink.py

+"""Various setups for uplink (WAN) configuration of router.
+"""
+import abc
+import ipaddress
+import typing
+
+from .. import cli
+from ._setup import Setup as _Setup
+
+
+class CommonWAN(_Setup):
+    """Abstract base for all WAN configuring setups."""
+
+    def __init__(self, shell: cli.Shell, interface: str = "wan", restart: bool = True):
+        self._sh = shell
+        self._restart = restart
+        self._interface = interface
+        self._config: dict[str, str] = {}
+        self._previous: dict[str, typing.Optional[str]] = {}
+
+    def prepare(self, no_revert_needed: bool = False):
+        for key, value in self._config.items():
+            if not no_revert_needed:
+                self._sh.command(f"uci -q get network.{self._interface}.{key}")
+                self._previous[key] = None if self._sh.prompt() != 0 else self._sh.output
+            self._sh.run(f"uci set network.{self._interface}.{key}={value}")
+        self._sh.run(f"uci commit network.{self._interface}")
+        if self._restart:
+            self._sh.run("/etc/init.d/network restart")
+
+    def revert(self):
+        for key, value in self._config.items():
+            if value is None:
+                self._sh.run(f"uci del network.{self._interface}.{key}")
+            else:
+                self._sh.run(f"uci set network.{self._interface}.{key}={value}")
+        self._sh.run(f"uci commit network.{self._interface}")
+        if self._restart:
+            self._sh.run("/etc/init.d/network restart")
+
+    def wait4ping(self, target_ip: str = "172.16.1.1"):
+        """Uses ping to wait for network to be ready."""
+        self._sh.run(f"while ! ping -c1 -w1 '{target_ip}' >/dev/null; do true; done")
+
+
+class DHCP(CommonWAN):
+    """Configure WAN interface to obtain IP address from DHCP."""
+
+    def __init__(self, shell: cli.Shell, interface: str = "wan", restart: bool = True):
+        super().__init__(shell, interface, restart)
+        self._config = {"proto": "dhcp"}
+
+    def wait4route(self):
+        """Waiting for default route to be added by DHCP."""
+        self._sh.run("while ! ip route | grep -q default; do sleep 1; done")
+
+
+class StaticIP(CommonWAN):
+    """Configure WAN interface to use static IP."""
+
+    def __init__(
+        self,
+        shell: cli.Shell,
+        network: typing.Union[ipaddress.IPv4Interface, ipaddress.IPv6Interface],
+        gateway: typing.Union[ipaddress.IPv4Address, ipaddress.IPv6Address],
+        dns: typing.Union[ipaddress.IPv4Address, ipaddress.IPv6Address],
+        interface: str = "wan",
+        restart: bool = True,
+    ):
+        super().__init__(shell, interface, restart)
+        self._config = {
+            "proto": "static",
+            "ipadd": str(network.ip),
+            "netmask": str(network.network),
+            "gateway": str(gateway),
+            "dns": str(dns),
+        }
+
+
+class PPPoE(CommonWAN):
+    """Configure WAN interface to use PPPoE."""
+
+    def __init__(
+        self,
+        shell: cli.Shell,
+        username: str = "turris",
+        password: str = "turris",
+        interface: str = "wan",
+        restart: bool = True,
+    ):
+        super().__init__(shell, interface, restart)
+        self._config = {
+            "proto": "pppoe",
+            "username": username,
+            "password": password,
+        }

tests/conftest.py

 import collections.abc
+import ipaddress
 import random
 import string
 import time
@@ -235,22 +236,11 @@ def fixture_board_wan(client_board, isp_container):
     This configures static IP through ips_container.
     Returns wan IPv4 address of WAN interface.
     """
-    wan_ip = "172.16.1.142"
-    client_board.run("uci set network.wan.proto='static'")
-    client_board.run(f"uci set network.wan.ipaddr='{wan_ip}'")
-    client_board.run("uci set network.wan.netmask='255.240.0.0'")
-    client_board.run("uci set network.wan.gateway='172.16.1.1'")
-    client_board.run("uci set network.wan.dns='172.16.1.1'")
-    client_board.run("uci commit network")
-    client_board.run("/etc/init.d/network restart")
-    client_board.run("while ! ping -c1 -w1 172.16.1.1 >/dev/null; do true; done")
-    yield wan_ip
-    client_board.run("uci set network.wan.proto='none'")
-    client_board.run("uci delete network.wan.ipaddr")
-    client_board.run("uci delete network.wan.netmask")
-    client_board.run("uci delete network.wan.gateway")
-    client_board.run("uci delete network.wan.dns")
-    client_board.run("uci commit network")
+    network = ipaddress.ip_interface("172.16.1.142/24")
+    uplink = ipaddress.ip_address("172.16.1.1")
+    with nsfarm.setup.uplink.StaticIP(client_board, network, uplink, uplink) as wan:
+        wan.wait4ping()
+        yield network.ip
 
 
 @pytest.fixture(name="updater_branch", scope="package")

tests/network/test_wan.py

@@ -2,9 +2,12 @@
 
 This checks if we are able to support various ISP configurations.
 """
+import ipaddress
+
 import pytest
 
 import nsfarm.lxd
+import nsfarm.setup
 
 from . import common
 
@@ -20,24 +23,12 @@ class TestStatic(common.InternetTests):
     @pytest.fixture(name="client", scope="class", autouse=True)
     def fixture_client(self, lxd_client, device_map, client_board):
         """Configure WAN to use static IP"""
+        network = ipaddress.ip_network("172.16.1.42/255.240.0.0")
+        uplink = ipaddress.ip_address("172.16.1.1")
         with nsfarm.lxd.Container(lxd_client, "isp-common", device_map) as container:
-            # TODO implement some utility class to set and revert uci configs on router
-            client_board.run("uci set network.wan.proto='static'")
-            client_board.run("uci set network.wan.ipaddr='172.16.1.42'")
-            client_board.run("uci set network.wan.netmask='255.240.0.0'")
-            client_board.run("uci set network.wan.gateway='172.16.1.1'")
-            client_board.run("uci set network.wan.dns='172.16.1.1'")
-            client_board.run("uci commit network")
-            container.shell.run("wait4network")
-            client_board.run("/etc/init.d/network restart")
-            client_board.run("while ! ping -c1 -w1 172.16.1.1 >/dev/null; do true; done")
-            yield client_board
-            client_board.run("uci set network.wan.proto='none'")
-            client_board.run("uci delete network.wan.ipaddr")
-            client_board.run("uci delete network.wan.netmask")
-            client_board.run("uci delete network.wan.gateway")
-            client_board.run("uci delete network.wan.dns")
-            client_board.run("uci commit network")
+            with nsfarm.setup.uplink.StaticIP(client_board, network, uplink, uplink) as wan:
+                wan.wait4ping()
+                yield client_board
 
 
 @pytest.mark.deploy
@@ -48,14 +39,9 @@ class TestDHCP(common.InternetTests):
     def fixture_client(self, lxd_client, device_map, client_board):
         """Configure WAN to use DHCP"""
         with nsfarm.lxd.Container(lxd_client, "isp-dhcp", device_map) as container:
-            client_board.run("uci set network.wan.proto='dhcp'")
-            client_board.run("uci commit network")
-            container.shell.run("wait4network")
-            client_board.run("/etc/init.d/network restart")
-            client_board.run("while ! ip route | grep -q default; do sleep 1; done")
-            yield client_board
-            client_board.run("uci set network.wan.proto='none'")
-            client_board.run("uci commit network")
+            with nsfarm.setup.uplink.DHCP(client_board) as wan:
+                wan.wait4route()
+                yield client_board
 
 
 @pytest.mark.skip("The configuration here removes what is done in fixture board_wan and this removes its revert")
@@ -65,19 +51,6 @@ class TestPPPoE(common.InternetTests):
     @pytest.fixture(name="client", scope="class", autouse=True)
     def fixture_client(self, lxd_client, device_map, client_board):
         with nsfarm.lxd.Container(lxd_client, "isp-pppoe", device_map) as container:
-            client_board.run("uci set network.wan.proto='pppoe'")
-            client_board.run("uci set network.wan.username='turris'")
-            client_board.run("uci set network.wan.password='turris'")
-            client_board.run("uci del network.wan.auto", None)
-            client_board.run("uci commit network")
-            client_board.run("/etc/init.d/network restart")
-            client_board.run("while ! ping -c1 -w1 172.16.1.1 >/dev/null; do true; done")
-            yield client_board
-            commands = [
-                "uci set network.wan.proto='none'",
-                "uci del network.wan.username",
-                "uci del network.wan.password",
-                "uci set network.wan.auto='1'",
-                "uci commit network",
-            ]
-            client_board.run(" && ".join(commands))
+            with nsfarm.setup.uplink.PPPoE(client_board) as wan:
+                wan.wait4ping()
+                yield client_board