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