[PATCH v3 10/15] tasst/veth: Helpers for constructing veth devices between namespaces

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

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

diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index e7456e8b..4eab9157 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -19,6 +19,7 @@ MODULES = [
     'cmdsite',
     'ip',
     'unshare',
+    'veth',
 ]
 
 
diff --git a/test/tasst/veth.py b/test/tasst/veth.py
new file mode 100644
index 00000000..7fa5cbb5
--- /dev/null
+++ b/test/tasst/veth.py
@@ -0,0 +1,80 @@
+#! /usr/bin/env python3
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david@gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+selftest/veth.py - Test various veth configurations
+"""
+
+import contextlib
+from typing import Iterator, Literal
+import ipaddress
+
+import exeter
+
+from . import cmdsite, ip, unshare
+
+
+@contextlib.contextmanager
+def veth(site: cmdsite.CmdSite, ifname: str,
+         peername: str, peer: unshare.Unshare | None = None) -> Iterator[None]:
+    site.fg('ip', 'link', 'add', ifname, 'type', 'veth',
+            'peer', 'name', peername, privilege=True)
+    if peer is not None:
+        site.fg('ip', 'link', 'set', peername,
+                'netns', f'{peer.relative_pid(site)}', privilege=True)
+    yield
+    site.fg('ip', 'link', 'del', ifname, privilege=True)
+
+
+def selftests() -> None:
+    @contextlib.contextmanager
+    def veth_setup() -> Iterator[tuple[cmdsite.CmdSite, cmdsite.CmdSite]]:
+        with unshare.unshare('ns1', '-Un') as ns1:
+            with unshare.unshare('ns2', '-n', parent=ns1,
+                                 privilege=True) as ns2:
+                with veth(ns1, 'vetha', 'vethb', ns2):
+                    yield (ns1, ns2)
+
+    @exeter.test
+    def test_ifs() -> None:
+        with veth_setup() as (ns1, ns2):
+            exeter.assert_eq(set(ip.ifs(ns1)), set(['lo', 'vetha']))
+            exeter.assert_eq(set(ip.ifs(ns2)), set(['lo', 'vethb']))
+
+    @exeter.test
+    def test_mtu() -> None:
+        with veth_setup() as (ns1, ns2):
+            exeter.assert_eq(ip.mtu(ns1, 'vetha'), 1500)
+            exeter.assert_eq(ip.mtu(ns2, 'vethb'), 1500)
+
+    def test_slaac(dad: Literal['disable', 'optimistic', None]) -> None:
+        TESTMAC = '02:aa:bb:cc:dd:ee'
+        TESTIP = ipaddress.ip_interface('fe80::aa:bbff:fecc:ddee/64')
+
+        with veth_setup() as (ns1, ns2):
+            ns1.fg('ip', 'link', 'set', 'dev', 'vetha', 'address', TESTMAC,
+                   privilege=True)
+
+            ip.ifup(ns1, 'vetha', dad=dad)
+            ip.ifup(ns2, 'vethb')
+
+            addrs = ip.addr_wait(ns1, 'vetha', family='inet6', scope='link')
+        exeter.assert_eq(addrs, [TESTIP])
+
+    @exeter.test
+    def test_dad() -> None:
+        test_slaac(dad=None)
+
+    @exeter.test
+    def test_optimistic_dad() -> None:
+        test_slaac(dad='optimistic')
+
+    @exeter.test
+    def test_no_dad() -> None:
+        test_slaac(dad='disable')
-- 
@@ -0,0 +1,80 @@
+#! /usr/bin/env python3
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david@gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+selftest/veth.py - Test various veth configurations
+"""
+
+import contextlib
+from typing import Iterator, Literal
+import ipaddress
+
+import exeter
+
+from . import cmdsite, ip, unshare
+
+
+@contextlib.contextmanager
+def veth(site: cmdsite.CmdSite, ifname: str,
+         peername: str, peer: unshare.Unshare | None = None) -> Iterator[None]:
+    site.fg('ip', 'link', 'add', ifname, 'type', 'veth',
+            'peer', 'name', peername, privilege=True)
+    if peer is not None:
+        site.fg('ip', 'link', 'set', peername,
+                'netns', f'{peer.relative_pid(site)}', privilege=True)
+    yield
+    site.fg('ip', 'link', 'del', ifname, privilege=True)
+
+
+def selftests() -> None:
+    @contextlib.contextmanager
+    def veth_setup() -> Iterator[tuple[cmdsite.CmdSite, cmdsite.CmdSite]]:
+        with unshare.unshare('ns1', '-Un') as ns1:
+            with unshare.unshare('ns2', '-n', parent=ns1,
+                                 privilege=True) as ns2:
+                with veth(ns1, 'vetha', 'vethb', ns2):
+                    yield (ns1, ns2)
+
+    @exeter.test
+    def test_ifs() -> None:
+        with veth_setup() as (ns1, ns2):
+            exeter.assert_eq(set(ip.ifs(ns1)), set(['lo', 'vetha']))
+            exeter.assert_eq(set(ip.ifs(ns2)), set(['lo', 'vethb']))
+
+    @exeter.test
+    def test_mtu() -> None:
+        with veth_setup() as (ns1, ns2):
+            exeter.assert_eq(ip.mtu(ns1, 'vetha'), 1500)
+            exeter.assert_eq(ip.mtu(ns2, 'vethb'), 1500)
+
+    def test_slaac(dad: Literal['disable', 'optimistic', None]) -> None:
+        TESTMAC = '02:aa:bb:cc:dd:ee'
+        TESTIP = ipaddress.ip_interface('fe80::aa:bbff:fecc:ddee/64')
+
+        with veth_setup() as (ns1, ns2):
+            ns1.fg('ip', 'link', 'set', 'dev', 'vetha', 'address', TESTMAC,
+                   privilege=True)
+
+            ip.ifup(ns1, 'vetha', dad=dad)
+            ip.ifup(ns2, 'vethb')
+
+            addrs = ip.addr_wait(ns1, 'vetha', family='inet6', scope='link')
+        exeter.assert_eq(addrs, [TESTIP])
+
+    @exeter.test
+    def test_dad() -> None:
+        test_slaac(dad=None)
+
+    @exeter.test
+    def test_optimistic_dad() -> None:
+        test_slaac(dad='optimistic')
+
+    @exeter.test
+    def test_no_dad() -> None:
+        test_slaac(dad='disable')
-- 
2.46.0