""" Validate the virt module """ import logging from numbers import Number from xml.etree import ElementTree import pytest from saltfactories.utils import cli_scripts from tests.support.helpers import skip_if_binaries_missing, slowTest from tests.support.virt import SaltVirtMinionContainerFactory docker = pytest.importorskip("docker") log = logging.getLogger(__name__) @pytest.fixture(scope="module") def docker_client(): urllib3_connectionpool_handler = logging.getLogger("urllib3.connectionpool") urllib3_connectionpool_handler_level = urllib3_connectionpool_handler.level urllib3_connectionpool_handler.setLevel(logging.INFO) try: client = docker.from_env() connectable = SaltVirtMinionContainerFactory.client_connectable(client) if connectable is not True: # pragma: no cover pytest.skip(connectable) client.images.pull("quay.io/rst0git/virt-minion") yield client finally: urllib3_connectionpool_handler.setLevel(urllib3_connectionpool_handler_level) @pytest.fixture(scope="module") def salt_minion_script_path(salt_factories): return cli_scripts.generate_script( salt_factories.scripts_dir, "salt-minion", code_dir=salt_factories.code_dir, inject_coverage=salt_factories.inject_coverage, inject_sitecustomize=salt_factories.inject_sitecustomize, ) @pytest.fixture(scope="module") def virt_minion_0_id(): return "virt-minion-0" @pytest.fixture(scope="module") def virt_minion_1_id(): return "virt-minion-1" @pytest.fixture(scope="module") def virt_minion_0( salt_factories, salt_master, docker_client, salt_minion_script_path, virt_minion_0_id, virt_minion_1_id, ): root_dir = salt_factories.get_root_dir_for_daemon(virt_minion_0_id) config_defaults = { "root_dir": str(root_dir), "id": virt_minion_0_id, "open_mode": True, "transport": salt_master.config["transport"], } config_overrides = {"user": "root"} config = SaltVirtMinionContainerFactory.configure( factories_manager=salt_factories, daemon_id=virt_minion_0_id, root_dir=root_dir, config_defaults=config_defaults, config_overrides=config_overrides, master=salt_master, ) salt_factories.final_minion_config_tweaks(config) loaded_config = SaltVirtMinionContainerFactory.write_config(config) factory = SaltVirtMinionContainerFactory( name=virt_minion_0_id, image="quay.io/rst0git/virt-minion", docker_client=docker_client, config=loaded_config, cli_script_name=salt_minion_script_path, start_timeout=60, factories_manager=salt_factories, event_listener=salt_factories.event_listener, container_run_kwargs={ "extra_hosts": { virt_minion_0_id: "127.0.0.1", virt_minion_1_id: "127.0.0.1", } }, ) factory.register_after_terminate_callback( pytest.helpers.remove_stale_minion_key, salt_master, factory.id ) with factory.started(): yield factory @pytest.fixture(scope="module") def virt_minion_1( salt_factories, salt_master, docker_client, salt_minion_script_path, virt_minion_0_id, virt_minion_1_id, ): root_dir = salt_factories.get_root_dir_for_daemon(virt_minion_1_id) config_defaults = { "root_dir": str(root_dir), "id": virt_minion_1_id, "open_mode": True, "transport": salt_master.config["transport"], } config_overrides = {"user": "root"} config = SaltVirtMinionContainerFactory.configure( factories_manager=salt_factories, daemon_id=virt_minion_1_id, root_dir=root_dir, config_defaults=config_defaults, config_overrides=config_overrides, master=salt_master, ) salt_factories.final_minion_config_tweaks(config) loaded_config = SaltVirtMinionContainerFactory.write_config(config) factory = SaltVirtMinionContainerFactory( name=virt_minion_1_id, image="quay.io/rst0git/virt-minion", docker_client=docker_client, config=loaded_config, cli_script_name=salt_minion_script_path, start_timeout=60, factories_manager=salt_factories, event_listener=salt_factories.event_listener, container_run_kwargs={ "extra_hosts": { virt_minion_0_id: "127.0.0.1", virt_minion_1_id: "127.0.0.1", } }, ) factory.register_after_terminate_callback( pytest.helpers.remove_stale_minion_key, salt_master, factory.id ) with factory.started(): yield factory @pytest.fixture(scope="module") def salt_cli(salt_master, virt_minion_0, virt_minion_1): return salt_master.get_salt_cli() @skip_if_binaries_missing("docker") @slowTest class TestVirtTest: """ Test virt routines """ cpu_models = [ "none", "armv7l", "armv7b", "aarch64", "i686", "ppc64", "ppc64le", "riscv32", "riscv64", "s390", "s390x", "x86_64", ] def test_default_kvm_profile(self, salt_cli, virt_minion_0): """ Test virt.get_profiles with the KVM profile """ ret = salt_cli.run("virt.get_profiles", "kvm", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret profiles = ret.json assert isinstance(profiles, dict) nic = profiles["nic"]["default"][0] disk = profiles["disk"]["default"][0] assert nic["name"] == "eth0" assert nic["type"] == "bridge" assert nic["model"] == "virtio" assert nic["source"] == "br0" assert disk["name"] == "system" assert disk["model"] == "virtio" assert disk["size"] == 8192 def test_default_vmware_profile(self, salt_cli, virt_minion_0): """ Test virt.get_profiles with the VMware profile """ ret = salt_cli.run("virt.get_profiles", "vmware", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret profiles = ret.json assert isinstance(profiles, dict) nic = profiles["nic"]["default"][0] disk = profiles["disk"]["default"][0] assert nic["name"] == "eth0" assert nic["type"] == "bridge" assert nic["model"] == "e1000" assert nic["source"] == "DEFAULT" assert disk["name"] == "system" assert disk["model"] == "scsi" assert disk["format"] == "vmdk" assert disk["size"] == 8192 def test_default_xen_profile(self, salt_cli, virt_minion_0): """ Test virt.get_profiles with the XEN profile """ ret = salt_cli.run("virt.get_profiles", "xen", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret profiles = ret.json assert isinstance(profiles, dict) nic = profiles["nic"]["default"][0] disk = profiles["disk"]["default"][0] assert nic["name"] == "eth0" assert nic["type"] == "bridge" assert nic["model"] is None assert nic["source"] == "br0" assert disk["name"] == "system" assert disk["model"] == "xen" assert disk["size"] == 8192 def test_default_bhyve_profile(self, salt_cli, virt_minion_0): """ Test virt.get_profiles with the Bhyve profile """ ret = salt_cli.run("virt.get_profiles", "bhyve", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret profiles = ret.json assert isinstance(profiles, dict) nic = profiles["nic"]["default"][0] disk = profiles["disk"]["default"][0] assert nic["name"] == "eth0" assert nic["type"] == "bridge" assert nic["model"] == "virtio" assert nic["source"] == "bridge0" assert disk["name"] == "system" assert disk["model"] == "virtio" assert disk["format"] == "raw" assert disk["sparse_volume"] is False assert disk["size"] == 8192 def test_all_capabilities(self, salt_cli, virt_minion_0): """ Test virt.all_capabilities """ ret = salt_cli.run("virt.all_capabilities", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret caps = ret.json assert isinstance(caps, dict) assert isinstance(caps["host"]["host"]["uuid"], str) assert len(caps["host"]["host"]["uuid"]) == 36 assert "qemu" in [domainCaps["domain"] for domainCaps in caps["domains"]] def test_capabilities(self, salt_cli, virt_minion_0): """ Test virt.capabilities """ ret = salt_cli.run("virt.capabilities", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret caps = ret.json assert isinstance(caps, dict) assert isinstance(caps["host"]["uuid"], str) assert len(caps["host"]["uuid"]) == 36 assert len(caps["guests"]) >= 1 assert caps["guests"][0]["os_type"] in ["hvm", "xen", "xenpvh", "exe"] def test_cpu_baseline(self, salt_cli, virt_minion_0): """ Test virt.cpu_baseline """ vendors = ["Intel", "ARM", "AMD"] ret = salt_cli.run( "virt.cpu_baseline", out="libvirt", minion_tgt=virt_minion_0.id ) assert ret.exitcode == 0, ret cpu_baseline = ret.json assert isinstance(cpu_baseline, str) cpu_baseline = ElementTree.fromstring(cpu_baseline) assert cpu_baseline.find("vendor").text in vendors ret = salt_cli.run("virt.cpu_baseline", out="salt", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret cpu_baseline = ret.json assert isinstance(cpu_baseline, dict) assert cpu_baseline["vendor"] in vendors def test_freemem(self, salt_cli, virt_minion_0): """ Test virt.freemem """ ret = salt_cli.run("virt.freemem", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret available_memory = ret.json assert isinstance(available_memory, Number) assert available_memory > 0 def test_freecpu(self, salt_cli, virt_minion_0): """ Test virt.freecpu """ ret = salt_cli.run("virt.freecpu", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret available_cpus = ret.json assert isinstance(available_cpus, Number) assert available_cpus > 0 def test_full_info(self, salt_cli, virt_minion_0): """ Test virt.full_info """ ret = salt_cli.run("virt.full_info", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret info = ret.json assert isinstance(info, dict) assert isinstance(info["vm_info"], dict) assert isinstance(info["freecpu"], Number) assert isinstance(info["freemem"], Number) assert info["freecpu"] > 0 assert info["freemem"] > 0 assert isinstance(info["node_info"], dict) assert isinstance(info["node_info"]["cpucores"], Number) assert isinstance(info["node_info"]["cpumhz"], Number) assert isinstance(info["node_info"]["cpus"], Number) assert isinstance(info["node_info"]["cputhreads"], Number) assert isinstance(info["node_info"]["numanodes"], Number) assert isinstance(info["node_info"]["phymemory"], Number) assert info["node_info"]["cpumodel"] in self.cpu_models def test_node_info(self, salt_cli, virt_minion_0): """ Test virt.node_info """ ret = salt_cli.run("virt.node_info", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret info = ret.json assert isinstance(info, dict) assert isinstance(info["cpucores"], Number) assert isinstance(info["cpumhz"], Number) assert isinstance(info["cpus"], Number) assert isinstance(info["cputhreads"], Number) assert isinstance(info["numanodes"], Number) assert isinstance(info["phymemory"], Number) assert isinstance(info["sockets"], Number) assert info["cpumodel"] in self.cpu_models @pytest.fixture(scope="module") def virt_domain(): return "core-vm" @pytest.fixture def prep_virt(salt_cli, virt_minion_0, virt_minion_1, virt_domain): try: ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json for domain in domains: salt_cli.run("virt.stop", virt_domain, minion_tgt=virt_minion_0.id) salt_cli.run("virt.undefine", virt_domain, minion_tgt=virt_minion_0.id) ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json for domain in domains: salt_cli.run("virt.stop", virt_domain, minion_tgt=virt_minion_1.id) salt_cli.run("virt.undefine", virt_domain, minion_tgt=virt_minion_1.id) ret = salt_cli.run( "virt.define_xml_path", "/{}.xml".format(virt_domain), minion_tgt=virt_minion_0.id, ) assert ret.exitcode == 0, ret ret = salt_cli.run("virt.start", virt_domain, minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret # Run tests yield finally: ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json for domain in domains: salt_cli.run("virt.stop", virt_domain, minion_tgt=virt_minion_0.id) salt_cli.run("virt.undefine", virt_domain, minion_tgt=virt_minion_0.id) ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json for domain in domains: salt_cli.run("virt.stop", virt_domain, minion_tgt=virt_minion_1.id) salt_cli.run("virt.undefine", virt_domain, minion_tgt=virt_minion_1.id) @skip_if_binaries_missing("docker") @slowTest class TestVirtMigrateTest: def test_define_xml_path(self, salt_cli, virt_minion_0, virt_domain): """ Define a new domain with virt.define_xml_path, verify that the new domain is shown with virt.list_domains, remove the domain with virt.undefine, and verifies that domain is no longer shown with virt.list_domains. """ ret = salt_cli.run( "virt.define_xml_path", "/{}.xml".format(virt_domain), minion_tgt=virt_minion_0.id, ) assert ret.exitcode == 0, ret result = ret.json assert isinstance(result, bool), result assert result is True, result ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain] ret = salt_cli.run("virt.undefine", virt_domain, minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret result = ret.json assert isinstance(result, bool) assert result is True ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [] def test_ssh_migration( self, salt_cli, virt_minion_0, virt_minion_1, prep_virt, virt_domain ): """ Test domain migration over SSH, TCP and TLS transport protocol """ ret = salt_cli.run("virt.list_active_vms", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret ret = salt_cli.run("virt.list_active_vms", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret ret = salt_cli.run("virt.vm_info", virt_domain, minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret # Verify that the VM has been created ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain] ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [] ret = salt_cli.run( "virt.migrate", virt_domain, virt_minion_1.uri, ssh=True, minion_tgt=virt_minion_0.id, ) assert ret.exitcode == 0, ret result = ret.json assert isinstance(result, bool) assert result is True # Verify that the VM has been migrated ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [], "Failed to migrate VM" ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain], "Failed to migrate VM" def test_tcp_migration( self, salt_cli, virt_minion_0, virt_minion_1, prep_virt, virt_domain ): """ Test domain migration over SSH, TCP and TLS transport protocol """ # Verify that the VM has been created ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain] ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [] ret = salt_cli.run( "virt.migrate", virt_domain, virt_minion_1.tcp_uri, minion_tgt=virt_minion_0.id, ) assert ret.exitcode == 0, ret result = ret.json assert isinstance(result, bool) assert result is True # Verify that the VM has been migrated ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [], "Failed to migrate VM" ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain], "Failed to migrate VM" def test_tls_migration( self, salt_cli, virt_minion_0, virt_minion_1, prep_virt, virt_domain ): """ Test domain migration over SSH, TCP and TLS transport protocol """ # Verify that the VM has been created ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain] ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [] ret = salt_cli.run( "virt.migrate", virt_domain, virt_minion_1.tls_uri, minion_tgt=virt_minion_0.id, ) assert ret.exitcode == 0, ret result = ret.json assert isinstance(result, bool) assert result is True # Verify that the VM has been migrated ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_0.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [], "Failed to migrate VM" ret = salt_cli.run("virt.list_domains", minion_tgt=virt_minion_1.id) assert ret.exitcode == 0, ret domains = ret.json assert isinstance(domains, list) assert domains == [virt_domain], "Failed to migrate VM"