[RFC XEN PATCH 00/13] Running bare-metal tests with "pytest"

[RFC XEN PATCH 00/13] Running bare-metal tests with "pytest"

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Patch series available in this git branch:
https://xenbits.xenproject.org/git-http/people/aperard/xen-unstable.git br.ci.bare-metal-with-pytest-v1

Hi,

I wanted to add some new machine to our GitLab CI, but I didn't want to
duplicate yet another time the existing shell script. They are good to get the
ball rolling but are getting harder to maintain as more test are been added,
and don't share common code between machines.

So, I've start work on something based on `pytest` which I hope will be easier
to maintain and extend and use to add new machines.

It's still Work In Progress.

Here is the current result:
    https://gitlab.com/xen-project/people/anthonyper/xen/-/jobs/16529107619

In this pipeline, the artifact use are a bit different from master; linux is
built with network driver, and vlan; the rootfs is from Marek's patch series,
with networking and dropbear service enabled.

Next:

I'd like to at least have a new fixture that give a started Host that a test
can ssh into. Then maybe a new class "Host" which could have functions for ssh
and getting the serial console output.

Don't go to much into detail review, I'm more interested of a general overview,
and to present what I have so far.

Some runes:
    - uv run mypy .
      to check the code, at least the types used.
      other code check could be added, and run by the pipeline
    - uv run pytest
      to run the whole test suite
    - uv run pytest test_boot.py::test_xen_tools_tests
      to run just one test

I did create an Arch Linux docker image with pytest, but paramiko got too new
to be used, so I've started to use `uv` to deal with the dependencies. `mypy`
always needed to be run with `uv` due to missing types for paramiko (not
packaged in Arch Linux). So the image is useless, and I'll look for a different
image later, for now just Arch Linux with `uv` installed.

Thanks,

Anthony PERARD (13):
  pytest: Add fixture to prepare rootfs
  Creating new python project
  pytest: Add fixture to prepare the tftp server
  pytest: Add fixture that boot a machine
  pytest: Introduce reading the console, and add a test
  We need an older version of paramiko.
  pytest: Add random string to look for in the logs
  pytest: Run run-tools-tests over ssh on the test machine
  pytest: Generate a rootfs for a guest
  pytest: Add guest test
  pytest: Add config to enable live log, and disable ssh/paramiko
    logging
  pytest: Add example .env for options for the tests
  gitlab-ci: Add job for moonshot with pytest

 automation/gitlab-ci/test.yaml       |  21 +++
 automation/pytest/.env               |  19 ++
 automation/pytest/conftest.py        |   6 +
 automation/pytest/lib/__init__.py    |   0
 automation/pytest/lib/boot_binary.py | 261 +++++++++++++++++++++++++++
 automation/pytest/lib/ipmi.py        | 155 ++++++++++++++++
 automation/pytest/lib/net_boot.py    |  60 ++++++
 automation/pytest/pyproject.toml     |  17 ++
 automation/pytest/pytest.toml        |  13 ++
 automation/pytest/test_boot.py       | 237 ++++++++++++++++++++++++
 10 files changed, 789 insertions(+)
 create mode 100644 automation/pytest/.env
 create mode 100644 automation/pytest/conftest.py
 create mode 100644 automation/pytest/lib/__init__.py
 create mode 100644 automation/pytest/lib/boot_binary.py
 create mode 100644 automation/pytest/lib/ipmi.py
 create mode 100644 automation/pytest/lib/net_boot.py
 create mode 100644 automation/pytest/pyproject.toml
 create mode 100644 automation/pytest/pytest.toml
 create mode 100644 automation/pytest/test_boot.py

-- 
Anthony PERARD

[RFC XEN PATCH 01/13] pytest: Add fixture to prepare rootfs

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

It's slyghtly more complicated than the QubeOS's test script as we try
here to create a small rootfs with just enough for setting up the
network, then once Linux is booted we load the rest, that is Xen's
toolstack.

Loading the full rootfs at netboot on the moonshot takes more than 10
minutes, while download it once Linux is booted is much faster.

The network configuration should be flexible enough, we can select
static IP or DHCP, and setup VLAN if needed. I've mainly only tested
static IP + VLAN.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/lib/__init__.py    |   0
 automation/pytest/lib/boot_binary.py | 212 +++++++++++++++++++++++++++
 automation/pytest/test_boot.py       |  32 ++++
 3 files changed, 244 insertions(+)
 create mode 100644 automation/pytest/lib/__init__.py
 create mode 100644 automation/pytest/lib/boot_binary.py
 create mode 100644 automation/pytest/test_boot.py

