From be65833724589266cf2ca401684c336062edecba Mon Sep 17 00:00:00 2001 From: Felix Matouschek Date: Thu, 27 Jun 2024 13:34:07 +0200 Subject: [PATCH] cleanup(tests): Major rework of inventory unit tests Rework the inventory unit tests by splitting up tests/unit/plugins/inventory/test_kubevirt.py into multiple files, by trying to simplify the test code and making it more robust and by using appropriate fixtures. This also adds new tests or test cases to improve code coverage. Tests that work from the black box perspective are now located in a subdirectory. Signed-off-by: Felix Matouschek --- .../test_kubevirt_ansible_connection_winrm.py | 80 ++ .../test_kubevirt_set_composable_vars.py | 63 + tests/unit/plugins/inventory/conftest.py | 151 +++ .../inventory/constants.py} | 3 + tests/unit/plugins/inventory/test_kubevirt.py | 1137 +++-------------- .../inventory/test_kubevirt_add_host.py | 82 ++ .../test_kubevirt_composable_vars.py | 119 -- .../inventory/test_kubevirt_fetch_objects.py | 193 +++ .../test_kubevirt_format_dynamic_api_exc.py | 56 + .../inventory/test_kubevirt_get_resources.py | 106 ++ ...kubevirt_get_ssh_services_for_namespace.py | 168 +++ .../test_kubevirt_inventory_options.py | 57 + ...evirt_populate_inventory_from_namespace.py | 163 +++ ...test_kubevirt_set_ansible_host_and_port.py | 286 +++++ .../test_kubevirt_set_common_vars.py | 122 ++ .../test_kubevirt_set_vars_from_vmi.py | 120 ++ .../plugins/inventory/test_kubevirt_setup.py | 63 + .../inventory/test_kubevirt_verify_file.py | 45 + 18 files changed, 1927 insertions(+), 1087 deletions(-) create mode 100644 tests/unit/plugins/inventory/blackbox/test_kubevirt_ansible_connection_winrm.py create mode 100644 tests/unit/plugins/inventory/blackbox/test_kubevirt_set_composable_vars.py create mode 100644 tests/unit/plugins/inventory/conftest.py rename tests/unit/{utils/merge_dicts.py => plugins/inventory/constants.py} (89%) create mode 100644 tests/unit/plugins/inventory/test_kubevirt_add_host.py delete mode 100644 tests/unit/plugins/inventory/test_kubevirt_composable_vars.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_fetch_objects.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_format_dynamic_api_exc.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_get_resources.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_get_ssh_services_for_namespace.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_inventory_options.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_populate_inventory_from_namespace.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_set_ansible_host_and_port.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_set_common_vars.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_set_vars_from_vmi.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_setup.py create mode 100644 tests/unit/plugins/inventory/test_kubevirt_verify_file.py diff --git a/tests/unit/plugins/inventory/blackbox/test_kubevirt_ansible_connection_winrm.py b/tests/unit/plugins/inventory/blackbox/test_kubevirt_ansible_connection_winrm.py new file mode 100644 index 0000000..22cc9c8 --- /dev/null +++ b/tests/unit/plugins/inventory/blackbox/test_kubevirt_ansible_connection_winrm.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, + merge_dicts, +) + +BASE_VMI = { + "metadata": { + "name": "testvmi", + "namespace": "default", + }, + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + }, +} +WINDOWS_VMI_1 = merge_dicts( + BASE_VMI, + { + "status": { + "guestOSInfo": {"id": "mswindows"}, + } + }, +) +WINDOWS_VMI_2 = merge_dicts( + BASE_VMI, + { + "metadata": { + "annotations": {"kubevirt.io/cluster-preference-name": "windows.2k22"} + }, + }, +) +WINDOWS_VMI_3 = merge_dicts( + BASE_VMI, + { + "metadata": {"annotations": {"kubevirt.io/preference-name": "windows.2k22"}}, + }, +) +WINDOWS_VMI_4 = merge_dicts( + BASE_VMI, + { + "metadata": {"annotations": {"vm.kubevirt.io/os": "windows2k22"}}, + }, +) + + +@pytest.mark.parametrize( + "client,vmi,expected", + [ + ({"vmis": [BASE_VMI]}, BASE_VMI, False), + ({"vmis": [WINDOWS_VMI_1]}, WINDOWS_VMI_1, True), + ({"vmis": [WINDOWS_VMI_2]}, WINDOWS_VMI_2, True), + ({"vmis": [WINDOWS_VMI_3]}, WINDOWS_VMI_3, True), + ({"vmis": [WINDOWS_VMI_4]}, WINDOWS_VMI_4, True), + ], + indirect=["client"], +) +def test_ansible_connection_winrm(inventory, hosts, client, vmi, expected): + inventory.populate_inventory_from_namespace( + client, "", DEFAULT_NAMESPACE, InventoryOptions() + ) + + host = f"{DEFAULT_NAMESPACE}-{vmi['metadata']['name']}" + if expected: + assert hosts[host]["ansible_connection"] == "winrm" + else: + assert "ansible_connection" not in hosts[host] diff --git a/tests/unit/plugins/inventory/blackbox/test_kubevirt_set_composable_vars.py b/tests/unit/plugins/inventory/blackbox/test_kubevirt_set_composable_vars.py new file mode 100644 index 0000000..01bf367 --- /dev/null +++ b/tests/unit/plugins/inventory/blackbox/test_kubevirt_set_composable_vars.py @@ -0,0 +1,63 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, +) + + +VMI = { + "metadata": { + "name": "testvmi", + "namespace": "default", + }, + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + "migrationMethod": "BlockMigration", + "nodeName": "test-node", + "guestOSInfo": { + "id": "fedora", + "versionId": "40", + }, + }, +} + + +@pytest.mark.parametrize( + "client", + [{"vmis": [VMI]}], + indirect=["client"], +) +def test_set_composable_vars( + inventory, + groups, + hosts, + client, +): + inventory._options = { + "compose": {"set_from_another_var": "vmi_node_name"}, + "groups": {"block_migratable_vmis": "vmi_migration_method == 'BlockMigration'"}, + "keyed_groups": [{"prefix": "fedora", "key": "vmi_guest_os_info.versionId"}], + "strict": True, + } + inventory.populate_inventory_from_namespace( + client, "", DEFAULT_NAMESPACE, InventoryOptions() + ) + + host = f"{DEFAULT_NAMESPACE}-testvmi" + assert hosts[host]["set_from_another_var"] == "test-node" + assert "block_migratable_vmis" in groups + assert host in groups["block_migratable_vmis"]["children"] + assert "fedora_40" in groups + assert host in groups["fedora_40"]["children"] diff --git a/tests/unit/plugins/inventory/conftest.py b/tests/unit/plugins/inventory/conftest.py new file mode 100644 index 0000000..f51606c --- /dev/null +++ b/tests/unit/plugins/inventory/conftest.py @@ -0,0 +1,151 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from addict import Dict + +from ansible.template import Templar + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryModule, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_BASE_DOMAIN, + DEFAULT_NAMESPACE, +) + + +@pytest.fixture(scope="function") +def inventory(mocker): + inventory = InventoryModule() + inventory.inventory = mocker.Mock() + inventory.templar = Templar(loader=None) + inventory._options = { + "compose": {}, + "groups": {}, + "keyed_groups": [], + "strict": True, + } + return inventory + + +@pytest.fixture(scope="function") +def inventory_data(mocker, inventory): + groups = {} + hosts = {} + + def add_group(group): + if group not in groups: + groups[group] = {"children": [], "vars": {}} + return group + + def add_child(group, name): + if name not in groups[group]["children"]: + groups[group]["children"].append(name) + + def add_host(host, group=None): + if host not in hosts: + hosts[host] = {} + if group is not None: + add_child(group, host) + + def get_host(hostname): + host = mocker.Mock() + host.get_vars = mocker.Mock(return_value=hosts[hostname]) + return host + + def set_variable(name, key, value): + if name in groups: + groups[name]["vars"][key] = value + else: + hosts[name][key] = value + + mocker.patch.object(inventory.inventory, "add_group", add_group) + mocker.patch.object(inventory.inventory, "add_child", add_child) + mocker.patch.object(inventory.inventory, "add_host", add_host) + mocker.patch.object(inventory.inventory, "get_host", get_host) + mocker.patch.object(inventory.inventory, "set_variable", set_variable) + return groups, hosts + + +@pytest.fixture(scope="function") +def groups(inventory_data): + return inventory_data[0] + + +@pytest.fixture(scope="function") +def hosts(inventory_data): + return inventory_data[1] + + +@pytest.fixture(scope="function") +def client(mocker, request): + param = {} + if hasattr(request, "param"): + param = request.param + + namespaces = mocker.Mock() + if "namespaces" in param: + items = param["namespaces"] + else: + items = [{"metadata": {"name": DEFAULT_NAMESPACE}}] + namespaces.items = [Dict(item) for item in items] + + vms = mocker.Mock() + vms.items = [Dict(item) for item in param.get("vms", [])] + vmis = mocker.Mock() + vmis.items = [Dict(item) for item in param.get("vmis", [])] + services = mocker.Mock() + services.items = [Dict(item) for item in param.get("services", [])] + + dns = mocker.Mock() + if "base_domain" in param: + base_domain = param["base_domain"] + else: + base_domain = DEFAULT_BASE_DOMAIN + dns_obj = Dict({"spec": {"baseDomain": base_domain}}) + dns.items = [dns_obj] + + namespace_client = mocker.Mock() + namespace_client.get = mocker.Mock(return_value=namespaces) + vm_client = mocker.Mock() + vm_client.get = mocker.Mock(return_value=vms) + vmi_client = mocker.Mock() + vmi_client.get = mocker.Mock(return_value=vmis) + service_client = mocker.Mock() + service_client.get = mocker.Mock(return_value=services) + + def dns_client_get(**kwargs): + if "name" in kwargs: + return dns_obj + return dns + + dns_client = mocker.Mock() + dns_client.get = dns_client_get + + def resources_get(api_version="", kind=""): + if api_version.lower() == "v1": + if kind.lower() == "namespace": + return namespace_client + if kind.lower() == "service": + return service_client + elif api_version.lower() == "config.openshift.io/v1" and kind.lower() == "dns": + return dns_client + elif "kubevirt.io/" in api_version.lower(): + if kind.lower() == "virtualmachine": + return vm_client + if kind.lower() == "virtualmachineinstance": + return vmi_client + + return None + + client = mocker.Mock() + client.resources.get = resources_get + return client diff --git a/tests/unit/utils/merge_dicts.py b/tests/unit/plugins/inventory/constants.py similarity index 89% rename from tests/unit/utils/merge_dicts.py rename to tests/unit/plugins/inventory/constants.py index c82863c..4d47f86 100644 --- a/tests/unit/utils/merge_dicts.py +++ b/tests/unit/plugins/inventory/constants.py @@ -8,6 +8,9 @@ __metaclass__ = type from copy import deepcopy +DEFAULT_NAMESPACE = "default" +DEFAULT_BASE_DOMAIN = "example.com" + def merge_dicts(dict1, dict2): merged = deepcopy(dict1) diff --git a/tests/unit/plugins/inventory/test_kubevirt.py b/tests/unit/plugins/inventory/test_kubevirt.py index 2617b58..477abf6 100644 --- a/tests/unit/plugins/inventory/test_kubevirt.py +++ b/tests/unit/plugins/inventory/test_kubevirt.py @@ -6,424 +6,168 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type -from json import dumps - import pytest from addict import Dict - from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( - DynamicApiError, - GetVmiOptions, InventoryModule, - KubeVirtInventoryException, ) -from ansible_collections.kubevirt.core.tests.unit.utils.merge_dicts import ( - merge_dicts, +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_BASE_DOMAIN, ) -from ansible_collections.kubevirt.core.plugins.inventory import kubevirt - -DEFAULT_NAMESPACE = "default" -DEFAULT_BASE_DOMAIN = "example.com" - -BASE_VMI = { - "apiVersion": "kubevirt.io/v1", - "kind": "VirtualMachineInstance", - "metadata": { - "name": "testvmi", - "namespace": "default", - "uid": "f8abae7c-d792-4b9b-af95-62d322ae5bc1", - "labels": {"kubevirt.io/domain": "test-domain"}, - }, - "spec": { - "domain": {"devices": {}}, - }, - "status": { - "interfaces": [{"ipAddress": "10.10.10.10"}], - }, -} -NO_STATUS_VMI = merge_dicts( - BASE_VMI, - { - "status": None, - }, -) -VMI_WITH_INTERFACE_NO_IPADDRESS = merge_dicts( - BASE_VMI, {"status": {"interfaces": [{"ipAddress": None}]}} -) -WINDOWS_VMI_1 = merge_dicts( - BASE_VMI, - { - "status": { - "guestOSInfo": {"id": "mswindows"}, - } - }, -) -WINDOWS_VMI_2 = merge_dicts( - BASE_VMI, - { - "metadata": { - "annotations": {"kubevirt.io/cluster-preference-name": "windows.2k22"} - }, - }, -) -WINDOWS_VMI_3 = merge_dicts( - BASE_VMI, - { - "metadata": {"annotations": {"kubevirt.io/preference-name": "windows.2k22"}}, - }, -) -WINDOWS_VMI_4 = merge_dicts( - BASE_VMI, - { - "metadata": {"annotations": {"vm.kubevirt.io/os": "windows2k22"}}, - }, -) -COMPLETE_VMI = merge_dicts( - BASE_VMI, - { - "metadata": { - "annotations": {"test-annotation": "test-annotation"}, - "clusterName": {"test-cluster"}, - "resourceVersion": {"42"}, - }, - "status": { - "activePods": {"d5b85485-354b-40d3-b6a0-23e18b685310": "node01"}, - "conditions": [ - { - "status": True, - "type": "Ready", - "lastProbeTime": "null", - "lastTransitionTime": "null", - }, - ], - "guestOSInfo": { - "id": "fedora", - "version": "39", - }, - "launcherContainerImageVersion": {"quay.io/kubevirt/virt-launcher:v1.1.0"}, - "migrationMethod": "BlockMigration", - "migrationTransport": "Unix", - "nodeName": "node01", - "phase": "Running", - "phaseTransitionTimestamps": [ - { - "phase": "Running", - "phaseTransitionTimestamp": "null", - }, - ], - "qosClass": "Burstable", - "virtualMachineRevisionName": "revision-start-vm-12345", - "volumeStatus": [ - {"name": "cloudinit", "size": 1048576, "target": "vdb"}, - { - "name": "containerdisk", - "target": "vda", - }, - ], - }, - }, -) -COMPLETE_VMI_WITH_NETWORK_NAME = merge_dicts( - COMPLETE_VMI, - {"status": {"interfaces": [{"ipAddress": "10.10.10.10", "name": "test-network"}]}}, -) - -BASE_SERVICE = { - "apiVersion": "v1", - "kind": "Service", - "metadata": {"name": "test-service"}, - "spec": {}, -} -BASE_LOADBALANCER_SERVICE = merge_dicts( - BASE_SERVICE, {"spec": {"type": "LoadBalancer"}} -) -LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT = merge_dicts( - BASE_LOADBALANCER_SERVICE, - { - "spec": { - "ports": [ - { - "protocol": "TCP", - "port": 80, - "targetPort": 80, - }, - ], - "type": "LoadBalancer", - } - }, -) -LOADBALANCER_SERVICE_WITHOUT_SELECTOR = merge_dicts( - BASE_LOADBALANCER_SERVICE, - { - "spec": { - "ports": [ - { - "protocol": "TCP", - "port": 22, - "targetPort": 22, - }, - ], - "type": "LoadBalancer", - } - }, -) -LOADBALANCER_SERVICE = merge_dicts( - LOADBALANCER_SERVICE_WITHOUT_SELECTOR, - {"spec": {"selector": {"kubevirt.io/domain": "test-domain"}}}, -) -NODEPORT_SERVICE = merge_dicts( - LOADBALANCER_SERVICE, - { - "spec": { - "type": "NodePort", - } - }, -) - - -BASIC_VMI_HOST_VARS = { - "default-testvmi": { - "object_type": "vmi", - "labels": {"kubevirt.io/domain": "test-domain"}, - "annotations": {}, - "cluster_name": {}, - "resource_version": {}, - "uid": "f8abae7c-d792-4b9b-af95-62d322ae5bc1", - "vmi_active_pods": {}, - "vmi_conditions": [], - "vmi_guest_os_info": {}, - "vmi_interfaces": [{"ipAddress": "10.10.10.10"}], - "vmi_launcher_container_image_version": {}, - "vmi_migration_method": {}, - "vmi_migration_transport": {}, - "vmi_node_name": {}, - "vmi_phase": {}, - "vmi_phase_transition_timestamps": [], - "vmi_qos_class": {}, - "vmi_virtual_machine_revision_name": {}, - "vmi_volume_status": [], - } -} -COMPLETE_VMI_HOST_VARS = { - "default-testvmi": merge_dicts( - BASIC_VMI_HOST_VARS["default-testvmi"], - { - "labels": {"kubevirt.io/domain": "test-domain"}, - "annotations": {"test-annotation": "test-annotation"}, - "cluster_name": {"test-cluster"}, - "resource_version": {"42"}, - "vmi_active_pods": {"d5b85485-354b-40d3-b6a0-23e18b685310": "node01"}, - "vmi_conditions": [ - { - "status": True, - "type": "Ready", - "lastProbeTime": "null", - "lastTransitionTime": "null", - }, - ], - "vmi_guest_os_info": {"id": "fedora", "version": "39"}, - "vmi_launcher_container_image_version": { - "quay.io/kubevirt/virt-launcher:v1.1.0" - }, - "vmi_migration_method": "BlockMigration", - "vmi_migration_transport": "Unix", - "vmi_node_name": "node01", - "vmi_phase": "Running", - "vmi_phase_transition_timestamps": [ - { - "phase": "Running", - "phaseTransitionTimestamp": "null", - }, - ], - "vmi_qos_class": "Burstable", - "vmi_virtual_machine_revision_name": "revision-start-vm-12345", - "vmi_volume_status": [ - {"name": "cloudinit", "size": 1048576, "target": "vdb"}, - {"name": "containerdisk", "target": "vda"}, - ], - }, - ) -} -WINDOWS_VMI_HOST_VARS = { - "default-testvmi": merge_dicts( - BASIC_VMI_HOST_VARS["default-testvmi"], - { - "ansible_connection": "winrm", - "vmi_guest_os_info": {"id": "mswindows"}, - }, - ) -} -COMPLETE_VMI_HOST_VARS_WITH_NETWORK = { - "default-testvmi": merge_dicts( - COMPLETE_VMI_HOST_VARS["default-testvmi"], - { - "vmi_interfaces": [ - { - "ipAddress": "10.10.10.10", - "name": "test-network", - }, - ], - }, - ) -} - - -@pytest.fixture(scope="function") -def inventory(mocker): - inventory = InventoryModule() - inventory.inventory = mocker.Mock() - mocker.patch.object(inventory, "set_composable_vars") - return inventory - - -@pytest.fixture(scope="function") -def host_vars(monkeypatch, inventory): - host_vars = {} - - def set_variable(host, key, value): - if host not in host_vars: - host_vars[host] = {} - host_vars[host][key] = value - - monkeypatch.setattr(inventory.inventory, "set_variable", set_variable) - return host_vars - - -@pytest.fixture(scope="function") -def add_group(monkeypatch, inventory): - groups = [] - - def add_group(name): - if name not in groups: - groups.append(name) - return name - - monkeypatch.setattr(inventory.inventory, "add_group", add_group) - return groups - - -@pytest.fixture(scope="function") -def add_host(monkeypatch, inventory): - hosts = [] - - def add_host(name, group=None): - if name not in hosts: - hosts.append(name) - if group is not None and group not in hosts: - hosts.append(group) - - monkeypatch.setattr(inventory.inventory, "add_host", add_host) - return hosts - - -@pytest.fixture(scope="function") -def add_child(monkeypatch, inventory): - children = {} - - def add_child(group, name): - if group not in children: - children[group] = [] - if name not in children[group]: - children[group].append(name) - - monkeypatch.setattr(inventory.inventory, "add_child", add_child) - return children - - -@pytest.fixture(scope="function") -def client(mocker, request): - namespaces = mocker.Mock() - namespaces.items = [ - Dict(item) - for item in request.param.get( - "namespaces", [{"metadata": {"name": DEFAULT_NAMESPACE}}] - ) - ] - vmis = mocker.Mock() - vmis.items = [Dict(item) for item in request.param.get("vmis", [])] - services = mocker.Mock() - services.items = [Dict(item) for item in request.param.get("services", [])] - dns = Dict( - {"spec": {"baseDomain": request.param.get("base_domain", DEFAULT_BASE_DOMAIN)}} - ) - - namespace_client = mocker.Mock() - namespace_client.get = mocker.Mock(return_value=namespaces) - vmi_client = mocker.Mock() - vmi_client.get = mocker.Mock(return_value=vmis) - service_client = mocker.Mock() - service_client.get = mocker.Mock(return_value=services) - dns_client = mocker.Mock() - dns_client.get = mocker.Mock(return_value=dns) - - def resources_get(api_version="", kind=""): - if api_version.lower() == "v1": - if kind.lower() == "namespace": - return namespace_client - elif kind.lower() == "service": - return service_client - elif api_version.lower() == "config.openshift.io/v1" and kind.lower() == "dns": - return dns_client - elif ( - "kubevirt.io/" in api_version.lower() - and kind.lower() == "virtualmachineinstance" - ): - return vmi_client - - return None - - client = mocker.Mock() - client.resources.get = resources_get - return client - @pytest.mark.parametrize( - "file_name,expected", + "host,expected", [ - ("inventory.kubevirt.yml", True), - ("inventory.kubevirt.yaml", True), - ("something.kubevirt.yml", True), - ("something.kubevirt.yaml", True), - ("inventory.somethingelse.yml", False), - ("inventory.somethingelse.yaml", False), - ("something.somethingelse.yml", False), - ("something.somethingelse.yaml", False), + ("https://example", "example"), + ("http://example", "example"), + ("example.com", "example-com"), + ("https://example:8080", "example_8080"), + ("https://example.com:8080", "example-com_8080"), ], ) -def test_verify_file(tmp_path, inventory, file_name, expected): - file = tmp_path / file_name - file.touch() - assert inventory.verify_file(str(file)) is expected +def test_get_default_host_name(host, expected): + assert InventoryModule.get_default_host_name(host) == expected @pytest.mark.parametrize( - "file_name", + "name,expected", [ - "inventory.kubevirt.yml", - "inventory.kubevirt.yaml", - "something.kubevirt.yml", - "something.kubevirt.yaml", - "inventory.somethingelse.yml", - "inventory.somethingelse.yaml", - "something.somethingelse.yml", - "something.somethingelse.yaml", + ("lowercase", "lowercase"), + ("ONLYUPPERCASE", "onlyuppercase"), + ("snake_case", "snake_case"), + ("snake_CASE", "snake_case"), + ("TestNameOne", "test_name_one"), + ("TestNameTWO", "test_name_two"), ], ) -def test_verify_file_bad_config(inventory, file_name): - assert inventory.verify_file(file_name) is False +def test_format_var_name(name, expected): + assert InventoryModule.format_var_name(name) == expected + + +@pytest.mark.parametrize( + "service,node_name,expected", + [ + ({}, None, None), + ({"spec": {"something": "something"}}, None, None), + ({"spec": {"type": "ClusterIP"}}, None, None), + ({"spec": {"type": "LoadBalancer"}}, None, None), + ({"spec": {"type": "LoadBalancer"}, "status": {}}, None, None), + ( + {"spec": {"type": "LoadBalancer"}, "status": {"loadBalancer": {}}}, + None, + None, + ), + ( + { + "spec": { + "type": "LoadBalancer", + }, + "status": {"loadBalancer": {"ingress": [{"ip": "192.168.1.100"}]}}, + }, + None, + "192.168.1.100", + ), + ( + { + "spec": { + "type": "LoadBalancer", + }, + "status": { + "loadBalancer": {"ingress": [{"hostname": "test-hostname"}]}, + }, + }, + None, + "test-hostname", + ), + ( + { + "spec": { + "type": "LoadBalancer", + }, + "status": { + "loadBalancer": { + "ingress": [ + {"hostname": "test-hostname", "ip": "192.168.1.100"} + ] + }, + }, + }, + None, + "test-hostname", + ), + ({"spec": {"type": "NodePort"}}, "test-nodename", "test-nodename"), + ], +) +def test_get_host_from_service(service, node_name, expected): + assert InventoryModule.get_host_from_service(service, node_name) == expected + + +@pytest.mark.parametrize( + "service,expected", + [ + ({}, None), + ({"spec": {"type": "LoadBalancer", "ports": []}}, None), + ({"spec": {"type": "ClusterIP"}}, None), + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"garbage": "80"}], + }, + }, + None, + ), + ( + { + "spec": { + "type": "NodePort", + "ports": [{"garbage": "8080"}], + } + }, + None, + ), + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": "80"}], + }, + }, + "80", + ), + ( + { + "spec": { + "type": "NodePort", + "ports": [{"nodePort": "8080"}], + } + }, + "8080", + ), + ], +) +def test_port_from_service(service, expected): + assert InventoryModule.get_port_from_service(service) == expected @pytest.mark.parametrize( "guest_os_info,annotations,expected", [ - ({"id": "mswindows"}, {}, True), - ({}, {"kubevirt.io/preference-name": "windows.2k22"}, True), - ({}, {"vm.kubevirt.io/os": "windows2k22"}, True), + (None, None, False), ({}, {}, False), + ({"id": "mswindows"}, None, True), + ({"id": "mswindows"}, {}, True), + (None, {"kubevirt.io/preference-name": "windows.2k22"}, True), + ({}, {"kubevirt.io/preference-name": "windows.2k22"}, True), + (None, {"kubevirt.io/cluster-preference-name": "windows.2k22"}, True), + ({}, {"kubevirt.io/cluster-preference-name": "windows.2k22"}, True), + (None, {"vm.kubevirt.io/os": "windows2k22"}, True), + ({}, {"vm.kubevirt.io/os": "windows2k22"}, True), + ({"id": "fedora"}, None, False), ({"id": "fedora"}, {}, False), ( {"id": "fedora"}, @@ -458,594 +202,51 @@ def test_verify_file_bad_config(inventory, file_name): ), ], ) -def test_is_windows(inventory, guest_os_info, annotations, expected): - assert inventory.is_windows(guest_os_info, annotations) is expected +def test_is_windows(guest_os_info, annotations, expected): + assert InventoryModule.is_windows(guest_os_info, annotations) == expected -@pytest.mark.parametrize( - "client,vmi,expected", - [ - ({"vmis": [BASE_VMI]}, BASE_VMI, False), - ({"vmis": [WINDOWS_VMI_1]}, WINDOWS_VMI_1, True), - ({"vmis": [WINDOWS_VMI_2]}, WINDOWS_VMI_2, True), - ({"vmis": [WINDOWS_VMI_3]}, WINDOWS_VMI_3, True), - ({"vmis": [WINDOWS_VMI_4]}, WINDOWS_VMI_4, True), - ], - indirect=["client"], -) -def test_ansible_connection_winrm(inventory, host_vars, client, vmi, expected): - inventory.get_vmis_for_namespace(client, "", DEFAULT_NAMESPACE, GetVmiOptions()) +def test_parse(mocker, inventory): + path = "/testpath" + cache_prefix = "test-prefix" + host_format = "test-format" + config_data = {"host_format": host_format} + cache = True - host = f"{DEFAULT_NAMESPACE}-{vmi['metadata']['name']}" - if expected: - assert host_vars[host]["ansible_connection"] == "winrm" - else: - assert "ansible_connection" not in host_vars[host] - - -@pytest.mark.parametrize( - "url,host_name", - [ - ("https://example", "example"), - ("http://example", "example"), - ("example.com", "example-com"), - ("https://example:8080", "example_8080"), - ("https://example.com:8080", "example-com_8080"), - ], -) -def test_get_default_host_name(inventory, url, host_name): - result = inventory.get_default_host_name(url) - assert result == host_name - - -@pytest.mark.parametrize( - "service,node_name,expected", - [ - ({"spec": {"type": "ClusterIP"}}, None, None), - ( - { - "spec": { - "type": "LoadBalancer", - }, - "status": {"loadBalancer": {"ingress": [{"ip": "192.168.1.100"}]}}, - }, - None, - "192.168.1.100", - ), - ( - { - "spec": { - "type": "LoadBalancer", - }, - "status": { - "loadBalancer": {"ingress": [{"hostname": "test-hostname"}]}, - }, - }, - None, - "test-hostname", - ), - ({"spec": {"type": "NodePort"}}, "test-nodename", "test-nodename"), - ], -) -def test_get_host_from_service(inventory, service, node_name, expected): - assert inventory.get_host_from_service(service, node_name) == expected - - -@pytest.mark.parametrize( - "service,port", - [ - ({"spec": {"type": "ClusterIP"}}, None), - ( - { - "spec": { - "type": "LoadBalancer", - "ports": [{"port": "80"}], - }, - }, - "80", - ), - ( - { - "spec": { - "type": "NodePort", - "ports": [{"nodePort": "8080"}], - } - }, - "8080", - ), - ], -) -def test_port_from_service(inventory, service, port): - assert port == inventory.get_port_from_service(service) - - -def test_parse(monkeypatch, inventory): - monkeypatch.setattr( - inventory, "_read_config_data", lambda path: {"host_format": "default-test"} + get_cache_prefix = mocker.patch.object( + inventory, "_get_cache_prefix", return_value=cache_prefix ) - monkeypatch.setattr(inventory, "_get_cache_prefix", lambda _: None) - monkeypatch.setattr(inventory, "setup", lambda a, b, c: None) - - inventory.parse(inventory, None, "test") - assert inventory.host_format == "default-test" - - -@pytest.mark.parametrize( - "connections,result,default_namespace", - [ - ( - [ - { - "name": "test", - }, - ], - {"name": "test", "namespace": DEFAULT_NAMESPACE, "opts": GetVmiOptions()}, - True, - ), - ( - [ - {"name": "test", "namespaces": ["test"]}, - ], - {"name": "test", "namespace": "test", "opts": GetVmiOptions()}, - False, - ), - ( - [ - { - "name": "test", - "namespaces": ["test"], - "use_service": True, - "create_groups": True, - "append_base_domain": True, - "base_domain": "test-domain", - }, - ], - { - "name": "test", - "namespace": "test", - "opts": GetVmiOptions( - use_service=True, - create_groups=True, - append_base_domain=True, - base_domain="test-domain", - ), - }, - False, - ), - ( - [ - { - "name": "test", - "namespaces": ["test"], - "use_service": True, - "create_groups": True, - "append_base_domain": True, - "base_domain": "test-domain", - "network_name": "test-network", - }, - ], - { - "name": "test", - "namespace": "test", - "opts": GetVmiOptions( - use_service=True, - create_groups=True, - append_base_domain=True, - base_domain="test-domain", - network_name="test-network", - ), - }, - False, - ), - ( - [ - { - "name": "test", - "namespaces": ["test"], - "use_service": True, - "create_groups": True, - "append_base_domain": True, - "base_domain": "test-domain", - "interface_name": "test-interface", - }, - ], - { - "name": "test", - "namespace": "test", - "opts": GetVmiOptions( - use_service=True, - create_groups=True, - append_base_domain=True, - base_domain="test-domain", - network_name="test-interface", - ), - }, - False, - ), - ( - None, - { - "name": "default-hostname", - "namespace": DEFAULT_NAMESPACE, - "opts": GetVmiOptions(), - }, - True, - ), - ], -) -def test_fetch_objects( - mocker, monkeypatch, inventory, connections, result, default_namespace -): - monkeypatch.setattr(kubevirt, "get_api_client", lambda **_: mocker.Mock()) - monkeypatch.setattr( - inventory, "get_default_host_name", lambda _: "default-hostname" + read_config_data = mocker.patch.object( + inventory, "_read_config_data", return_value=config_data ) - monkeypatch.setattr(inventory, "get_cluster_domain", lambda _: None) + setup = mocker.patch.object(inventory, "setup") - get_vmis_for_namespace = mocker.patch.object(inventory, "get_vmis_for_namespace") - get_available_namespaces = mocker.patch.object( - inventory, "get_available_namespaces" - ) + inventory.parse(None, None, path, cache) - if default_namespace: - get_available_namespaces.return_value = [DEFAULT_NAMESPACE] - inventory.fetch_objects(connections) - get_available_namespaces.assert_called_once_with(mocker.ANY) - else: - inventory.fetch_objects(connections) - get_available_namespaces.assert_not_called() + get_cache_prefix.assert_called_once_with(path) + read_config_data.assert_called_once_with(path) + setup.assert_called_once_with(config_data, cache, cache_prefix) + assert inventory.host_format == host_format - get_vmis_for_namespace.assert_called_once_with( - mocker.ANY, result["name"], result["namespace"], result["opts"] - ) + +def test_get_cluster_domain(inventory, client): + assert inventory.get_cluster_domain(client) == DEFAULT_BASE_DOMAIN @pytest.mark.parametrize( - "connections,result", + "labels,expected", [ - ("test", "Expecting connections to be a list."), - (["test", "test"], "Expecting connection to be a dictionary."), + ({}, []), + ({"testkey": "testval"}, ["label_testkey_testval"]), + ( + {"testkey1": "testval", "testkey2": "testval"}, + ["label_testkey1_testval", "label_testkey2_testval"], + ), ], ) -def test_fetch_objects_exceptions(inventory, connections, result): - with pytest.raises(KubeVirtInventoryException, match=result): - inventory.fetch_objects(connections) - - -@pytest.mark.parametrize( - "client,result", - [ - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - DEFAULT_BASE_DOMAIN, - ) - ], - indirect=["client"], -) -def test_get_cluster_domain(inventory, client, result): - assert result == inventory.get_cluster_domain(client) - - -@pytest.mark.parametrize( - "client,result", - [ - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - [DEFAULT_NAMESPACE], - ), - ( - { - "namespaces": [ - {"metadata": {"name": DEFAULT_NAMESPACE}}, - {"metadata": {"name": "test"}}, - ] - }, - [DEFAULT_NAMESPACE, "test"], - ), - ], - indirect=["client"], -) -def test_get_available_namespaces(request, inventory, client, result): - assert result == inventory.get_available_namespaces(client) - - -@pytest.mark.parametrize( - "client,vmi,groups,vmi_group,child_group,create_groups,network_name,expected_host_vars,call_functions,windows", - [ - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - None, - [], - [], - {}, - False, - None, - {}, - False, - False, - ), - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - None, - [], - [], - {}, - False, - None, - {}, - False, - False, - ), - ( - {"vmis": [NO_STATUS_VMI]}, - NO_STATUS_VMI, - ["test", "namespace_default"], - [], - {"test": ["namespace_default"]}, - False, - None, - {}, - False, - False, - ), - ( - {"vmis": [VMI_WITH_INTERFACE_NO_IPADDRESS]}, - BASE_VMI, - ["test", "namespace_default"], - [], - {"test": ["namespace_default"]}, - False, - None, - {}, - False, - False, - ), - ( - {"vmis": [BASE_VMI], "services": [LOADBALANCER_SERVICE]}, - BASE_VMI, - ["test", "namespace_default"], - ["default-testvmi"], - {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, - False, - None, - BASIC_VMI_HOST_VARS, - True, - False, - ), - ( - {"vmis": [COMPLETE_VMI], "services": [LOADBALANCER_SERVICE]}, - COMPLETE_VMI, - ["test", "namespace_default"], - ["default-testvmi"], - {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, - False, - None, - COMPLETE_VMI_HOST_VARS, - True, - False, - ), - ( - {"vmis": [COMPLETE_VMI], "services": [LOADBALANCER_SERVICE]}, - COMPLETE_VMI, - ["test", "namespace_default", "label_kubevirt_io_domain_test_domain"], - ["default-testvmi"], - { - "test": ["namespace_default"], - "namespace_default": ["default-testvmi"], - "label_kubevirt_io_domain_test_domain": ["default-testvmi"], - }, - True, - None, - COMPLETE_VMI_HOST_VARS, - True, - False, - ), - ( - {"vmis": [WINDOWS_VMI_1]}, - WINDOWS_VMI_1, - ["test", "namespace_default"], - ["default-testvmi"], - {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, - False, - None, - WINDOWS_VMI_HOST_VARS, - True, - True, - ), - ( - { - "vmis": [COMPLETE_VMI_WITH_NETWORK_NAME], - "services": [LOADBALANCER_SERVICE], - }, - COMPLETE_VMI_WITH_NETWORK_NAME, - ["test", "namespace_default", "label_kubevirt_io_domain_test_domain"], - ["default-testvmi"], - { - "test": ["namespace_default"], - "namespace_default": ["default-testvmi"], - "label_kubevirt_io_domain_test_domain": ["default-testvmi"], - }, - True, - "test-network", - COMPLETE_VMI_HOST_VARS_WITH_NETWORK, - True, - False, - ), - ], - indirect=["client"], -) -def test_get_vmis_for_namespace( - mocker, - inventory, - vmi, - host_vars, - add_group, - add_host, - add_child, - client, - groups, - vmi_group, - child_group, - create_groups, - network_name, - expected_host_vars, - call_functions, - windows, -): - set_ansible_host_and_port = mocker.patch.object( - inventory, "set_ansible_host_and_port" - ) - set_composable_vars = mocker.patch.object(inventory, "set_composable_vars") - - inventory.get_vmis_for_namespace( - client, - "test", - DEFAULT_NAMESPACE, - GetVmiOptions(create_groups=create_groups, network_name=network_name), - ) - - assert groups == add_group - assert vmi_group == add_host - assert child_group == add_child - assert expected_host_vars == host_vars - - if call_functions: - vmi_name = f"{DEFAULT_NAMESPACE}-{vmi['metadata']['name']}" - service = None if windows else LOADBALANCER_SERVICE - set_ansible_host_and_port.assert_called_once_with( - vmi, - vmi_name, - vmi["status"]["interfaces"][0]["ipAddress"], - service, - GetVmiOptions(create_groups=create_groups, network_name=network_name), - ) - set_composable_vars.assert_called_once_with(vmi_name) - else: - set_composable_vars.assert_not_called() - set_ansible_host_and_port.asser_not_called() - - -@pytest.mark.parametrize( - "client,result", - [ - ( - { - "services": [LOADBALANCER_SERVICE], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - { - "test-domain": { - "apiVersion": "v1", - "kind": "Service", - "metadata": { - "name": "test-service", - }, - "spec": { - "selector": { - "kubevirt.io/domain": "test-domain", - }, - "ports": [{"protocol": "TCP", "port": 22, "targetPort": 22}], - "type": "LoadBalancer", - }, - }, - }, - ), - ( - { - "services": [NODEPORT_SERVICE], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - { - "test-domain": { - "apiVersion": "v1", - "kind": "Service", - "metadata": { - "name": "test-service", - }, - "spec": { - "selector": { - "kubevirt.io/domain": "test-domain", - }, - "ports": [{"protocol": "TCP", "port": 22, "targetPort": 22}], - "type": "NodePort", - }, - }, - }, - ), - ( - { - "services": [BASE_SERVICE], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - {}, - ), - ( - { - "services": [BASE_LOADBALANCER_SERVICE], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - {}, - ), - ( - { - "services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - {}, - ), - ( - { - "services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR], - "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], - }, - {}, - ), - ], - indirect=["client"], -) -def test_get_ssh_services_for_namespace(inventory, client, result): - assert result == inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) - - -@pytest.fixture(scope="function") -def body_error(mocker): - error = DynamicApiError(e=mocker.Mock()) - error.headers = None - - body = "This is a test error" - error.body = body - - return error - - -@pytest.fixture(scope="function") -def message_error(mocker): - error = DynamicApiError(e=mocker.Mock()) - error.headers = {"Content-Type": "application/json"} - - error.body = dumps({"message": "This is a test error"}).encode("utf-8") - - return error - - -@pytest.fixture(scope="function") -def status_reason_error(mocker): - error = DynamicApiError(e=mocker.Mock()) - error.body = None - error.status = 404 - error.reason = "This is a test error" - return error - - -@pytest.mark.parametrize( - "error_object,expected_error_msg", - [ - ("body_error", "This is a test error"), - ("message_error", "This is a test error"), - ("status_reason_error", "404 Reason: This is a test error"), - ], -) -def test_format_dynamic_api_exc(request, inventory, error_object, expected_error_msg): - - result = inventory.format_dynamic_api_exc(request.getfixturevalue(error_object)) - assert expected_error_msg == result +def test_set_groups_from_labels(inventory, groups, labels, expected): + hostname = "default-testvm" + inventory.set_groups_from_labels(hostname, Dict(labels)) + for group in expected: + assert group in groups + assert hostname in groups[group]["children"] diff --git a/tests/unit/plugins/inventory/test_kubevirt_add_host.py b/tests/unit/plugins/inventory/test_kubevirt_add_host.py new file mode 100644 index 0000000..fd1cb97 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_add_host.py @@ -0,0 +1,82 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from addict import Dict + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, +) + + +@pytest.mark.parametrize( + "host_format,expected", + [ + ("static", "static"), + ("{name}", "testvm"), + ("{name}-static", "testvm-static"), + ("{namespace}", "default"), + ("{uid}", "f8abae7c-d792-4b9b-af95-62d322ae5bc1"), + ("{name}-{namespace}", "testvm-default"), + ("{name}-{namespace}-static", "testvm-default-static"), + ("{name}-{uid}", "testvm-f8abae7c-d792-4b9b-af95-62d322ae5bc1"), + ("{namespace}-{name}", "default-testvm"), + ("{namespace}-{uid}", "default-f8abae7c-d792-4b9b-af95-62d322ae5bc1"), + ("{uid}-{name}", "f8abae7c-d792-4b9b-af95-62d322ae5bc1-testvm"), + ("{uid}-{namespace}", "f8abae7c-d792-4b9b-af95-62d322ae5bc1-default"), + ( + "{name}-{namespace}-{uid}", + "testvm-default-f8abae7c-d792-4b9b-af95-62d322ae5bc1", + ), + ( + "{name}-{namespace}-{uid}-static", + "testvm-default-f8abae7c-d792-4b9b-af95-62d322ae5bc1-static", + ), + ( + "{name}-{uid}-{namespace}", + "testvm-f8abae7c-d792-4b9b-af95-62d322ae5bc1-default", + ), + ( + "{namespace}-{name}-{uid}", + "default-testvm-f8abae7c-d792-4b9b-af95-62d322ae5bc1", + ), + ( + "{namespace}-{uid}-{name}", + "default-f8abae7c-d792-4b9b-af95-62d322ae5bc1-testvm", + ), + ( + "{uid}-{namespace}-{name}", + "f8abae7c-d792-4b9b-af95-62d322ae5bc1-default-testvm", + ), + ( + "{uid}-{name}-{namespace}", + "f8abae7c-d792-4b9b-af95-62d322ae5bc1-testvm-default", + ), + ], +) +def test_add_host(inventory, groups, hosts, host_format, expected): + namespace_group = "namespace_default" + inventory.inventory.add_group(namespace_group) + + inventory.add_host( + Dict( + { + "metadata": { + "name": "testvm", + "namespace": DEFAULT_NAMESPACE, + "uid": "f8abae7c-d792-4b9b-af95-62d322ae5bc1", + } + } + ), + host_format, + namespace_group, + ) + + assert expected in hosts + assert expected in groups[namespace_group]["children"] diff --git a/tests/unit/plugins/inventory/test_kubevirt_composable_vars.py b/tests/unit/plugins/inventory/test_kubevirt_composable_vars.py deleted file mode 100644 index c087226..0000000 --- a/tests/unit/plugins/inventory/test_kubevirt_composable_vars.py +++ /dev/null @@ -1,119 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2024 Red Hat, Inc. -# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) - -from __future__ import absolute_import, division, print_function - -__metaclass__ = type - -import pytest - -from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( - InventoryModule, -) - -from ansible.template import Templar - - -@pytest.fixture(scope="function") -def inventory_composable_vars(mocker): - inventory = InventoryModule() - - inventory.templar = Templar(loader=None) - - inventory._options = { - "compose": {"block_migratable_vmis": "vmi_migration_method"}, - "strict": True, - "groups": {"vmi_node_groups": "cluster_name"}, - "keyed_groups": [{"prefix": "fedora", "key": "vmi_guest_os_info.version"}], - } - - inventory.inventory = mocker.Mock() - - return inventory - - -@pytest.fixture(scope="function") -def host_vars(monkeypatch, inventory_composable_vars): - host_vars = {} - - def set_variable(host, key, value): - if host not in host_vars: - host_vars[host] = {} - host_vars[host][key] = value - - monkeypatch.setattr( - inventory_composable_vars.inventory, "set_variable", set_variable - ) - return host_vars - - -@pytest.fixture(scope="function") -def add_group(monkeypatch, inventory_composable_vars): - groups = [] - - def add_group(name): - if name not in groups: - groups.append(name) - return name - - monkeypatch.setattr(inventory_composable_vars.inventory, "add_group", add_group) - return groups - - -@pytest.fixture(scope="function") -def add_host(monkeypatch, inventory_composable_vars): - hosts = [] - - def add_host(name, group=None): - if name not in hosts: - hosts.append(name) - if group is not None and group not in hosts: - hosts.append(group) - - monkeypatch.setattr(inventory_composable_vars.inventory, "add_host", add_host) - return hosts - - -@pytest.fixture(scope="function") -def add_child(monkeypatch, inventory_composable_vars): - children = {} - - def add_child(group, name): - if group not in children: - children[group] = [] - if name not in children[group]: - children[group].append(name) - - monkeypatch.setattr(inventory_composable_vars.inventory, "add_child", add_child) - return children - - -@pytest.fixture(scope="module") -def vmi_host_vars(): - return { - "vmi_migration_method": "BlockMigration", - "vmi_guest_os_info": {"id": "fedora", "version": "39"}, - "cluster_name": {"test-cluster"}, - } - - -def test_set_composable_vars( - inventory_composable_vars, - mocker, - host_vars, - add_group, - add_child, - add_host, - vmi_host_vars, -): - get_vars = mocker.patch.object( - inventory_composable_vars.inventory.get_host(), "get_vars" - ) - get_vars.return_value = vmi_host_vars - inventory_composable_vars.set_composable_vars("testvmi") - - assert {"testvmi": {"block_migratable_vmis": "BlockMigration"}} == host_vars - assert ["vmi_node_groups", "fedora_39"] == add_group - assert {"vmi_node_groups": ["testvmi"]} == add_child - assert ["testvmi", "fedora_39"] == add_host diff --git a/tests/unit/plugins/inventory/test_kubevirt_fetch_objects.py b/tests/unit/plugins/inventory/test_kubevirt_fetch_objects.py new file mode 100644 index 0000000..14ed667 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_fetch_objects.py @@ -0,0 +1,193 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, + KubeVirtInventoryException, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, +) + +from ansible_collections.kubevirt.core.plugins.inventory import kubevirt + + +@pytest.mark.parametrize( + "connections,expected", + [ + ( + None, + [ + { + "name": "default-hostname", + "namespace": DEFAULT_NAMESPACE, + "opts": InventoryOptions(), + }, + ], + ), + ( + [ + { + "name": "test", + }, + ], + [ + { + "name": "test", + "namespace": DEFAULT_NAMESPACE, + "opts": InventoryOptions(), + }, + ], + ), + ( + [ + {"name": "test", "namespaces": ["test"]}, + ], + [ + {"name": "test", "namespace": "test", "opts": InventoryOptions()}, + ], + ), + ( + [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain", + }, + ], + [ + { + "name": "test", + "namespace": "test", + "opts": InventoryOptions( + use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain", + ), + }, + ], + ), + ( + [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain", + "network_name": "test-network", + }, + ], + [ + { + "name": "test", + "namespace": "test", + "opts": InventoryOptions( + use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain", + network_name="test-network", + ), + }, + ], + ), + ( + [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain", + "interface_name": "test-interface", + }, + ], + [ + { + "name": "test", + "namespace": "test", + "opts": InventoryOptions( + use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain", + network_name="test-interface", + ), + }, + ], + ), + ( + [ + { + "name": "test", + }, + {"name": "test", "namespaces": ["test"]}, + ], + [ + { + "name": "test", + "namespace": DEFAULT_NAMESPACE, + "opts": InventoryOptions(), + }, + {"name": "test", "namespace": "test", "opts": InventoryOptions()}, + ], + ), + ], +) +def test_fetch_objects(mocker, inventory, connections, expected): + mocker.patch.object(kubevirt, "get_api_client") + mocker.patch.object( + inventory, "get_default_host_name", return_value="default-hostname" + ) + + cluster_domain = "test.com" + mocker.patch.object(inventory, "get_cluster_domain", return_value=cluster_domain) + for e in expected: + e["opts"].base_domain = e["opts"].base_domain or cluster_domain + + get_available_namespaces = mocker.patch.object( + inventory, "get_available_namespaces", return_value=[DEFAULT_NAMESPACE] + ) + populate_inventory_from_namespace = mocker.patch.object( + inventory, "populate_inventory_from_namespace" + ) + + inventory.fetch_objects(connections) + + get_available_namespaces.assert_has_calls( + [mocker.call(mocker.ANY) for c in connections or [{}] if "namespaces" not in c] + ) + populate_inventory_from_namespace.assert_has_calls( + [ + mocker.call(mocker.ANY, e["name"], e["namespace"], e["opts"]) + for e in expected + ] + ) + + +@pytest.mark.parametrize( + "connections,expected", + [ + ("test", "Expecting connections to be a list."), + (["test", "test"], "Expecting connection to be a dictionary."), + ], +) +def test_fetch_objects_exceptions(inventory, connections, expected): + with pytest.raises(KubeVirtInventoryException, match=expected): + inventory.fetch_objects(connections) diff --git a/tests/unit/plugins/inventory/test_kubevirt_format_dynamic_api_exc.py b/tests/unit/plugins/inventory/test_kubevirt_format_dynamic_api_exc.py new file mode 100644 index 0000000..1e1feb3 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_format_dynamic_api_exc.py @@ -0,0 +1,56 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from json import dumps + +import pytest + +from kubernetes.dynamic.exceptions import DynamicApiError + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryModule, +) + + +@pytest.fixture(scope="function") +def body_error(mocker): + error = DynamicApiError(e=mocker.Mock()) + error.headers = None + error.body = "This is a test error" + return error + + +@pytest.fixture(scope="function") +def message_error(mocker): + error = DynamicApiError(e=mocker.Mock()) + error.headers = {"Content-Type": "application/json"} + error.body = dumps({"message": "This is a test error"}).encode("utf-8") + return error + + +@pytest.fixture(scope="function") +def status_reason_error(mocker): + error = DynamicApiError(e=mocker.Mock()) + error.body = None + error.status = 404 + error.reason = "This is a test error" + return error + + +@pytest.mark.parametrize( + "exc,expected", + [ + ("body_error", "This is a test error"), + ("message_error", "This is a test error"), + ("status_reason_error", "404 Reason: This is a test error"), + ], +) +def test_format_dynamic_api_exc(request, exc, expected): + assert ( + InventoryModule.format_dynamic_api_exc(request.getfixturevalue(exc)) == expected + ) diff --git a/tests/unit/plugins/inventory/test_kubevirt_get_resources.py b/tests/unit/plugins/inventory/test_kubevirt_get_resources.py new file mode 100644 index 0000000..bbd039f --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_get_resources.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, + DEFAULT_BASE_DOMAIN, +) + + +@pytest.mark.parametrize( + "client", + [ + { + "vms": [{"metadata": {"name": "testvm"}}], + "vmis": [{"metadata": {"name": "testvmi"}}], + "services": [{"metadata": {"name": "testsvc"}}], + }, + ], + indirect=["client"], +) +def test_get_resources(inventory, client): + assert inventory.get_resources(client, "v1", "Namespace") == [ + {"metadata": {"name": DEFAULT_NAMESPACE}} + ] + assert inventory.get_resources(client, "v1", "Service") == [ + {"metadata": {"name": "testsvc"}} + ] + assert inventory.get_resources(client, "config.openshift.io/v1", "DNS") == [ + {"spec": {"baseDomain": DEFAULT_BASE_DOMAIN}} + ] + assert inventory.get_resources(client, "kubevirt.io/v1", "VirtualMachine") == [ + {"metadata": {"name": "testvm"}} + ] + assert inventory.get_resources( + client, "kubevirt.io/v1", "VirtualMachineInstance" + ) == [{"metadata": {"name": "testvmi"}}] + + +@pytest.mark.parametrize( + "client,expected", + [ + ( + {}, + [DEFAULT_NAMESPACE], + ), + ( + { + "namespaces": [ + {"metadata": {"name": DEFAULT_NAMESPACE}}, + {"metadata": {"name": "test"}}, + ] + }, + [DEFAULT_NAMESPACE, "test"], + ), + ], + indirect=["client"], +) +def test_get_available_namespaces(inventory, client, expected): + assert inventory.get_available_namespaces(client) == expected + + +@pytest.mark.parametrize( + "client", + [ + { + "vms": [ + {"metadata": {"name": "testvm1"}}, + {"metadata": {"name": "testvm2"}}, + ], + }, + ], + indirect=["client"], +) +def test_get_vms_for_namespace(inventory, client): + assert inventory.get_vms_for_namespace( + client, DEFAULT_NAMESPACE, InventoryOptions() + ) == [{"metadata": {"name": "testvm1"}}, {"metadata": {"name": "testvm2"}}] + + +@pytest.mark.parametrize( + "client", + [ + { + "vmis": [ + {"metadata": {"name": "testvmi1"}}, + {"metadata": {"name": "testvmi2"}}, + ], + }, + ], + indirect=["client"], +) +def test_get_vmis_for_namespace(inventory, client): + assert inventory.get_vmis_for_namespace( + client, DEFAULT_NAMESPACE, InventoryOptions() + ) == [{"metadata": {"name": "testvmi1"}}, {"metadata": {"name": "testvmi2"}}] diff --git a/tests/unit/plugins/inventory/test_kubevirt_get_ssh_services_for_namespace.py b/tests/unit/plugins/inventory/test_kubevirt_get_ssh_services_for_namespace.py new file mode 100644 index 0000000..5efa2f0 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_get_ssh_services_for_namespace.py @@ -0,0 +1,168 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, +) + +SVC_LB_SSH = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-lb-ssh"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "type": "LoadBalancer", + "selector": {"kubevirt.io/domain": "test-lb-ssh"}, + }, +} + +SVC_NP_SSH = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-np-ssh"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "type": "NodePort", + "selector": {"kubevirt.io/domain": "test-np-ssh"}, + }, +} + + +@pytest.mark.parametrize( + "client", + [ + { + "services": [SVC_LB_SSH, SVC_NP_SSH], + }, + ], + indirect=["client"], +) +def test_get_ssh_services_for_namespace(inventory, client): + assert inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) == { + "test-lb-ssh": SVC_LB_SSH, + "test-np-ssh": SVC_NP_SSH, + } + + +SVC_CLUSTERIP = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-clusterip"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "type": "ClusterIP", + "selector": {"kubevirt.io/domain": "test-clusterip"}, + }, +} + + +SVC_HTTP = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-http"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 80, + "targetPort": 80, + }, + ], + "type": "LoadBalancer", + "selector": {"kubevirt.io/domain": "test-http"}, + }, +} + + +SVC_NO_SPEC = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-no-spec"}, +} + +SVC_NO_TYPE = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-no-type"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "selector": {"kubevirt.io/domain": "test-no-type"}, + }, +} + +SVC_NO_PORTS = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-no-ports"}, + "spec": { + "type": "LoadBalancer", + "selector": {"kubevirt.io/domain": "test-no-ports"}, + }, +} + +SVC_NO_SELECTOR = { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "test-no-selector"}, + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "type": "LoadBalancer", + }, +} + + +@pytest.mark.parametrize( + "client", + [ + { + "services": [ + SVC_HTTP, + SVC_CLUSTERIP, + SVC_NO_SPEC, + SVC_NO_TYPE, + SVC_NO_PORTS, + SVC_NO_SELECTOR, + ], + }, + ], + indirect=["client"], +) +def test_ignore_unwanted_services(inventory, client): + assert not inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) diff --git a/tests/unit/plugins/inventory/test_kubevirt_inventory_options.py b/tests/unit/plugins/inventory/test_kubevirt_inventory_options.py new file mode 100644 index 0000000..fa2437b --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_inventory_options.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + + +def test_inventory_options_defaults(): + opts = InventoryOptions() + assert opts.api_version == "kubevirt.io/v1" + assert opts.label_selector is None + assert opts.network_name is None + assert opts.kube_secondary_dns is False + assert opts.use_service is True + assert opts.create_groups is False + assert opts.base_domain is None + assert opts.append_base_domain is False + assert opts.host_format == "{namespace}-{name}" + + +def test_inventory_options_override_defaults(): + api_version = "test/v1" + label_selector = "test-selector" + network_name = "test-network" + kube_secondary_dns = True + use_service = False + create_groups = True + base_domain = "test-domain.com" + append_base_domain = True + host_format = "{name}-testhost" + + opts = InventoryOptions( + api_version=api_version, + label_selector=label_selector, + network_name=network_name, + kube_secondary_dns=kube_secondary_dns, + use_service=use_service, + create_groups=create_groups, + base_domain=base_domain, + append_base_domain=append_base_domain, + host_format=host_format, + ) + assert opts.api_version == api_version + assert opts.label_selector == label_selector + assert opts.network_name == network_name + assert opts.kube_secondary_dns == kube_secondary_dns + assert opts.use_service == use_service + assert opts.create_groups == create_groups + assert opts.base_domain == base_domain + assert opts.append_base_domain == append_base_domain + assert opts.host_format == host_format diff --git a/tests/unit/plugins/inventory/test_kubevirt_populate_inventory_from_namespace.py b/tests/unit/plugins/inventory/test_kubevirt_populate_inventory_from_namespace.py new file mode 100644 index 0000000..1077a39 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_populate_inventory_from_namespace.py @@ -0,0 +1,163 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from addict import Dict + + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + +from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants import ( + DEFAULT_NAMESPACE, +) + +VM1 = { + "metadata": { + "name": "testvm1", + "namespace": DEFAULT_NAMESPACE, + "uid": "940003aa-0160-4b7e-9e55-8ec3df72047f", + }, +} + +VM2 = { + "metadata": { + "name": "testvm2", + "namespace": DEFAULT_NAMESPACE, + "uid": "c2c68de5-b9d7-4c25-872f-462e7245b3e6", + }, +} + +VMI1 = { + "metadata": { + "name": "testvm1", + "namespace": DEFAULT_NAMESPACE, + "uid": "a84319a9-db31-4a36-9b66-3e387578f871", + }, + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + }, +} + +VMI2 = { + "metadata": { + "name": "testvm2", + "namespace": DEFAULT_NAMESPACE, + "uid": "fd35700a-9cbe-488b-8f32-7adbe57eadc2", + }, + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + }, +} + + +@pytest.mark.parametrize( + "vms,vmis,expected", + [ + ([], [], 0), + ([VM1], [], 1), + ([VM1, VM2], [], 2), + ([VM1], [VMI1], 1), + ([VM2], [VMI2], 1), + ([VM1], [VMI2], 2), + ([VM2], [VMI1], 2), + ([VM1], [VMI1, VMI2], 2), + ([VM2], [VMI1, VMI2], 2), + ([VM1, VM2], [VMI1, VMI2], 2), + ([], [VMI1], 1), + ([], [VMI1, VMI2], 2), + ], +) +def test_populate_inventory_from_namespace( + mocker, inventory, groups, vms, vmis, expected +): + _vms = {vm["metadata"]["name"]: Dict(vm) for vm in vms} + _vmis = {vmi["metadata"]["name"]: Dict(vmi) for vmi in vmis} + opts = InventoryOptions() + + def format_hostname(obj): + return opts.host_format.format( + namespace=obj.metadata.namespace, + name=obj.metadata.name, + uid=obj.metadata.uid, + ) + + def add_host_call(obj): + return mocker.call( + obj, + opts.host_format, + f"namespace_{DEFAULT_NAMESPACE}", + ) + + add_host_side_effects = [] + add_host_calls = [] + set_vars_from_vm_calls = [] + set_vars_from_vmi_calls = [] + set_composable_vars_calls = [] + + # For each VM add the expected calls + # Also add expected calls for VMIs for which a VM exists + for name, vm in _vms.items(): + hostname = format_hostname(vm) + add_host_side_effects.append(hostname) + add_host_calls.append(add_host_call(vm)) + set_vars_from_vm_calls.append(mocker.call(hostname, vm, opts)) + if name in _vmis.keys(): + set_vars_from_vmi_calls.append(mocker.call(hostname, _vmis[name], [], opts)) + set_composable_vars_calls.append(mocker.call(hostname)) + + # For each VMI add the expected calls + # Do not add for VMIs for which a VM exists + for name, vmi in _vmis.items(): + if name not in _vms.keys(): + hostname = format_hostname(vmi) + add_host_side_effects.append(hostname) + add_host_calls.append(add_host_call(vmi)) + set_vars_from_vmi_calls.append(mocker.call(hostname, vmi, [], opts)) + set_composable_vars_calls.append(mocker.call(hostname)) + + get_vms_for_namespace = mocker.patch.object( + inventory, "get_vms_for_namespace", return_value=_vms.values() + ) + get_vmis_for_namespace = mocker.patch.object( + inventory, "get_vmis_for_namespace", return_value=_vmis.values() + ) + get_ssh_services_for_namespace = mocker.patch.object( + inventory, "get_ssh_services_for_namespace", return_value=[] + ) + add_host = mocker.patch.object( + inventory, "add_host", side_effect=add_host_side_effects + ) + set_vars_from_vm = mocker.patch.object(inventory, "set_vars_from_vm") + set_vars_from_vmi = mocker.patch.object(inventory, "set_vars_from_vmi") + set_composable_vars = mocker.patch.object(inventory, "set_composable_vars") + + inventory.populate_inventory_from_namespace(None, "test", DEFAULT_NAMESPACE, opts) + + # These should always get called once + get_vms_for_namespace.assert_called_once_with(None, DEFAULT_NAMESPACE, opts) + get_vmis_for_namespace.assert_called_once_with(None, DEFAULT_NAMESPACE, opts) + + # Assert it tries to add the expected vars for all provided VMs/VMIs + set_vars_from_vm.assert_has_calls(set_vars_from_vm_calls) + set_vars_from_vmi.assert_has_calls(set_vars_from_vmi_calls) + set_composable_vars.assert_has_calls(set_composable_vars_calls) + + # If no VMs or VMIs were provided the function should not add any groups + if vms or vmis: + get_ssh_services_for_namespace.assert_called_once_with(None, DEFAULT_NAMESPACE) + assert list(groups.keys()) == ["test", f"namespace_{DEFAULT_NAMESPACE}"] + else: + get_ssh_services_for_namespace.assert_not_called() + assert not list(groups.keys()) + + # Assert the expected amount of hosts was added + add_host.assert_has_calls(add_host_calls) + assert len(add_host_calls) == expected diff --git a/tests/unit/plugins/inventory/test_kubevirt_set_ansible_host_and_port.py b/tests/unit/plugins/inventory/test_kubevirt_set_ansible_host_and_port.py new file mode 100644 index 0000000..ab20265 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_set_ansible_host_and_port.py @@ -0,0 +1,286 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from addict import Dict + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + + +@pytest.mark.parametrize( + "opts", + [ + InventoryOptions(), + InventoryOptions(kube_secondary_dns=True), # needs network_name + InventoryOptions(use_service=True), # needs service + ], +) +def test_use_ip_address_by_default(mocker, inventory, opts): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + ip_address = "1.1.1.1" + + inventory.set_ansible_host_and_port(Dict(), hostname, ip_address, None, opts) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", ip_address), + mocker.call(hostname, "ansible_port", None), + ] + ) + + +@pytest.mark.parametrize( + "base_domain", + [ + True, + False, + ], +) +def test_kube_secondary_dns(mocker, inventory, base_domain): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + vmi = Dict( + { + "metadata": {"name": "testvm", "namespace": "default"}, + "status": {"interfaces": [{"name": "awesome"}]}, + } + ) + + inventory.set_ansible_host_and_port( + vmi, + hostname, + "1.1.1.1", + None, + InventoryOptions( + kube_secondary_dns=True, + network_name="awesome", + base_domain="example.com" if base_domain else None, + ), + ) + + ansible_host = "awesome.testvm.default.vm" + if base_domain: + ansible_host += ".example.com" + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", ansible_host), + mocker.call(hostname, "ansible_port", None), + ] + ) + + +def test_kube_secondary_dns_precedence_over_service(mocker, inventory): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + vmi = Dict( + { + "metadata": {"name": "testvm", "namespace": "default"}, + "status": {"interfaces": [{"name": "awesome"}]}, + } + ) + + inventory.set_ansible_host_and_port( + vmi, + hostname, + "1.1.1.1", + {"metadata": {"name": "testsvc"}}, + InventoryOptions( + kube_secondary_dns=True, network_name="awesome", use_service=True + ), + ) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", "awesome.testvm.default.vm"), + mocker.call(hostname, "ansible_port", None), + ] + ) + + +@pytest.mark.parametrize( + "service,expected_host,expected_port", + [ + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": 22}], + }, + "status": { + "loadBalancer": {"ingress": [{"hostname": "testhost.example.com"}]} + }, + }, + "testhost.example.com", + 22, + ), + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": 23}], + }, + "status": { + "loadBalancer": { + "ingress": [ + {"hostname": "testhost.example.com", "ip": "2.2.2.2"} + ] + } + }, + }, + "testhost.example.com", + 23, + ), + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": 24}], + }, + "status": {"loadBalancer": {"ingress": [{"ip": "2.2.2.2"}]}}, + }, + "2.2.2.2", + 24, + ), + ( + { + "spec": { + "type": "NodePort", + "ports": [{"nodePort": 25}], + }, + }, + "testnode.example.com", + 25, + ), + ], +) +def test_service(mocker, inventory, service, expected_host, expected_port): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + vmi = Dict( + { + "status": { + "nodeName": "testnode.example.com", + }, + } + ) + + inventory.set_ansible_host_and_port( + vmi, + hostname, + "1.1.1.1", + service, + InventoryOptions(use_service=True), + ) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", expected_host), + mocker.call(hostname, "ansible_port", expected_port), + ] + ) + + +def test_service_append_base_domain(mocker, inventory): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + vmi = Dict( + { + "status": { + "nodeName": "testnode", + }, + } + ) + service = { + "spec": { + "type": "NodePort", + "ports": [{"nodePort": 25}], + }, + } + inventory.set_ansible_host_and_port( + vmi, + hostname, + "1.1.1.1", + service, + InventoryOptions( + use_service=True, append_base_domain=True, base_domain="awesome.com" + ), + ) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", "testnode.awesome.com"), + mocker.call(hostname, "ansible_port", 25), + ] + ) + + +@pytest.mark.parametrize( + "host,port", + [ + ("testhost.com", None), + (None, 22), + (None, None), + ], +) +def test_service_fallback(mocker, inventory, host, port): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + mocker.patch.object(inventory, "get_host_from_service", return_value=host) + mocker.patch.object(inventory, "get_port_from_service", return_value=port) + + hostname = "default-testvm" + vmi = Dict( + { + "status": { + "nodeName": "testnode", + }, + } + ) + inventory.set_ansible_host_and_port( + vmi, + hostname, + "1.1.1.1", + {"something": "something"}, + InventoryOptions(use_service=True), + ) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", "1.1.1.1"), + mocker.call(hostname, "ansible_port", None), + ] + ) + + +def test_no_service_if_network_name(mocker, inventory): + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + inventory.set_ansible_host_and_port( + Dict(), + hostname, + "1.2.3.4", + {"something": "something"}, + InventoryOptions(use_service=True, network_name="awesome"), + ) + + set_variable.assert_has_calls( + [ + mocker.call(hostname, "ansible_host", "1.2.3.4"), + mocker.call(hostname, "ansible_port", None), + ] + ) diff --git a/tests/unit/plugins/inventory/test_kubevirt_set_common_vars.py b/tests/unit/plugins/inventory/test_kubevirt_set_common_vars.py new file mode 100644 index 0000000..6e76032 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_set_common_vars.py @@ -0,0 +1,122 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from random import choice +from string import ascii_lowercase + +import pytest + +from addict import Dict + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, +) + + +@pytest.mark.parametrize( + "obj,expected", + [ + ( + { + "metadata": { + "something": "idontcare", + }, + }, + {}, + ), + ( + { + "metadata": { + "annotations": {"testanno": "testval"}, + }, + }, + {"annotations": {"testanno": "testval"}}, + ), + ( + { + "metadata": { + "labels": {"testlabel": "testval"}, + }, + }, + {"labels": {"testlabel": "testval"}}, + ), + ( + { + "metadata": { + "resourceVersion": "123", + }, + }, + {"resource_version": "123"}, + ), + ( + { + "metadata": { + "uid": "48e6ed2c-d8a2-4172-844d-0fe7056aa180", + }, + }, + {"uid": "48e6ed2c-d8a2-4172-844d-0fe7056aa180"}, + ), + ( + { + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + }, + }, + {"interfaces": [{"ipAddress": "10.10.10.10"}]}, + ), + ], +) +def test_set_common_vars(inventory, hosts, obj, expected): + hostname = "default-testvm" + prefix = "".join(choice(ascii_lowercase) for i in range(5)) + inventory.inventory.add_host(hostname) + inventory.set_common_vars(hostname, prefix, Dict(obj), InventoryOptions()) + + for key, value in expected.items(): + prefixed_key = f"{prefix}_{key}" + assert prefixed_key in hosts[hostname] + assert hosts[hostname][prefixed_key] == value + + +@pytest.mark.parametrize( + "create_groups", + [ + True, + False, + ], +) +def test_set_common_vars_create_groups(mocker, inventory, create_groups): + mocker.patch.object(inventory.inventory, "set_variable") + set_groups_from_labels = mocker.patch.object(inventory, "set_groups_from_labels") + + hostname = "default-testvm" + labels = {"testkey": "testval"} + opts = InventoryOptions(create_groups=create_groups) + + inventory.set_common_vars( + hostname, "prefix", Dict({"metadata": {"labels": labels}}), opts + ) + + if create_groups: + set_groups_from_labels.assert_called_once_with(hostname, labels) + else: + set_groups_from_labels.assert_not_called() + + +def test_called_by_set_vars_from(mocker, inventory): + hostname = "default-testvm" + opts = InventoryOptions() + + set_common_vars = mocker.patch.object(inventory, "set_common_vars") + + inventory.set_vars_from_vm(hostname, Dict(), opts) + inventory.set_vars_from_vmi(hostname, Dict(), {}, opts) + + set_common_vars.assert_has_calls( + [mocker.call(hostname, "vm", {}, opts), mocker.call(hostname, "vmi", {}, opts)] + ) diff --git a/tests/unit/plugins/inventory/test_kubevirt_set_vars_from_vmi.py b/tests/unit/plugins/inventory/test_kubevirt_set_vars_from_vmi.py new file mode 100644 index 0000000..db7dc1f --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_set_vars_from_vmi.py @@ -0,0 +1,120 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from addict import Dict + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + InventoryOptions, + LABEL_KUBEVIRT_IO_DOMAIN, +) + + +def test_ignore_vmi_without_interface(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + set_ansible_host_and_port = mocker.patch.object( + inventory, "set_ansible_host_and_port" + ) + + vmi = Dict({"status": {}}) + inventory.set_vars_from_vmi("default-testvm", vmi, {}, InventoryOptions()) + + set_ansible_host_and_port.assert_not_called() + + +def test_use_first_interface_by_default(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + set_ansible_host_and_port = mocker.patch.object( + inventory, "set_ansible_host_and_port" + ) + + hostname = "default-testvm" + vmi = Dict( + {"status": {"interfaces": [{"ipAddress": "1.1.1.1"}, {"ipAddress": "2.2.2.2"}]}} + ) + opts = InventoryOptions() + inventory.set_vars_from_vmi(hostname, vmi, {}, opts) + + set_ansible_host_and_port.assert_called_once_with( + vmi, hostname, "1.1.1.1", None, opts + ) + + +def test_use_named_interface(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + set_ansible_host_and_port = mocker.patch.object( + inventory, "set_ansible_host_and_port" + ) + + hostname = "default-testvm" + vmi = Dict( + { + "status": { + "interfaces": [ + {"name": "first", "ipAddress": "1.1.1.1"}, + {"name": "second", "ipAddress": "2.2.2.2"}, + ] + } + } + ) + opts = InventoryOptions(network_name="second") + inventory.set_vars_from_vmi(hostname, vmi, {}, opts) + + set_ansible_host_and_port.assert_called_once_with( + vmi, hostname, "2.2.2.2", None, opts + ) + + +def test_ignore_vmi_without_named_interface(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + set_ansible_host_and_port = mocker.patch.object( + inventory, "set_ansible_host_and_port" + ) + + vmi = Dict( + {"status": {"interfaces": [{"name": "somename", "ipAddress": "1.1.1.1"}]}} + ) + inventory.set_vars_from_vmi( + "default-testvm", vmi, {}, InventoryOptions(network_name="awesome") + ) + + set_ansible_host_and_port.assert_not_called() + + +def test_set_winrm_if_windows(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + mocker.patch.object(inventory, "is_windows", return_value=True) + mocker.patch.object(inventory, "set_ansible_host_and_port") + set_variable = mocker.patch.object(inventory.inventory, "set_variable") + + hostname = "default-testvm" + vmi = Dict({"status": {"interfaces": [{"ipAddress": "1.1.1.1"}]}}) + inventory.set_vars_from_vmi(hostname, vmi, {}, InventoryOptions()) + + set_variable.assert_called_once_with(hostname, "ansible_connection", "winrm") + + +def test_service_lookup(mocker, inventory): + mocker.patch.object(inventory, "set_common_vars") + set_ansible_host_and_port = mocker.patch.object( + inventory, "set_ansible_host_and_port" + ) + + hostname = "default-testvm" + vmi = Dict( + { + "metadata": {"labels": {LABEL_KUBEVIRT_IO_DOMAIN: "testdomain"}}, + "status": {"interfaces": [{"name": "somename", "ipAddress": "1.1.1.1"}]}, + } + ) + opts = InventoryOptions() + service = {"metadata": {"name": "testsvc"}} + inventory.set_vars_from_vmi(hostname, vmi, {"testdomain": service}, opts) + + set_ansible_host_and_port.assert_called_once_with( + vmi, hostname, "1.1.1.1", service, opts + ) diff --git a/tests/unit/plugins/inventory/test_kubevirt_setup.py b/tests/unit/plugins/inventory/test_kubevirt_setup.py new file mode 100644 index 0000000..6aa6edd --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_setup.py @@ -0,0 +1,63 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.kubevirt.core.plugins.inventory import ( + kubevirt, +) + +from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( + KubeVirtInventoryException, +) + + +@pytest.mark.parametrize( + "cache", + [ + True, + False, + ], +) +def test_cache_is_used(mocker, inventory, cache): + connections = [{"test-conn": {}}] + config_data = {"connections": connections} + cache_key = "test-prefix" + + mocker.patch.dict(inventory._cache, {cache_key: {"test-key": "test-value"}}) + fetch_objects = mocker.patch.object(inventory, "fetch_objects") + + inventory.setup(config_data, cache, cache_key) + + if cache: + fetch_objects.assert_not_called() + else: + fetch_objects.assert_called_once_with(connections) + + +@pytest.mark.parametrize( + "present", + [ + True, + False, + ], +) +def test_k8s_client_missing(mocker, inventory, present): + mocker.patch.object(kubevirt, "HAS_K8S_MODULE_HELPER", present) + fetch_objects = mocker.patch.object(inventory, "fetch_objects") + + if present: + inventory.setup({}, False, "test") + fetch_objects.assert_called_once() + else: + with pytest.raises( + KubeVirtInventoryException, + match="This module requires the Kubernetes Python client. Try `pip install kubernetes`. Detail: None", + ): + inventory.setup({}, False, "test") + fetch_objects.assert_not_called() diff --git a/tests/unit/plugins/inventory/test_kubevirt_verify_file.py b/tests/unit/plugins/inventory/test_kubevirt_verify_file.py new file mode 100644 index 0000000..8b3bbf2 --- /dev/null +++ b/tests/unit/plugins/inventory/test_kubevirt_verify_file.py @@ -0,0 +1,45 @@ +# -*- coding: utf-8 -*- +# Copyright 2024 Red Hat, Inc. +# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0) + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + + +@pytest.mark.parametrize( + "file_name,expected", + [ + ("inventory.kubevirt.yml", True), + ("inventory.kubevirt.yaml", True), + ("something.kubevirt.yml", True), + ("something.kubevirt.yaml", True), + ("inventory.somethingelse.yml", False), + ("inventory.somethingelse.yaml", False), + ("something.somethingelse.yml", False), + ("something.somethingelse.yaml", False), + ], +) +def test_verify_file(tmp_path, inventory, file_name, expected): + file = tmp_path / file_name + file.touch() + assert inventory.verify_file(str(file)) is expected + + +@pytest.mark.parametrize( + "file_name", + [ + "inventory.kubevirt.yml", + "inventory.kubevirt.yaml", + "something.kubevirt.yml", + "something.kubevirt.yaml", + "inventory.somethingelse.yml", + "inventory.somethingelse.yaml", + "something.somethingelse.yml", + "something.somethingelse.yaml", + ], +) +def test_verify_file_bad_config(inventory, file_name): + assert inventory.verify_file(file_name) is False