diff --git a/tests/guest_tools/unix/conftest.py b/tests/guest_tools/unix/conftest.py new file mode 100644 index 000000000..89795d48e --- /dev/null +++ b/tests/guest_tools/unix/conftest.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import pytest + +_GITLAB_API = 'https://gitlab.com/api/v4' +# Numeric ID avoids %2F in the URL, which APT decodes and breaks. +_GITLAB_PROJECT_ID = 28547076 + +_PKG_REGISTRY = f'{_GITLAB_API}/projects/{_GITLAB_PROJECT_ID}/packages/generic' + +_RPM_REPO_URL = 'https://xen-project.gitlab.io/xen-guest-agent/rpm-x86_64/' +# DEB uses the numeric project ID so APT doesn't mangle the URL. +_DEB_REPO_URL = f'{_PKG_REGISTRY}/deb-amd64/' + + +@pytest.fixture(scope="module") +def xen_guest_agent_urls() -> dict[str, str]: + return { + 'rpm_repo': _RPM_REPO_URL, + 'deb_repo': _DEB_REPO_URL, + } diff --git a/tests/guest_tools/unix/test_xen_guest_agent.py b/tests/guest_tools/unix/test_xen_guest_agent.py new file mode 100644 index 000000000..c82004c8a --- /dev/null +++ b/tests/guest_tools/unix/test_xen_guest_agent.py @@ -0,0 +1,118 @@ +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: + rpm_repo = xen_guest_agent_urls['rpm_repo'] + vm.ssh(f"echo -e '[xen-guest-agent]\\nbaseurl={rpm_repo}main/\\ngpgcheck=0'" + f" > /etc/yum.repos.d/xen-guest-agent.repo") + vm.ssh('dnf install -y xen-guest-agent') + 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})"