diff --git a/tests/unit/plugins/inventory/test_kubevirt.py b/tests/unit/plugins/inventory/test_kubevirt.py index 446a127..3641527 100644 --- a/tests/unit/plugins/inventory/test_kubevirt.py +++ b/tests/unit/plugins/inventory/test_kubevirt.py @@ -9,16 +9,23 @@ __metaclass__ = type import pytest from addict import Dict +import copy from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import ( GetVmiOptions, InventoryModule, + KubeVirtInventoryException, ) from ansible_collections.kubevirt.core.tests.unit.utils.merge_dicts import ( merge_dicts, ) +from ansible_collections.kubevirt.core.plugins.inventory import ( + kubevirt +) + + DEFAULT_NAMESPACE = "default" DEFAULT_BASE_DOMAIN = "example.com" @@ -29,16 +36,31 @@ BASE_VMI = { "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"}], - }, } -WINDOWS_VMI_1 = merge_dicts( +# Note, if "interfaces": [{}] or "interfaces": [{"ipAddress": {}}] the if on line 517, problem? +VMI_WITH_INTERFACE_NO_IPADDRESS = merge_dicts( BASE_VMI, + { + "status": { + "interfaces": [{"ipAddress": None}] + } + } +) +BASIC_VMI = merge_dicts( + BASE_VMI, + { + "status": { + "interfaces": [{"ipAddress": "10.10.10.10"}], + }, + } +) +WINDOWS_VMI_1 = merge_dicts( + BASIC_VMI, { "status": { "guestOSInfo": {"id": "mswindows"}, @@ -46,7 +68,7 @@ WINDOWS_VMI_1 = merge_dicts( }, ) WINDOWS_VMI_2 = merge_dicts( - BASE_VMI, + BASIC_VMI, { "metadata": { "annotations": {"kubevirt.io/cluster-preference-name": "windows.2k22"} @@ -54,17 +76,142 @@ WINDOWS_VMI_2 = merge_dicts( }, ) WINDOWS_VMI_3 = merge_dicts( - BASE_VMI, + BASIC_VMI, { "metadata": {"annotations": {"kubevirt.io/preference-name": "windows.2k22"}}, }, ) WINDOWS_VMI_4 = merge_dicts( - BASE_VMI, + BASIC_VMI, { "metadata": {"annotations": {"vm.kubevirt.io/os": "windows2k22"}}, }, ) +COMPLETE_VMI = merge_dicts( + BASIC_VMI, + { + "metadata": { + # {"test-label": "test-label"} does not work + "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", + } + ], + } + }, +) + +BASE_SERVICE = { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": "test-service" + }, + "spec": {} +} +LOADBALANCER_SERVICE_WITHOUT_AND_SELECTOR_PORTS = merge_dicts( + BASE_SERVICE, + { + "spec": { + "type": "LoadBalancer" + } + } +) +LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT = merge_dicts( + LOADBALANCER_SERVICE_WITHOUT_AND_SELECTOR_PORTS, + { + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 80, + "targetPort": 80, + }, + ], + "type": "LoadBalancer", + } + } +) +LOADBALANCER_SERVICE_WITHOUT_SELECTOR = merge_dicts( + LOADBALANCER_SERVICE_WITHOUT_AND_SELECTOR_PORTS, + { + "spec": { + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22, + }, + ], + "type": "LoadBalancer", + } + } +) +LOADBALANCER_SERVICE = { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": "test-service" + }, + "spec": { + "selector": { + "kubevirt.io/domain": "test-domain" + }, + "ports": [ + { + "protocol": "TCP", + "port": 22, + "targetPort": 22 + } + ], + "type": "LoadBalancer" + } +} +NODEPORT_SERVICE = merge_dicts( + LOADBALANCER_SERVICE, + { + "spec": { + "type": "NodePort", + } + } +) @pytest.fixture(scope="function") @@ -88,6 +235,44 @@ def host_vars(monkeypatch, inventory): 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) + + monkeypatch.setattr(inventory.inventory, "add_group", add_group) + return groups + + +@pytest.fixture(scope="function") +def add_host(monkeypatch, inventory): + hosts = [] + + def add_host(name): + if name not in hosts: + hosts.append(name) + + 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() @@ -219,7 +404,7 @@ def test_is_windows(inventory, guest_os_info, annotations, expected): @pytest.mark.parametrize( "client,vmi,expected", [ - ({"vmis": [BASE_VMI]}, BASE_VMI, False), + ({"vmis": [BASIC_VMI]}, BASIC_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), @@ -235,3 +420,724 @@ def test_ansible_connection_winrm(inventory, host_vars, client, vmi, 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.fixture(scope="module") +def load_balancer_ingress_ip(): + return { + "spec": { + "type": "LoadBalancer", + }, + "status": { + "loadBalancer": { + "ingress": [{"ip": "192.168.1.100"}] + }, + } + } + + +@pytest.fixture(scope="module") +def load_balancer_ingress_hostname(): + return { + "spec": { + "type": "LoadBalancer", + }, + "status": { + "loadBalancer": { + "ingress": [{"hostname": "test-hostname"}] + }, + } + } + + +@pytest.fixture(scope="module") +def node_port(): + return { + "spec": { + "type": "NodePort" + } + } + + +@pytest.fixture(scope="module") +def service_other(): + return { + "spec": { + "type": "ClusterIP" + } + } + + +@pytest.mark.parametrize( + "service,node_name,expected", + [ + ("service_other", None, None), + ("load_balancer_ingress_ip", None, "192.168.1.100"), + ("load_balancer_ingress_hostname", None, "test-hostname"), + ("node_port", "test-nodename", "test-nodename"), + ], +) +def test_get_host_from_service(request, inventory, service, node_name, expected): + result = inventory.get_host_from_service(request.getfixturevalue(service), node_name) + assert result == expected + + +@pytest.fixture(scope="module") +def load_balancer_port(): + return { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": "80"}], + }, + } + + +@pytest.fixture(scope="module") +def node_port_port(): + return { + "spec": { + "type": "NodePort", + "ports": [{"nodePort": "8080"}], + } + } + + +@pytest.mark.parametrize( + "service,port", + [ + ("service_other", None), + ("load_balancer_port", "80"), + ("node_port_port", "8080"), + ], +) +def test_port_from_service(request, inventory, service, port): + assert port == inventory.get_port_from_service(request.getfixturevalue(service)) + + +def test_parse(monkeypatch, inventory): + monkeypatch.setattr(inventory, "_read_config_data", lambda path: {"host_format": "default-test"}) + 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.fixture(scope="module") +def connection_none(): + return None + + +@pytest.fixture(scope="module") +def connection_not_list(): + return "test" + + +@pytest.fixture(scope="module") +def connection_list_not_dict(): + return [ + "test", + "test" + ] + + +@pytest.fixture(scope="module") +def connection_list(): + return [ + { + "name": "test", + }, + ] + + +@pytest.fixture(scope="module") +def connection_list_namespace(): + return [ + { + "name": "test", + "namespaces": ["test"] + }, + ] + + +@pytest.fixture(scope="module") +def connection_with_base_values(): + return [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain" + }, + ] + + +@pytest.fixture(scope="module") +def connection_with_network(): + return [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain", + "network_name": "test-network" + }, + ] + + +@pytest.fixture(scope="module") +def connection_with_interface(): + return [ + { + "name": "test", + "namespaces": ["test"], + "use_service": True, + "create_groups": True, + "append_base_domain": True, + "base_domain": "test-domain", + "interface_name": "test-interface" + }, + ] + + +def vmi_options_base_values(): + return GetVmiOptions(use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain" + ) + + +def vmi_options_with_network(): + return GetVmiOptions(use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain", + network_name="test-network" + ) + + +def vmi_options_with_interface(): + return GetVmiOptions(use_service=True, + create_groups=True, + append_base_domain=True, + base_domain="test-domain", + network_name="test-interface" + ) + + +@pytest.mark.parametrize( + "connections,result,default_namespace", + [ + + ("connection_list", {"name": "test", "namespace": DEFAULT_NAMESPACE, "opts": GetVmiOptions()}, True), + ("connection_list_namespace", {"name": "test", "namespace": "test", "opts": GetVmiOptions()}, False), + ("connection_with_base_values", {"name": "test", "namespace": "test", "opts": vmi_options_base_values()}, False), + ("connection_with_network", {"name": "test", "namespace": "test", "opts": vmi_options_with_network()}, False), + ("connection_with_interface", {"name": "test", "namespace": "test", "opts": vmi_options_with_interface()}, False), + ("connection_none", {"name": "default-hostname", "namespace": DEFAULT_NAMESPACE, "opts": GetVmiOptions()}, True), + ], +) +def test_fetch_objects(request, 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") + monkeypatch.setattr(inventory, "get_cluster_domain", lambda _: None) + + get_vmis_for_namespace = mocker.patch.object(inventory, "get_vmis_for_namespace") + get_available_namespaces = mocker.patch.object(inventory, "get_available_namespaces") + + if default_namespace: + get_available_namespaces.return_value = [DEFAULT_NAMESPACE] + inventory.fetch_objects(request.getfixturevalue(connections)) + get_available_namespaces.assert_called_once_with(mocker.ANY) + else: + inventory.fetch_objects(request.getfixturevalue(connections)) + get_available_namespaces.assert_not_called() + + get_vmis_for_namespace.assert_called_once_with(mocker.ANY, + result["name"], + result["namespace"], + result["opts"]) + + +@pytest.mark.parametrize( + "connections,result", + [ + ("connection_not_list", "Expecting connections to be a list."), + ("connection_list_not_dict", "Expecting connection to be a dictionary."), + ], +) +def test_fetch_objects_exceptions(request, inventory, connections, result): + with pytest.raises(KubeVirtInventoryException, match=result): + inventory.fetch_objects(request.getfixturevalue(connections)) + + +# Exceptions? +@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.fixture(scope="module") +def get_default_namespace_list(): + return [ + DEFAULT_NAMESPACE + ] + + +@pytest.fixture(scope="module") +def get_multiple_namespace_list(): + return [ + DEFAULT_NAMESPACE, + "test" + ] + + +# Exceptions? +@pytest.mark.parametrize( + "client,result", + [ + ({"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, "get_default_namespace_list"), + ({"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}, {"metadata": {"name": "test"}}]}, "get_multiple_namespace_list") + ], + indirect=["client"], +) +def test_get_available_namespaces(request, inventory, client, result): + assert request.getfixturevalue(result) == inventory.get_available_namespaces(client) + + +@pytest.fixture(scope="module") +def inventory_empty(): + return [] + + +@pytest.fixture(scope="module") +def inventory_groups(): + return [ + "test", + "namespace_default", + ] + + +@pytest.fixture(scope="module") +def inventory_vmi_group(): + return ["default-testvmi"] + + +@pytest.fixture(scope="module") +def children_group_without_vmi(): + return { + "test": [ + "namespace_default" + ], + } + + +@pytest.fixture(scope="module") +def children_group_with_vmi(children_group_without_vmi): + return children_group_without_vmi | { + "namespace_default": [ + "default-testvmi" + ] + } + + +@pytest.fixture(scope="module") +def children_group_with_vmi_create_groups_option(children_group_with_vmi): + return children_group_with_vmi | { + "label_kubevirt_io_domain_test_domain": [ + "default-testvmi" + ] + } + + +@pytest.fixture(scope="function") +def inventory_groups_create_groups_option(inventory_groups): + inv = copy.deepcopy(inventory_groups) + inv.append("label_kubevirt_io_domain_test_domain") + return inv + + +@pytest.fixture(scope="module") +def base_vmi_host_vars(): + return {} + + +@pytest.fixture(scope="module") +def basic_vmi_host_vars(): + return { + "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": [], + } + } + + +@pytest.fixture(scope="module") +def complete_vmi_host_vars(basic_vmi_host_vars): + vmi = 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" + }, + ], + } + vmi = { + "default-testvmi": vmi + } + return vmi + + +@pytest.fixture(scope="module") +def windows_vmi_host_vars(basic_vmi_host_vars): + vmi = basic_vmi_host_vars["default-testvmi"] | { + "ansible_connection": "winrm", + "vmi_guest_os_info": {"id": "mswindows"}, + } + vmi = { + "default-testvmi": vmi + } + return vmi + + +@pytest.mark.parametrize( + "client,vmi,groups,vmi_group,child_group,create_groups,expected_host_vars,call_functions,windows", + [ + ( + {"vmis": [BASE_VMI]}, + BASE_VMI, + "inventory_groups", + "inventory_empty", + "children_group_without_vmi", + False, "base_vmi_host_vars", + False, + False, + ), + ( + {"vmis": [VMI_WITH_INTERFACE_NO_IPADDRESS]}, + BASE_VMI, + "inventory_groups", + "inventory_empty", + "children_group_without_vmi", + False, + "base_vmi_host_vars", + False, + False, + ), + ( + {"vmis": [BASIC_VMI], "services": [LOADBALANCER_SERVICE]}, + BASIC_VMI, + "inventory_groups", + "inventory_vmi_group", + "children_group_with_vmi", + False, + "basic_vmi_host_vars", + True, + False, + ), + ( + {"vmis": [COMPLETE_VMI], "services": [LOADBALANCER_SERVICE]}, + COMPLETE_VMI, + "inventory_groups", + "inventory_vmi_group", + "children_group_with_vmi", + False, + "complete_vmi_host_vars", + True, + False, + ), + ( + {"vmis": [COMPLETE_VMI], "services": [LOADBALANCER_SERVICE]}, + COMPLETE_VMI, + "inventory_groups_create_groups_option", + "inventory_vmi_group", + "children_group_with_vmi_create_groups_option", + True, + "complete_vmi_host_vars", + True, + False, + ), + ( + {"vmis": [WINDOWS_VMI_1]}, + WINDOWS_VMI_1, + "inventory_groups", + "inventory_vmi_group", + "children_group_with_vmi", + False, + "windows_vmi_host_vars", + True, + True, + ), + ], + indirect=["client"], +) +def test_get_vmis_for_namespace(mocker, + request, + inventory, + vmi, + host_vars, + add_group, + add_host, + add_child, + client, + groups, + vmi_group, + child_group, + create_groups, + 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)) + + assert request.getfixturevalue(groups) == add_group + assert request.getfixturevalue(vmi_group) == add_host + assert request.getfixturevalue(child_group) == add_child + assert request.getfixturevalue(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) + ) + set_composable_vars.assert_called_once_with(vmi_name) + + +@pytest.fixture(scope="module") +def loadbalancer(): + return { + "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" + }, + }, + } + + +@pytest.fixture(scope="module") +def nodeport(loadbalancer): + np = copy.deepcopy(loadbalancer) + np["test-domain"]["spec"]["type"] = "NodePort" + return np + + +@pytest.fixture(scope="module") +def empty_service(): + return {} + + +@pytest.mark.parametrize( + "client,result", + [ + ( + {"services": [LOADBALANCER_SERVICE], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "loadbalancer", + ), + ( + {"services": [NODEPORT_SERVICE], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "nodeport", + ), + ( + {"services": [BASE_SERVICE], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "empty_service", + ), + ( + {"services": [LOADBALANCER_SERVICE_WITHOUT_AND_SELECTOR_PORTS], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "empty_service", + ), + ( + {"services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "empty_service", + ), + ( + {"services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, + "empty_service", + ), + ], + indirect=["client"] +) +def test_get_ssh_services_for_namespace(request, inventory, client, result): + assert request.getfixturevalue(result) == inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) + + +@pytest.fixture(scope="module") +def ansible_host_and_port_network(): + return { + "default-testvmi": { + "ansible_host": "test-network.testvmi.default.vm", + "ansible_port": None, + } + } + + +@pytest.fixture(scope="module") +def ansible_host_and_port_network_with_base_domain(): + return { + "default-testvmi": { + "ansible_host": "test-network.testvmi.default.vm.example.com", + "ansible_port": None, + } + } + + +@pytest.fixture(scope="module") +def ansible_host_and_port_service(): + return { + "default-testvmi": { + "ansible_host": "test-host", + "ansible_port": "8080", + } + } + + +@pytest.fixture(scope="module") +def ansible_host_and_port_service_ip(): + return { + "default-testvmi": { + "ansible_host": "10.10.10.10", + "ansible_port": None, + } + } + + +@pytest.mark.parametrize( + "vmi,use_service,opts,result", + [ + ( + BASIC_VMI, + None, + GetVmiOptions(kube_secondary_dns=True, network_name="test-network"), + "ansible_host_and_port_network", + ), + ( + BASIC_VMI, + None, + GetVmiOptions(kube_secondary_dns=True, network_name="test-network", base_domain=DEFAULT_BASE_DOMAIN), + "ansible_host_and_port_network_with_base_domain", + ), + ( + BASIC_VMI, + {"host": "test-host", "port": "8080"}, GetVmiOptions(use_service=True), + "ansible_host_and_port_service", + ), + ( + BASIC_VMI, + {"host": None, "port": None}, GetVmiOptions(use_service=True), + "ansible_host_and_port_service_ip", + ), + ], +) +def test_set_ansible_host_and_port(monkeypatch, request, inventory, host_vars, vmi, use_service, opts, result): + vmi_name = f"{DEFAULT_NAMESPACE}-{vmi['metadata']['name']}" + service = {} + if use_service is not None: + monkeypatch.setattr(inventory, "get_host_from_service", lambda a, b: use_service["host"]) + monkeypatch.setattr(inventory, "get_port_from_service", lambda _: use_service["port"]) + + inventory.set_ansible_host_and_port(Dict(vmi), vmi_name, "10.10.10.10", service, opts) + + assert request.getfixturevalue(result) == host_vars