internet/kubevirt.core
3
0
Fork 0
mirror of https://github.com/kubevirt/kubevirt.core.git synced 2026-07-29 02:44:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

cleanup(inventory): Mark methods as private

Browse Source 
Mark all methods of the inventory which should be private with a underscore
prefix in their name.

Signed-off-by: Felix Matouschek <fmatouschek@redhat.com>
This commit is contained in:
Felix Matouschek
2024-07-11 16:03:27 +02:00
parent 54bbdb4d73
commit 9029574f7c
16 changed files with 179 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
plugins/inventory/kubevirt.py
View File

@@ -256,12 +256,12 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
NAME = "kubevirt.core.kubevirt" NAME = "kubevirt.core.kubevirt"
# Used to convert camel case variable names into snake case # Used to convert camel case variable names into snake case
snake_case_pattern = re_compile(r"(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])") _snake_case_pattern = re_compile(r"(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
@staticmethod @staticmethod
def get_default_hostname(host: str) -> str: def _get_default_hostname(host: str) -> str:
""" """
get_default_host_name strips URL schemes from the host name and _get_default_host_name strips URL schemes from the host name and
replaces invalid characters. replaces invalid characters.
""" """
return ( return (
@@ -272,9 +272,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
) )
@staticmethod @staticmethod
def format_dynamic_api_exc(exc: DynamicApiError) -> str: def _format_dynamic_api_exc(exc: DynamicApiError) -> str:
""" """
format_dynamic_api_exc tries to extract the message from the JSON body _format_dynamic_api_exc tries to extract the message from the JSON body
of a DynamicApiError. of a DynamicApiError.
""" """
if exc.body: if exc.body:
@@ -287,17 +287,17 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
return f"{exc.status} Reason: {exc.reason}" return f"{exc.status} Reason: {exc.reason}"
@staticmethod @staticmethod
def format_var_name(name: str) -> str: def _format_var_name(name: str) -> str:
""" """
format_var_name formats a CamelCase variable name into a snake_case name _format_var_name formats a CamelCase variable name into a snake_case name
suitable for use as a inventory variable name. suitable for use as a inventory variable name.
""" """
return InventoryModule.snake_case_pattern.sub("_", name).lower() return InventoryModule._snake_case_pattern.sub("_", name).lower()
@staticmethod @staticmethod
def obj_is_valid(obj: Dict) -> bool: def _obj_is_valid(obj: Dict) -> bool:
""" """
obj_is_valid ensures commonly used keys are present in the passed object. _obj_is_valid ensures commonly used keys are present in the passed object.
""" """
return bool( return bool(
"spec" in obj "spec" in obj
@@ -309,9 +309,11 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
) )
@staticmethod @staticmethod
def get_host_from_service(service: Dict, node_name: Optional[str]) -> Optional[str]: def _get_host_from_service(
service: Dict, node_name: Optional[str]
) -> Optional[str]:
""" """
get_host_from_service extracts the hostname to be used from the _get_host_from_service extracts the hostname to be used from the
passed in service. passed in service.
""" """
service_type = service.get("spec", {}).get("type") service_type = service.get("spec", {}).get("type")
@@ -329,9 +331,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
return None return None
@staticmethod @staticmethod
def get_port_from_service(service: Dict) -> Optional[str]: def _get_port_from_service(service: Dict) -> Optional[str]:
""" """
get_port_from_service extracts the port to be used from the _get_port_from_service extracts the port to be used from the
passed in service. passed in service.
""" """
ports = service.get("spec", {}).get("ports", []) ports = service.get("spec", {}).get("ports", [])
@@ -349,9 +351,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
return None return None
@staticmethod @staticmethod
def is_windows(guest_os_info: Optional[Dict], annotations: Optional[Dict]) -> bool: def _is_windows(guest_os_info: Optional[Dict], annotations: Optional[Dict]) -> bool:
""" """
is_windows checkes whether a given VM is running a Windows guest _is_windows checks whether a given VM is running a Windows guest
by checking its GuestOSInfo and annotations. by checking its GuestOSInfo and annotations.
""" """
if guest_os_info and "id" in guest_os_info: if guest_os_info and "id" in guest_os_info:
@@ -404,7 +406,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
attempt_to_read_cache = user_cache_setting and cache attempt_to_read_cache = user_cache_setting and cache
cache_needs_update = user_cache_setting and not cache cache_needs_update = user_cache_setting and not cache
self.connections_compatibility(config_data) self._connections_compatibility(config_data)
opts = InventoryOptions(config_data=config_data) opts = InventoryOptions(config_data=config_data)
results = {} results = {}
@@ -414,13 +416,13 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
except KeyError: except KeyError:
cache_needs_update = True cache_needs_update = True
if not attempt_to_read_cache or cache_needs_update: if not attempt_to_read_cache or cache_needs_update:
results = self.fetch_objects(get_api_client(**config_data), opts) results = self._fetch_objects(get_api_client(**config_data), opts)
if cache_needs_update: if cache_needs_update:
self._cache[cache_key] = results self._cache[cache_key] = results
self.populate_inventory(results, opts) self._populate_inventory(results, opts)
def connections_compatibility(self, config_data: Dict) -> None: def _connections_compatibility(self, config_data: Dict) -> None:
collection_name = "kubevirt.core" collection_name = "kubevirt.core"
version_removed_in = "3.0.0" version_removed_in = "3.0.0"
@@ -457,7 +459,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
removed=True, removed=True,
) )
def fetch_objects(self, client: Any, opts: InventoryOptions) -> Dict: def _fetch_objects(self, client: Any, opts: InventoryOptions) -> Dict:
""" """
fetch_objects fetches all relevant objects from the K8S API. fetch_objects fetches all relevant objects from the K8S API.
""" """
@@ -465,10 +467,10 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
for namespace in ( for namespace in (
opts.namespaces opts.namespaces
if opts.namespaces if opts.namespaces
else self.get_available_namespaces(client) else self._get_available_namespaces(client)
): ):
vms = self.get_vms_for_namespace(client, namespace, opts) vms = self._get_vms_for_namespace(client, namespace, opts)
vmis = self.get_vmis_for_namespace(client, namespace, opts) vmis = self._get_vmis_for_namespace(client, namespace, opts)
if not vms and not vmis: if not vms and not vmis:
# Continue if no VMs and VMIs were found to avoid adding empty groups. # Continue if no VMs and VMIs were found to avoid adding empty groups.
@@ -477,18 +479,18 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
namespaces[namespace] = { namespaces[namespace] = {
"vms": vms, "vms": vms,
"vmis": vmis, "vmis": vmis,
"services": self.get_ssh_services_for_namespace(client, namespace), "services": self._get_ssh_services_for_namespace(client, namespace),
} }
return { return {
"default_hostname": self.get_default_hostname(client.configuration.host), "default_hostname": self._get_default_hostname(client.configuration.host),
"cluster_domain": self.get_cluster_domain(client), "cluster_domain": self._get_cluster_domain(client),
"namespaces": namespaces, "namespaces": namespaces,
} }
def get_cluster_domain(self, client: K8SClient) -> Optional[str]: def _get_cluster_domain(self, client: K8SClient) -> Optional[str]:
""" """
get_cluster_domain tries to get the base domain of an OpenShift cluster. _get_cluster_domain tries to get the base domain of an OpenShift cluster.
""" """
try: try:
v1_dns = client.resources.get( v1_dns = client.resources.get(
@@ -501,16 +503,16 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
obj = v1_dns.get(name="cluster") obj = v1_dns.get(name="cluster")
except DynamicApiError as exc: except DynamicApiError as exc:
self.display.debug( self.display.debug(
f"Failed to fetch cluster DNS config: {self.format_dynamic_api_exc(exc)}" f"Failed to fetch cluster DNS config: {self._format_dynamic_api_exc(exc)}"
) )
return None return None
return obj.get("spec", {}).get("baseDomain") return obj.get("spec", {}).get("baseDomain")
def get_resources( def _get_resources(
self, client: K8SClient, api_version: str, kind: str, **kwargs self, client: K8SClient, api_version: str, kind: str, **kwargs
) -> List[Dict]: ) -> List[Dict]:
""" """
get_resources uses a dynamic K8SClient to fetch resources from the K8S API. _get_resources uses a dynamic K8SClient to fetch resources from the K8S API.
""" """
client = client.resources.get(api_version=api_version, kind=kind) client = client.resources.get(api_version=api_version, kind=kind)
try: try:
@@ -518,29 +520,29 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
except DynamicApiError as exc: except DynamicApiError as exc:
self.display.debug(exc) self.display.debug(exc)
raise KubeVirtInventoryException( raise KubeVirtInventoryException(
f"Error fetching {kind} list: {self.format_dynamic_api_exc(exc)}" f"Error fetching {kind} list: {self._format_dynamic_api_exc(exc)}"
) from exc ) from exc
return [item.to_dict() for item in result.items] return [item.to_dict() for item in result.items]
def get_available_namespaces(self, client: K8SClient) -> List[str]: def _get_available_namespaces(self, client: K8SClient) -> List[str]:
""" """
get_available_namespaces lists all namespaces accessible with the _get_available_namespaces lists all namespaces accessible with the
configured credentials and returns them. configured credentials and returns them.
""" """
return [ return [
namespace["metadata"]["name"] namespace["metadata"]["name"]
for namespace in self.get_resources(client, "v1", "Namespace") for namespace in self._get_resources(client, "v1", "Namespace")
if "metadata" in namespace and "name" in namespace["metadata"] if "metadata" in namespace and "name" in namespace["metadata"]
] ]
def get_vms_for_namespace( def _get_vms_for_namespace(
self, client: K8SClient, namespace: str, opts: InventoryOptions self, client: K8SClient, namespace: str, opts: InventoryOptions
) -> List[Dict]: ) -> List[Dict]:
""" """
get_vms_for_namespace returns a list of all VirtualMachines in a namespace. _get_vms_for_namespace returns a list of all VirtualMachines in a namespace.
""" """
return self.get_resources( return self._get_resources(
client, client,
opts.api_version, opts.api_version,
"VirtualMachine", "VirtualMachine",
@@ -548,13 +550,13 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
label_selector=opts.label_selector, label_selector=opts.label_selector,
) )
def get_vmis_for_namespace( def _get_vmis_for_namespace(
self, client: K8SClient, namespace: str, opts: InventoryOptions self, client: K8SClient, namespace: str, opts: InventoryOptions
) -> List[Dict]: ) -> List[Dict]:
""" """
get_vmis_for_namespace returns a list of all VirtualMachineInstances in a namespace. _get_vmis_for_namespace returns a list of all VirtualMachineInstances in a namespace.
""" """
return self.get_resources( return self._get_resources(
client, client,
opts.api_version, opts.api_version,
"VirtualMachineInstance", "VirtualMachineInstance",
@@ -562,12 +564,14 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
label_selector=opts.label_selector, label_selector=opts.label_selector,
) )
def get_ssh_services_for_namespace(self, client: K8SClient, namespace: str) -> Dict: def _get_ssh_services_for_namespace(
self, client: K8SClient, namespace: str
) -> Dict:
""" """
get_ssh_services_for_namespace retrieves all services of a namespace exposing port 22/ssh. _get_ssh_services_for_namespace retrieves all services of a namespace exposing port 22/ssh.
The services are mapped to the name of the corresponding domain. The services are mapped to the name of the corresponding domain.
""" """
items = self.get_resources( items = self._get_resources(
client, client,
"v1", "v1",
"Service", "Service",
@@ -601,9 +605,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
return services return services
def populate_inventory(self, results: Dict, opts: InventoryOptions) -> None: def _populate_inventory(self, results: Dict, opts: InventoryOptions) -> None:
""" """
populate_inventory populates the inventory by completing the InventoryOptions _populate_inventory populates the inventory by completing the InventoryOptions
and invoking populate_inventory_from_namespace for every namespace in results. and invoking populate_inventory_from_namespace for every namespace in results.
""" """
if opts.base_domain is None: if opts.base_domain is None:
@@ -611,22 +615,22 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
if opts.name is None: if opts.name is None:
opts.name = results["default_hostname"] opts.name = results["default_hostname"]
for namespace, data in results["namespaces"].items(): for namespace, data in results["namespaces"].items():
self.populate_inventory_from_namespace(namespace, data, opts) self._populate_inventory_from_namespace(namespace, data, opts)
def populate_inventory_from_namespace( def _populate_inventory_from_namespace(
self, namespace: str, data: Dict, opts: InventoryOptions self, namespace: str, data: Dict, opts: InventoryOptions
) -> None: ) -> None:
""" """
populate_inventory_from_namespace adds groups and hosts from a _populate_inventory_from_namespace adds groups and hosts from a
namespace to the inventory. namespace to the inventory.
""" """
vms = { vms = {
vm["metadata"]["name"]: vm for vm in data["vms"] if self.obj_is_valid(vm) vm["metadata"]["name"]: vm for vm in data["vms"] if self._obj_is_valid(vm)
} }
vmis = { vmis = {
vmi["metadata"]["name"]: vmi vmi["metadata"]["name"]: vmi
for vmi in data["vmis"] for vmi in data["vmis"]
if self.obj_is_valid(vmi) if self._obj_is_valid(vmi)
} }
if not vms and not vmis: if not vms and not vmis:
@@ -636,7 +640,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
services = { services = {
domain: service domain: service
for domain, service in data["services"].items() for domain, service in data["services"].items()
if self.obj_is_valid(service) if self._obj_is_valid(service)
} }
name = self._sanitize_group_name(opts.name) name = self._sanitize_group_name(opts.name)
@@ -648,23 +652,25 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
# Add found VMs and optionally enhance with VMI data # Add found VMs and optionally enhance with VMI data
for name, vm in vms.items(): for name, vm in vms.items():
hostname = self.add_host(vm["metadata"], opts.host_format, namespace_group) hostname = self._add_host(vm["metadata"], opts.host_format, namespace_group)
self.set_vars_from_vm(hostname, vm, opts) self._set_vars_from_vm(hostname, vm, opts)
if name in vmis: if name in vmis:
self.set_vars_from_vmi(hostname, vmis[name], services, opts) self._set_vars_from_vmi(hostname, vmis[name], services, opts)
self.set_composable_vars(hostname) self._set_composable_vars(hostname)
# Add remaining VMIs without VM # Add remaining VMIs without VM
for name, vmi in vmis.items(): for name, vmi in vmis.items():
if name in vms: if name in vms:
continue continue
hostname = self.add_host(vmi["metadata"], opts.host_format, namespace_group) hostname = self._add_host(
self.set_vars_from_vmi(hostname, vmi, services, opts) vmi["metadata"], opts.host_format, namespace_group
self.set_composable_vars(hostname) )
self._set_vars_from_vmi(hostname, vmi, services, opts)
self._set_composable_vars(hostname)
def add_host(self, metadata: Dict, host_format: str, namespace_group: str) -> str: def _add_host(self, metadata: Dict, host_format: str, namespace_group: str) -> str:
""" """
add_hosts adds a host to the inventory. _add_host adds a host to the inventory.
""" """
hostname = host_format.format( hostname = host_format.format(
namespace=metadata["namespace"], namespace=metadata["namespace"],
@@ -676,20 +682,22 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
return hostname return hostname
def set_vars_from_vm(self, hostname: str, vm: Dict, opts: InventoryOptions) -> None: def _set_vars_from_vm(
self, hostname: str, vm: Dict, opts: InventoryOptions
) -> None:
""" """
set_vars_from_vm sets inventory variables from a VM prefixed with vm_. _set_vars_from_vm sets inventory variables from a VM prefixed with vm_.
""" """
self.set_common_vars(hostname, "vm", vm, opts) self._set_common_vars(hostname, "vm", vm, opts)
def set_vars_from_vmi( def _set_vars_from_vmi(
self, hostname: str, vmi: Dict, services: Dict, opts: InventoryOptions self, hostname: str, vmi: Dict, services: Dict, opts: InventoryOptions
) -> None: ) -> None:
""" """
set_vars_from_vmi sets inventory variables from a VMI prefixed with vmi_ and _set_vars_from_vmi sets inventory variables from a VMI prefixed with vmi_ and
looks up the interface to set ansible_host and ansible_port. looks up the interface to set ansible_host and ansible_port.
""" """
self.set_common_vars(hostname, "vmi", vmi, opts) self._set_common_vars(hostname, "vmi", vmi, opts)
if not (interfaces := vmi["status"].get("interfaces")): if not (interfaces := vmi["status"].get("interfaces")):
return return
@@ -710,7 +718,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
# Set up the connection # Set up the connection
service = None service = None
if self.is_windows( if self._is_windows(
vmi["status"].get("guestOSInfo", {}), vmi["status"].get("guestOSInfo", {}),
vmi["metadata"].get("annotations", {}), vmi["metadata"].get("annotations", {}),
): ):
@@ -719,7 +727,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
service = services.get( service = services.get(
vmi["metadata"].get("labels", {}).get(LABEL_KUBEVIRT_IO_DOMAIN) vmi["metadata"].get("labels", {}).get(LABEL_KUBEVIRT_IO_DOMAIN)
) )
self.set_ansible_host_and_port( self._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
interface["ipAddress"], interface["ipAddress"],
@@ -727,11 +735,11 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
opts, opts,
) )
def set_common_vars( def _set_common_vars(
self, hostname: str, prefix: str, obj: Dict, opts: InventoryOptions self, hostname: str, prefix: str, obj: Dict, opts: InventoryOptions
): ):
""" """
set_common_vars sets common inventory variables from VMs or VMIs. _set_common_vars sets common inventory variables from VMs or VMIs.
""" """
# Add hostvars from metadata # Add hostvars from metadata
if annotations := obj["metadata"].get("annotations"): if annotations := obj["metadata"].get("annotations"):
@@ -740,7 +748,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
self.inventory.set_variable(hostname, f"{prefix}_labels", labels) self.inventory.set_variable(hostname, f"{prefix}_labels", labels)
# Create label groups and add vm to it if enabled # Create label groups and add vm to it if enabled
if opts.create_groups: if opts.create_groups:
self.set_groups_from_labels(hostname, labels) self._set_groups_from_labels(hostname, labels)
if resource_version := obj["metadata"].get("resourceVersion"): if resource_version := obj["metadata"].get("resourceVersion"):
self.inventory.set_variable( self.inventory.set_variable(
hostname, f"{prefix}_resource_version", resource_version hostname, f"{prefix}_resource_version", resource_version
@@ -751,12 +759,12 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
# Add hostvars from status # Add hostvars from status
for key, value in obj["status"].items(): for key, value in obj["status"].items():
self.inventory.set_variable( self.inventory.set_variable(
hostname, f"{prefix}_{self.format_var_name(key)}", value hostname, f"{prefix}_{self._format_var_name(key)}", value
) )
def set_groups_from_labels(self, hostname: str, labels: Dict) -> None: def _set_groups_from_labels(self, hostname: str, labels: Dict) -> None:
""" """
set_groups_from_labels adds groups for each label of a VM or VMI and _set_groups_from_labels adds groups for each label of a VM or VMI and
adds the host to each group. adds the host to each group.
""" """
groups = [] groups = []
@@ -769,7 +777,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
self.inventory.add_group(group) self.inventory.add_group(group)
self.inventory.add_child(group, hostname) self.inventory.add_child(group, hostname)
def set_ansible_host_and_port( def _set_ansible_host_and_port(
self, self,
vmi: Dict, vmi: Dict,
hostname: str, hostname: str,
@@ -778,7 +786,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
opts: InventoryOptions, opts: InventoryOptions,
) -> None: ) -> None:
""" """
set_ansible_host_and_port sets the ansible_host and possibly the ansible_port var. _set_ansible_host_and_port sets the ansible_host and possibly the ansible_port var.
Secondary interfaces have priority over a service exposing SSH Secondary interfaces have priority over a service exposing SSH
""" """
ansible_host = None ansible_host = None
@@ -795,8 +803,8 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
node_name = vmi["status"].get("nodeName") node_name = vmi["status"].get("nodeName")
if node_name and opts.append_base_domain and opts.base_domain: if node_name and opts.append_base_domain and opts.base_domain:
node_name += f".{opts.base_domain}" node_name += f".{opts.base_domain}"
host = self.get_host_from_service(service, node_name) host = self._get_host_from_service(service, node_name)
port = self.get_port_from_service(service) port = self._get_port_from_service(service)
if host is not None and port is not None: if host is not None and port is not None:
ansible_host = host ansible_host = host
ansible_port = port ansible_port = port
@@ -808,9 +816,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
self.inventory.set_variable(hostname, "ansible_host", ansible_host) self.inventory.set_variable(hostname, "ansible_host", ansible_host)
self.inventory.set_variable(hostname, "ansible_port", ansible_port) self.inventory.set_variable(hostname, "ansible_port", ansible_port)
def set_composable_vars(self, hostname: str) -> None: def _set_composable_vars(self, hostname: str) -> None:
""" """
set_composable_vars sets vars per _set_composable_vars sets vars per
https://docs.ansible.com/ansible/latest/dev_guide/developing_inventory.html https://docs.ansible.com/ansible/latest/dev_guide/developing_inventory.html
""" """
hostvars = self.inventory.get_host(hostname).get_vars() hostvars = self.inventory.get_host(hostname).get_vars()

2
tests/unit/plugins/inventory/blackbox/test_kubevirt_ansible_connection_winrm.py
View File

@@ -69,7 +69,7 @@ WINDOWS_VMI_4 = merge_dicts(
], ],
) )
def test_ansible_connection_winrm(inventory, hosts, vmi, expected): def test_ansible_connection_winrm(inventory, hosts, vmi, expected):
inventory.populate_inventory( inventory._populate_inventory(
{ {
"default_hostname": "test", "default_hostname": "test",
"cluster_domain": "test.com", "cluster_domain": "test.com",

2
tests/unit/plugins/inventory/blackbox/test_kubevirt_set_composable_vars.py
View File

@@ -45,7 +45,7 @@ def test_set_composable_vars(
"keyed_groups": [{"prefix": "fedora", "key": "vmi_guest_os_info.versionId"}], "keyed_groups": [{"prefix": "fedora", "key": "vmi_guest_os_info.versionId"}],
"strict": True, "strict": True,
} }
inventory.populate_inventory( inventory._populate_inventory(
{ {
"default_hostname": "test", "default_hostname": "test",
"cluster_domain": "test.com", "cluster_domain": "test.com",

2
tests/unit/plugins/inventory/blackbox/test_kubevirt_stopped_vm.py
View File

@@ -44,7 +44,7 @@ VMI1 = {
def test_stopped_vm(inventory, hosts): def test_stopped_vm(inventory, hosts):
inventory.populate_inventory( inventory._populate_inventory(
{ {
"default_hostname": "test", "default_hostname": "test",
"cluster_domain": "test.com", "cluster_domain": "test.com",

20
tests/unit/plugins/inventory/test_kubevirt.py
View File

@@ -29,7 +29,7 @@ from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants im
], ],
) )
def test_get_default_hostname(host, expected): def test_get_default_hostname(host, expected):
assert InventoryModule.get_default_hostname(host) == expected assert InventoryModule._get_default_hostname(host) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -44,7 +44,7 @@ def test_get_default_hostname(host, expected):
], ],
) )
def test_format_var_name(name, expected): def test_format_var_name(name, expected):
assert InventoryModule.format_var_name(name) == expected assert InventoryModule._format_var_name(name) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -93,7 +93,7 @@ def test_format_var_name(name, expected):
], ],
) )
def test_obj_is_valid(obj, expected): def test_obj_is_valid(obj, expected):
assert InventoryModule.obj_is_valid(obj) == expected assert InventoryModule._obj_is_valid(obj) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -151,7 +151,7 @@ def test_obj_is_valid(obj, expected):
], ],
) )
def test_get_host_from_service(service, node_name, expected): def test_get_host_from_service(service, node_name, expected):
assert InventoryModule.get_host_from_service(service, node_name) == expected assert InventoryModule._get_host_from_service(service, node_name) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -199,7 +199,7 @@ def test_get_host_from_service(service, node_name, expected):
], ],
) )
def test_port_from_service(service, expected): def test_port_from_service(service, expected):
assert InventoryModule.get_port_from_service(service) == expected assert InventoryModule._get_port_from_service(service) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -251,11 +251,11 @@ def test_port_from_service(service, expected):
], ],
) )
def test_is_windows(guest_os_info, annotations, expected): def test_is_windows(guest_os_info, annotations, expected):
assert InventoryModule.is_windows(guest_os_info, annotations) == expected assert InventoryModule._is_windows(guest_os_info, annotations) == expected
def test_get_cluster_domain(inventory, client): def test_get_cluster_domain(inventory, client):
assert inventory.get_cluster_domain(client) == DEFAULT_BASE_DOMAIN assert inventory._get_cluster_domain(client) == DEFAULT_BASE_DOMAIN
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -292,10 +292,10 @@ def test_get_cluster_domain(inventory, client):
) )
def test_populate_inventory(mocker, inventory, results, expected): def test_populate_inventory(mocker, inventory, results, expected):
populate_inventory_from_namespace = mocker.patch.object( populate_inventory_from_namespace = mocker.patch.object(
inventory, "populate_inventory_from_namespace" inventory, "_populate_inventory_from_namespace"
) )
inventory.populate_inventory(results, InventoryOptions()) inventory._populate_inventory(results, InventoryOptions())
opts = InventoryOptions( opts = InventoryOptions(
base_domain=results["cluster_domain"], name=results["default_hostname"] base_domain=results["cluster_domain"], name=results["default_hostname"]
@@ -321,7 +321,7 @@ def test_populate_inventory(mocker, inventory, results, expected):
) )
def test_set_groups_from_labels(inventory, groups, labels, expected): def test_set_groups_from_labels(inventory, groups, labels, expected):
hostname = "default-testvm" hostname = "default-testvm"
inventory.set_groups_from_labels(hostname, labels) inventory._set_groups_from_labels(hostname, labels)
for group in expected: for group in expected:
assert group in groups assert group in groups
assert hostname in groups[group]["children"] assert hostname in groups[group]["children"]

