tests: update munet to 0.15.4

- add readline and waitline functions for use with popen objects
- other non-topotest (munet native) run changes
  - vm/qemu support booting cloud images (rocky, ubuntu, debian)
  - native topology init commands

Signed-off-by: Christian Hopps <chopps@labn.net>

tests/topotests/munet/base.py

@@ -332,6 +332,10 @@ class Commander:  # pylint: disable=R0904
         self.last = None
         self.exec_paths = {}
 
+        # For running commands one time only (deals with asyncio)
+        self.cmd_once_done = {}
+        self.cmd_once_locks = {}
+
         if not logger:
             logname = f"munet.{self.__class__.__name__.lower()}.{name}"
             self.logger = logging.getLogger(logname)
@@ -1189,7 +1193,7 @@ class Commander:  # pylint: disable=R0904
         return stdout
 
     # Run a command in a new window (gnome-terminal, screen, tmux, xterm)
-    def run_in_window(
+    def run_in_window(  # pylint: disable=too-many-positional-arguments
         self,
         cmd,
         wait_for=False,
@@ -1205,7 +1209,7 @@ class Commander:  # pylint: disable=R0904
 
         Args:
             cmd: string to execute.
-            wait_for: True to wait for exit from command or `str` as channel neme to
+            wait_for: True to wait for exit from command or `str` as channel name to
                 signal on exit, otherwise False
             background: Do not change focus to new window.
             title: Title for new pane (tmux) or window (xterm).
@@ -1405,6 +1409,26 @@ class Commander:  # pylint: disable=R0904
 
         return pane_info
 
+    async def async_cmd_raises_once(self, cmd, **kwargs):
+        if cmd in self.cmd_once_done:
+            return self.cmd_once_done[cmd]
+
+        if cmd not in self.cmd_once_locks:
+            self.cmd_once_locks[cmd] = asyncio.Lock()
+
+        async with self.cmd_once_locks[cmd]:
+            if cmd not in self.cmd_once_done:
+                self.logger.info("Running command once: %s", cmd)
+                self.cmd_once_done[cmd] = await commander.async_cmd_raises(
+                    cmd, **kwargs
+                )
+        return self.cmd_once_done[cmd]
+
+    def cmd_raises_once(self, cmd, **kwargs):
+        if cmd not in self.cmd_once_done:
+            self.cmd_once_done[cmd] = commander.cmd_raises(cmd, **kwargs)
+        return self.cmd_once_done[cmd]
+
     def delete(self):
         """Calls self.async_delete within an exec loop."""
         asyncio.run(self.async_delete())

tests/topotests/munet/munet-schema.json

@@ -117,6 +117,12 @@
               "bios": {
                 "type": "string"
               },
+              "cloud-init": {
+                "type": "boolean"
+              },
+              "cloud-init-disk": {
+                "type": "string"
+              },
               "disk": {
                 "type": "string"
               },
@@ -129,7 +135,7 @@
               "initial-cmd": {
                 "type": "string"
               },
-              "kerenel": {
+              "kernel": {
                 "type": "string"
               },
               "initrd": {
@@ -373,6 +379,12 @@
         "networks-autonumber": {
           "type": "boolean"
         },
+        "initial-setup-cmd": {
+          "type": "string"
+        },
+        "initial-setup-host-cmd": {
+          "type": "string"
+        },
         "networks": {
           "type": "array",
           "items": {
@@ -452,6 +464,12 @@
                   "bios": {
                     "type": "string"
                   },
+                  "cloud-init": {
+                    "type": "boolean"
+                  },
+                  "cloud-init-disk": {
+                    "type": "string"
+                  },
                   "disk": {
                     "type": "string"
                   },
@@ -464,7 +482,7 @@
                   "initial-cmd": {
                     "type": "string"
                   },
-                  "kerenel": {
+                  "kernel": {
                     "type": "string"
                   },
                   "initrd": {

tests/topotests/munet/mutest/userapi.py

@@ -180,7 +180,7 @@ class TestCase:
 
     # sum_hfmt = "{:5.5s} {:4.4s} {:>6.6s} {}"
     # sum_dfmt = "{:5s} {:4.4s} {:^6.6s} {}"
-    sum_fmt = "%-8.8s %4.4s %{}s %6s  %s"
+    sum_fmt = "%-10s %4.4s %{}s %6s  %s"
 
     def __init__(
         self,

tests/topotests/munet/native.py

@@ -24,6 +24,13 @@ import time
 
 from pathlib import Path
 
+
+try:
+    # We only want to require yaml for the gen cloud image feature
+    import yaml
+except ImportError:
+    pass
+
 from . import cli
 from .base import BaseMunet
 from .base import Bridge
@@ -749,9 +756,11 @@ class L3NodeMixin(NodeMixin):
             # Disable IPv6
             self.cmd_raises("sysctl -w net.ipv6.conf.all.autoconf=0")
             self.cmd_raises("sysctl -w net.ipv6.conf.all.disable_ipv6=1")
+            self.cmd_raises("sysctl -w net.ipv6.conf.all.forwarding=0")
         else:
             self.cmd_raises("sysctl -w net.ipv6.conf.all.autoconf=1")
             self.cmd_raises("sysctl -w net.ipv6.conf.all.disable_ipv6=0")
+            self.cmd_raises("sysctl -w net.ipv6.conf.all.forwarding=1")
 
         self.next_p2p_network = ipaddress.ip_network(f"10.254.{self.id}.0/31")
         self.next_p2p_network6 = ipaddress.ip_network(f"fcff:ffff:{self.id:02x}::/127")
@@ -2265,6 +2274,164 @@ class L3QemuVM(L3NodeMixin, LinuxNamespace):
             tid = self.cpu_thread_map[i]
             self.cmd_raises_nsonly(f"taskset -cp {aff} {tid}")
 
+    def _gen_network_config(self):
+        intfs = sorted(self.intfs)
+        if not intfs:
+            return ""
+
+        self.logger.debug("Generating cloud-init interface config")
+        config = {}
+        config["version"] = 2
+        enets = config["ethernets"] = {}
+
+        for ifname in sorted(self.intfs):
+            self.logger.debug("Interface %s", ifname)
+            conn = find_with_kv(self.config["connections"], "name", ifname)
+
+            index = self.config["connections"].index(conn)
+            to = conn["to"]
+            switch = self.unet.switches.get(to)
+            mtu = conn.get("mtu")
+            if not mtu and switch:
+                mtu = switch.config.get("mtu")
+
+            devaddr = conn.get("physical", "")
+            # Eventually we should get the MAC from /sys
+            if not devaddr:
+                mac = self.tapmacs.get(ifname, f"02:aa:aa:aa:{index:02x}:{self.id:02x}")
+                nic = {
+                    "match": {"macaddress": str(mac)},
+                    "set-name": ifname,
+                }
+                if mtu:
+                    nic["mtu"] = str(mtu)
+                enets[f"nic-{ifname}"] = nic
+
+            ifaddr4 = self.get_intf_addr(ifname, ipv6=False)
+            ifaddr6 = self.get_intf_addr(ifname, ipv6=True)
+            if not ifaddr4 and not ifaddr6:
+                continue
+            net = {
+                "dhcp4": False,
+                "dhcp6": False,
+                "accept-ra": False,
+                "addresses": [],
+            }
+            if ifaddr4:
+                net["addresses"].append(str(ifaddr4))
+            if ifaddr6:
+                net["addresses"].append(str(ifaddr6))
+            if switch and hasattr(switch, "is_nat") and switch.is_nat:
+                net["nameservers"] = {"addresses": []}
+                nameservers = net["nameservers"]["addresses"]
+                if hasattr(switch, "ip6_address"):
+                    net["gateway6"] = str(switch.ip6_address)
+                    nameservers.append("2001:4860:4860::8888")
+                if switch.ip_address:
+                    net["gateway4"] = str(switch.ip_address)
+                    nameservers.append("8.8.8.8")
+            enets[ifname] = net
+
+        return yaml.safe_dump(config)
+
+    def _gen_cloud_init(self):
+        qc = self.qemu_config
+        cc = qc.get("console", {})
+        cipath = self.rundir.joinpath("cloud-init.img")
+
+        geniso = get_exec_path_host("genisoimage")
+        if not geniso:
+            mfbin = get_exec_path_host("mkfs.vfat")
+            mcbin = get_exec_path_host("mcopy")
+            assert (
+                mfbin and mcbin
+            ), "genisoimage or mkfs.vfat,mcopy needed to gen cloud-init disk"
+
+        #
+        # cloud-init: meta-data
+        #
+        mdata = f"""
+instance-id: "munet-{self.id}"
+local-hostname: "{self.name}"
+"""
+        #
+        # cloud-init: user-data
+        #
+        ssh_auth_s = ""
+        if bool(self.ssh_keyfile):
+            pubkey = commander.cmd_raises(f"ssh-keygen -y -f {self.ssh_keyfile}")
+            assert pubkey, f"Can't extract public key from {self.ssh_keyfile}"
+            pubkey = pubkey.strip()
+            ssh_auth_s = f'ssh_authorized_keys: ["{pubkey}"]'
+
+        user = cc.get("user", "root")
+        password = cc.get("password", "admin")
+        if user != "root":
+            root_password = "admin"
+        else:
+            root_password = password
+
+        udata = f"""#cloud-config
+disable_root: 0
+ssh_pwauth: 1
+hostname: {self.name}
+runcmd:
+  - systemctl enable serial-getty@ttyS1.service
+  - systemctl start serial-getty@ttyS1.service
+  - systemctl enable serial-getty@ttyS2.service
+  - systemctl start serial-getty@ttyS2.service
+  - systemctl enable serial-getty@hvc0.service
+  - systemctl start serial-getty@hvc0.service
+  - systemctl enable serial-getty@hvc1.service
+  - systemctl start serial-getty@hvc1.service
+users:
+  - name: root
+    lock_passwd: false
+    plain_text_passwd: "{root_password}"
+    {ssh_auth_s}
+"""
+        if user != "root":
+            udata += """
+  - name: {user}
+    lock_passwd: false
+    plain_text_passwd: "{password}"
+    {ssh_auth_s}
+"""
+        #
+        # cloud-init: network-config
+        #
+        ndata = self._gen_network_config()
+
+        #
+        # Generate cloud-init files
+        #
+        cidir = self.rundir.joinpath("ci-data")
+        commander.cmd_raises(f"mkdir -p {cidir}")
+
+        with open(cidir.joinpath("meta-data"), "w+", encoding="utf-8") as f:
+            f.write(mdata)
+        with open(cidir.joinpath("user-data"), "w+", encoding="utf-8") as f:
+            f.write(udata)
+        files = "meta-data user-data"
+        if ndata:
+            files += " network-config"
+            with open(cidir.joinpath("network-config"), "w+", encoding="utf-8") as f:
+                f.write(ndata)
+        if geniso:
+            commander.cmd_raises(
+                f"cd {cidir} && "
+                f'genisoimage -output "{cipath}" -volid cidata'
+                f" -joliet -rock {files}"
+            )
+        else:
+            commander.cmd_raises(f'cd {cidir} && mkfs.vfat -n cidata "{cipath}"')
+            commander.cmd_raises(f'cd {cidir} && mcopy -oi "{cipath}" {files}')
+
+        #
+        # Generate cloud-init disk
+        #
+        return cipath
+
     async def launch(self):
         """Launch qemu."""
         self.logger.info("%s: Launch Qemu", self)
@@ -2367,11 +2534,21 @@ class L3QemuVM(L3NodeMixin, LinuxNamespace):
                 diskpath = os.path.join(self.unet.config_dirname, diskpath)
 
         if dtpl and (not disk or not os.path.exists(diskpath)):
+            basename = os.path.basename(dtpl)
+            confdir = self.unet.config_dirname
+            if re.match("(https|http|ftp|tftp):.*", dtpl):
+                await self.unet.async_cmd_raises_once(
+                    f"cd {confdir} && (test -e {basename} || curl -fLO {dtpl})"
+                )
+                dtplpath = os.path.join(confdir, basename)
+
             if not disk:
-                disk = qc["disk"] = f"{self.name}-{os.path.basename(dtpl)}"
+                disk = qc["disk"] = f"{self.name}-{basename}"
                 diskpath = os.path.join(self.rundir, disk)
+
             if self.path_exists(diskpath):
                 logging.debug("Disk '%s' file exists, using.", diskpath)
+
             else:
                 if dtplpath[0] != "/":
                     dtplpath = os.path.join(self.unet.config_dirname, dtpl)
@@ -2392,11 +2569,15 @@ class L3QemuVM(L3NodeMixin, LinuxNamespace):
                 args.extend(["-device", "ahci,id=ahci"])
                 args.extend(["-device", "ide-hd,bus=ahci.0,drive=sata-disk0"])
 
-        cidiskpath = qc.get("cloud-init-disk")
+        if qc.get("cloud-init"):
-        if cidiskpath:
+            cidiskpath = qc.get("cloud-init-disk")
-            if cidiskpath[0] != "/":
+            if cidiskpath:
-                cidiskpath = os.path.join(self.unet.config_dirname, cidiskpath)
+                if cidiskpath[0] != "/":
-            args.extend(["-drive", f"file={cidiskpath},if=virtio,format=qcow2"])
+                    cidiskpath = os.path.join(self.unet.config_dirname, cidiskpath)
+            else:
+                cidiskpath = self._gen_cloud_init()
+            diskfmt = "qcow2" if str(cidiskpath).endswith("qcow2") else "raw"
+            args.extend(["-drive", f"file={cidiskpath},if=virtio,format={diskfmt}"])
 
         # args.extend(["-display", "vnc=0.0.0.0:40"])
 
@@ -2488,7 +2669,7 @@ class L3QemuVM(L3NodeMixin, LinuxNamespace):
         if use_cmdcon:
             confiles.append("_cmdcon")
 
-        password = cc.get("password", "")
+        password = cc.get("password", "admin")
         if self.disk_created:
             password = cc.get("initial-password", password)
 
@@ -2764,9 +2945,11 @@ ff02::2\tip6-allrouters
                 # Disable IPv6
                 self.cmd_raises("sysctl -w net.ipv6.conf.all.autoconf=0")
                 self.cmd_raises("sysctl -w net.ipv6.conf.all.disable_ipv6=1")
+                self.cmd_raises("sysctl -w net.ipv6.conf.all.forwarding=0")
             else:
                 self.cmd_raises("sysctl -w net.ipv6.conf.all.autoconf=1")
                 self.cmd_raises("sysctl -w net.ipv6.conf.all.disable_ipv6=0")
+                self.cmd_raises("sysctl -w net.ipv6.conf.all.forwarding=1")
 
         # we really need overlay, but overlay-layers (used by overlay-images)
         # counts on things being present in overlay so this temp stuff doesn't work.
@@ -2774,6 +2957,24 @@ ff02::2\tip6-allrouters
         #     # Let's hide podman details
         #     self.tmpfs_mount("/var/lib/containers/storage/overlay-containers")
 
+        def run_init_cmds(unet, key, on_host):
+            cmds = unet.topoconf.get(key, "")
+            cmds = cmds.replace("%CONFIGDIR%", str(unet.config_dirname))
+            cmds = cmds.replace("%RUNDIR%", str(unet.rundir))
+            cmds = cmds.strip()
+            if not cmds:
+                return
+
+            cmds += "\n"
+            c = commander if on_host else unet
+            o = c.cmd_raises(cmds)
+            self.logger.debug(
+                "run_init_cmds (on-host: %s): %s", on_host, cmd_error(0, o, "")
+            )
+
+        run_init_cmds(self, "initial-setup-host-cmd", True)
+        run_init_cmds(self, "initial-setup-cmd", False)
+
         shellopt = self.cfgopt.getoption("--shell")
         shellopt = shellopt if shellopt else ""
         if shellopt == "all" or "." in shellopt.split(","):
@@ -3061,7 +3262,8 @@ done"""
             if not rc:
                 continue
             logging.info("Pulling missing image %s", image)
-            aw = self.rootcmd.async_cmd_raises(f"podman pull {image}")
+
+            aw = self.rootcmd.async_cmd_raises_once(f"podman pull {image}")
             tasks.append(asyncio.create_task(aw))
         if not tasks:
             return

tests/topotests/munet/testing/util.py

@@ -8,12 +8,17 @@
 """Utility functions useful when using munet testing functionailty in pytest."""
 import asyncio
 import datetime
+import fcntl
 import functools
 import logging
+import os
+import re
+import select
 import sys
 import time
 
 from ..base import BaseMunet
+from ..base import Timeout
 from ..cli import async_cli
 
 
@@ -23,6 +28,7 @@ from ..cli import async_cli
 
 
 async def async_pause_test(desc=""):
+    """Pause the running of a test offering options for CLI or PDB."""
     isatty = sys.stdout.isatty()
     if not isatty:
         desc = f" for {desc}" if desc else ""
@@ -49,11 +55,12 @@ async def async_pause_test(desc=""):
 
 
 def pause_test(desc=""):
+    """Pause the running of a test offering options for CLI or PDB."""
     asyncio.run(async_pause_test(desc))
 
 
 def retry(retry_timeout, initial_wait=0, retry_sleep=2, expected=True):
-    """decorator: retry while functions return is not None or raises an exception.
+    """Retry decorated function until it returns None, raises an exception, or timeout.
 
     * `retry_timeout`: Retry for at least this many seconds; after waiting
                        initial_wait seconds
@@ -116,3 +123,91 @@ def retry(retry_timeout, initial_wait=0, retry_sleep=2, expected=True):
         return func_retry
 
     return _retry
+
+
+def readline(f, timeout=None):
+    """Read a line or timeout.
+
+    This function will take over the file object, the file object should not be used
+    outside of calling this function once you begin.
+
+    Return: A line, remaining buffer if EOF (subsequent calls will return ""), or None
+    for timeout.
+    """
+    fd = f.fileno()
+    if not hasattr(f, "munet_non_block_set"):
+        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+        f.munet_non_block_set = True
+        f.munet_lines = []
+        f.munet_buf = ""
+
+    if f.munet_lines:
+        return f.munet_lines.pop(0)
+
+    timeout = Timeout(timeout)
+    remaining = timeout.remaining()
+    while remaining > 0:
+        ready, _, _ = select.select([fd], [], [], remaining)
+        if not ready:
+            return None
+
+        c = f.read()
+        if c is None:
+            logging.error("munet readline: unexpected None during read")
+            return None
+
+        if not c:
+            logging.debug("munet readline: got eof")
+            c = f.munet_buf
+            f.munet_buf = ""
+            return c
+
+        f.munet_buf += c
+        while "\n" in f.munet_buf:
+            a, f.munet_buf = f.munet_buf.split("\n", 1)
+            f.munet_lines.append(a + "\n")
+
+        if f.munet_lines:
+            return f.munet_lines.pop(0)
+
+        remaining = timeout.remaining()
+    return None
+
+
+def waitline(f, regex, timeout=120):
+    """Match a regex within lines from a file with a timeout.
+
+    This function will take over the file object (by calling `readline` above), the file
+    object should not be used outside of calling these functions once you begin.
+
+    Return: the match object or None.
+    """
+    timeo = Timeout(timeout)
+    while not timeo.is_expired():
+        line = readline(f, timeo.remaining())
+        if line is None:
+            break
+
+        if line == "":
+            logging.warning("waitline: got eof while matching '%s'", regex)
+            return None
+
+        assert line[-1] == "\n"
+        line = line[:-1]
+        if not line:
+            continue
+
+        logging.debug("waitline: searching: '%s' for '%s'", line, regex)
+        m = re.search(regex, line)
+        if m:
+            logging.debug("waitline: matched '%s'", m.group(0))
+            return m
+
+    logging.warning(
+        "Timeout while getting output matching '%s' within %ss (actual %ss)",
+        regex,
+        timeout,
+        timeo.elapsed(),
+    )
+    return None