[PATCH v3 13/15] tasst: Helpers for testing DHCP & DHCPv6 behaviour

[David Gibson <david@gibson.dropbear.id.au>]

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 test/tasst/__main__.py |   1 +
 test/tasst/dhcp.py     | 190 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 191 insertions(+)
 create mode 100644 test/tasst/dhcp.py

diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index 92319d46..f94001e7 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -17,6 +17,7 @@ import exeter
 
 MODULES = [
     'cmdsite',
+    'dhcp',
     'ip',
     'ndp',
     'transfer',
diff --git a/test/tasst/dhcp.py b/test/tasst/dhcp.py
new file mode 100644
index 00000000..9231b086
--- /dev/null
+++ b/test/tasst/dhcp.py
@@ -0,0 +1,190 @@
+#! /usr/bin/env avocado-runner-avocado-classless
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david@gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+dhcp.py - Helpers for testing DHCP & DHCPv6
+"""
+
+import contextlib
+import dataclasses
+import ipaddress
+import os
+import tempfile
+from typing import Iterator, Literal
+
+import exeter
+
+from . import cmdsite, ip, unshare, veth
+
+
+DHCLIENT = '/sbin/dhclient'
+
+
+def _dhclient(site: cmdsite.CmdSite, ifname: str, ipv: Literal['4', '6']) \
+        -> Iterator[None]:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        pidfile = os.path.join(tmpdir, 'dhclient.pid')
+        leasefile = os.path.join(tmpdir, 'dhclient.leases')
+
+        # We need '-nc' because we may be running with
+        # capabilities but not UID 0.  Without -nc dhclient drops
+        # capabilities before invoking dhclient-script, so it's
+        # unable to actually configure the interface
+        opts = [f'-{ipv}', '-v', '-nc', '-pf', f'{pidfile}',
+                '-lf', f'{leasefile}', f'{ifname}']
+        site.fg(f'{DHCLIENT}', *opts, privilege=True)
+        yield
+        site.fg(f'{DHCLIENT}', '-x', '-pf', f'{pidfile}', privilege=True)
+
+
+@contextlib.contextmanager
+def dhclient4(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+    yield from _dhclient(site, ifname, '4')
+
+
+@contextlib.contextmanager
+def dhclient6(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+    yield from _dhclient(site, ifname, '6')
+
+
+@dataclasses.dataclass
+class Dhcp4Scenario(exeter.Scenario):
+    client: cmdsite.CmdSite
+    ifname: str
+    addr: ip.Addr
+    gateway: ip.Addr
+    mtu: int
+
+    @exeter.scenariotest
+    def dhcp_addr(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            (actual_addr,) = ip.addrs(self.client, self.ifname,
+                                      family='inet', scope='global')
+            exeter.assert_eq(actual_addr.ip, self.addr)
+
+    @exeter.scenariotest
+    def dhcp_route(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            (defroute,) = ip.routes4(self.client, dst='default')
+            exeter.assert_eq(ipaddress.ip_address(defroute['gateway']),
+                             self.gateway)
+
+    @exeter.scenariotest
+    def dhcp_mtu(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            exeter.assert_eq(ip.mtu(self.client, self.ifname), self.mtu)
+
+
+DHCPD = 'dhcpd'
+IFNAME = 'clientif'
+
+SUBNET4 = ip.TEST_NET_1
+ipa4 = ip.IpiAllocator(SUBNET4)
+(SERVER_IP4,) = ipa4.next_ipis()
+(CLIENT_IP4,) = ipa4.next_ipis()
+
+
+@contextlib.contextmanager
+def setup_dhcpd_common(ifname: str, server_ifname: str) \
+        -> Iterator[tuple[cmdsite.CmdSite, cmdsite.CmdSite, str]]:
+    with unshare.unshare('client', '-Un') as client, \
+         unshare.unshare('server', '-n',
+                         parent=client, privilege=True) as server, \
+         veth.veth(client, ifname, server_ifname, server), \
+         tempfile.TemporaryDirectory() as tmpdir:
+        yield (client, server, tmpdir)
+
+
+def setup_dhcpd4() -> Iterator[Dhcp4Scenario]:
+    server_ifname = 'serverif'
+
+    with setup_dhcpd_common(IFNAME, server_ifname) as (client, server, tmpdir):
+        # Configure dhcpd
+        confpath = os.path.join(tmpdir, 'dhcpd.conf')
+        open(confpath, 'w', encoding='UTF-8').write(
+            f'''subnet {SUBNET4.network_address} netmask {SUBNET4.netmask} {{
+            option routers {SERVER_IP4.ip};
+            range {CLIENT_IP4.ip} {CLIENT_IP4.ip};
+            }}'''
+        )
+        pidfile = os.path.join(tmpdir, 'dhcpd.pid')
+        leasepath = os.path.join(tmpdir, 'dhcpd.leases')
+        open(leasepath, 'wb').write(b'')
+
+        ip.ifup(server, 'lo')
+        ip.ifup(server, server_ifname, SERVER_IP4)
+
+        opts = ['-f', '-d', '-4', '-cf', f'{confpath}',
+                '-lf', f'{leasepath}', '-pf', f'{pidfile}']
+        server.fg(f'{DHCPD}', '-t', *opts)  # test config
+        with server.bg(f'{DHCPD}', *opts, privilege=True,
+                       check=False) as dhcpd:
+            # Configure the client
+            ip.ifup(client, 'lo')
+            yield Dhcp4Scenario(client=client, ifname=IFNAME,
+                                addr=CLIENT_IP4.ip,
+                                gateway=SERVER_IP4.ip, mtu=1500)
+            dhcpd.terminate()
+
+
+@dataclasses.dataclass
+class Dhcp6Scenario(exeter.Scenario):
+    client: cmdsite.CmdSite
+    ifname: str
+    addr: ip.Addr
+
+    @exeter.scenariotest
+    def dhcp6_addr(self) -> None:
+        with dhclient6(self.client, self.ifname):
+            addrs = [a.ip for a in ip.addrs(self.client, self.ifname,
+                                            family='inet6',
+                                            scope='global')]
+            assert self.addr in addrs  # Might also have a SLAAC address
+
+
+SUBNET6 = ip.TEST_NET6_TASST_A
+ipa6 = ip.IpiAllocator(SUBNET6)
+(SERVER_IP6,) = ipa6.next_ipis()
+(CLIENT_IP6,) = ipa6.next_ipis()
+
+
+def setup_dhcpd6() -> Iterator[Dhcp6Scenario]:
+    server_ifname = 'serverif'
+
+    with setup_dhcpd_common(IFNAME, server_ifname) as (client, server, tmpdir):
+        # Sort out link local addressing
+        ip.ifup(server, 'lo')
+        ip.ifup(server, server_ifname, SERVER_IP6)
+        ip.ifup(client, 'lo')
+        ip.ifup(client, IFNAME)
+        ip.addr_wait(server, server_ifname, family='inet6', scope='link')
+
+        # Configure the DHCP server
+        confpath = os.path.join(tmpdir, 'dhcpd.conf')
+        open(confpath, 'w', encoding='UTF-8').write(
+            f'''subnet6 {SUBNET6} {{
+            range6 {CLIENT_IP6.ip} {CLIENT_IP6.ip};
+            }}''')
+        pidfile = os.path.join(tmpdir, 'dhcpd.pid')
+        leasepath = os.path.join(tmpdir, 'dhcpd.leases')
+        open(leasepath, 'wb').write(b'')
+
+        opts = ['-f', '-d', '-6', '-cf', f'{confpath}',
+                '-lf', f'{leasepath}', '-pf', f'{pidfile}']
+        server.fg(f'{DHCPD}', '-t', *opts)  # test config
+        with server.bg(f'{DHCPD}', *opts, privilege=True,
+                       check=False) as dhcpd:
+            yield Dhcp6Scenario(client=client, ifname=IFNAME,
+                                addr=CLIENT_IP6.ip)
+            dhcpd.terminate()
+
+
+def selftests() -> None:
+    Dhcp4Scenario.test(setup_dhcpd4)
+    Dhcp6Scenario.test(setup_dhcpd6)
-- 
@@ -0,0 +1,190 @@
+#! /usr/bin/env avocado-runner-avocado-classless
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david@gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+dhcp.py - Helpers for testing DHCP & DHCPv6
+"""
+
+import contextlib
+import dataclasses
+import ipaddress
+import os
+import tempfile
+from typing import Iterator, Literal
+
+import exeter
+
+from . import cmdsite, ip, unshare, veth
+
+
+DHCLIENT = '/sbin/dhclient'
+
+
+def _dhclient(site: cmdsite.CmdSite, ifname: str, ipv: Literal['4', '6']) \
+        -> Iterator[None]:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        pidfile = os.path.join(tmpdir, 'dhclient.pid')
+        leasefile = os.path.join(tmpdir, 'dhclient.leases')
+
+        # We need '-nc' because we may be running with
+        # capabilities but not UID 0.  Without -nc dhclient drops
+        # capabilities before invoking dhclient-script, so it's
+        # unable to actually configure the interface
+        opts = [f'-{ipv}', '-v', '-nc', '-pf', f'{pidfile}',
+                '-lf', f'{leasefile}', f'{ifname}']
+        site.fg(f'{DHCLIENT}', *opts, privilege=True)
+        yield
+        site.fg(f'{DHCLIENT}', '-x', '-pf', f'{pidfile}', privilege=True)
+
+
+@contextlib.contextmanager
+def dhclient4(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+    yield from _dhclient(site, ifname, '4')
+
+
+@contextlib.contextmanager
+def dhclient6(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+    yield from _dhclient(site, ifname, '6')
+
+
+@dataclasses.dataclass
+class Dhcp4Scenario(exeter.Scenario):
+    client: cmdsite.CmdSite
+    ifname: str
+    addr: ip.Addr
+    gateway: ip.Addr
+    mtu: int
+
+    @exeter.scenariotest
+    def dhcp_addr(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            (actual_addr,) = ip.addrs(self.client, self.ifname,
+                                      family='inet', scope='global')
+            exeter.assert_eq(actual_addr.ip, self.addr)
+
+    @exeter.scenariotest
+    def dhcp_route(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            (defroute,) = ip.routes4(self.client, dst='default')
+            exeter.assert_eq(ipaddress.ip_address(defroute['gateway']),
+                             self.gateway)
+
+    @exeter.scenariotest
+    def dhcp_mtu(self) -> None:
+        with dhclient4(self.client, self.ifname):
+            exeter.assert_eq(ip.mtu(self.client, self.ifname), self.mtu)
+
+
+DHCPD = 'dhcpd'
+IFNAME = 'clientif'
+
+SUBNET4 = ip.TEST_NET_1
+ipa4 = ip.IpiAllocator(SUBNET4)
+(SERVER_IP4,) = ipa4.next_ipis()
+(CLIENT_IP4,) = ipa4.next_ipis()
+
+
+@contextlib.contextmanager
+def setup_dhcpd_common(ifname: str, server_ifname: str) \
+        -> Iterator[tuple[cmdsite.CmdSite, cmdsite.CmdSite, str]]:
+    with unshare.unshare('client', '-Un') as client, \
+         unshare.unshare('server', '-n',
+                         parent=client, privilege=True) as server, \
+         veth.veth(client, ifname, server_ifname, server), \
+         tempfile.TemporaryDirectory() as tmpdir:
+        yield (client, server, tmpdir)
+
+
+def setup_dhcpd4() -> Iterator[Dhcp4Scenario]:
+    server_ifname = 'serverif'
+
+    with setup_dhcpd_common(IFNAME, server_ifname) as (client, server, tmpdir):
+        # Configure dhcpd
+        confpath = os.path.join(tmpdir, 'dhcpd.conf')
+        open(confpath, 'w', encoding='UTF-8').write(
+            f'''subnet {SUBNET4.network_address} netmask {SUBNET4.netmask} {{
+            option routers {SERVER_IP4.ip};
+            range {CLIENT_IP4.ip} {CLIENT_IP4.ip};
+            }}'''
+        )
+        pidfile = os.path.join(tmpdir, 'dhcpd.pid')
+        leasepath = os.path.join(tmpdir, 'dhcpd.leases')
+        open(leasepath, 'wb').write(b'')
+
+        ip.ifup(server, 'lo')
+        ip.ifup(server, server_ifname, SERVER_IP4)
+
+        opts = ['-f', '-d', '-4', '-cf', f'{confpath}',
+                '-lf', f'{leasepath}', '-pf', f'{pidfile}']
+        server.fg(f'{DHCPD}', '-t', *opts)  # test config
+        with server.bg(f'{DHCPD}', *opts, privilege=True,
+                       check=False) as dhcpd:
+            # Configure the client
+            ip.ifup(client, 'lo')
+            yield Dhcp4Scenario(client=client, ifname=IFNAME,
+                                addr=CLIENT_IP4.ip,
+                                gateway=SERVER_IP4.ip, mtu=1500)
+            dhcpd.terminate()
+
+
+@dataclasses.dataclass
+class Dhcp6Scenario(exeter.Scenario):
+    client: cmdsite.CmdSite
+    ifname: str
+    addr: ip.Addr
+
+    @exeter.scenariotest
+    def dhcp6_addr(self) -> None:
+        with dhclient6(self.client, self.ifname):
+            addrs = [a.ip for a in ip.addrs(self.client, self.ifname,
+                                            family='inet6',
+                                            scope='global')]
+            assert self.addr in addrs  # Might also have a SLAAC address
+
+
+SUBNET6 = ip.TEST_NET6_TASST_A
+ipa6 = ip.IpiAllocator(SUBNET6)
+(SERVER_IP6,) = ipa6.next_ipis()
+(CLIENT_IP6,) = ipa6.next_ipis()
+
+
+def setup_dhcpd6() -> Iterator[Dhcp6Scenario]:
+    server_ifname = 'serverif'
+
+    with setup_dhcpd_common(IFNAME, server_ifname) as (client, server, tmpdir):
+        # Sort out link local addressing
+        ip.ifup(server, 'lo')
+        ip.ifup(server, server_ifname, SERVER_IP6)
+        ip.ifup(client, 'lo')
+        ip.ifup(client, IFNAME)
+        ip.addr_wait(server, server_ifname, family='inet6', scope='link')
+
+        # Configure the DHCP server
+        confpath = os.path.join(tmpdir, 'dhcpd.conf')
+        open(confpath, 'w', encoding='UTF-8').write(
+            f'''subnet6 {SUBNET6} {{
+            range6 {CLIENT_IP6.ip} {CLIENT_IP6.ip};
+            }}''')
+        pidfile = os.path.join(tmpdir, 'dhcpd.pid')
+        leasepath = os.path.join(tmpdir, 'dhcpd.leases')
+        open(leasepath, 'wb').write(b'')
+
+        opts = ['-f', '-d', '-6', '-cf', f'{confpath}',
+                '-lf', f'{leasepath}', '-pf', f'{pidfile}']
+        server.fg(f'{DHCPD}', '-t', *opts)  # test config
+        with server.bg(f'{DHCPD}', *opts, privilege=True,
+                       check=False) as dhcpd:
+            yield Dhcp6Scenario(client=client, ifname=IFNAME,
+                                addr=CLIENT_IP6.ip)
+            dhcpd.terminate()
+
+
+def selftests() -> None:
+    Dhcp4Scenario.test(setup_dhcpd4)
+    Dhcp6Scenario.test(setup_dhcpd6)
-- 
2.46.0