2
tests/unit/plugins/inventory/test_kubevirt_add_host.py
View File

@@ -62,7 +62,7 @@ def test_add_host(inventory, groups, hosts, host_format, expected):
namespace_group = "namespace_default" namespace_group = "namespace_default"
inventory.inventory.add_group(namespace_group) inventory.inventory.add_group(namespace_group)
inventory.add_host( inventory._add_host(
{ {
"name": "testvm", "name": "testvm",
"namespace": DEFAULT_NAMESPACE, "namespace": DEFAULT_NAMESPACE,

8
tests/unit/plugins/inventory/test_kubevirt_connections_compatibility.py
View File

@@ -23,7 +23,7 @@ def test_config_data_without_connections_ignored(inventory):
"label_selector": "app=test", "label_selector": "app=test",
} }
inventory.connections_compatibility(config_data) inventory._connections_compatibility(config_data)
assert config_data["name"] == "connection-1" assert config_data["name"] == "connection-1"
assert config_data["namespaces"] == ["default"] assert config_data["namespaces"] == ["default"]
@@ -46,7 +46,7 @@ def test_single_connection_supported(inventory):
"label_selector": "app=something", "label_selector": "app=something",
} }
inventory.connections_compatibility(config_data) inventory._connections_compatibility(config_data)
assert config_data["name"] == "connection-1" assert config_data["name"] == "connection-1"
assert config_data["namespaces"] == ["default"] assert config_data["namespaces"] == ["default"]
@@ -58,7 +58,7 @@ def test_multiple_connections_not_supported(inventory):
with pytest.raises( with pytest.raises(
AnsibleError, match="Split your connections into multiple configuration files." AnsibleError, match="Split your connections into multiple configuration files."
): ):
inventory.connections_compatibility( inventory._connections_compatibility(
{ {
"connections": [ "connections": [
{ {
@@ -84,4 +84,4 @@ def test_multiple_connections_not_supported(inventory):
) )
def test_connections_exceptions(inventory, config_data, expected): def test_connections_exceptions(inventory, config_data, expected):
with pytest.raises(KubeVirtInventoryException, match=expected): with pytest.raises(KubeVirtInventoryException, match=expected):
inventory.connections_compatibility(config_data) inventory._connections_compatibility(config_data)

28
tests/unit/plugins/inventory/test_kubevirt_fetch_objects.py
View File

@@ -37,25 +37,25 @@ from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants im
) )
def test_fetch_objects(mocker, inventory, opts, namespaces): def test_fetch_objects(mocker, inventory, opts, namespaces):
get_available_namespaces = mocker.patch.object( get_available_namespaces = mocker.patch.object(
inventory, "get_available_namespaces", return_value=[DEFAULT_NAMESPACE] inventory, "_get_available_namespaces", return_value=[DEFAULT_NAMESPACE]
) )
get_vms_for_namespace = mocker.patch.object( get_vms_for_namespace = mocker.patch.object(
inventory, "get_vms_for_namespace", return_value=[{}] inventory, "_get_vms_for_namespace", return_value=[{}]
) )
get_vmis_for_namespace = mocker.patch.object( get_vmis_for_namespace = mocker.patch.object(
inventory, "get_vmis_for_namespace", return_value=[{}] inventory, "_get_vmis_for_namespace", return_value=[{}]
) )
get_ssh_services_for_namespace = mocker.patch.object( get_ssh_services_for_namespace = mocker.patch.object(
inventory, "get_ssh_services_for_namespace", return_value=[] inventory, "_get_ssh_services_for_namespace", return_value=[]
) )
get_default_hostname = mocker.patch.object( get_default_hostname = mocker.patch.object(
inventory, "get_default_hostname", return_value="default-hostname" inventory, "_get_default_hostname", return_value="default-hostname"
) )
get_cluster_domain = mocker.patch.object( get_cluster_domain = mocker.patch.object(
inventory, "get_cluster_domain", return_value="test.com" inventory, "_get_cluster_domain", return_value="test.com"
) )
inventory.fetch_objects(mocker.Mock(), opts) inventory._fetch_objects(mocker.Mock(), opts)
if opts.namespaces: if opts.namespaces:
get_available_namespaces.assert_not_called() get_available_namespaces.assert_not_called()
@@ -77,25 +77,25 @@ def test_fetch_objects(mocker, inventory, opts, namespaces):
def test_fetch_objects_early_return(mocker, inventory): def test_fetch_objects_early_return(mocker, inventory):
get_available_namespaces = mocker.patch.object( get_available_namespaces = mocker.patch.object(
inventory, "get_available_namespaces", return_value=[DEFAULT_NAMESPACE] inventory, "_get_available_namespaces", return_value=[DEFAULT_NAMESPACE]
) )
get_vms_for_namespace = mocker.patch.object( get_vms_for_namespace = mocker.patch.object(
inventory, "get_vms_for_namespace", return_value=[] inventory, "_get_vms_for_namespace", return_value=[]
) )
get_vmis_for_namespace = mocker.patch.object( get_vmis_for_namespace = mocker.patch.object(
inventory, "get_vmis_for_namespace", return_value=[] inventory, "_get_vmis_for_namespace", return_value=[]
) )
get_ssh_services_for_namespace = mocker.patch.object( get_ssh_services_for_namespace = mocker.patch.object(
inventory, "get_ssh_services_for_namespace" inventory, "_get_ssh_services_for_namespace"
) )
get_default_hostname = mocker.patch.object( get_default_hostname = mocker.patch.object(
inventory, "get_default_hostname", return_value="default-hostname" inventory, "_get_default_hostname", return_value="default-hostname"
) )
get_cluster_domain = mocker.patch.object( get_cluster_domain = mocker.patch.object(
inventory, "get_cluster_domain", return_value="test.com" inventory, "_get_cluster_domain", return_value="test.com"
) )
inventory.fetch_objects(mocker.Mock(), InventoryOptions()) inventory._fetch_objects(mocker.Mock(), InventoryOptions())
get_available_namespaces.assert_called_once() get_available_namespaces.assert_called_once()
get_vms_for_namespace.assert_called_once_with( get_vms_for_namespace.assert_called_once_with(

3
tests/unit/plugins/inventory/test_kubevirt_format_dynamic_api_exc.py
View File

@@ -52,5 +52,6 @@ def status_reason_error(mocker):
) )
def test_format_dynamic_api_exc(request, exc, expected): def test_format_dynamic_api_exc(request, exc, expected):
assert ( assert (
InventoryModule.format_dynamic_api_exc(request.getfixturevalue(exc)) == expected InventoryModule._format_dynamic_api_exc(request.getfixturevalue(exc))
== expected
) )

16
tests/unit/plugins/inventory/test_kubevirt_get_resources.py
View File

@@ -30,19 +30,19 @@ from ansible_collections.kubevirt.core.tests.unit.plugins.inventory.constants im
indirect=["client"], indirect=["client"],
) )
def test_get_resources(inventory, client): def test_get_resources(inventory, client):
assert inventory.get_resources(client, "v1", "Namespace") == [ assert inventory._get_resources(client, "v1", "Namespace") == [
{"metadata": {"name": DEFAULT_NAMESPACE}} {"metadata": {"name": DEFAULT_NAMESPACE}}
] ]
assert inventory.get_resources(client, "v1", "Service") == [ assert inventory._get_resources(client, "v1", "Service") == [
{"metadata": {"name": "testsvc"}} {"metadata": {"name": "testsvc"}}
] ]
assert inventory.get_resources(client, "config.openshift.io/v1", "DNS") == [ assert inventory._get_resources(client, "config.openshift.io/v1", "DNS") == [
{"spec": {"baseDomain": DEFAULT_BASE_DOMAIN}} {"spec": {"baseDomain": DEFAULT_BASE_DOMAIN}}
] ]
assert inventory.get_resources(client, "kubevirt.io/v1", "VirtualMachine") == [ assert inventory._get_resources(client, "kubevirt.io/v1", "VirtualMachine") == [
{"metadata": {"name": "testvm"}} {"metadata": {"name": "testvm"}}
] ]
assert inventory.get_resources( assert inventory._get_resources(
client, "kubevirt.io/v1", "VirtualMachineInstance" client, "kubevirt.io/v1", "VirtualMachineInstance"
) == [{"metadata": {"name": "testvmi"}}] ) == [{"metadata": {"name": "testvmi"}}]
@@ -67,7 +67,7 @@ def test_get_resources(inventory, client):
indirect=["client"], indirect=["client"],
) )
def test_get_available_namespaces(inventory, client, expected): def test_get_available_namespaces(inventory, client, expected):
assert inventory.get_available_namespaces(client) == expected assert inventory._get_available_namespaces(client) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -83,7 +83,7 @@ def test_get_available_namespaces(inventory, client, expected):
indirect=["client"], indirect=["client"],
) )
def test_get_vms_for_namespace(inventory, client): def test_get_vms_for_namespace(inventory, client):
assert inventory.get_vms_for_namespace( assert inventory._get_vms_for_namespace(
client, DEFAULT_NAMESPACE, InventoryOptions() client, DEFAULT_NAMESPACE, InventoryOptions()
) == [{"metadata": {"name": "testvm1"}}, {"metadata": {"name": "testvm2"}}] ) == [{"metadata": {"name": "testvm1"}}, {"metadata": {"name": "testvm2"}}]
@@ -101,6 +101,6 @@ def test_get_vms_for_namespace(inventory, client):
indirect=["client"], indirect=["client"],
) )
def test_get_vmis_for_namespace(inventory, client): def test_get_vmis_for_namespace(inventory, client):
assert inventory.get_vmis_for_namespace( assert inventory._get_vmis_for_namespace(
client, DEFAULT_NAMESPACE, InventoryOptions() client, DEFAULT_NAMESPACE, InventoryOptions()
) == [{"metadata": {"name": "testvmi1"}}, {"metadata": {"name": "testvmi2"}}] ) == [{"metadata": {"name": "testvmi1"}}, {"metadata": {"name": "testvmi2"}}]

