[PATCH v2 20/22] 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/Makefile          |   4 +-
 test/tasst/__main__.py |   2 +-
 test/tasst/dhcp.py     | 132 +++++++++++++++++++++++++++++++++++++++++
 test/tasst/dhcpv6.py   |  89 +++++++++++++++++++++++++++
 4 files changed, 224 insertions(+), 3 deletions(-)
 create mode 100644 test/tasst/dhcp.py
 create mode 100644 test/tasst/dhcpv6.py

diff --git a/test/Makefile b/test/Makefile
index 248329e5..0eeaf82e 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -72,8 +72,8 @@ EXETER_PY = build/build.py
 EXETER_JOBS = $(EXETER_SH:%.sh=%.json) $(EXETER_PY:%.py=%.json)
 AVOCADO_JOBS = $(EXETER_JOBS) avocado/static_checkers.json
 
-TASST_SRCS = __init__.py __main__.py address.py ndp.py nstool.py snh.py \
-	transfer.py \
+TASST_SRCS = __init__.py __main__.py address.py dhcp.py dhcpv6.py ndp.py \
+	nstool.py snh.py transfer.py \
 	selftest/__init__.py selftest/static_ifup.py selftest/veth.py
 
 EXETER_META = meta/lint.json meta/tasst.json
diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index 6a95eec1..8c4efd74 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -13,7 +13,7 @@ library of test helpers for passt & pasta
 import exeter
 
 # We import just to get the exeter tests, which flake8 can't see
-from . import ndp, nstool, snh, transfer  # noqa: F401
+from . import dhcp, dhcpv6, ndp, nstool, snh, transfer  # noqa: F401
 from .selftest import static_ifup, veth  # noqa: F401
 
 
