X-Git-Url: https://tinc-vpn.org/git/browse?p=tinc;a=blobdiff_plain;f=test%2Fintegration%2Ftestlib%2Fexternal.py;fp=test%2Fintegration%2Ftestlib%2Fexternal.py;h=af330601669b2ea26743a6f27c5c4044c40e6e39;hp=b90d779f5e4b1f367e670097ab0bd973466ef290;hb=c8402791b82947c49ba1d04f855dab04191607ca;hpb=66eb66ec8f872db3dc12e1d01101772918d69a4a

diff --git a/test/integration/testlib/external.py b/test/integration/testlib/external.py
index b90d779f..af330601 100755
--- a/test/integration/testlib/external.py
+++ b/test/integration/testlib/external.py
@@ -5,16 +5,22 @@ import atexit
 import typing as T
 
 from .log import log
+from .util import random_string
 
 _netns_created: T.Set[str] = set()
+_iface_created: T.Set[str] = set()
 
 
-def _netns_cleanup() -> None:
+def _cleanup() -> None:
     for namespace in _netns_created.copy():
         netns_delete(namespace)
 
+    # Ignore errors since device may have been moved to a different netns
+    for iface in _iface_created.copy():
+        subp.run(["ip", "link", "delete", iface], check=False)
 
-atexit.register(_netns_cleanup)
+
+atexit.register(_cleanup)
 
 
 def _netns_action(action: str, namespace: str) -> bool:
@@ -43,3 +49,46 @@ def netns_add(namespace: str) -> bool:
     if success:
         _netns_created.add(namespace)
     return success
+
+
+def netns_exec(netns: str, *args: str, check: bool = False) -> subp.CompletedProcess:
+    """Execute command in the network namespace."""
+    return subp.run(["ip", "netns", "exec", netns, *args], check=check)
+
+
+def ping(address: str, netns: T.Optional[str] = None) -> bool:
+    """Ping the address from inside the network namespace."""
+    args = ["ping", "-l1", "-W1", "-i0.1", "-c10", address]
+    if netns:
+        proc = netns_exec(netns, *args)
+    else:
+        proc = subp.run(args, check=False)
+    return proc.returncode == 0
+
+
+def move_dev(netns: str, device: str, ip_addr: str) -> None:
+    """Move device to the network namespace."""
+    if netns not in _netns_created:
+        netns_add(netns)
+    subp.run(["ip", "link", "set", device, "netns", netns], check=True)
+    netns_exec(netns, "ip", "addr", "add", ip_addr, "dev", device, check=True)
+    netns_exec(netns, "ip", "link", "set", device, "up", check=True)
+
+
+def veth_add(name0: str, name1: str) -> None:
+    """Create a veth link pair."""
+    subp.run(
+        ["ip", "link", "add", name0, "type", "veth", "peer", "name", name1], check=True
+    )
+    _iface_created.add(name0)
+
+
+def link_add(link_type: str) -> str:
+    """Create a virtual link."""
+    name = random_string(10)
+    if link_type in ("tun", "tap"):
+        subp.run(["ip", "tuntap", "add", "mode", link_type, "dev", name], check=True)
+    else:
+        subp.run(["ip", "link", "add", name, "type", link_type], check=True)
+    _iface_created.add(name)
+    return name