4
tests/unit/plugins/inventory/test_kubevirt_get_ssh_services_for_namespace.py
View File

@@ -57,7 +57,7 @@ SVC_NP_SSH = {
indirect=["client"], indirect=["client"],
) )
def test_get_ssh_services_for_namespace(inventory, client): def test_get_ssh_services_for_namespace(inventory, client):
assert inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) == { assert inventory._get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) == {
"test-lb-ssh": SVC_LB_SSH, "test-lb-ssh": SVC_LB_SSH,
"test-np-ssh": SVC_NP_SSH, "test-np-ssh": SVC_NP_SSH,
} }
@@ -165,4 +165,4 @@ SVC_NO_SELECTOR = {
indirect=["client"], indirect=["client"],
) )
def test_ignore_unwanted_services(inventory, client): def test_ignore_unwanted_services(inventory, client):
assert not inventory.get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE) assert not inventory._get_ssh_services_for_namespace(client, DEFAULT_NAMESPACE)

10
tests/unit/plugins/inventory/test_kubevirt_parse.py
View File

@@ -102,8 +102,8 @@ def test_config_data_to_opts(mocker, inventory, config_data, expected):
mocker.patch.object(inventory, "get_cache_key") mocker.patch.object(inventory, "get_cache_key")
mocker.patch.object(inventory, "get_option") mocker.patch.object(inventory, "get_option")
mocker.patch.object(kubevirt, "get_api_client") mocker.patch.object(kubevirt, "get_api_client")
mocker.patch.object(inventory, "fetch_objects") mocker.patch.object(inventory, "_fetch_objects")
populate_inventory = mocker.patch.object(inventory, "populate_inventory") populate_inventory = mocker.patch.object(inventory, "_populate_inventory")
inventory.parse(None, None, "", False) inventory.parse(None, None, "", False)
@@ -139,8 +139,8 @@ def test_use_of_cache(
) )
get_option = mocker.patch.object(inventory, "get_option", return_value=cache_option) get_option = mocker.patch.object(inventory, "get_option", return_value=cache_option)
get_api_client = mocker.patch.object(kubevirt, "get_api_client") get_api_client = mocker.patch.object(kubevirt, "get_api_client")
fetch_objects = mocker.patch.object(inventory, "fetch_objects") fetch_objects = mocker.patch.object(inventory, "_fetch_objects")
populate_inventory = mocker.patch.object(inventory, "populate_inventory") populate_inventory = mocker.patch.object(inventory, "_populate_inventory")
if cache_parse is None: if cache_parse is None:
inventory.parse(None, None, path) inventory.parse(None, None, path)
@@ -173,7 +173,7 @@ def test_k8s_client_missing(mocker, inventory, present):
mocker.patch.object(inventory, "_read_config_data", return_value={}) mocker.patch.object(inventory, "_read_config_data", return_value={})
mocker.patch.object(inventory, "get_cache_key") mocker.patch.object(inventory, "get_cache_key")
mocker.patch.object(inventory, "get_option") mocker.patch.object(inventory, "get_option")
fetch_objects = mocker.patch.object(inventory, "fetch_objects") fetch_objects = mocker.patch.object(inventory, "_fetch_objects")
if present: if present:
inventory.parse(None, None, "", False) inventory.parse(None, None, "", False)