diff --git a/test/tasst/dhcp.py b/test/tasst/dhcp.py
new file mode 100644
index 00000000..d86df2de
--- /dev/null
+++ b/test/tasst/dhcp.py
@@ -0,0 +1,132 @@
+#! /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
+"""
+
+import contextlib
+import ipaddress
+import os
+import tempfile
+
+import exeter
+
+from . import address, nstool
+
+
+DHCLIENT = '/sbin/dhclient'
+
+
+@contextlib.contextmanager
+def dhclient(snh, ifname, ipv='4'):
+    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}']
+        snh.fg(f'{DHCLIENT}', *opts, capable=True)
+        yield
+        snh.fg(f'{DHCLIENT}', '-x', '-pf', f'{pidfile}', capable=True)
+
+
+class DhcpTestScenario:
+    def __init__(self, *, client, ifname, addr, gateway, mtu):
+        self.client = client
+        self.ifname = ifname
+        self.addr = addr
+        self.gateway = gateway
+        self.mtu = mtu
+
+
+def test_dhcp_addr(setup):
+    with setup as scn, dhclient(scn.client, scn.ifname):
+        (actual_addr,) = scn.client.addrs(scn.ifname,
+                                          family='inet', scope='global')
+        exeter.assert_eq(actual_addr.ip, scn.addr)
+
+
+def test_dhcp_route(setup):
+    with setup as scn, dhclient(scn.client, scn.ifname):
+        (defroute,) = scn.client.routes4(dst='default')
+        exeter.assert_eq(ipaddress.ip_address(defroute['gateway']),
+                         scn.gateway)
+
+
+def test_dhcp_mtu(setup):
+    with setup as scn, dhclient(scn.client, scn.ifname):
+        exeter.assert_eq(scn.client.mtu(scn.ifname), scn.mtu)
+
+
+DHCP_TESTS = [test_dhcp_addr, test_dhcp_route, test_dhcp_mtu]
+
+
+def dhcp_tests(setup):
+    for t in DHCP_TESTS:
+        testid = f'{setup.__qualname__}|{t.__qualname__}'
+        exeter.register_pipe(testid, setup, t)
+
+
+DHCPD = 'dhcpd'
+SUBNET = address.TEST_NET_1
+ipa = address.IpiAllocator(SUBNET)
+(SERVER_IP4,) = ipa.next_ipis()
+(CLIENT_IP4,) = ipa.next_ipis()
+IFNAME = 'clientif'
+
+
+@contextlib.contextmanager
+def setup_dhcpd_common(ifname, server_ifname):
+    with nstool.unshare_snh('client', '-Un') as client, \
+         nstool.unshare_snh('server', '-n',
+                            parent=client, capable=True) as server:
+        client.veth(ifname, server_ifname, server)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            yield (client, server, tmpdir)
+
+
+@contextlib.contextmanager
+def setup_dhcpd():
+    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 {SUBNET.network_address} netmask {SUBNET.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'')
+
+        server.ifup('lo')
+        server.ifup(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, capable=True, check=False) as dhcpd:
+            # Configure the client
+            client.ifup('lo')
+            yield DhcpTestScenario(client=client, ifname=IFNAME,
+                                   addr=CLIENT_IP4.ip,
+                                   gateway=SERVER_IP4.ip, mtu=1500)
+            dhcpd.terminate()
+
+
+dhcp_tests(setup_dhcpd)
diff --git a/test/tasst/dhcpv6.py b/test/tasst/dhcpv6.py
new file mode 100644
index 00000000..ab119ae7
--- /dev/null
+++ b/test/tasst/dhcpv6.py
@@ -0,0 +1,89 @@
+#! /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
+
+dhcpv6.py - Helpers for testing DHCPv6
+"""
+
+import contextlib
+import os
+
+import exeter
+
+from . import address, dhcp
+
+
+def dhclientv6(snh, ifname):
+    return dhcp.dhclient(snh, ifname, '6')
+
+
+class Dhcpv6TestScenario:
+    def __init__(self, *, client, ifname, addr):
+        self.client = client
+        self.ifname = ifname
+        self.addr = addr
+
+
+def test_dhcp6_addr(setup):
+    with setup as scn, dhclientv6(scn.client, scn.ifname):
+        addrs = [a.ip for a in scn.client.addrs(scn.ifname, family='inet6',
+                                                scope='global')]
+        assert scn.addr in addrs  # Might also have a SLAAC address
+
+
+DHCP6_TESTS = [test_dhcp6_addr]
+
+
+def dhcp6_tests(setup):
+    for t in DHCP6_TESTS:
+        testid = f'{setup.__qualname__}|{t.__qualname__}'
+        exeter.register_pipe(testid, setup, t)
+
+
+DHCPD = 'dhcpd'
+SUBNET = address.TEST_NET6_TASST_A
+ipa = address.IpiAllocator(SUBNET)
+(SERVER_IP6,) = ipa.next_ipis()
+(CLIENT_IP6,) = ipa.next_ipis()
+IFNAME = 'clientif'
+
+
+@contextlib.contextmanager
+def setup_dhcpdv6():
+    server_ifname = 'serverif'
+
+    with dhcp.setup_dhcpd_common(IFNAME, server_ifname) \
+         as (client, server, tmpdir):
+        # Sort out link local addressing
+        server.ifup('lo')
+        server.ifup(server_ifname, SERVER_IP6)
+        client.ifup('lo')
+        client.ifup(IFNAME)
+        server.addr_wait(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 {SUBNET} {{
+            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, capable=True, check=False) as dhcpd:
+            yield Dhcpv6TestScenario(client=client, ifname=IFNAME,
+                                     addr=CLIENT_IP6.ip)
+            dhcpd.terminate()
+
+
+dhcp6_tests(setup_dhcpdv6)
-- 
@@ -0,0 +1,89 @@
+#! /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
+
+dhcpv6.py - Helpers for testing DHCPv6
+"""
+
+import contextlib
+import os
+
+import exeter
+
+from . import address, dhcp
+
+
+def dhclientv6(snh, ifname):
+    return dhcp.dhclient(snh, ifname, '6')
+
+
+class Dhcpv6TestScenario:
+    def __init__(self, *, client, ifname, addr):
+        self.client = client
+        self.ifname = ifname
+        self.addr = addr
+
+
+def test_dhcp6_addr(setup):
+    with setup as scn, dhclientv6(scn.client, scn.ifname):
+        addrs = [a.ip for a in scn.client.addrs(scn.ifname, family='inet6',
+                                                scope='global')]
+        assert scn.addr in addrs  # Might also have a SLAAC address
+
+
+DHCP6_TESTS = [test_dhcp6_addr]
+
+
+def dhcp6_tests(setup):
+    for t in DHCP6_TESTS:
+        testid = f'{setup.__qualname__}|{t.__qualname__}'
+        exeter.register_pipe(testid, setup, t)
+
+
+DHCPD = 'dhcpd'
+SUBNET = address.TEST_NET6_TASST_A
+ipa = address.IpiAllocator(SUBNET)
+(SERVER_IP6,) = ipa.next_ipis()
+(CLIENT_IP6,) = ipa.next_ipis()
+IFNAME = 'clientif'
+
+
+@contextlib.contextmanager
+def setup_dhcpdv6():
+    server_ifname = 'serverif'
+
+    with dhcp.setup_dhcpd_common(IFNAME, server_ifname) \
+         as (client, server, tmpdir):
+        # Sort out link local addressing
+        server.ifup('lo')
+        server.ifup(server_ifname, SERVER_IP6)
+        client.ifup('lo')
+        client.ifup(IFNAME)
+        server.addr_wait(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 {SUBNET} {{
+            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, capable=True, check=False) as dhcpd:
+            yield Dhcpv6TestScenario(client=client, ifname=IFNAME,
+                                     addr=CLIENT_IP6.ip)
+            dhcpd.terminate()
+
+
+dhcp6_tests(setup_dhcpdv6)
-- 
2.45.2