diff --git a/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object.py b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object.py index fe34c8f3..a4d32ebc 100644 --- a/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object.py +++ b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object.py @@ -13,126 +13,91 @@ class OpenStackEndpointListObject: self.url = None def set_id(self, id: str): - """ - Setter for id + """Setter for id + Args: - id (): the id - - Returns: - + id (str): the id """ self.id = id def get_id(self) -> str: - """ - Getter for id - Returns: the id + """Getter for id + Returns: the id """ return self.id def set_region(self, region: str): - """ - Setter for region - Args: - region (): the region + """Setter for region - Returns: + Args: + region (str): the region """ self.region = region def get_region(self) -> str: - """ - Getter for region - Returns: - - """ + """Getter for region""" return self.region def set_service_name(self, service_name: str): - """ - Setter for service name + """Setter for service name + Args: - service_name (): the service name - - Returns: - + service_name (str): the service name """ self.service_name = service_name def get_service_name(self) -> str: - """ - Getter for service name - Returns: - - """ + """Getter for service name""" return self.service_name - def set_service_type(self, service_type: str): - """ - Setter for service type - Args: - service_type (): the service type + def get_service_type(self) -> str: + """Getter for service type""" + return self.service_type - Returns: + def set_service_type(self, service_type: str): + """Setter for service type + + Args: + service_type (str): the service type """ self.service_type = service_type def set_enabled(self, enabled: bool): - """ - Setter for enabled + """Setter for enabled + Args: - enabled (): True or False - - Returns: - + enabled (bool): True or False """ self.enabled = enabled def get_enabled(self) -> bool: - """ - Getter for enabled - Returns: - - """ + """Getter for enabled""" return self.enabled def set_interface(self, interface: str): - """ - Setter for interface - Args: - interface (): the interface + """Setter for interface - Returns: + Args: + interface (str): the interface """ self.interface = interface def get_interface(self) -> str: - """ - Getter for interface - Returns: - - """ + """Getter for interface""" return self.interface def set_url(self, url: str): - """ - Setter for url + """Setter for url + Args: - url (): the url - - Returns: - + url (str): the url """ self.url = url def get_url(self) -> str: - """ - Getter for url - Returns: - - """ + """Getter for url""" return self.url diff --git a/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object_filter.py b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object_filter.py new file mode 100644 index 00000000..90846a89 --- /dev/null +++ b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object_filter.py @@ -0,0 +1,77 @@ +from typing import Optional + + +class OpenStackEndpointListObjectFilter: + """ + Class to filter OpenStack Endpoint List objects. + """ + + def __init__(self): + self.region: Optional[str] = None + self.service_name: Optional[str] = None + self.service_type: Optional[str] = None + self.interface: Optional[str] = None + + def get_region(self) -> Optional[str]: + """ + Getter for the filter by Region. + """ + return self.region + + def set_region(self, region: str): + """ + Setter for the filter by Region. + """ + self.region = region + return self + + def get_service_name(self) -> Optional[str]: + """ + Getter for the filter by Service Name. + """ + return self.service_name + + def set_service_name(self, service_name: str): + """ + Setter for the filter by Service Name. + """ + self.service_name = service_name + return self + + def get_service_type(self) -> Optional[str]: + """ + Getter for the filter by Service Type. + """ + return self.service_type + + def set_service_type(self, service_type: str): + """ + Setter for the filter by Service Type. + """ + self.service_type = service_type + return self + + def get_interface(self) -> Optional[str]: + """ + Getter for the filter by Interface. + """ + return self.interface + + def set_interface(self, interface: str): + """ + Setter for the filter by Interface. + """ + self.interface = interface + return self + + def __str__(self) -> str: + """ + String representation of the filter object. + """ + return f"OpenstackEndpointListObjectFilter(region={self.region}, " f"service_name={self.service_name}, service_type={self.service_type}, " f"interface={self.interface})" + + def __repr__(self) -> str: + """ + String representation of the filter object for debugging. + """ + return self.__str__() diff --git a/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_output.py b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_output.py index fecb7e6e..e3d23c28 100644 --- a/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_output.py +++ b/keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_output.py @@ -1,5 +1,9 @@ +import copy +from typing import List, Optional + from framework.exceptions.keyword_exception import KeywordException from keywords.cloud_platform.openstack.endpoint.objects.openstack_endpoint_list_object import OpenStackEndpointListObject +from keywords.cloud_platform.openstack.endpoint.objects.openstack_endpoint_list_object_filter import OpenStackEndpointListObjectFilter from keywords.cloud_platform.openstack.openstack_table_parser import OpenStackTableParser @@ -15,31 +19,61 @@ class OpenStackEndpointListOutput: for value in output_values: endpoint_object = OpenStackEndpointListObject() - endpoint_object.set_id(value['ID']) - endpoint_object.set_region(value['Region']) - endpoint_object.set_service_name(value['Service Name']) - endpoint_object.set_service_type(value['Service Type']) - endpoint_object.set_enabled(value['Enabled']) - endpoint_object.set_interface(value['Interface']) - endpoint_object.set_url(value['URL']) + endpoint_object.set_id(value["ID"]) + endpoint_object.set_region(value["Region"]) + endpoint_object.set_service_name(value["Service Name"]) + endpoint_object.set_service_type(value["Service Type"]) + endpoint_object.set_enabled(value["Enabled"]) + endpoint_object.set_interface(value["Interface"]) + endpoint_object.set_url(value["URL"]) self.openstack_endpoint_list_objects.append(endpoint_object) def get_endpoints(self) -> [OpenStackEndpointListObject]: - """ - Getter for endpoints + """Getter for endpoints""" + return self.get_endpoints_list_objects_filtered(None) + + def get_endpoints_list_objects_filtered(self, filter_object: Optional[OpenStackEndpointListObjectFilter]) -> Optional[List[OpenStackEndpointListObjectFilter]]: + """Gets the list of endpoints filtered by the filter object. + + Args: + filter_object (Optional[OpenStackEndpointListObjectFilter]): OpenStackEndpointListObjectFilter object + Returns: + Optional[List[OpenStackEndpointListObjectFilter]]: List of OpenStackEndpointListObject objects """ - return self.openstack_endpoint_list_objects + if filter_object is None: + return self.openstack_endpoint_list_objects + + os_ep_list_objects_copy = copy.deepcopy(self.openstack_endpoint_list_objects) + + # Filters the copied list of OpenStackEndpointListObjectFilter by Region. + if filter_object.get_region() is not None: + os_ep_list_objects_copy = [ep for ep in os_ep_list_objects_copy if ep.get_region() == filter_object.get_region()] + + # Filters the copied list of OpenStackEndpointListObjectFilter by Service Name. + if filter_object.get_service_name() is not None: + os_ep_list_objects_copy = [ep for ep in os_ep_list_objects_copy if ep.get_service_name() == filter_object.get_service_name()] + + # Filters the copied list of OpenStackEndpointListObjectFilter by Service Type. + if filter_object.get_service_type() is not None: + os_ep_list_objects_copy = [ep for ep in os_ep_list_objects_copy if ep.get_service_type() == filter_object.get_service_type()] + + # Filters the copied list of OpenStackEndpointListObjectFilter by Interface. + if filter_object.get_interface() is not None: + os_ep_list_objects_copy = [ep for ep in os_ep_list_objects_copy if ep.get_interface() == filter_object.get_interface()] + + return os_ep_list_objects_copy def is_endpoint(self, service_name: str) -> bool: - """ - This function will return true if the 'service_name' endpoint exists in the list. - Args: - service_name: Name of the service that we are looking for. + """This function will return true if the 'service_name' endpoint exists in the list. - Returns: Boolean + Args: + service_name (str): Name of the service that we are looking for. + + Returns: + bool: True if the endpoint exists, False otherwise. """ for endpoint in self.openstack_endpoint_list_objects: @@ -47,17 +81,17 @@ class OpenStackEndpointListOutput: return True return False - def get_endpoint(self, service_name: str, interface) -> OpenStackEndpointListObject: - """ - Gets the endpoint with the given service and interface + def get_endpoint(self, service_name: str, interface: str) -> OpenStackEndpointListObject: + """Gets the endpoint with the given service and interface + Args: - service_name (): service name - interface (): interface + service_name (str): service name + interface (str): interface Returns: + OpenStackEndpointListObject: The endpoint object with the given service name and interface. """ - endpoints = list(filter(lambda endpoint: endpoint.get_service_name() == service_name and endpoint.get_interface() == interface, self.openstack_endpoint_list_objects)) if len(endpoints) < 1: @@ -65,20 +99,18 @@ class OpenStackEndpointListOutput: return endpoints[0] - def get_horizon_url(self): - """ - Gets the Horizon Endpoint URL for this lab. - Args: - ssh_connection: Connection to the active controller. + def get_horizon_url(self) -> str: + """Gets the Horizon Endpoint URL for this lab. Returns: + str: Horizon URL for this lab. """ # Remove port from orignal url and then add 8443 for https or 8080 for http - url = self.get_endpoint('keystone', 'public').get_url().rsplit(':', 1)[0] - if 'https' in url: - url += ':8443/' + url = self.get_endpoint("keystone", "public").get_url().rsplit(":", 1)[0] + if "https" in url: + url += ":8443/" else: - url += ':8080/' + url += ":8080/" return url