12
tests/unit/plugins/inventory/test_kubevirt_populate_inventory_from_namespace.py
View File

@@ -129,15 +129,15 @@ def test_populate_inventory_from_namespace(
set_vars_from_vmi_calls.append(mocker.call(hostname, vmi, {}, opts)) set_vars_from_vmi_calls.append(mocker.call(hostname, vmi, {}, opts))
set_composable_vars_calls.append(mocker.call(hostname)) set_composable_vars_calls.append(mocker.call(hostname))
obj_is_valid = mocker.patch.object(inventory, "obj_is_valid", return_value=True) obj_is_valid = mocker.patch.object(inventory, "_obj_is_valid", return_value=True)
add_host = mocker.patch.object( add_host = mocker.patch.object(
inventory, "add_host", side_effect=add_host_side_effects inventory, "_add_host", side_effect=add_host_side_effects
) )
set_vars_from_vm = mocker.patch.object(inventory, "set_vars_from_vm") 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_vars_from_vmi = mocker.patch.object(inventory, "_set_vars_from_vmi")
set_composable_vars = mocker.patch.object(inventory, "set_composable_vars") set_composable_vars = mocker.patch.object(inventory, "_set_composable_vars")
inventory.populate_inventory_from_namespace( inventory._populate_inventory_from_namespace(
DEFAULT_NAMESPACE, {"vms": vms, "vmis": vmis, "services": {}}, opts DEFAULT_NAMESPACE, {"vms": vms, "vmis": vmis, "services": {}}, opts
) )

18
tests/unit/plugins/inventory/test_kubevirt_set_ansible_host_and_port.py
View File

@@ -27,7 +27,7 @@ def test_use_ip_address_by_default(mocker, inventory, opts):
hostname = "default-testvm" hostname = "default-testvm"
ip_address = "1.1.1.1" ip_address = "1.1.1.1"
inventory.set_ansible_host_and_port({}, hostname, ip_address, None, opts) inventory._set_ansible_host_and_port({}, hostname, ip_address, None, opts)
set_variable.assert_has_calls( set_variable.assert_has_calls(
[ [
@@ -53,7 +53,7 @@ def test_kube_secondary_dns(mocker, inventory, base_domain):
"status": {"interfaces": [{"name": "awesome"}]}, "status": {"interfaces": [{"name": "awesome"}]},
} }
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
"1.1.1.1", "1.1.1.1",
@@ -86,7 +86,7 @@ def test_kube_secondary_dns_precedence_over_service(mocker, inventory):
"status": {"interfaces": [{"name": "awesome"}]}, "status": {"interfaces": [{"name": "awesome"}]},
} }
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
"1.1.1.1", "1.1.1.1",
@@ -170,7 +170,7 @@ def test_service(mocker, inventory, service, expected_host, expected_port):
}, },
} }
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
"1.1.1.1", "1.1.1.1",
@@ -201,7 +201,7 @@ def test_service_append_base_domain(mocker, inventory):
"ports": [{"nodePort": 25}], "ports": [{"nodePort": 25}],
}, },
} }
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
"1.1.1.1", "1.1.1.1",
@@ -229,8 +229,8 @@ def test_service_append_base_domain(mocker, inventory):
) )
def test_service_fallback(mocker, inventory, host, port): def test_service_fallback(mocker, inventory, host, port):
set_variable = mocker.patch.object(inventory.inventory, "set_variable") 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_host_from_service", return_value=host)
mocker.patch.object(inventory, "get_port_from_service", return_value=port) mocker.patch.object(inventory, "_get_port_from_service", return_value=port)
hostname = "default-testvm" hostname = "default-testvm"
vmi = { vmi = {
@@ -238,7 +238,7 @@ def test_service_fallback(mocker, inventory, host, port):
"nodeName": "testnode", "nodeName": "testnode",
}, },
} }
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
vmi, vmi,
hostname, hostname,
"1.1.1.1", "1.1.1.1",
@@ -258,7 +258,7 @@ def test_no_service_if_network_name(mocker, inventory):
set_variable = mocker.patch.object(inventory.inventory, "set_variable") set_variable = mocker.patch.object(inventory.inventory, "set_variable")
hostname = "default-testvm" hostname = "default-testvm"
inventory.set_ansible_host_and_port( inventory._set_ansible_host_and_port(
{}, {},
hostname, hostname,
"1.2.3.4", "1.2.3.4",

12
tests/unit/plugins/inventory/test_kubevirt_set_common_vars.py
View File

@@ -85,7 +85,7 @@ def test_set_common_vars(inventory, hosts, obj, expected):
hostname = "default-testvm" hostname = "default-testvm"
prefix = "".join(choice(ascii_lowercase) for i in range(5)) prefix = "".join(choice(ascii_lowercase) for i in range(5))
inventory.inventory.add_host(hostname) inventory.inventory.add_host(hostname)
inventory.set_common_vars(hostname, prefix, obj, InventoryOptions()) inventory._set_common_vars(hostname, prefix, obj, InventoryOptions())
for key, value in expected.items(): for key, value in expected.items():
prefixed_key = f"{prefix}_{key}" prefixed_key = f"{prefix}_{key}"
@@ -102,13 +102,13 @@ def test_set_common_vars(inventory, hosts, obj, expected):
) )
def test_set_common_vars_create_groups(mocker, inventory, create_groups): def test_set_common_vars_create_groups(mocker, inventory, create_groups):
mocker.patch.object(inventory.inventory, "set_variable") mocker.patch.object(inventory.inventory, "set_variable")
set_groups_from_labels = mocker.patch.object(inventory, "set_groups_from_labels") set_groups_from_labels = mocker.patch.object(inventory, "_set_groups_from_labels")
hostname = "default-testvm" hostname = "default-testvm"
labels = {"testkey": "testval"} labels = {"testkey": "testval"}
opts = InventoryOptions(create_groups=create_groups) opts = InventoryOptions(create_groups=create_groups)
inventory.set_common_vars( inventory._set_common_vars(
hostname, "prefix", {"metadata": {"labels": labels}, "status": {}}, opts hostname, "prefix", {"metadata": {"labels": labels}, "status": {}}, opts
) )
@@ -123,10 +123,10 @@ def test_called_by_set_vars_from(mocker, inventory):
opts = InventoryOptions() opts = InventoryOptions()
obj = {"status": {}} obj = {"status": {}}
set_common_vars = mocker.patch.object(inventory, "set_common_vars") set_common_vars = mocker.patch.object(inventory, "_set_common_vars")
inventory.set_vars_from_vm(hostname, obj, opts) inventory._set_vars_from_vm(hostname, obj, opts)
inventory.set_vars_from_vmi(hostname, obj, {}, opts) inventory._set_vars_from_vmi(hostname, obj, {}, opts)
set_common_vars.assert_has_calls( set_common_vars.assert_has_calls(
[ [

38
tests/unit/plugins/inventory/test_kubevirt_set_vars_from_vmi.py
View File

@@ -13,21 +13,21 @@ from ansible_collections.kubevirt.core.plugins.inventory.kubevirt import (
def test_ignore_vmi_without_interface(mocker, inventory): def test_ignore_vmi_without_interface(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
set_ansible_host_and_port = mocker.patch.object( set_ansible_host_and_port = mocker.patch.object(
inventory, "set_ansible_host_and_port" inventory, "_set_ansible_host_and_port"
) )
vmi = {"status": {}} vmi = {"status": {}}
inventory.set_vars_from_vmi("default-testvm", vmi, {}, InventoryOptions()) inventory._set_vars_from_vmi("default-testvm", vmi, {}, InventoryOptions())
set_ansible_host_and_port.assert_not_called() set_ansible_host_and_port.assert_not_called()
def test_use_first_interface_by_default(mocker, inventory): def test_use_first_interface_by_default(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
set_ansible_host_and_port = mocker.patch.object( set_ansible_host_and_port = mocker.patch.object(
inventory, "set_ansible_host_and_port" inventory, "_set_ansible_host_and_port"
) )
hostname = "default-testvm" hostname = "default-testvm"
@@ -36,7 +36,7 @@ def test_use_first_interface_by_default(mocker, inventory):
"status": {"interfaces": [{"ipAddress": "1.1.1.1"}, {"ipAddress": "2.2.2.2"}]}, "status": {"interfaces": [{"ipAddress": "1.1.1.1"}, {"ipAddress": "2.2.2.2"}]},
} }
opts = InventoryOptions() opts = InventoryOptions()
inventory.set_vars_from_vmi(hostname, vmi, {}, opts) inventory._set_vars_from_vmi(hostname, vmi, {}, opts)
set_ansible_host_and_port.assert_called_once_with( set_ansible_host_and_port.assert_called_once_with(
vmi, hostname, "1.1.1.1", None, opts vmi, hostname, "1.1.1.1", None, opts
@@ -44,9 +44,9 @@ def test_use_first_interface_by_default(mocker, inventory):
def test_use_named_interface(mocker, inventory): def test_use_named_interface(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
set_ansible_host_and_port = mocker.patch.object( set_ansible_host_and_port = mocker.patch.object(
inventory, "set_ansible_host_and_port" inventory, "_set_ansible_host_and_port"
) )
hostname = "default-testvm" hostname = "default-testvm"
@@ -60,7 +60,7 @@ def test_use_named_interface(mocker, inventory):
}, },
} }
opts = InventoryOptions(network_name="second") opts = InventoryOptions(network_name="second")
inventory.set_vars_from_vmi(hostname, vmi, {}, opts) inventory._set_vars_from_vmi(hostname, vmi, {}, opts)
set_ansible_host_and_port.assert_called_once_with( set_ansible_host_and_port.assert_called_once_with(
vmi, hostname, "2.2.2.2", None, opts vmi, hostname, "2.2.2.2", None, opts
@@ -68,16 +68,16 @@ def test_use_named_interface(mocker, inventory):
def test_ignore_vmi_without_named_interface(mocker, inventory): def test_ignore_vmi_without_named_interface(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
set_ansible_host_and_port = mocker.patch.object( set_ansible_host_and_port = mocker.patch.object(
inventory, "set_ansible_host_and_port" inventory, "_set_ansible_host_and_port"
) )
vmi = { vmi = {
"metadata": {}, "metadata": {},
"status": {"interfaces": [{"name": "somename", "ipAddress": "1.1.1.1"}]}, "status": {"interfaces": [{"name": "somename", "ipAddress": "1.1.1.1"}]},
} }
inventory.set_vars_from_vmi( inventory._set_vars_from_vmi(
"default-testvm", vmi, {}, InventoryOptions(network_name="awesome") "default-testvm", vmi, {}, InventoryOptions(network_name="awesome")
) )
@@ -85,22 +85,22 @@ def test_ignore_vmi_without_named_interface(mocker, inventory):
def test_set_winrm_if_windows(mocker, inventory): def test_set_winrm_if_windows(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
mocker.patch.object(inventory, "is_windows", return_value=True) mocker.patch.object(inventory, "_is_windows", return_value=True)
mocker.patch.object(inventory, "set_ansible_host_and_port") mocker.patch.object(inventory, "_set_ansible_host_and_port")
set_variable = mocker.patch.object(inventory.inventory, "set_variable") set_variable = mocker.patch.object(inventory.inventory, "set_variable")
hostname = "default-testvm" hostname = "default-testvm"
vmi = {"metadata": {}, "status": {"interfaces": [{"ipAddress": "1.1.1.1"}]}} vmi = {"metadata": {}, "status": {"interfaces": [{"ipAddress": "1.1.1.1"}]}}
inventory.set_vars_from_vmi(hostname, vmi, {}, InventoryOptions()) inventory._set_vars_from_vmi(hostname, vmi, {}, InventoryOptions())
set_variable.assert_called_once_with(hostname, "ansible_connection", "winrm") set_variable.assert_called_once_with(hostname, "ansible_connection", "winrm")
def test_service_lookup(mocker, inventory): def test_service_lookup(mocker, inventory):
mocker.patch.object(inventory, "set_common_vars") mocker.patch.object(inventory, "_set_common_vars")
set_ansible_host_and_port = mocker.patch.object( set_ansible_host_and_port = mocker.patch.object(
inventory, "set_ansible_host_and_port" inventory, "_set_ansible_host_and_port"
) )
hostname = "default-testvm" hostname = "default-testvm"
@@ -110,7 +110,7 @@ def test_service_lookup(mocker, inventory):
} }
opts = InventoryOptions() opts = InventoryOptions()
service = {"metadata": {"name": "testsvc"}} service = {"metadata": {"name": "testsvc"}}
inventory.set_vars_from_vmi(hostname, vmi, {"testdomain": service}, opts) inventory._set_vars_from_vmi(hostname, vmi, {"testdomain": service}, opts)
set_ansible_host_and_port.assert_called_once_with( set_ansible_host_and_port.assert_called_once_with(
vmi, hostname, "1.1.1.1", service, opts vmi, hostname, "1.1.1.1", service, opts