Signed-off-by: David Gibson <david(a)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(a)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)
+
+
+(a)contextlib.contextmanager
+def dhclient4(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+ yield from _dhclient(site, ifname, '4')
+
+
+(a)contextlib.contextmanager
+def dhclient6(site: cmdsite.CmdSite, ifname: str) -> Iterator[None]:
+ yield from _dhclient(site, ifname, '6')
+
+
+(a)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()
+
+
+(a)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()
+
+
+(a)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