diff --git a/automation/pytest/lib/__init__.py b/automation/pytest/lib/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/automation/pytest/lib/boot_binary.py b/automation/pytest/lib/boot_binary.py
new file mode 100644
index 000000000000..d8f503ead9f2
--- /dev/null
+++ b/automation/pytest/lib/boot_binary.py
@@ -0,0 +1,212 @@
+from pathlib import Path
+import shlex
+import stat
+import subprocess
+from textwrap import dedent
+
+
+def cpio_archive_from_dir(root: Path, output: Path) -> None:
+    subprocess.run(f"find . | cpio -H newc --owner=0:0 -o | gzip >> {output}",
+                   shell=True,
+                   cwd=root,
+                   check=True)
+
+
+def concatenate_cpios(archives: list[Path], output: Path) -> None:
+    paths = shlex.join([str(p) for p in archives])
+    subprocess.run(f"cat {paths} > {output}",
+                   shell=True,
+                   check=True)
+
+
+class BootBinary:
+    def __init__(self, tmp_path: Path, xenroot: Path,
+                 pxe_host: str,
+                 http_pxe_path: str,
+                 vlan_id: int | None = None,
+                 static_ip: str | None = None,
+                 static_gateway: str | None = None,
+                 static_dns: str | None = None,
+                 static_ipv6: str | None = None,
+                 boot_script: str | None = None,
+                 ) -> None:
+        self.vlan_id = vlan_id
+        self.pxe_host = pxe_host
+        self.http_pxe_path = http_pxe_path
+        self.static_ip = static_ip
+        self.static_gateway = static_gateway
+        self.static_dns = static_dns
+        self.static_ipv6 = static_ipv6
+        self.boot_script = boot_script
+
+        # Minimal rootfs
+        rootfs_path = tmp_path / "rootfs_minimal"
+        rootfs_path.mkdir()
+        self.prepare_minimal_rootfs(rootfs_path, xenroot)
+        self.rootfs_dir = rootfs_path
+
+        # Rootfs for dom0
+        rootfs_path = tmp_path / "rootfs_dom0"
+        rootfs_path.mkdir()
+        self.prepare_dom0_rootfs(rootfs_path, xenroot)
+        self.rootfs_dom0_dir = rootfs_path
+
+
+    def prepare_minimal_rootfs(self, rootfs_path: Path, xenroot: Path) -> None:
+        script_pre_xen = dedent(f"""\
+            #!/bin/ash
+            set -xe
+
+            wget http://{self.pxe_host}/{self.http_pxe_path}/dom0-rootfs.cpio.gz -O /root/dom0-rootfs.cpio.gz
+            cd /
+            gzip -d < /root/dom0-rootfs.cpio.gz | while cpio -iu; do :; done
+            /etc/local.d/xen.start
+        """)
+
+        bridge_port = "eth0"
+        cfg_vlan = ""
+        if self.vlan_id:
+            bridge_port = f"eth0.{self.vlan_id}"
+            cfg_vlan = f"""\
+                auto {bridge_port}
+                iface {bridge_port} inet manual
+                    vlan-raw-device eth0
+                    # Remove all ipv6 route, they are duplicates from
+                    # the one added to the bridge interface
+                    post-up ip -6 route flush dev $IFACE
+            """
+
+        cfg_addr = ""
+        if self.static_ip and self.static_gateway:
+            cfg_addr = f"""\
+                use static
+                address {self.static_ip}
+                gateway {self.static_gateway}
+            """
+        else:
+            cfg_addr = """\
+                use dhcp
+                """
+
+        if self.static_ip and not self.static_gateway:
+            cfg_addr += f"""\
+                address {self.static_ip}
+            """
+
+        cfg_addr_v6 = ""
+        if self.static_ipv6:
+            cfg_addr_v6 = f"""\
+                address {self.static_ipv6}
+            """
+
+        network_interfaces = dedent(f"""\
+            {cfg_vlan}
+
+            auto xenbr0
+            iface xenbr0
+                {cfg_addr}
+                bridge-ports {bridge_port}
+                bridge-stp 0
+                {cfg_addr_v6}
+        """)
+
+        etc_networkd_d = rootfs_path / "etc/network"
+        etc_networkd_d.mkdir(parents=True)
+        (etc_networkd_d / "interfaces").write_text(network_interfaces)
+
+        # if static
+        if self.static_dns:
+            dns_config = dedent(f"""\
+                nameserver {self.static_dns}
+            """)
+            (rootfs_path / "etc/resolv.conf").write_text(dns_config)
+
+        # rc-update add networking boot
+        runlevel_boot_d = rootfs_path / "etc/runlevels/boot"
+        runlevel_boot_d.mkdir(parents=True)
+        (runlevel_boot_d / "networking").symlink_to("/etc/init.d/networking")
+
+        local_d_dir = rootfs_path / "etc/local.d"
+        local_d_dir.mkdir(parents=True)
+
+        pre_xen_start_script = local_d_dir / "pre-xen.start"
+        pre_xen_start_script.write_text(script_pre_xen)
+        pre_xen_start_script.chmod(stat.S_IRUSR | stat.S_IXUSR)
+
+        # rootfs archive
+        rootfs = []
+        # Uncompressed first! Then compressed cpio archives!
+        rootfs.append(xenroot / "binaries/ucode.cpio")
+        rootfs.append(xenroot / "binaries/rootfs.cpio.gz")
+        cpio_rootfs = xenroot / "binaries/pre-dom0-rootfs.cpio.gz"
+        concatenate_cpios(rootfs, cpio_rootfs)
+        cpio_archive_from_dir(rootfs_path, cpio_rootfs)
+
+        self.minimal_rootfs = cpio_rootfs
+
+    def prepare_dom0_rootfs(self, rootfs_path: Path, xenroot: Path) -> None:
+        etc = rootfs_path / "etc"
+        etc.mkdir()
+
+        local_d_dir = etc / "local.d"
+        local_d_dir.mkdir()
+
+        script_xen = dedent("""\
+            #!/bin/bash
+
+            bash /etc/init.d/xencommons start
+        """)
+        if self.boot_script:
+            script_xen += self.boot_script
+
+        xen_start_script = local_d_dir / "xen.start"
+        xen_start_script.write_text(script_xen)
+        xen_start_script.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+
+        etc_default = etc / "default"
+        etc_default.mkdir()
+
+        xencommons_default = etc_default / "xencommons"
+        xencommons_default.write_text(dedent("""
+            XENCONSOLED_TRACE=all
+            QEMU_XEN=/bin/false
+            """))
+
+        (rootfs_path / "var/log/xen/console").mkdir(parents=True)
+
+        root_home_dir = rootfs_path / "root"
+        root_home_dir.mkdir(parents=True)
+        run_tool_tests_script = xenroot / "automation/scripts/run-tools-tests"
+        run_tool_tests_script.copy_into(root_home_dir, preserve_metadata=True)
+
+        # VM config file
+        etc_xen = etc / "xen"
+        etc_xen.mkdir()
+        domU_type = "hvm"
+        domU_vif = "'bridge=xenbr0'"
+        (etc_xen / "domU.cfg").write_text(dedent(f"""\
+            type = '{domU_type}'
+            name = 'domU'
+            kernel = '/boot/vmlinuz-domU'
+            ramdisk = '/boot/initrd-domU'
+            cmdline = 'root=/dev/ram0 console=hvc0'
+            memory = 512
+            vif = [ {domU_vif} ]
+            disk = [ ]
+            """))
+
+        # rootfs archive
+        rootfs_dom0 = []
+        # Uncompressed first! Then compressed cpio archives!
+        rootfs_dom0.append(xenroot / "binaries/xen-tools.cpio.gz")
+        cpio_rootfs = xenroot / "binaries/dom0-rootfs.cpio.gz"
+        concatenate_cpios(rootfs_dom0, cpio_rootfs)
+        cpio_archive_from_dir(rootfs_path, cpio_rootfs)
+
+        self.xen_tools_rootfs = cpio_rootfs
+
+    def minimal(self) -> Path:
+        return self.minimal_rootfs
+
+    def dom0_tools(self) -> Path:
+        return self.xen_tools_rootfs
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
new file mode 100644
index 000000000000..230cee96d0ba
--- /dev/null
+++ b/automation/pytest/test_boot.py
@@ -0,0 +1,32 @@
+import os
+from pathlib import Path
+import pytest
+
+from lib.boot_binary import BootBinary
+
+
+@pytest.fixture
+def xenroot_path() -> Path:
+    return Path("../..").resolve()
+
+
+@pytest.fixture
+def prepare_rootfs(tmp_path: Path,
+                   xenroot_path: Path,
+                   ) -> BootBinary:
+    static_gateway = os.environ['HOST_IP_GATEWAY']
+    http_pxe_path = os.environ["PXE_HTTP_PATH"]
+
+    boot_script = f"echo Dom0 ready\n"
+
+    return BootBinary(tmp_path,
+                      xenroot_path,
+                      vlan_id=int(os.environ['HOST_IP_VLANID']),
+                      pxe_host=os.environ["PXE_HOST"],
+                      http_pxe_path=http_pxe_path,
+                      static_ip=os.environ['HOST_IP4'],
+                      static_gateway=static_gateway,
+                      static_dns=static_gateway,
+                      static_ipv6=os.environ['HOST_IP6'],
+                      boot_script=boot_script,
+                      )
-- 
Anthony PERARD

[RFC XEN PATCH 02/13] Creating new python project

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

This can by use by `uv` for example.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
uv run mypy .
---
 automation/pytest/pyproject.toml | 12 ++++++++++++
 1 file changed, 12 insertions(+)
 create mode 100644 automation/pytest/pyproject.toml

diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
new file mode 100644
index 000000000000..3180b2555029
--- /dev/null
+++ b/automation/pytest/pyproject.toml
@@ -0,0 +1,12 @@
+[project]
+name = "xen-ci-intree"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = [
+    "pip>=26.0.1",
+    "pytest>=9.0.2",
+]
+[dependency-groups]
+dev = [
+    "mypy",
+]
-- 
Anthony PERARD

[RFC XEN PATCH 03/13] pytest: Add fixture to prepare the tftp server

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Prepare the netboot config, and upload the necessary files.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/lib/net_boot.py | 60 +++++++++++++++++++++++++++++++
 automation/pytest/pyproject.toml  |  2 ++
 automation/pytest/test_boot.py    | 11 ++++++
 3 files changed, 73 insertions(+)
 create mode 100644 automation/pytest/lib/net_boot.py

diff --git a/automation/pytest/lib/net_boot.py b/automation/pytest/lib/net_boot.py
new file mode 100644
index 000000000000..50d2cfaa5aaa
--- /dev/null
+++ b/automation/pytest/lib/net_boot.py
@@ -0,0 +1,60 @@
+import io
+import logging
+import os
+import paramiko
+from pathlib import Path
+import textwrap
+
+
+class NetBoot:
+    def grub_script(self) -> str:
+        http_pxe_path = os.environ['PXE_HTTP_PATH']
+        tftp_path = f"(http)/{http_pxe_path}"
+        console_opts = "console=vga,com1"
+
+        xen_cmdline_opts = [
+            console_opts,
+            "loglvl=all",
+            "guest_loglvl=all",
+            "dom0_mem=4G",
+            "console_timestamps=boot",
+            "watchdog",
+            "noreboot",
+            ]
+        xen_cmdline = " ".join(xen_cmdline_opts)
+        linux_cmdline = "console=hvc0 root=/dev/ram0 earlyprintk=xen"
+        script = textwrap.dedent(f"""\
+            echo \"Loading xen\"
+            multiboot2 {tftp_path}/xen {xen_cmdline}
+            echo \"Loading vmlinuz\"
+            module2 {tftp_path}/bzImage {linux_cmdline}
+            echo \"Loading initrd\"
+            module2 --nounzip {tftp_path}/pre-dom0-rootfs.cpio.gz
+            echo \"Booting\"
+            boot
+            """)
+
+        return script
+
+    def prepare_tftp(self, xenroot: Path, rootfss: list[Path]) -> None:
+        username = os.environ['PXE_SSH_USER']
+        hostname = os.environ['PXE_HOST']
+        remote_path = Path(os.environ['PXE_SSH_PATH'])
+
+        binaries_path = xenroot / "binaries"
+        binaries = []
+        for b1 in ["xen", "bzImage"]:
+            binaries.append(binaries_path / b1)
+        binaries += rootfss
+
+        ssh = paramiko.client.SSHClient()
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        ssh.connect(hostname, username=username)
+        with ssh.open_sftp() as sftp:
+            for b2 in binaries:
+                logging.debug(f" put {b2}, to {remote_path / b2.name}")
+                sftp.put(str(b2), str(remote_path / b2.name), confirm=True)
+            grub_script = io.BytesIO(self.grub_script().encode('utf-8'))
+            sftp.putfo(grub_script, str(remote_path / "grub.cfg"),
+                       confirm=True)
+        ssh.close()
diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
index 3180b2555029..78f60c4f8e77 100644
--- a/automation/pytest/pyproject.toml
+++ b/automation/pytest/pyproject.toml
@@ -3,8 +3,10 @@ name = "xen-ci-intree"
 version = "0.1.0"
 requires-python = ">=3.14"
 dependencies = [
+    "paramiko>=4.0.0",
     "pip>=26.0.1",
     "pytest>=9.0.2",
+    "types-paramiko>=4.0.0.20260322",
 ]
 [dependency-groups]
 dev = [
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 230cee96d0ba..661c2a59e4fe 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -3,6 +3,7 @@ from pathlib import Path
 import pytest
 
 from lib.boot_binary import BootBinary
+from lib.net_boot import NetBoot
 
 
 @pytest.fixture
@@ -30,3 +31,13 @@ def prepare_rootfs(tmp_path: Path,
                       static_ipv6=os.environ['HOST_IP6'],
                       boot_script=boot_script,
                       )
+
+
+@pytest.fixture
+def prepare_netboot(prepare_rootfs: BootBinary, xenroot_path: Path) -> None:
+    netboot = NetBoot()
+    rootfs_paths = [
+        prepare_rootfs.minimal(),
+        prepare_rootfs.dom0_tools(),
+    ]
+    netboot.prepare_tftp(xenroot_path, rootfs_paths)
-- 
Anthony PERARD

[RFC XEN PATCH 04/13] pytest: Add fixture that boot a machine

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

.. and power it off once a test using the fixture is done.

This IPMI class is using the `ssh` console of an HP Moonshot 1500
Chassis.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
Maybe we could rename the class to Machine as a lib to control the
machine behavior, and access it's console.
---
 automation/pytest/lib/ipmi.py  | 69 ++++++++++++++++++++++++++++++++++
 automation/pytest/test_boot.py | 17 +++++++++
 2 files changed, 86 insertions(+)
 create mode 100644 automation/pytest/lib/ipmi.py

diff --git a/automation/pytest/lib/ipmi.py b/automation/pytest/lib/ipmi.py
new file mode 100644
index 000000000000..0d3b085f704c
--- /dev/null
+++ b/automation/pytest/lib/ipmi.py
@@ -0,0 +1,69 @@
+import logging
+import paramiko
+import re
+
+
+class IPMI:
+    def __init__(self, *,
+                 hostname: str,
+                 username: str,
+                 password: str,
+                 blade_node: str,
+                 ) -> None:
+        self.hostname = hostname
+        self.username = username
+        self.password = password
+        self.blade_node = blade_node
+
+        self.channel = None
+
+    def _new_client(self):
+        ssh = paramiko.client.SSHClient()
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        ssh.connect(self.hostname,
+                    username=self.username,
+                    password=self.password)
+        return ssh
+
+    def _run_cmd(self, command: str, debug_print: bool = True) -> str:
+        with self._new_client() as ssh:
+            stdin, stdout, stderr = ssh.exec_command(command)
+            output = stdout.read().decode()
+            if debug_print:
+                for line in output.splitlines():
+                    if line in ['']:
+                        continue
+                    logging.debug(line.rstrip())
+            # the command's exit status (0 = success, 1 = failure)
+            # stdout.channel.recv_exit_status()
+            return output
+
+    def _power(self, state: str) -> None:
+        assert state in ["on", "off force"]
+        logging.info(f"Set power {state.split(' ')[0]} {self.blade_node}")
+        command = f"set node power {state} {self.blade_node}"
+        output = self._run_cmd(command, debug_print=False)
+        m = re.search(f"{self.blade_node}: #Node 1 (.*)\\r$",
+                      output,
+                      flags=re.M)
+        if m:
+            logging.info(f"-> {m[1]}")
+        else:
+            logging.warning(f"Failed to parse 'set power' command output: '{output}'")
+
+    def power_status(self):
+        command = f"show node power {self.blade_node}"
+        output = self._run_cmd(command, debug_print=False)
+        m = re.search(f"{self.blade_node}: #Node 1\\r\\n\\s+Power State: (\\S+)\\r$",
+                      output,
+                      flags=re.M)
+        if m:
+            logging.info(f"Machine is {m[1]}")
+        else:
+            logging.warning(f"Failed to match for \"Power State\" in '{output}'")
+
+    def power_on(self):
+        self._power('on')
+
+    def power_off(self):
+        self._power('off force')
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 661c2a59e4fe..b7610736fa25 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -1,9 +1,11 @@
 import os
 from pathlib import Path
 import pytest
+from typing import Generator
 
 from lib.boot_binary import BootBinary
 from lib.net_boot import NetBoot
+from lib.ipmi import IPMI
 
 
 @pytest.fixture
@@ -41,3 +43,18 @@ def prepare_netboot(prepare_rootfs: BootBinary, xenroot_path: Path) -> None:
         prepare_rootfs.dom0_tools(),
     ]
     netboot.prepare_tftp(xenroot_path, rootfs_paths)
