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)
- Check that balloning feature return 1

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

Author: d3athjest3r

tests/guest_tools/unix/conftest.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import pytest
+
+_GITLAB_API = 'https://gitlab.com/api/v4'
+_GITLAB_PROJECT_PATH = 'xen-project%2Fxen-guest-agent'
+# Numeric ID avoids %2F in the URL, which APT decodes and breaks.
+_GITLAB_PROJECT_ID = 28547076
+
+# RPM: no hosted yum repo exists yet, so we fall back to CI artifact download.
+_RPM_ARTIFACT_URL = (
+    f'{_GITLAB_API}/projects/{_GITLAB_PROJECT_PATH}'
+    '/jobs/artifacts/main/download'
+    '?search_recent_successful_pipelines=true&job=pkg-rpm-x86_64'
+)
+
+# DEB: packages are deployed to the GitLab Generic Package Registry after each
+# push to main, forming a proper APT repo with stable URLs.
+# Uses the numeric project ID so APT does not mangle the URL.
+_DEB_REPO_URL = f'{_GITLAB_API}/projects/{_GITLAB_PROJECT_ID}/packages/generic/deb-amd64/'
+
+
+@pytest.fixture(scope="module")
+def xen_guest_agent_urls() -> dict[str, str]:
+    return {
+        'rpm_artifact': _RPM_ARTIFACT_URL,
+        'deb_repo': _DEB_REPO_URL,
+    }

tests/guest_tools/unix/test_xen_guest_agent.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import pytest
+
+import logging
+
+from lib.common import PackageManagerEnum, wait_for
+from lib.host import Host
+from lib.vm import VM
+
+# 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: Host, xs_prefix: str, vif_id: int, proto: str) -> list[str]:
+    """Return all IPs published under attr/vif/{vif_id}/{proto}/* in Xenstore."""
+    parent = f'{xs_prefix}/attr/vif/{vif_id}/{proto}'
+    res = host.ssh_with_result(f'xenstore-list {parent}')
+    if res.returncode != 0:
+        return []
+    return [
+        host.ssh(f'xenstore-read {parent}/{slot}').strip()
+        for slot in res.stdout.split()
+    ]
+
+
+@pytest.mark.multi_vms
+@pytest.mark.usefixtures("unix_vm")
+class TestXenGuestAgent:
+    @pytest.fixture(scope="class", autouse=True)
+    def agent_install(self, running_vm: VM, xen_guest_agent_urls: dict[str, str]) -> None:
+        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')
+        elif 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')
+
+        if pkg_mgr == PackageManagerEnum.RPM:
+            # No hosted yum repo exists yet; download the artifact zip via Python's
+            # standard library to avoid requiring curl/wget/unzip in the guest.
+            url = xen_guest_agent_urls['rpm_artifact']
+            py_script = (
+                f"import urllib.request, zipfile, io; "
+                f"z = zipfile.ZipFile(io.BytesIO(urllib.request.urlopen('{url}').read())); "
+                f"pkg = next(n for n in z.namelist() "
+                f"if n.endswith('.rpm') and 'debug' not in n and 'dbgsym' not in n); "
+                f"z.extract(pkg, '/tmp/xga/')"
+            )
+            vm.ssh(f'python3 -c "{py_script}"')
+            vm.ssh("yum install -y $(find /tmp/xga -name '*.rpm')")
+        elif pkg_mgr == PackageManagerEnum.APT_GET:
+            # DEB packages are published to a stable APT repo in the GitLab
+            # Generic Package Registry after each push to main.
+            deb_repo = xen_guest_agent_urls['deb_repo']
+            vm.ssh(f"echo 'deb [trusted=yes] {deb_repo} main/' "
+                   f"> /etc/apt/sources.list.d/xen-guest-agent.list")
+            vm.ssh('apt-get update')
+            vm.ssh('apt-get install -y xen-guest-agent')
+
+        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_after_reboot(self, running_vm: VM) -> None:
+        running_vm.reboot(verify=True)
+        running_vm.ssh('systemctl is-active xen-guest-agent')
+
+    def test_xenstore_version(self, running_vm: VM) -> None:
+        host = running_vm.host
+        xs_prefix = f'/local/domain/{running_vm.param_get("dom-id")}'
+        host.ssh(f'xenstore-read {xs_prefix}/attr/PVAddons/MajorVersion')
+        host.ssh(f'xenstore-read {xs_prefix}/attr/PVAddons/BuildVersion')
+
+    def test_xenstore_os_info(self, running_vm: VM) -> None:
+        host = running_vm.host
+        xs_prefix = f'/local/domain/{running_vm.param_get("dom-id")}'
+        host.ssh(f'xenstore-read {xs_prefix}/data/os_distro')
+        host.ssh(f'xenstore-read {xs_prefix}/data/os_uname')
+
+    def test_xenstore_memory(self, running_vm: VM) -> None:
+        host = running_vm.host
+        xs_prefix = f'/local/domain/{running_vm.param_get("dom-id")}'
+        host.ssh(f'xenstore-read {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(f'xenstore-read {xs_prefix}/data/meminfo_free').returncode == 0,
+            "Wait for meminfo_free in Xenstore",
+            timeout_secs=90,
+        )
+
+    def test_xenstore_feature_balloon(self, running_vm: VM) -> None:
+        host = running_vm.host
+        xs_prefix = f'/local/domain/{running_vm.param_get("dom-id")}'
+        res = host.ssh_with_result(f'xenstore-read {xs_prefix}/control/feature-balloon')
+        if res.returncode != 0:
+            pytest.skip("control/feature-balloon not present — agent may lack write permission on this host")
+        assert res.stdout.strip() == '1', \
+            f"Expected control/feature-balloon to be '1', got {res.stdout.strip()!r}"
+
+    def test_xenstore_vif_ip(self, running_vm: VM) -> None:
+        vm = running_vm
+        host = vm.host
+        xs_prefix = f'/local/domain/{vm.param_get("dom-id")}'
+        if host.ssh_with_result(f'xenstore-exists {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})"