From 280a291902f4eec78ed46bb7dce78edd2d863a89 Mon Sep 17 00:00:00 2001 From: Abhishek jaiswal Date: Wed, 23 Jul 2025 01:51:05 -0400 Subject: [PATCH] Add OpenStack endpoint filtering capability This change improves the OpenStack endpoint handling by: 1. Refactoring the OpenStackEndpointListObject class to have cleaner docstrings and proper type annotations 2. Adding a new OpenStackEndpointListObjectFilter class to enable filtering of endpoints by region, service name, service type, and interface 3. Enhancing the OpenStackEndpointListOutput class with filtering functionality and improved type annotations The refactoring improves code readability and maintainability while the new filtering capability provides a more flexible way to query endpoints based on different criteria. Change-Id: Ibf7e3b9111b5fca03c210f545013b3b702133eb3 Signed-off-by: Abhishek jaiswal --- .../objects/openstack_endpoint_list_object.py | 101 ++++++------------ .../openstack_endpoint_list_object_filter.py | 77 +++++++++++++ .../objects/openstack_endpoint_list_output.py | 92 ++++++++++------ 3 files changed, 172 insertions(+), 98 deletions(-) create mode 100644 keywords/cloud_platform/openstack/endpoint/objects/openstack_endpoint_list_object_filter.py 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