tests: add xen-guest-agent integration tests

Add a test suite that validates the xen-guest-agent daemon running
inside a guest VM correctly publishes data to Xenstore. The conftest
downloads RPM and DEB packages from the xen-guest-agent GitLab CI
artifacts and SCPs them to the VM for installation.
The test class (TestXenGuestAgent) then installs the agent on the VM via
yum (RPM distros) or dpkg (APT distros), then verifies:
- the systemd service is active after install and after reboot
- Xenstore paths for version, OS info, memory and VIF IP are populated
  (meminfo_free is polled with a 90s timeout as it is only published on
  a 60s timer; the VIF/IP check is skipped if no Xen PV NIC is present)

Signed-off-by: Julian Vetter <julian.vetter@vates.tech>

Author: d3athjest3r

tests/guest_tools/unix/conftest.py

@@ -0,0 +1,52 @@
+import pytest
+
+import os
+import tempfile
+import zipfile
+
+import requests
+
+from lib.common import url_download
+
+_GITLAB_API = 'https://gitlab.com/api/v4/projects/xen-project%2Fxen-guest-' \
+              'agent/jobs/artifacts/main/download?search_recent_successful_' \
+              'pipelines=true&job='
+
+
+def _extract_from_zip(zip_path, suffix, dest_dir):
+    """
+    Extract the main package matching suffix from zip_path into dest_dir.
+    Excludes debug/dbgsym packages which may also be present.
+    """
+    with zipfile.ZipFile(zip_path) as zf:
+        matches = [
+            n for n in zf.namelist()
+            if n.endswith(suffix) and not any(
+                kw in os.path.basename(n) for kw in ('debug', 'dbgsym')
+            )
+        ]
+        assert len(matches) == 1, \
+            f"Expected exactly one non-debug *{suffix} in artifact zip, found: {matches}"
+        zf.extract(matches[0], dest_dir)
+        return os.path.join(dest_dir, matches[0])
+
+
+@pytest.fixture(scope="module")
+def xen_guest_agent_packages():
+    """
+    Download the latest xen-guest-agent RPM and DEB from GitLab CI artifacts.
+    Yields a dict with keys 'rpm' and 'deb' pointing to the local file paths.
+    """
+    artifact_urls = { 'rpm': f'{_GITLAB_API}pkg-rpm-x86_64',
+                      'deb': f'{_GITLAB_API}pkg-deb-amd64' }
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        zip_path = os.path.join(tmpdir, 'artifacts.zip')
+
+        url_download(artifact_urls['rpm'], zip_path)
+        rpm_path = _extract_from_zip(zip_path, '.rpm', tmpdir)
+
+        url_download(artifact_urls['deb'], zip_path)
+        deb_path = _extract_from_zip(zip_path, '.deb', tmpdir)
+
+        yield {'rpm': rpm_path, 'deb': deb_path}

tests/guest_tools/unix/test_xen_guest_agent.py

