Add a helper to the Site() class to wait for an address with specified
characteristics to be ready on an interface. In particular this is useful
for waiting for IPv6 SLAAC & DAD (Duplicate Address Detection) to complete.
Because DAD is not going to be useful in many of our scenarios, also extend
Site.ifup() to allow DAD to be switched to optimistic mode or disabled.
Signed-off-by: David Gibson <david(a)gibson.dropbear.id.au>
---
test/Makefile | 2 +-
test/tasst/__main__.py | 2 +-
test/tasst/selftest/static_ifup.py | 40 ++++++++++++++++++++++++++++++
test/tasst/selftest/veth.py | 27 ++++++++++++++++++++
test/tasst/snh.py | 24 +++++++++++++++++-
5 files changed, 92 insertions(+), 3 deletions(-)
create mode 100644 test/tasst/selftest/static_ifup.py
diff --git a/test/Makefile b/test/Makefile
index e13c49c8..139a0b14 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -73,7 +73,7 @@ EXETER_JOBS = $(EXETER_SH:%.sh=%.json) $(EXETER_PY:%.py=%.json)
AVOCADO_JOBS = $(EXETER_JOBS) avocado/static_checkers.json
TASST_SRCS = __init__.py __main__.py nstool.py snh.py \
- selftest/__init__.py selftest/veth.py
+ selftest/__init__.py selftest/static_ifup.py selftest/veth.py
EXETER_META = meta/lint.json meta/tasst.json
META_JOBS = $(EXETER_META)
diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index d52f9c55..f3f88424 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -14,7 +14,7 @@ import exeter
# We import just to get the exeter tests, which flake8 can't see
from . import nstool, snh # noqa: F401
-from .selftest import veth # noqa: F401
+from .selftest import static_ifup, veth # noqa: F401
if __name__ == '__main__':
diff --git a/test/tasst/selftest/static_ifup.py b/test/tasst/selftest/static_ifup.py
new file mode 100644
index 00000000..0c6375d4
--- /dev/null
+++ b/test/tasst/selftest/static_ifup.py
@@ -0,0 +1,40 @@
+#! /usr/bin/env python3
+
+# 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
+
+meta/static_ifup - Static address configuration
+"""
+
+import contextlib
+import ipaddress
+
+import exeter
+
+from tasst import nstool
+
+
+IFNAME = 'testveth'
+IFNAME_PEER = 'vethpeer'
+TEST_IPS = set([ipaddress.ip_interface('192.0.2.1/24'),
+ ipaddress.ip_interface('2001:db8:9a55::1/112'),
+ ipaddress.ip_interface('10.1.2.3/8')])
+
+
+(a)contextlib.contextmanager
+def setup_ns():
+ with nstool.unshare_snh('ns', '-Un') as ns:
+ ns.veth(IFNAME, IFNAME_PEER)
+ ns.ifup(IFNAME, *TEST_IPS, dad='disable')
+ yield ns
+
+
+(a)exeter.test
+def test_addr():
+ with setup_ns() as ns:
+ exeter.assert_eq(set(ns.addrs(IFNAME, scope='global')), TEST_IPS)
diff --git a/test/tasst/selftest/veth.py b/test/tasst/selftest/veth.py
index 5c8f0c0b..24bbdc27 100644
--- a/test/tasst/selftest/veth.py
+++ b/test/tasst/selftest/veth.py
@@ -12,6 +12,7 @@ selftest/veth.py - Test various veth configurations
"""
import contextlib
+import ipaddress
import exeter
@@ -38,3 +39,29 @@ def test_mtu():
with unconfigured_veth() as (ns1, ns2):
exeter.assert_eq(ns1.mtu('veth1'), 1500)
exeter.assert_eq(ns2.mtu('veth2'), 1500)
+
+
+(a)exeter.test
+def test_slaac(dad=None):
+ TESTMAC = '02:aa:bb:cc:dd:ee'
+ TESTIP = ipaddress.ip_interface('fe80::aa:bbff:fecc:ddee/64')
+
+ with unconfigured_veth() as (ns1, ns2):
+ ns1.fg('ip', 'link', 'set', 'dev',
'veth1', 'address', f'{TESTMAC}',
+ capable=True)
+
+ ns1.ifup('veth1', dad=dad)
+ ns2.ifup('veth2')
+
+ addrs = ns1.addr_wait('veth1', family='inet6',
scope='link')
+ exeter.assert_eq(addrs, [TESTIP])
+
+
+(a)exeter.test
+def test_optimistic_dad():
+ test_slaac(dad='optimistic')
+
+
+(a)exeter.test
+def test_no_dad():
+ test_slaac(dad='disable')
diff --git a/test/tasst/snh.py b/test/tasst/snh.py
index 0554fbd0..a1225ff0 100644
--- a/test/tasst/snh.py
+++ b/test/tasst/snh.py
@@ -111,7 +111,23 @@ class SimNetHost(contextlib.AbstractContextManager):
info = json.loads(self.output('ip', '-j', 'link',
'show'))
return [i['ifname'] for i in info]
- def ifup(self, ifname):
+ def ifup(self, ifname, *addrs, dad=None):
+ if dad == 'disable':
+ self.fg('sysctl',
f'net.ipv6.conf.{ifname}.accept_dad=0',
+ capable=True)
+ elif dad == 'optimistic':
+ self.fg('sysctl',
f'net.ipv6.conf.{ifname}.optimistic_dad=1',
+ capable=True)
+ elif dad is not None:
+ raise ValueError
+
+ for a in addrs:
+ if not isinstance(a, ipaddress.IPv4Interface) \
+ and not isinstance(a, ipaddress.IPv6Interface):
+ raise TypeError
+ self.fg('ip', 'addr', 'add',
f'{a.with_prefixlen}',
+ 'dev', f'{ifname}', capable=True)
+
self.fg('ip', 'link', 'set', f'{ifname}',
'up', capable=True)
def addrinfos(self, ifname, **criteria):
@@ -135,6 +151,12 @@ class SimNetHost(contextlib.AbstractContextManager):
(info,) = json.loads(self.output(*cmd))
return info['mtu']
+ def addr_wait(self, ifname, **criteria):
+ while True:
+ addrs = self.addrs(ifname, **criteria)
+ if addrs:
+ return addrs
+
# Internal tests
def test_true(self):
with self as snh:
--
2.45.2