+
+
+@pytest.fixture
+def machine_on_ipmi(prepare_netboot: None) -> Generator[IPMI]:
+    ipmi = IPMI(
+            hostname=os.environ['IPMI_HOSTNAME'],
+            username=os.environ['MOON_CONSOLE_USER'],
+            password=os.environ['MOON_CONSOLE_PASS'],
+            blade_node=os.environ['IPMI_BLADE_NODE'],
+            )
+    ipmi.power_off()
+    ipmi.power_status()
+    ipmi.power_on()
+    yield ipmi
+    ipmi.power_off()
-- 
Anthony PERARD

[RFC XEN PATCH 05/13] pytest: Introduce reading the console, and add a test

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

The test will do more than just checking dom0 booted in a later patch.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/lib/ipmi.py    | 86 ++++++++++++++++++++++++++++++++
 automation/pytest/pyproject.toml |  1 +
 automation/pytest/test_boot.py   | 24 +++++++++
 3 files changed, 111 insertions(+)

diff --git a/automation/pytest/lib/ipmi.py b/automation/pytest/lib/ipmi.py
index 0d3b085f704c..bda5d8077577 100644
--- a/automation/pytest/lib/ipmi.py
+++ b/automation/pytest/lib/ipmi.py
@@ -1,6 +1,8 @@
 import logging
 import paramiko
 import re
+import time
+from typing import Generator
 
 
 class IPMI:
@@ -67,3 +69,87 @@ class IPMI:
 
     def power_off(self):
         self._power('off force')
+
+    def get_console(self):
+        logging.info(f"Start console acquisition on {self.blade_node}")
+        ssh = self._new_client()
+        channel = ssh.invoke_shell()
+
+        output = ""
+        logging.debug("wait for prompt")
+        while True:
+            if channel.recv_ready():
+                output += channel.recv(9999).decode('utf-8')
+                # wait for prompt
+                logging.debug(output)
+                if output.strip().endswith('hpiLO->'):
+                    logging.debug("got prompt")
+                    break
+            elif channel.exit_status_ready():
+                raise Exception("Shell closed unexpectedly.")
+            else:
+                time.sleep(0.1)  # Wait a bit for more output
+
+        logging.debug("run connect node vsp acquire")
+        cmd = f"connect node vsp acquire {self.blade_node}\n"
+        channel.send(cmd)
+
+        self.channel = channel
+
+    def console_output_generator(self) -> Generator[str]:
+        if not self.channel:
+            raise Exception("Must call get_console")
+        channel = self.channel
+        previous_output = ''
+        # `count` when this reach 0, we had a line from the console that isn't
+        # finished, like a prompt, for sometime already, so display it in the
+        # debug.
+        count = 0
+        ansi_escape = re.compile(r'''
+            \x1B  # ESC
+            (?:   # 7-bit C1 Fe (except CSI)
+                [@-Z\\-_]
+            |     # or [ for CSI, followed by a control sequence
+                \[
+                [0-?]*  # Parameter bytes
+                [ -/]*  # Intermediate bytes
+                [@-~]   # Final byte
+            )
+        ''', re.VERBOSE)
+        while True:
+            if channel.recv_ready():
+                output = previous_output
+                output += channel.recv(9999).decode('utf-8')
+                output_lines = output.splitlines(keepends=True)
+                for line in output_lines[:-1]:
+                    line = ansi_escape.sub('', line)
+                    line = line.rstrip()
+                    if line in ['', '\r', '\r\n']:
+                        continue
+                    logging.debug(line)
+                    yield line
+                previous_output = output_lines[-1]
+                if previous_output.endswith('\n'):
+                    line = previous_output
+                    line = ansi_escape.sub('', line)
+                    line = line.rstrip()
+                    if line not in ['', '\r', '\r\n']:
+                        logging.debug(line)
+                    previous_output = ''
+                    count = 0
+                else:
+                    count = 20
+                # previous_output = output[-len(expect):]
+            # Check if the shell closed unexpectedly
+            elif channel.exit_status_ready():
+                raise Exception("Shell closed unexpectedly.")
+            else:
+                if count > 1:
+                    count -= 1
+                elif count == 1:
+                    line = previous_output
+                    line = ansi_escape.sub('', line)
+                    if line not in ['\r']:
+                        logging.debug("XXXXX output buffer: %s XXXXXX" % repr(line))
+                    count -= 1
+                time.sleep(0.5)  # Wait a bit for more output
diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
index 78f60c4f8e77..4ab7094e0d84 100644
--- a/automation/pytest/pyproject.toml
+++ b/automation/pytest/pyproject.toml
@@ -6,6 +6,7 @@ dependencies = [
     "paramiko>=4.0.0",
     "pip>=26.0.1",
     "pytest>=9.0.2",
+    "pytest-timeout>=2.4.0",
     "types-paramiko>=4.0.0.20260322",
 ]
 [dependency-groups]
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index b7610736fa25..158f1a30dcb6 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -1,3 +1,4 @@
+import logging
 import os
 from pathlib import Path
 import pytest
@@ -53,8 +54,31 @@ def machine_on_ipmi(prepare_netboot: None) -> Generator[IPMI]:
             password=os.environ['MOON_CONSOLE_PASS'],
             blade_node=os.environ['IPMI_BLADE_NODE'],
             )
+    # start console
+    ipmi.get_console()
     ipmi.power_off()
     ipmi.power_status()
     ipmi.power_on()
     yield ipmi
     ipmi.power_off()
+
+
+@pytest.mark.timeout(600)
+def test_xen_tools_tests(machine_on_ipmi: IPMI,
+                         ) -> None:
+    expect_string = "Dom0 ready"
+    for line in machine_on_ipmi.console_output_generator():
+        #
+        # Wait for `expect_string` for the test to succeed
+        #
+        if expect_string in line:
+            logging.info(f"got expect_string '{line}'.")
+            break
+        #
+        # Some common expected or not console outputs.
+        #
+        if "Latest ChangeSet:" in line:
+            logging.info(f"Test xen build from '{line}'.")
+        if "Manual reset required ('noreboot' specified)" in line:
+            # Test failed, could wait a bit before failing the test.
+            logging.error("Xen panic")
-- 
Anthony PERARD

