From 81fc608b44cffa233fd747f6281e1c76e6175fb1 Mon Sep 17 00:00:00 2001 From: Javier Cano Cano Date: Mon, 17 Jun 2024 11:46:09 +0200 Subject: [PATCH] feat: Refactor kubevirt unit test suite It refactors the test suite to reduce the amount of fixtures used on it. Signed-off-by: Javier Cano Cano --- tests/unit/plugins/inventory/test_kubevirt.py | 1047 +++++++---------- 1 file changed, 405 insertions(+), 642 deletions(-) diff --git a/tests/unit/plugins/inventory/test_kubevirt.py b/tests/unit/plugins/inventory/test_kubevirt.py index 5a90327..e7cb673 100644 --- a/tests/unit/plugins/inventory/test_kubevirt.py +++ b/tests/unit/plugins/inventory/test_kubevirt.py @@ -218,6 +218,93 @@ NODEPORT_SERVICE = merge_dicts( ) +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() @@ -445,96 +532,65 @@ def test_get_default_host_name(inventory, url, host_name): 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"), + ({"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(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"}], - } - } +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", [ - ("service_other", None), - ("load_balancer_port", "80"), - ("node_port_port", "8080"), + ({"spec": {"type": "ClusterIP"}}, None), + ( + { + "spec": { + "type": "LoadBalancer", + "ports": [{"port": "80"}], + }, + }, + "80", + ), + ( + { + "spec": { + "type": "NodePort", + "ports": [{"nodePort": "8080"}], + } + }, + "8080", + ), ], ) -def test_port_from_service(request, inventory, service, port): - assert port == inventory.get_port_from_service(request.getfixturevalue(service)) +def test_port_from_service(inventory, service, port): + assert port == inventory.get_port_from_service(service) def test_parse(monkeypatch, inventory): @@ -546,160 +602,149 @@ def test_parse(monkeypatch, inventory): assert inventory.host_format == "default-test" -@pytest.fixture(scope="module") -def connection_none(): - return None - - -@pytest.fixture(scope="module") -def connection_list(): - return [ - { - "name": "connection-list", - }, - ] - - -@pytest.fixture(scope="module") -def connection_list_namespace(): - return [ - { - "name": "connection-list-namespace", - "namespaces": ["test"] - }, - ] - - -@pytest.fixture(scope="module") -def connection_with_base_values(): - return [ - { - "name": "connection-with-base-values", - "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": "connection-with-network", - "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": "connection-with-interface", - "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": "connection-list", "namespace": DEFAULT_NAMESPACE, "opts": GetVmiOptions()}, True), - ("connection_list_namespace", {"name": "connection-list-namespace", "namespace": "test", "opts": GetVmiOptions()}, False), - ("connection_with_base_values", {"name": "connection-with-base-values", "namespace": "test", "opts": vmi_options_base_values()}, False), - ("connection_with_network", {"name": "connection-with-network", "namespace": "test", "opts": vmi_options_with_network()}, False), - ("connection_with_interface", {"name": "connection-with-interface", "namespace": "test", "opts": vmi_options_with_interface()}, False), - ("connection_none", {"name": "default-hostname", "namespace": DEFAULT_NAMESPACE, "opts": GetVmiOptions()}, True), + ( + [ + { + "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(request, mocker, monkeypatch, inventory, connections, result, default_namespace): +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") + 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") + 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)) + inventory.fetch_objects(connections) get_available_namespaces.assert_called_once_with(mocker.ANY) else: - inventory.fetch_objects(request.getfixturevalue(connections)) + inventory.fetch_objects(connections) get_available_namespaces.assert_not_called() - get_vmis_for_namespace.assert_called_once_with(mocker.ANY, - result["name"], - result["namespace"], - result["opts"]) - - -@pytest.fixture(scope="module") -def connection_not_list(): - return "test" - - -@pytest.fixture(scope="module") -def connection_list_not_dict(): - return [ - "test", - "test" - ] + 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."), + ("test", "Expecting connections to be a list."), + (["test", "test"], "Expecting connection to be a dictionary."), ], ) -def test_fetch_objects_exceptions(request, inventory, connections, result): +def test_fetch_objects_exceptions(inventory, connections, result): with pytest.raises(KubeVirtInventoryException, match=result): - inventory.fetch_objects(request.getfixturevalue(connections)) + inventory.fetch_objects(connections) + -# Exceptions? @pytest.mark.parametrize( "client,result", [ @@ -712,213 +757,16 @@ 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" - ] - - @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") + ({"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 request.getfixturevalue(result) == inventory.get_available_namespaces(client) - - -@pytest.fixture(scope="module") -def inventory_empty(): - return [] - - -@pytest.fixture(scope="module") -def children_group_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 = deepcopy(inventory_groups) - inv.append("label_kubevirt_io_domain_test_domain") - return inv - - -@pytest.fixture(scope="module") -def empty_host_vars(): - return {} - - -@pytest.fixture(scope="module") -def base_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(base_vmi_host_vars): - vmi = base_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 complete_vmi_host_vars_with_network(complete_vmi_host_vars): - vmi = complete_vmi_host_vars["default-testvmi"] | { - "vmi_interfaces": [ - { - "ipAddress": "10.10.10.10", - "name": "test-network" - }, - ], - } - vmi = { - "default-testvmi": vmi - } - return vmi - - -@pytest.fixture(scope="module") -def complete_vmi_host_vars_with_network(complete_vmi_host_vars): - vmi = complete_vmi_host_vars["default-testvmi"] | { - "vmi_interfaces": [ - { - "ipAddress": "10.10.10.10", - "name": "test-network" - }, - ], - } - vmi = { - "default-testvmi": vmi - } - return vmi - - -@pytest.fixture(scope="module") -def windows_vmi_host_vars(base_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 + assert result == inventory.get_available_namespaces(client) @pytest.mark.parametrize( @@ -927,343 +775,258 @@ def windows_vmi_host_vars(base_vmi_host_vars): ( {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, None, - "inventory_empty", - "inventory_empty", - "children_group_empty", + [], + [], + {}, False, None, - "base_vmi_host_vars", + {}, False, False, ), ( {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, None, - "inventory_empty", - "inventory_empty", - "children_group_empty", + [], + [], + {}, False, None, - "base_vmi_host_vars", - False, - False, - ), - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - None, - "inventory_empty", - "inventory_empty", - "children_group_empty", - False, - None, - "base_vmi_host_vars", - False, - False, - ), - ( - {"namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - None, - "inventory_empty", - "inventory_empty", - "children_group_empty", - False, - None, - "base_vmi_host_vars", + {}, False, False, ), ( {"vmis": [NO_STATUS_VMI]}, NO_STATUS_VMI, - "inventory_groups", - "inventory_empty", - "children_group_without_vmi", + ["test", "namespace_default"], + [], + {"test": ["namespace_default"]}, False, None, - "base_vmi_host_vars", - False, - "empty_host_vars", + {}, False, False, ), ( {"vmis": [VMI_WITH_INTERFACE_NO_IPADDRESS]}, BASE_VMI, - "inventory_groups", - "inventory_empty", - "children_group_without_vmi", + ["test", "namespace_default"], + [], + {"test": ["namespace_default"]}, False, None, - "base_vmi_host_vars", - "empty_host_vars", + {}, False, False, ), ( {"vmis": [BASE_VMI], "services": [LOADBALANCER_SERVICE]}, BASE_VMI, - "inventory_groups", - "inventory_vmi_group", - "children_group_with_vmi", + ["test", "namespace_default"], + ["default-testvmi"], + {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, False, None, - "basic_vmi_host_vars", - "base_vmi_host_vars", + BASIC_VMI_HOST_VARS, True, False, ), ( {"vmis": [COMPLETE_VMI], "services": [LOADBALANCER_SERVICE]}, COMPLETE_VMI, - "inventory_groups", - "inventory_vmi_group", - "children_group_with_vmi", + ["test", "namespace_default"], + ["default-testvmi"], + {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, False, None, - "complete_vmi_host_vars", + 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", + ["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", + COMPLETE_VMI_HOST_VARS, True, False, ), ( {"vmis": [WINDOWS_VMI_1]}, WINDOWS_VMI_1, - "inventory_groups", - "inventory_vmi_group", - "children_group_with_vmi", + ["test", "namespace_default"], + ["default-testvmi"], + {"test": ["namespace_default"], "namespace_default": ["default-testvmi"]}, False, None, - "windows_vmi_host_vars", + WINDOWS_VMI_HOST_VARS, True, True, ), ( - {"vmis": [COMPLETE_VMI_WITH_NETWORK_NAME], "services": [LOADBALANCER_SERVICE]}, + { + "vmis": [COMPLETE_VMI_WITH_NETWORK_NAME], + "services": [LOADBALANCER_SERVICE], + }, COMPLETE_VMI_WITH_NETWORK_NAME, - "inventory_groups_create_groups_option", - "inventory_vmi_group", - "children_group_with_vmi_create_groups_option", + ["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", + COMPLETE_VMI_HOST_VARS_WITH_NETWORK, True, False, ), ], 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, - network_name, - expected_host_vars, - call_functions, - windows, - ): - set_ansible_host_and_port = mocker.patch.object(inventory, "set_ansible_host_and_port") +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)) + inventory.get_vmis_for_namespace( + client, + "test", + DEFAULT_NAMESPACE, + GetVmiOptions(create_groups=create_groups, network_name=network_name), + ) - 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 + 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 + 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_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.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 = 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": [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}}]}, - "nodeport", + { + "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}}]}, - "empty_service", + { + "services": [BASE_SERVICE], + "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], + }, + {}, ), ( - {"services": [BASE_LOADBALANCER_SERVICE], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - "empty_service", + { + "services": [BASE_LOADBALANCER_SERVICE], + "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], + }, + {}, ), ( - {"services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - "empty_service", + { + "services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR_AND_SSH_PORT], + "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], + }, + {}, ), ( - {"services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR], "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}]}, - "empty_service", + { + "services": [LOADBALANCER_SERVICE_WITHOUT_SELECTOR], + "namespaces": [{"metadata": {"name": DEFAULT_NAMESPACE}}], + }, + {}, ), ], - indirect=["client"] + 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", - [ - ( - BASE_VMI, - None, - GetVmiOptions(kube_secondary_dns=True, network_name="test-network"), - "ansible_host_and_port_network", - ), - ( - BASE_VMI, - None, - GetVmiOptions(kube_secondary_dns=True, network_name="test-network", base_domain=DEFAULT_BASE_DOMAIN), - "ansible_host_and_port_network_with_base_domain", - ), - ( - BASE_VMI, - {"host": "test-host", "port": "8080"}, GetVmiOptions(use_service=True), - "ansible_host_and_port_service", - ), - ( - BASE_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 +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")