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 <abhishek.jaiswal@windriver.com>
This commit is contained in:
Abhishek jaiswal
@@ -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
|
||||
|
||||
@@ -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__()
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user