[RFC XEN PATCH 06/13] We need an older version of paramiko.

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

To connect to the moonshot controller over ssh.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/pyproject.toml | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
index 4ab7094e0d84..c54c1338f656 100644
--- a/automation/pytest/pyproject.toml
+++ b/automation/pytest/pyproject.toml
@@ -3,7 +3,8 @@ name = "xen-ci-intree"
 version = "0.1.0"
 requires-python = ">=3.14"
 dependencies = [
-    "paramiko>=4.0.0",
+    # Need paramiko with support for SHA-1 key exchange
+    "paramiko<5.0.0",
     "pip>=26.0.1",
     "pytest>=9.0.2",
     "pytest-timeout>=2.4.0",
-- 
Anthony PERARD

[RFC XEN PATCH 07/13] pytest: Add random string to look for in the logs

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

This allow to check that we are testing the expected rootfs.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/test_boot.py | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 158f1a30dcb6..90d8e00e90d9 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -2,6 +2,8 @@ import logging
 import os
 from pathlib import Path
 import pytest
+import random
+import string
 from typing import Generator
 
 from lib.boot_binary import BootBinary
@@ -14,14 +16,20 @@ def xenroot_path() -> Path:
     return Path("../..").resolve()
 
 
+@pytest.fixture
+def boot_string() -> str:
+    return ''.join(random.choice(string.ascii_letters) for _ in range(16))
+
+
 @pytest.fixture
 def prepare_rootfs(tmp_path: Path,
                    xenroot_path: Path,
+                   boot_string: str,
                    ) -> BootBinary:
     static_gateway = os.environ['HOST_IP_GATEWAY']
     http_pxe_path = os.environ["PXE_HTTP_PATH"]
 
-    boot_script = f"echo Dom0 ready\n"
+    boot_script = f"echo Dom0 ready {boot_string}\n"
 
     return BootBinary(tmp_path,
                       xenroot_path,
@@ -65,8 +73,9 @@ def machine_on_ipmi(prepare_netboot: None) -> Generator[IPMI]:
 
 @pytest.mark.timeout(600)
 def test_xen_tools_tests(machine_on_ipmi: IPMI,
+                         boot_string: str,
                          ) -> None:
-    expect_string = "Dom0 ready"
+    expect_string = boot_string
     for line in machine_on_ipmi.console_output_generator():
         #
         # Wait for `expect_string` for the test to succeed
-- 
Anthony PERARD

[RFC XEN PATCH 08/13] pytest: Run run-tools-tests over ssh on the test machine

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Add a new random ssh public key to the rootfs.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/lib/boot_binary.py |  7 +++++
 automation/pytest/test_boot.py       | 39 ++++++++++++++++++++++++++++
 2 files changed, 46 insertions(+)

diff --git a/automation/pytest/lib/boot_binary.py b/automation/pytest/lib/boot_binary.py
index d8f503ead9f2..1910d1f88fe6 100644
--- a/automation/pytest/lib/boot_binary.py
+++ b/automation/pytest/lib/boot_binary.py
@@ -23,6 +23,7 @@ class BootBinary:
     def __init__(self, tmp_path: Path, xenroot: Path,
                  pxe_host: str,
                  http_pxe_path: str,
+                 ssh_pub_keys: str | None = None,
                  vlan_id: int | None = None,
                  static_ip: str | None = None,
                  static_gateway: str | None = None,
@@ -37,6 +38,7 @@ class BootBinary:
         self.static_gateway = static_gateway
         self.static_dns = static_dns
         self.static_ipv6 = static_ipv6
+        self.ssh_pub_keys = ssh_pub_keys
         self.boot_script = boot_script
 
         # Minimal rootfs
@@ -126,6 +128,11 @@ class BootBinary:
         runlevel_boot_d.mkdir(parents=True)
         (runlevel_boot_d / "networking").symlink_to("/etc/init.d/networking")
 
+        if self.ssh_pub_keys:
+            home_root_ssh = rootfs_path / "root/.ssh"
+            home_root_ssh.mkdir(parents=True)
+            (home_root_ssh / "authorized_keys").write_text(self.ssh_pub_keys)
+
         local_d_dir = rootfs_path / "etc/local.d"
         local_d_dir.mkdir(parents=True)
 
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 90d8e00e90d9..039cc36c1a04 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import paramiko
 from pathlib import Path
 import pytest
 import random
@@ -16,6 +17,11 @@ def xenroot_path() -> Path:
     return Path("../..").resolve()
 
 
+@pytest.fixture
+def sshkey_for_test() -> paramiko.pkey.PKey:
+    return paramiko.ecdsakey.ECDSAKey.generate()
+
+
 @pytest.fixture
 def boot_string() -> str:
     return ''.join(random.choice(string.ascii_letters) for _ in range(16))
@@ -24,6 +30,7 @@ def boot_string() -> str:
 @pytest.fixture
 def prepare_rootfs(tmp_path: Path,
                    xenroot_path: Path,
+                   sshkey_for_test: paramiko.pkey.PKey,
                    boot_string: str,
                    ) -> BootBinary:
     static_gateway = os.environ['HOST_IP_GATEWAY']
@@ -31,6 +38,9 @@ def prepare_rootfs(tmp_path: Path,
 
     boot_script = f"echo Dom0 ready {boot_string}\n"
 
+    ssh_pub_keys = os.environ['SSH_PUB_KEYS']
+    ssh_pub_keys += f"\n{sshkey_for_test.get_name()} {sshkey_for_test.get_base64()}\n"
+
     return BootBinary(tmp_path,
                       xenroot_path,
                       vlan_id=int(os.environ['HOST_IP_VLANID']),
@@ -40,6 +50,7 @@ def prepare_rootfs(tmp_path: Path,
                       static_gateway=static_gateway,
                       static_dns=static_gateway,
                       static_ipv6=os.environ['HOST_IP6'],
+                      ssh_pub_keys=ssh_pub_keys,
                       boot_script=boot_script,
                       )
 
@@ -73,6 +84,7 @@ def machine_on_ipmi(prepare_netboot: None) -> Generator[IPMI]:
 
 @pytest.mark.timeout(600)
 def test_xen_tools_tests(machine_on_ipmi: IPMI,
+                         sshkey_for_test: paramiko.pkey.PKey,
                          boot_string: str,
                          ) -> None:
     expect_string = boot_string
@@ -91,3 +103,30 @@ def test_xen_tools_tests(machine_on_ipmi: IPMI,
         if "Manual reset required ('noreboot' specified)" in line:
             # Test failed, could wait a bit before failing the test.
             logging.error("Xen panic")
+    ssh = paramiko.client.SSHClient()
+    ssh.set_missing_host_key_policy(paramiko.client.WarningPolicy)
+    logging.debug(f"connecting to {os.environ['HOST_IP4'].split('/')[0]}")
+    ssh.connect(os.environ['HOST_IP4'].split('/')[0],
+                username='root',
+                pkey=sshkey_for_test,
+                timeout=10,
+                )
+
+    cmd = '/root/run-tools-tests /usr/lib/xen/tests /tmp/tests-junit.xml'
+    logging.info(f"Exec command: {cmd}")
+    stdin, stdout, stderr = ssh.exec_command(cmd)
+    for output in stdout:
+        logging.debug(output.strip())
+    errormsg = stderr.read()
+    if errormsg:
+        logging.info(f"cmd errro: {errormsg!r}")
+    exit_status = stderr.channel.recv_exit_status()
+    logging.debug(f"exit status: {exit_status}")
+
+    cmd = 'cat /tmp/tests-junit.xml'
+    stdin, stdout, stderr = ssh.exec_command(cmd)
+    junit = stdout.read()
+    logging.debug(f"got junit result: {junit!r}")
+    errormsg = stderr.read()
+    if errormsg:
+        logging.info(f"cmd errro: {errormsg!r}")
-- 
Anthony PERARD

[RFC XEN PATCH 09/13] pytest: Generate a rootfs for a guest

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/conftest.py        |  6 ++++
 automation/pytest/lib/boot_binary.py | 42 ++++++++++++++++++++++++++++
 automation/pytest/test_boot.py       |  8 ++++++
 3 files changed, 56 insertions(+)
 create mode 100644 automation/pytest/conftest.py

diff --git a/automation/pytest/conftest.py b/automation/pytest/conftest.py
new file mode 100644
index 000000000000..304241ae06ba
--- /dev/null
+++ b/automation/pytest/conftest.py
@@ -0,0 +1,6 @@
+import pytest
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line("markers",
+                            "guest_boot_script: script to run when the guest boot")
diff --git a/automation/pytest/lib/boot_binary.py b/automation/pytest/lib/boot_binary.py
index 1910d1f88fe6..9e79b1798908 100644
--- a/automation/pytest/lib/boot_binary.py
+++ b/automation/pytest/lib/boot_binary.py
@@ -30,6 +30,7 @@ class BootBinary:
                  static_dns: str | None = None,
                  static_ipv6: str | None = None,
                  boot_script: str | None = None,
+                 guest_boot_script: str | None = None,
                  ) -> None:
         self.vlan_id = vlan_id
         self.pxe_host = pxe_host
@@ -40,6 +41,7 @@ class BootBinary:
         self.static_ipv6 = static_ipv6
         self.ssh_pub_keys = ssh_pub_keys
         self.boot_script = boot_script
+        self.guest_boot_script = guest_boot_script
 
         # Minimal rootfs
         rootfs_path = tmp_path / "rootfs_minimal"
@@ -53,6 +55,12 @@ class BootBinary:
         self.prepare_dom0_rootfs(rootfs_path, xenroot)
         self.rootfs_dom0_dir = rootfs_path
 
+        # Guest kernel + rootfs
+        rootfs_path = tmp_path / "rootfs_domU"
+        rootfs_path.mkdir()
+        self.prepare_guest_rootfs(rootfs_path, xenroot)
+        self.rootfs_guest_dir = rootfs_path
+
 
     def prepare_minimal_rootfs(self, rootfs_path: Path, xenroot: Path) -> None:
         script_pre_xen = dedent(f"""\
@@ -212,8 +220,42 @@ class BootBinary:
 
         self.xen_tools_rootfs = cpio_rootfs
 
+    def prepare_guest_rootfs(self, rootfs_path: Path, xenroot: Path) -> None:
+        etc = rootfs_path / "etc"
+        etc.mkdir()
+
+        local_d_dir = etc / "local.d"
+        local_d_dir.mkdir()
+
+        xen_start_script = local_d_dir / "xen.start"
+        xen_start_script.write_text(dedent(f"""\
+            #!/bin/sh
+            echo 8 > /proc/sys/kernel/printk
+            {self.guest_boot_script}
+            """))
+        xen_start_script.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+
+        (etc / "issue").write_text(dedent("""\
+            domU Welcome to Alpine Linux
+            Kernel \\r on an \\m (\\l)
+
+            """))
+
+        # rootfs archive
+        rootfs = []
+        # Uncompressed first! Then compressed cpio archives!
+        rootfs.append(xenroot / "binaries/rootfs.cpio.gz")
+        cpio_rootfs = xenroot / "binaries/guest-rootfs.cpio.gz"
+        concatenate_cpios(rootfs, cpio_rootfs)
+        cpio_archive_from_dir(rootfs_path, cpio_rootfs)
+
+        self.guest_rootfs = cpio_rootfs
+
     def minimal(self) -> Path:
         return self.minimal_rootfs
 
     def dom0_tools(self) -> Path:
         return self.xen_tools_rootfs
+
+    def guest(self) -> Path:
+        return self.guest_rootfs
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 039cc36c1a04..5a7ec43fa52b 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -32,11 +32,17 @@ def prepare_rootfs(tmp_path: Path,
                    xenroot_path: Path,
                    sshkey_for_test: paramiko.pkey.PKey,
                    boot_string: str,
+                   request: pytest.FixtureRequest
                    ) -> BootBinary:
     static_gateway = os.environ['HOST_IP_GATEWAY']
     http_pxe_path = os.environ["PXE_HTTP_PATH"]
 
     boot_script = f"echo Dom0 ready {boot_string}\n"
+    marker = request.node.get_closest_marker("guest_boot_script")
+    if marker:
+        guest_boot_script = marker.args[0]
+    else:
+        guest_boot_script = None
 
     ssh_pub_keys = os.environ['SSH_PUB_KEYS']
     ssh_pub_keys += f"\n{sshkey_for_test.get_name()} {sshkey_for_test.get_base64()}\n"
@@ -52,6 +58,7 @@ def prepare_rootfs(tmp_path: Path,
                       static_ipv6=os.environ['HOST_IP6'],
                       ssh_pub_keys=ssh_pub_keys,
                       boot_script=boot_script,
+                      guest_boot_script=guest_boot_script,
                       )
 
 
@@ -61,6 +68,7 @@ def prepare_netboot(prepare_rootfs: BootBinary, xenroot_path: Path) -> None:
     rootfs_paths = [
         prepare_rootfs.minimal(),
         prepare_rootfs.dom0_tools(),
+        prepare_rootfs.guest(),
     ]
     netboot.prepare_tftp(xenroot_path, rootfs_paths)
 
-- 
Anthony PERARD

[RFC XEN PATCH 10/13] pytest: Add guest test

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Adding a second test, to help show how to improve the fixtures, and
needed code in libraries

That test copies part of the run tools-test tests, but we could shared
that better.

Then, we ssh to the host, to both follow the guest log files and to
prepare and start the guest, over several SSH channel.

tail_log() could go into a lib.
And I don't know yet what to do with the collected threads object.

I also want to get rid of the guest_boot_script marker, and ssh to the
guest instead.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/test_boot.py | 97 ++++++++++++++++++++++++++++++++++
 1 file changed, 97 insertions(+)

diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index 5a7ec43fa52b..089aa1e6c063 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -3,8 +3,11 @@ import os
 import paramiko
 from pathlib import Path
 import pytest
+from queue import Queue
 import random
 import string
+from textwrap import dedent
+from threading import Thread
 from typing import Generator
 
 from lib.boot_binary import BootBinary
@@ -138,3 +141,97 @@ def test_xen_tools_tests(machine_on_ipmi: IPMI,
     errormsg = stderr.read()
     if errormsg:
         logging.info(f"cmd errro: {errormsg!r}")
+
+
+@pytest.mark.timeout(600)
+@pytest.mark.guest_boot_script(dedent(f"""
+    ifconfig eth0 10.1.105.101
+    until ping -c 10 {os.environ["HOST_IP4"].partition('/')[0]}; do
+        sleep 1
+    done
+    echo \"guest ping test passed\"
+    """))
+def test_guest(
+        machine_on_ipmi: IPMI,
+        sshkey_for_test: paramiko.pkey.PKey,
+        boot_string: str,
+        ) -> None:
+    expect_string = boot_string
+    for line in machine_on_ipmi.console_output_generator():
+        #
+        # Wait for `expect_string` for the test to succeed
+        #
+        if expect_string in line:
+            logging.info(f"got expect_string '{line}'.")
+            break
+        #
+        # Some common expected or not console outputs.
+        #
+        if "Latest ChangeSet:" in line:
+            logging.info(f"Test xen build from '{line}'.")
+        if "Manual reset required ('noreboot' specified)" in line:
+            # Test failed, could wait a bit before failing the test.
+            logging.error("Xen panic")
+
+    with paramiko.client.SSHClient() as ssh:
+        threads = []
+        ssh_logs: Queue[str] = Queue()
+        ssh.set_missing_host_key_policy(paramiko.client.WarningPolicy)
+        logging.debug(f"connecting to {os.environ['HOST_IP4'].split('/')[0]}")
+        ssh.connect(os.environ['HOST_IP4'].split('/')[0],
+                    username='root',
+                    pkey=sshkey_for_test,
+                    timeout=10,
+                    )
+
+        def ssh_channel_read(stdout, prefix):
+            for output in stdout:
+                ssh_logs.put(f"{prefix}{output.strip()}")
+
+        def tail_log(file: Path, prefix: str):
+            stdin, stdout, stderr = ssh.exec_command(f"tail -n0 -F {file}")
+            stdin.close()
+            # Dirty, should keep `stderr` to clean it up
+            t = Thread(target=ssh_channel_read, args=(stdout, prefix))
+            t.start()
+            return t
+
+        threads += [
+            tail_log(Path("/var/log/xen/console/guest-domU.log"), "(domU) "),
+            tail_log(Path("/var/log/xen/qemu-dm-domU.log"), "(qemu-dm) "),
+         ]
+
+        cmd = dedent(f"""
+            set -x
+            pxe_host="{os.environ["PXE_HOST"]}"
+            http_pxe_path="{os.environ["PXE_HTTP_PATH"]}"
+            wget http://${{pxe_host}}/${{http_pxe_path}}/bzImage -O /boot/vmlinuz-domU
+            wget http://${{pxe_host}}/${{http_pxe_path}}/guest-rootfs.cpio.gz -O /boot/initrd-domU
+            xl -Tvvv create /etc/xen/domU.cfg 2>&1
+            """)
+        logging.info(f"Exec command: {cmd}")
+        stdin, stdout, stderr = ssh.exec_command(cmd)
+
+        # everything is on stderr, both wget and xl output
+        t = Thread(target=ssh_channel_read, args=(stderr, ""))
+        t.start()
+        threads.append(t)
+        t = Thread(target=ssh_channel_read, args=(stdout, ""))
+        t.start()
+        threads.append(t)
+
+        # Log while `xl` is preparing the guest to start
+        while t.is_alive():
+            line = ssh_logs.get()
+            logging.debug(line)
+
+        xl_exit_status = stderr.channel.recv_exit_status()
+        assert xl_exit_status == 0
+
+        # `xl` have exited, continue logging the guest output
+        while True:
+            line = ssh_logs.get()
+            logging.debug(line)
+            if "guest ping test passed" in line:
+                logging.info("found expected guest line in logs")
+                break
-- 
Anthony PERARD

[RFC XEN PATCH 11/13] pytest: Add config to enable live log, and disable ssh/paramiko logging

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/pytest.toml | 9 +++++++++
 1 file changed, 9 insertions(+)
 create mode 100644 automation/pytest/pytest.toml

diff --git a/automation/pytest/pytest.toml b/automation/pytest/pytest.toml
new file mode 100644
index 000000000000..ef61ff74ab47
--- /dev/null
+++ b/automation/pytest/pytest.toml
@@ -0,0 +1,9 @@
+[pytest]
+addopts = [
+    "--log-disable=paramiko.transport",
+    "--log-disable=paramiko.transport.sftp",
+]
+log_cli_format = "%(asctime)s %(levelname)s %(message)s"
+log_cli_date_format = "%H:%M:%S"
+log_cli_level = "DEBUG"
+log_cli = true
-- 
Anthony PERARD

[RFC XEN PATCH 12/13] pytest: Add example .env for options for the tests

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

All thees could be instead pass to the environment of a gitlab-runner
job instead. But that can make developing easier.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/.env           | 19 +++++++++++++++++++
 automation/pytest/pyproject.toml |  1 +
 automation/pytest/pytest.toml    |  4 ++++
 3 files changed, 24 insertions(+)
 create mode 100644 automation/pytest/.env

diff --git a/automation/pytest/.env b/automation/pytest/.env
new file mode 100644
index 000000000000..6ac8b6158de8
--- /dev/null
+++ b/automation/pytest/.env
@@ -0,0 +1,19 @@
+# Test machine
+HOST_MAC="ff:ff:ff:ff:ff:ff"
+HOST_IP_GATEWAY="203.0.113.1"
+HOST_IP_VLANID=42
+HOST_IP4="203.0.113.42/24"
+HOST_IP6="2001:db8::1:42/32"
+
+# IPMI
+IPMI_HOSTNAME='moonshot.example.org'
+IPMI_BLADE_NODE="c42n1"
+
+# Network environment
+# XXX default value probably not needed with env_files_skip_if_set
+PXE_HOST=${PXE_HOST:-pxe.example.org}
+PXE_HTTP_PATH="configs/custom/${HOST_MAC}"
+PXE_SSH_USER=pxe
+PXE_SSH_PATH="pxe/moonshot-c42"
+
+SSH_PUB_KEYS="ssh-algo AAAArandomstring Not a real key\n"
diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
index c54c1338f656..b62befaa07d3 100644
--- a/automation/pytest/pyproject.toml
+++ b/automation/pytest/pyproject.toml
@@ -7,6 +7,7 @@ dependencies = [
     "paramiko<5.0.0",
     "pip>=26.0.1",
     "pytest>=9.0.2",
+    "pytest-env>=1.6.0",
     "pytest-timeout>=2.4.0",
     "types-paramiko>=4.0.0.20260322",
 ]
diff --git a/automation/pytest/pytest.toml b/automation/pytest/pytest.toml
index ef61ff74ab47..405a56a83301 100644
--- a/automation/pytest/pytest.toml
+++ b/automation/pytest/pytest.toml
@@ -7,3 +7,7 @@ log_cli_format = "%(asctime)s %(levelname)s %(message)s"
 log_cli_date_format = "%H:%M:%S"
 log_cli_level = "DEBUG"
 log_cli = true
+
+[pytest_env]
+env_files = [".env"]
+env_files_skip_if_set = true
-- 
Anthony PERARD

[RFC XEN PATCH 13/13] gitlab-ci: Add job for moonshot with pytest

[Anthony PERARD]

From: Anthony PERARD <anthony.perard@vates.tech>

WIP, use alternative branchs for artefacts

- moonshot need kernel with driver for the network card
  and for vlan
- alpine rootfs need network support and sshd

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/gitlab-ci/test.yaml | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/automation/gitlab-ci/test.yaml b/automation/gitlab-ci/test.yaml
index 61adc1baff30..1f760e24811f 100644
--- a/automation/gitlab-ci/test.yaml
+++ b/automation/gitlab-ci/test.yaml
@@ -205,6 +205,7 @@
     - qubes-hw11
 
 # Test jobs
+
 build-each-commit-gcc:
   extends: .test-jobs-common
   variables:
@@ -777,6 +778,26 @@ qemu-xtf-argo-x86_64-gcc-debug:
   needs:
     - alpine-3.24-x86_64-gcc-debug
 
+broadwell-tools-tests-pv-x86-64-gcc-debug:
+  extends: .test-jobs-common
+  variables:
+    XEN_REGISTRY: registry.gitlab.com/xen-project/people/anthonyper/xen
+    CONTAINER: archlinux:current-uv
+  artifacts:
+    paths:
+      - smoke.serial
+      - '*.log'
+    when: always
+    reports:
+      junit: tests-junit.xml
+  tags:
+    - vates-moonshot-1
+  needs:
+    - *x86_64-test-needs
+    - alpine-3.24-x86_64-gcc-debug
+  script:
+    - cd automation/pytest && uv run pytest
+
 qemu-smoke-riscv64-gcc:
   extends: .qemu-riscv64
   script:
-- 
Anthony PERARD

Re: [RFC XEN PATCH 00/13] Running bare-metal tests with "pytest"

[Marek Marczykowski-Górecki]

[-- Attachment #1: Type: text/plain, Size: 3638 bytes --]

On Tue, Sep 22, 2026 at 12:08:47PM +0200, Anthony PERARD wrote:
> From: Anthony PERARD <anthony.perard@vates.tech>
> 
> Patch series available in this git branch:
> https://xenbits.xenproject.org/git-http/people/aperard/xen-unstable.git br.ci.bare-metal-with-pytest-v1
> 
> Hi,
> 
> I wanted to add some new machine to our GitLab CI, but I didn't want to
> duplicate yet another time the existing shell script. They are good to get the
> ball rolling but are getting harder to maintain as more test are been added,
> and don't share common code between machines.
>
> So, I've start work on something based on `pytest` which I hope will be easier
> to maintain and extend and use to add new machines.

Nice!

> It's still Work In Progress.
> 
> Here is the current result:
>     https://gitlab.com/xen-project/people/anthonyper/xen/-/jobs/16529107619

Minor readability improvement idea: force color output of pytest.

> In this pipeline, the artifact use are a bit different from master; linux is
> built with network driver, and vlan; the rootfs is from Marek's patch series,
> with networking and dropbear service enabled.
> 
> Next:
> 
> I'd like to at least have a new fixture that give a started Host that a test
> can ssh into. Then maybe a new class "Host" which could have functions for ssh
> and getting the serial console output.
> 
> Don't go to much into detail review, I'm more interested of a general overview,
> and to present what I have so far.

We had a design session on this topic, and we discussed a structural
change: have one job per Xen boot flavor (mostly dom0 PV vs PVH, but
could be also other settings needing host reboot) and then have pytest
run several tests via SSH. The idea is to not waste 2-4 minutes for
every ~10 sec test. Yes, it does mean one failed test will likely
prevent others in the same job from running (especially in case of Xen
panic), but since the intended state is "all green", a single failure is
bad already, and this approach should still get you enough information
to debug that first failure.

In practical terms, it would require most fixtures to be class or
session scope, instead of the default test scope. And using more SSH
instead of baking tests into startup scripts (already partially done,
but may need extending to domU commands too).

Other than that, this approach looks nice, and I really like using
pytest here, as it makes the structure significantly better. This will
likely require a bit of adjustments to handle also other hosts (for
example for qubes runners tftp dir is mounted into gitlab's container so
there is no need to sftp boot files). But that should be easy to do
later, it doesn't change the overall structure much.


> Some runes:
>     - uv run mypy .
>       to check the code, at least the types used.
>       other code check could be added, and run by the pipeline
>     - uv run pytest
>       to run the whole test suite
>     - uv run pytest test_boot.py::test_xen_tools_tests
>       to run just one test
> 
> I did create an Arch Linux docker image with pytest, but paramiko got too new
> to be used, so I've started to use `uv` to deal with the dependencies. `mypy`
> always needed to be run with `uv` due to missing types for paramiko (not
> packaged in Arch Linux). So the image is useless, and I'll look for a different
> image later, for now just Arch Linux with `uv` installed.

FWIW Qubes tests use Alpine container, seems to have all what we need. I
see it also has paramiko 4.0.0

-- 
Best Regards,
Marek Marczykowski-Górecki
Invisible Things Lab

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [RFC XEN PATCH 00/13] Running bare-metal tests with "pytest"

[Baptiste Le Duc]

> From: Anthony PERARD <anthony.perard@vates.tech>
> 
> Patch series available in this git branch:
> https://xenbits.xenproject.org/git-http/people/aperard/xen-unstable.git br.ci.bare-metal-with-pytest-v1
> 
> Hi,
> 
> I wanted to add some new machine to our GitLab CI, but I didn't want to
> duplicate yet another time the existing shell script. They are good to get the
> ball rolling but are getting harder to maintain as more test are been added,
> and don't share common code between machines.

Hi Anthony,

Nice to hear! I reviewed it quickly and it seems to be a more maintainable
solution to add machines in the CI.

FWIW the patch series I sent [1] also uses a python-driven framework (QTB)
which aims to replace the existing shell smoke test scripts. It's QEMU-only
though as it drives QEMU over QMP/qtest to manually trigger some IRQs. So,
it doesn't cover your usecase for the moment as-is and for the moment I
don't see a possible common abstraction between them but worth flagging.

Moreover, do you think we could replace the QEMU smoke-tests by using your
framework? With a very quick review, it seems we could just add a pytest
fixture machine_on_qemu(). If yes, then it would definitely be an overlap
with the usage of QTB I described above.

[1] https://lore.kernel.org/xen-devel/1787823334.8631fc262581453bbf619ec5b2062170.1a04293255c000c4f3@vates.tech/

-- 
Baptiste Le Duc <baptiste.le-duc@vates.tech>