@@ -0,0 +1,101 @@
+import pytest
+
+import logging
+
+from lib.common import PackageManagerEnum, wait_for
+
+# Requirements:
+# From --hosts parameter:
+# - host(A1): first XCP-ng host >= 8.0
+# From --vm parameter:
+# - A Linux VM with systemd and a supported package manager (RPM or APT)
+
+
+def _vif_published_ips(host, xs_prefix, vif_id, proto):
+    """Return all IPs published under attr/vif/{vif_id}/{proto}/* in Xenstore."""
+    ips = []
+    for slot in range(10):  # NUM_IFACE_IPS = 10 in xen-guest-agent
+        res = host.ssh_with_result(
+            ['xenstore-read', f'{xs_prefix}/attr/vif/{vif_id}/{proto}/{slot}']
+        )
+        if res.returncode == 0:
+            ips.append(res.stdout.strip())
+    return ips
+
+
+@pytest.mark.multi_vms
+@pytest.mark.usefixtures("unix_vm")
+class TestXenGuestAgent:
+    @pytest.fixture(scope="class", autouse=True)
+    def agent_install(self, running_vm, xen_guest_agent_packages):
+        vm = running_vm
+
+        if vm.ssh_with_result(['which', 'systemctl']).returncode != 0:
+            pytest.skip("systemd not available on this VM")
+
+        pkg_mgr = vm.detect_package_manager()
+        if pkg_mgr not in (PackageManagerEnum.RPM, PackageManagerEnum.APT_GET):
+            pytest.skip(f"Package manager '{pkg_mgr}' not supported in this test")
+
+        # Remove conflicting xe-guest-utilities if present
+        logging.info("Removing xe-guest-utilities if present")
+        if pkg_mgr == PackageManagerEnum.RPM:
+            vm.ssh('rpm -qa | grep xe-guest-utilities | xargs --no-run-if-empty rpm -e')
+        if pkg_mgr == PackageManagerEnum.APT_GET and \
+           vm.ssh_with_result(['dpkg', '-l', 'xe-guest-utilities']).returncode == 0:
+            vm.ssh(['apt-get', 'remove', '-y', 'xe-guest-utilities'])
+
+        # Copy package to VM and install
+        if pkg_mgr == PackageManagerEnum.RPM:
+            vm.scp(xen_guest_agent_packages['rpm'], '/root/xen-guest-agent.rpm')
+            vm.ssh(['yum', 'install', '-y', '/root/xen-guest-agent.rpm'])
+        if pkg_mgr == PackageManagerEnum.APT_GET:
+            vm.scp(xen_guest_agent_packages['deb'], '/root/xen-guest-agent.deb')
+            vm.ssh(['dpkg', '-i', '/root/xen-guest-agent.deb'])
+
+        wait_for(
+            lambda: vm.ssh_with_result(['systemctl', 'is-active', 'xen-guest-agent']).returncode == 0,
+            "Wait for xen-guest-agent service to be active",
+        )
+
+    def test_agent_running(self, running_vm):
+        running_vm.ssh(['systemctl', 'is-active', 'xen-guest-agent'])
+
+    def test_agent_running_after_reboot(self, running_vm):
+        running_vm.reboot(verify=True)
+        running_vm.ssh(['systemctl', 'is-active', 'xen-guest-agent'])
+
+    def test_xenstore_data(self, running_vm):
+        vm = running_vm
+        domid = vm.param_get('dom-id')
+        host = vm.host
+        xs_prefix = f'/local/domain/{domid}'
+
+        logging.info("Check that xen-guest-agent published version info to Xenstore")
+        host.ssh(['xenstore-read', f'{xs_prefix}/attr/PVAddons/MajorVersion'])
+        host.ssh(['xenstore-read', f'{xs_prefix}/attr/PVAddons/BuildVersion'])
+
+        logging.info("Check that OS info is published to Xenstore")
+        host.ssh(['xenstore-read', f'{xs_prefix}/data/os_distro'])
+        host.ssh(['xenstore-read', f'{xs_prefix}/data/os_uname'])
+
+        logging.info("Check that memory info is published to Xenstore")
+        host.ssh(['xenstore-read', f'{xs_prefix}/data/meminfo_total'])
+        # meminfo_free is published on a 60s timer, wait for it to appear
+        wait_for(
+            lambda: host.ssh_with_result(['xenstore-read', f'{xs_prefix}/data/meminfo_free']).returncode == 0,
+            "Wait for meminfo_free in Xenstore",
+            timeout_secs=90,
+        )
+
+        logging.info("Check that the VM's IP is published under attr/vif")
+        # VIF detection requires a Xen PV NIC; skip if none was detected
+        if host.ssh_with_result(['xenstore-exists', f'{xs_prefix}/attr/vif']).returncode != 0:
+            pytest.skip("No VIF published in Xenstore — VM may not be using a Xen PV NIC")
+
+        ipv4s = _vif_published_ips(host, xs_prefix, vif_id=0, proto='ipv4')
+        ipv6s = _vif_published_ips(host, xs_prefix, vif_id=0, proto='ipv6')
+        logging.info("Published IPv4: %s, IPv6: %s", ipv4s, ipv6s)
+        assert ipv4s or ipv6s, "No IPs published in Xenstore under attr/vif/0"
+        assert vm.ip in ipv4s + ipv6s, \
+            f"VM IP {vm.ip!r} not found in Xenstore (ipv4: {ipv4s}, ipv6: {ipv6s})"