From 65d935a2b812b21f2b3c94bb56ade88963628fd6 Mon Sep 17 00:00:00 2001 From: aabhinav Date: Mon, 14 Jul 2025 06:29:22 -0400 Subject: [PATCH] pylint fix 01 Description: - Added a missing import for SSHConnection. - Added data types to function arguments and return values. - Removed Returns section in docstring when function return type is None Change-Id: Id96217573f629d40a52459a3b45d26ae0649aa03 Signed-off-by: aabhinav --- keywords/linux/ip/object/interface_object.py | 12 +- keywords/linux/ip/object/ip_br_addr_output.py | 18 +- .../linux/ip/object/ip_link_show_output.py | 14 +- keywords/linux/ip/object/ip_object.py | 48 ++-- keywords/linux/keyring/keyring_keywords.py | 5 +- .../process_status_args_keywords.py | 15 +- .../systemctl/systemctl_is_active_keywords.py | 17 +- .../object/openstack_stack_list_output.py | 66 +++--- .../stack/object/openstack_stack_output.py | 48 ++-- .../stack/openstack_stack_create_keywords.py | 57 ++--- .../stack/openstack_stack_delete_keywords.py | 28 ++- .../stack/openstack_stack_list_keywords.py | 28 +-- .../stack/openstack_stack_update_keywords.py | 51 ++--- keywords/ptp/cat/cat_ptp_cgu_keywords.py | 9 +- keywords/ptp/cat/cat_ptp_config_keywords.py | 22 +- .../cat/objects/clock_description_object.py | 66 +++--- .../cat/objects/clock_description_output.py | 30 ++- .../default_interface_options_object.py | 122 +++++----- .../default_interface_options_output.py | 50 ++--- .../ptp/cat/objects/port_data_set_object.py | 202 ++++++++--------- .../ptp/cat/objects/port_data_set_output.py | 78 ++++--- .../cat/objects/ptp_cgu_component_output.py | 3 +- .../cat/objects/run_time_options_object.py | 195 +++++++--------- .../cat/objects/run_time_options_output.py | 78 ++++--- .../ptp/cat/objects/servo_options_object.py | 211 ++++++++---------- .../ptp/cat/objects/servo_options_output.py | 82 ++++--- .../cat/objects/transport_options_object.py | 89 ++++---- .../cat/objects/transport_options_output.py | 38 ++-- keywords/ptp/phc_ctl_keywords.py | 21 +- 29 files changed, 793 insertions(+), 910 deletions(-) diff --git a/keywords/linux/ip/object/interface_object.py b/keywords/linux/ip/object/interface_object.py index 5c6d25fc..88ffa176 100644 --- a/keywords/linux/ip/object/interface_object.py +++ b/keywords/linux/ip/object/interface_object.py @@ -12,31 +12,35 @@ class InterfaceObject: def get_interface_name(self) -> str: """ Getter for interface name - Returns: + Returns: + str: the interface name value """ return self.interface_name def get_mtu(self) -> str: """ Getter for mtu name - Returns: + Returns: + str: the mtu name value """ return self.mtu def get_state(self) -> str: """ Getter for state - Returns: + Returns: + str: the state value """ return self.state def get_mode(self) -> str: """ Getter for mode - Returns: + Returns: + str: the mode value """ return self.mode diff --git a/keywords/linux/ip/object/ip_br_addr_output.py b/keywords/linux/ip/object/ip_br_addr_output.py index 63c27d12..3b40852a 100644 --- a/keywords/linux/ip/object/ip_br_addr_output.py +++ b/keywords/linux/ip/object/ip_br_addr_output.py @@ -11,21 +11,22 @@ class IPBrAddrOutput: def __init__(self, ip_br_addr_output: str): """ Constructor + Args: - ip_br_addr_output: a string representing the output of the command 'ip -br addr'. + ip_br_addr_output (str): a string representing the output of the command 'ip -br addr'. """ - self.ip_br_addr_objects: [IPBrAddrObject] = [] + self.ip_br_addr_objects: list[IPBrAddrObject] = [] ip_br_addr_parser = IPBrAddrParser(ip_br_addr_output) output_values = ip_br_addr_parser.get_output_values_list() for value in output_values: ip_br_addr_object = IPBrAddrObject() - ip_br_addr_object.set_network_interface_name(value.get('network_interface_name')) - ip_br_addr_object.set_network_interface_status(value.get('network_interface_status')) + ip_br_addr_object.set_network_interface_name(value.get("network_interface_name")) + ip_br_addr_object.set_network_interface_status(value.get("network_interface_status")) # The column 'ip_addresses' of the output of the command 'ip -br addr' can have zero or many IP addresses. - if value.get('ip_addresses') is not None: - for ip in value['ip_addresses']: + if value.get("ip_addresses") is not None: + for ip in value["ip_addresses"]: ip_object = IPObject(ip) ip_br_addr_object.get_ip_objects().append(ip_object) @@ -33,10 +34,7 @@ class IPBrAddrOutput: def get_ip_br_addr_objects(self) -> list[IPBrAddrObject]: """ - Getter for the list of instances of IPBrAddrObject. Each item of this list is an object the represents a row in - the table shown by the execution of the command 'ip -br addr'. - - Args: None. + Getter for the list of instances of IPBrAddrObject. Each item of this list is an object the represents a row in the table shown by the execution of the command 'ip -br addr'. Returns: list[IPBrAddrObject]: list of instances of IPBrAddrObject where each item represents a row in diff --git a/keywords/linux/ip/object/ip_link_show_output.py b/keywords/linux/ip/object/ip_link_show_output.py index ab590078..23d4cf60 100644 --- a/keywords/linux/ip/object/ip_link_show_output.py +++ b/keywords/linux/ip/object/ip_link_show_output.py @@ -15,22 +15,22 @@ class IPLinkShowOutput: def get_interface(self) -> InterfaceObject: """ Getter for interface + Returns: - + InterfaceObject: Parsed interface data. """ - # Regex is designed for an output of the shape: # 2: eth0: mtu 1500 qdisc fq_codel state UP mode DEFAULT group default qlen 1000 # This regex will extract the interface name, mode, mtu and state - regex = r'\d+:\s+(?P\S+):\s+<(?P\S+)>' r'\s+mtu\s+(?P\d+)[\s\S]+state\s+(?P\w+)' + regex = r"\d+:\s+(?P\S+):\s+<(?P\S+)>" r"\s+mtu\s+(?P\d+)[\s\S]+state\s+(?P\w+)" match = re.match(regex, self.ip_link_show_output[0]) if match: - interface_name = match.group('interface_name') - modes = match.group('mode') - mtu = match.group('mtu') - state = match.group('state') + interface_name = match.group("interface_name") + modes = match.group("mode") + mtu = match.group("mtu") + state = match.group("state") modes, self.interface_object = InterfaceObject(interface_name, mtu, state, modes) diff --git a/keywords/linux/ip/object/ip_object.py b/keywords/linux/ip/object/ip_object.py index 39a9ff73..1adad1f5 100644 --- a/keywords/linux/ip/object/ip_object.py +++ b/keywords/linux/ip/object/ip_object.py @@ -7,13 +7,13 @@ class IPObject: Class that represents an IP address, including the IP itself and its prefix length (subnet network). """ - def __init__(self, ip_with_prefix): - """ - Constructor. - Args: - ip_with_prefix: An IP address, either version 4 or 6, followed or not by /. + def __init__(self, ip_with_prefix: str): """ + Constructor + Args: + ip_with_prefix (str): An IP address, either version 4 or 6, followed or not by /. + """ # Gets the IP from an IP with prefix length. Example: gets '192.168.1.1' from '192.168.1.1/24'. self.ip_address = IPObject._extract_ip_address(ip_with_prefix) @@ -21,22 +21,23 @@ class IPObject: self.prefix_length = IPObject._extract_prefix_length(ip_with_prefix) if not ipaddress.ip_address(self.ip_address): - raise ValueError(f'The argument {ip_with_prefix} must be a valid IPv4 or IPv6 address.') + raise ValueError(f"The argument {ip_with_prefix} must be a valid IPv4 or IPv6 address.") - def set_ip_address(self, ip_address: str): + def set_ip_address(self, ip_address: str) -> None: """ - Setter for ip_address property. + Setter for ip_address property + Args: - ip_address (str): a string representing an IPv4 or IPv6 address either with prefix length or without prefix length. - The prefix length will be discarded if present. + ip_address (str): a string representing an IPv4 or IPv6 address either with prefix length or without prefix length, the prefix length will be discarded if present. """ if ipaddress.ip_address(ip_address): - raise ValueError(f'The argument {ip_address} must be a valid IPv4 or IPv6 address.') + raise ValueError(f"The argument {ip_address} must be a valid IPv4 or IPv6 address.") self.ip_address = IPObject._extract_ip_address(ip_address) def get_ip_address(self) -> str: """ - Getter for ip_address property. + Getter for ip_address property + Returns: str: ip_address """ @@ -44,15 +45,19 @@ class IPObject: def set_prefix_length(self, prefix_length: int): """ - Setter for prefix_length property. + Setter for prefix_length property + Args: prefix_length (int): an prefix length number. """ self.prefix_length = prefix_length - def get_prefix_length(self): + def get_prefix_length(self) -> int: """ - Getter for prefix length property. + Getter for prefix length property + + Returns: + int: prefix length number """ return self.prefix_length @@ -60,29 +65,32 @@ class IPObject: def _extract_ip_address(ip_address_with_prefix: str) -> str: """ Extracts the IP address from an IP with prefix length. + Example: gets '192.168.1.1' from '192.168.1.1/24'. + Args: - ip_address_with_prefix: an IP address with prefix length. + ip_address_with_prefix (str): an IP address with prefix length. Returns: str: an ip_address without prefix length. - """ - return re.sub(r'/\d+', '', ip_address_with_prefix) + return re.sub(r"/\d+", "", ip_address_with_prefix) @staticmethod def _extract_prefix_length(ip_address_with_prefix: str) -> int: """ Extracts the prefix length from an IP with prefix length. + Example: gets 24 from '192.168.1.1/24'. + Args: - ip_address_with_prefix: + ip_address_with_prefix (str): an IP address with prefix length. Returns: int: the prefix_length of the IP passed as argument. """ prefix_length = 0 - match = re.search(r'/(\d+)', ip_address_with_prefix) + match = re.search(r"/(\d+)", ip_address_with_prefix) if match: prefix_length = int(match.group(1)) return prefix_length diff --git a/keywords/linux/keyring/keyring_keywords.py b/keywords/linux/keyring/keyring_keywords.py index 00a67d70..835bdcd9 100644 --- a/keywords/linux/keyring/keyring_keywords.py +++ b/keywords/linux/keyring/keyring_keywords.py @@ -17,9 +17,10 @@ class KeyringKeywords(BaseKeyword): Args: service (str): keyring service identifier (str): keyring identifier + Returns: - The value from the keyring. + str: The value from the keyring. """ keyring_value = self.ssh_connection.send(f"keyring get {service} {identifier}") self.validate_success_return_code(self.ssh_connection) - return keyring_value[0].strip() \ No newline at end of file + return keyring_value[0].strip() diff --git a/keywords/linux/process_status/process_status_args_keywords.py b/keywords/linux/process_status/process_status_args_keywords.py index 9619a677..54122900 100644 --- a/keywords/linux/process_status/process_status_args_keywords.py +++ b/keywords/linux/process_status/process_status_args_keywords.py @@ -1,3 +1,4 @@ +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword @@ -6,25 +7,27 @@ class ProcessStatusArgsKeywords(BaseKeyword): Class for "ps -o args" keywords """ - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ Constructor + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection - def get_process_arguments_as_string(self, process_name) -> str: + def get_process_arguments_as_string(self, process_name: str) -> str: """ This function will return the command line arguments associated with the specified process. Args: - process_name: The name of the process for which we want to get the command line arguments. + process_name (str): The name of the process for which we want to get the command line arguments. - Returns: A string containing the process name with all command line arguments. + Returns: + str: A string containing the process name with all command line arguments. """ - output = self.ssh_connection.send(f'ps -C {process_name} -o args= | cat') + output = self.ssh_connection.send(f"ps -C {process_name} -o args= | cat") self.validate_success_return_code(self.ssh_connection) # output is a list with one value diff --git a/keywords/linux/systemctl/systemctl_is_active_keywords.py b/keywords/linux/systemctl/systemctl_is_active_keywords.py index ba7e11f0..c77bd9d2 100644 --- a/keywords/linux/systemctl/systemctl_is_active_keywords.py +++ b/keywords/linux/systemctl/systemctl_is_active_keywords.py @@ -1,3 +1,4 @@ +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword @@ -6,24 +7,26 @@ class SystemCTLIsActiveKeywords(BaseKeyword): Class for "systemctl is-active" keywords """ - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ Constructor + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection - def is_active(self, service_name) -> str: + def is_active(self, service_name: str) -> str: """ Checks if the service is active using "systemctl is-active " + Args: - service_name: The name of the service - - Returns: 'active' or 'inactive' + service_name (str): The name of the service + Returns: + str: 'active' or 'inactive' """ - output = self.ssh_connection.send(f'systemctl is-active {service_name}') + output = self.ssh_connection.send(f"systemctl is-active {service_name}") self.validate_success_return_code(self.ssh_connection) # output is a List of 1 string. "active/n" diff --git a/keywords/openstack/openstack/stack/object/openstack_stack_list_output.py b/keywords/openstack/openstack/stack/object/openstack_stack_list_output.py index 039560f9..59364e6a 100644 --- a/keywords/openstack/openstack/stack/object/openstack_stack_list_output.py +++ b/keywords/openstack/openstack/stack/object/openstack_stack_list_output.py @@ -1,12 +1,15 @@ +from typing import Any, Dict + from framework.exceptions.keyword_exception import KeywordException from framework.logging.automation_logger import get_logger -from keywords.openstack.openstack.stack.object.openstack_stack_list_object import OpenstackStackListObject from keywords.openstack.openstack.openstack_json_parser import OpenstackJsonParser +from keywords.openstack.openstack.stack.object.openstack_stack_list_object import OpenstackStackListObject class OpenstackStackListOutput: """ This class parses the output of the command 'openstack stack list' + The parsing result is a 'OpenstackStackListObject' instance. Example: @@ -19,13 +22,14 @@ class OpenstackStackListOutput: """ - def __init__(self, openstack_stack_list_output): + def __init__(self, openstack_stack_list_output: list[str]): """ - Constructor + Constructor for OpenstackStackListOutput class. + Args: - openstack_stack_list_output: the output of the command 'openstack stack list'. + openstack_stack_list_output (list[str]): the output of the command 'openstack stack list'. """ - self.openstack_stacks: [OpenstackStackListObject] = [] + self.openstack_stacks: list[OpenstackStackListObject] = [] openstack_json_parser = OpenstackJsonParser(openstack_stack_list_output) output_values = openstack_json_parser.get_output_values_list() @@ -33,33 +37,38 @@ class OpenstackStackListOutput: if self.is_valid_output(value): self.openstack_stacks.append( OpenstackStackListObject( - value['ID'], - value['Stack Name'], - value['Project'], - value['Stack Status'], - value['Creation Time'], - value['Updated Time'], + value["ID"], + value["Stack Name"], + value["Project"], + value["Stack Status"], + value["Creation Time"], + value["Updated Time"], ) ) else: raise KeywordException(f"The output line {value} was not valid") - def get_stacks(self) -> [OpenstackStackListObject]: + def get_stacks(self) -> list[OpenstackStackListObject]: """ Returns the list of stack objects - Returns: + Returns: + list[OpenstackStackListObject]: a list of OpenstackStackListObject instances representing the stacks """ return self.openstack_stacks def get_stack(self, stack_name: str) -> OpenstackStackListObject: """ Gets the given stack + Args: - stack_name (): the name of the stack + stack_name (str): the name of the stack - Returns: the stack object + Returns: + OpenstackStackListObject: the OpenstackStackListObject instance representing the stack with the name 'stack_name' + Raises: + KeywordException: if no stack with the name 'stack_name' was found. """ stacks = list(filter(lambda stack: stack.get_stack_name() == stack_name, self.openstack_stacks)) if len(stacks) == 0: @@ -68,30 +77,31 @@ class OpenstackStackListOutput: return stacks[0] @staticmethod - def is_valid_output(value): + def is_valid_output(value: Dict[str, Any]) -> bool: """ Checks to ensure the output has the correct keys + Args: - value (): the value to check + value (Dict[str, Any]): a dictionary representing a single row of the output Returns: - + bool: True if the output has the required keys; False otherwise """ valid = True - if 'ID' not in value: - get_logger().log_error(f'id is not in the output value: {value}') + if "ID" not in value: + get_logger().log_error(f"id is not in the output value: {value}") valid = False - if 'Stack Name' not in value: - get_logger().log_error(f'stack name is not in the output value: {value}') + if "Stack Name" not in value: + get_logger().log_error(f"stack name is not in the output value: {value}") valid = False - if 'Project' not in value: - get_logger().log_error(f'project is not in the output value: {value}') + if "Project" not in value: + get_logger().log_error(f"project is not in the output value: {value}") valid = False - if 'Stack Status' not in value: - get_logger().log_error(f'stack status is not in the output value: {value}') + if "Stack Status" not in value: + get_logger().log_error(f"stack status is not in the output value: {value}") valid = False - if 'Creation Time' not in value: - get_logger().log_error(f'creation time is not in the output value: {value}') + if "Creation Time" not in value: + get_logger().log_error(f"creation time is not in the output value: {value}") valid = False return valid diff --git a/keywords/openstack/openstack/stack/object/openstack_stack_output.py b/keywords/openstack/openstack/stack/object/openstack_stack_output.py index 7c89378f..e5c8308e 100644 --- a/keywords/openstack/openstack/stack/object/openstack_stack_output.py +++ b/keywords/openstack/openstack/stack/object/openstack_stack_output.py @@ -1,10 +1,11 @@ -from keywords.openstack.openstack.stack.object.openstack_stack_object import OpenstackStackObject from keywords.openstack.openstack.openstack_json_parser import OpenstackJsonParser +from keywords.openstack.openstack.stack.object.openstack_stack_object import OpenstackStackObject + class OpenstackStackOutput: """ - This class parses the output of commands such as 'openstack stack create' - that share the same output as shown in the example below. + This class parses the output of commands such as 'openstack stack create' that share the same output as shown in the example below. + The parsing result is a 'openstackStackObject' instance. Example: @@ -20,47 +21,44 @@ class OpenstackStackOutput: } """ - def __init__(self, openstack_stack_output): + def __init__(self, openstack_stack_output: list[str]): """ - Constructor. - Create an internal OpenstackStackCreateObject from the passed parameter. - Args: - openstack_stack_output (list[str]): a list of strings representing the output of the - 'openstack stack create' command. + Create an internal OpenstackStackCreateObject from the passed parameter. + Args: + openstack_stack_output (list[str]): a list of strings representing the output of the 'openstack stack create' command. """ openstack_json_parser = OpenstackJsonParser(openstack_stack_output) output_values = openstack_json_parser.get_output_values_list() self.openstack_stack_object = OpenstackStackObject() - if 'id' in output_values: - self.openstack_stack_object.set_id(output_values['id']) + if "id" in output_values: + self.openstack_stack_object.set_id(output_values["id"]) - if 'stack_name' in output_values: - self.openstack_stack_object.set_stack_name(output_values['stack_name']) + if "stack_name" in output_values: + self.openstack_stack_object.set_stack_name(output_values["stack_name"]) - if 'description' in output_values: - self.openstack_stack_object.set_description(output_values['description']) + if "description" in output_values: + self.openstack_stack_object.set_description(output_values["description"]) - if 'creation_time' in output_values: - self.openstack_stack_object.set_creation_time(output_values['creation_time']) + if "creation_time" in output_values: + self.openstack_stack_object.set_creation_time(output_values["creation_time"]) - if 'updated_time' in output_values: - self.openstack_stack_object.set_updated_time(output_values['updated_time']) + if "updated_time" in output_values: + self.openstack_stack_object.set_updated_time(output_values["updated_time"]) - if 'stack_status' in output_values: - self.openstack_stack_object.set_stack_status(output_values['stack_status']) + if "stack_status" in output_values: + self.openstack_stack_object.set_stack_status(output_values["stack_status"]) - if 'stack_status_reason' in output_values: - self.openstack_stack_object.set_stack_status_reason(output_values['stack_status_reason']) + if "stack_status_reason" in output_values: + self.openstack_stack_object.set_stack_status_reason(output_values["stack_status_reason"]) def get_openstack_stack_object(self) -> OpenstackStackObject: """ Getter for OpenstackStackObject object. Returns: - A OpenstackStackObject instance representing the output of commands sucha as 'openstack stack create'. - + OpenstackStackObject: the openstack_stack_object """ return self.openstack_stack_object diff --git a/keywords/openstack/openstack/stack/openstack_stack_create_keywords.py b/keywords/openstack/openstack/stack/openstack_stack_create_keywords.py index 2ddff121..e0e3611d 100644 --- a/keywords/openstack/openstack/stack/openstack_stack_create_keywords.py +++ b/keywords/openstack/openstack/stack/openstack_stack_create_keywords.py @@ -1,76 +1,69 @@ from framework.logging.automation_logger import get_logger +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword from keywords.openstack.command_wrappers import source_admin_openrc from keywords.openstack.openstack.stack.object.openstack_stack_create_input import OpenstackStackCreateInput -from keywords.openstack.openstack.stack.openstack_stack_list_keywords import OpenstackStackListKeywords from keywords.openstack.openstack.stack.object.openstack_stack_output import OpenstackStackOutput from keywords.openstack.openstack.stack.object.openstack_stack_status_enum import OpenstackStackStatusEnum +from keywords.openstack.openstack.stack.openstack_stack_list_keywords import OpenstackStackListKeywords from keywords.python.string import String class OpenstackStackCreateKeywords(BaseKeyword): - """ - Class for Openstack Stack Create - """ + """Class for Openstack Stack Create Keywords""" - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ - Constructor + Constructor for OpenstackStackCreateKeywords class. + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection def openstack_stack_create(self, openstack_stack_create_input: OpenstackStackCreateInput) -> OpenstackStackOutput: """ Openstack stack create function, creates a given stack + Args: openstack_stack_create_input (OpenstackStackCreateInput): an object with the required parameters Returns: - openstack_stack_output: An object with the openstack stack create output - + OpenstackStackOutput: the output of the openstack stack create command """ # Gets the command 'openstack stack create' with its parameters configured. cmd = self.get_command(openstack_stack_create_input) stack_name = openstack_stack_create_input.get_stack_name() # Executes the command 'openstack stack create'. - output = self.ssh_connection.send(source_admin_openrc(cmd),get_pty=True) + output = self.ssh_connection.send(source_admin_openrc(cmd), get_pty=True) self.validate_success_return_code(self.ssh_connection) openstack_stack_output = OpenstackStackOutput(output) # Tracks the execution of the command 'openstack stack create' until its completion or a timeout. openstack_stack_list_keywords = OpenstackStackListKeywords(self.ssh_connection) - openstack_stack_list_keywords.validate_stack_status(stack_name, 'CREATE_COMPLETE') + openstack_stack_list_keywords.validate_stack_status(stack_name, "CREATE_COMPLETE") # If the execution arrived here the status of the stack is 'created'. - openstack_stack_output.get_openstack_stack_object().set_stack_status('create_complete') + openstack_stack_output.get_openstack_stack_object().set_stack_status("create_complete") return openstack_stack_output def is_already_created(self, stack_name: str) -> bool: """ Verifies if the stack has already been created. + Args: stack_name (str): a string representing the name of the stack. Returns: bool: True if the stack named 'stack_name' has already been created; False otherwise. - """ try: openstack_stack_list_keywords = OpenstackStackListKeywords(self.ssh_connection) if openstack_stack_list_keywords.get_openstack_stack_list().is_in_stack_list(stack_name): stack = OpenstackStackListKeywords(self.ssh_connection).get_openstack_stack_list().get_stack(stack_name) - return ( - stack.get_stack_status() == OpenstackStackStatusEnum.CREATE_IN_PROGRESS.value - or stack.get_stack_status() == OpenstackStackStatusEnum.CREATE_COMPLETE.value - or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_IN_PROGRESS.value - or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_COMPLETE.value - or stack.get_stack_status() == OpenstackStackStatusEnum.ROLLBACK_IN_PROGRESS.value - or stack.get_stack_status() == OpenstackStackStatusEnum.ROLLBACK_COMPLETE.value - ) + return stack.get_stack_status() == OpenstackStackStatusEnum.CREATE_IN_PROGRESS.value or stack.get_stack_status() == OpenstackStackStatusEnum.CREATE_COMPLETE.value or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_IN_PROGRESS.value or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_COMPLETE.value or stack.get_stack_status() == OpenstackStackStatusEnum.ROLLBACK_IN_PROGRESS.value or stack.get_stack_status() == OpenstackStackStatusEnum.ROLLBACK_COMPLETE.value return False except Exception as ex: get_logger().log_exception(f"An error occurred while verifying whether the application named {stack_name} is already created.") @@ -78,18 +71,14 @@ class OpenstackStackCreateKeywords(BaseKeyword): def get_command(self, openstack_stack_create_input: OpenstackStackCreateInput) -> str: """ - Generates a string representing the 'openstack stack create' command with parameters based on the values in - the 'openstack_stack_create_input' argument. + Generates the 'openstack stack create' command using values from the given input object. + Args: - openstack_stack_create_input (OpenstackStackCreateInput): an instance of OpenstackStackCreateInput - configured with the parameters needed to execute the 'openstack stack create' command properly. + openstack_stack_create_input (OpenstackStackCreateInput): Input parameters for the create command Returns: - str: a string representing the 'openstack stack create' command, configured according to the parameters - in the 'openstack_stack_create_input' argument. - + str: a string representing the 'openstack stack create' command, configured according to the parameters in the 'openstack_stack_create_input' argument. """ - # 'template_file_path' and 'template_file_name' properties are required template_file_path = openstack_stack_create_input.get_template_file_path() template_file_name = openstack_stack_create_input.get_template_file_name() @@ -97,9 +86,7 @@ class OpenstackStackCreateKeywords(BaseKeyword): error_message = "Template path and name must be specified" get_logger().log_exception(error_message) raise ValueError(error_message) - template_file_path_as_param = ( - f'--template={openstack_stack_create_input.get_template_file_path()}/{openstack_stack_create_input.get_template_file_name()}' - ) + template_file_path_as_param = f"--template={openstack_stack_create_input.get_template_file_path()}/{openstack_stack_create_input.get_template_file_name()}" # 'stack_name' is required stack_name = openstack_stack_create_input.get_stack_name() @@ -107,11 +94,11 @@ class OpenstackStackCreateKeywords(BaseKeyword): error_message = "Stack name is required" get_logger().log_exception(error_message) raise ValueError(error_message) - + # 'return_format' property is optional. - return_format_as_param = f'-f {openstack_stack_create_input.get_return_format()}' + return_format_as_param = f"-f {openstack_stack_create_input.get_return_format()}" # Assembles the command. - cmd = f'openstack stack create {return_format_as_param} {template_file_path_as_param} {stack_name}' + cmd = f"openstack stack create {return_format_as_param} {template_file_path_as_param} {stack_name}" return cmd diff --git a/keywords/openstack/openstack/stack/openstack_stack_delete_keywords.py b/keywords/openstack/openstack/stack/openstack_stack_delete_keywords.py index ed5c248b..89bc7a22 100644 --- a/keywords/openstack/openstack/stack/openstack_stack_delete_keywords.py +++ b/keywords/openstack/openstack/stack/openstack_stack_delete_keywords.py @@ -1,24 +1,25 @@ +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword from keywords.openstack.command_wrappers import source_admin_openrc from keywords.openstack.openstack.stack.object.openstack_stack_delete_input import OpenstackStackDeleteInput class OpenstackStackDeleteKeywords(BaseKeyword): - """ - Class for Openstack stack delete keywords - """ + """Class for Openstack stack delete keywords""" - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ - Constructor + Constructor for OpenstackStackDeleteKeywords class + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection def get_openstack_stack_delete(self, openstack_stack_delete_input: OpenstackStackDeleteInput) -> str: """ Delete a stack specified in the parameter 'openstack_stack_delete_input'. + Args: openstack_stack_delete_input (OpenstackStackDeleteInput): defines the stack name @@ -28,23 +29,20 @@ class OpenstackStackDeleteKeywords(BaseKeyword): 'Stack-delete rejected: stack not found.\n' """ cmd = self.get_command(openstack_stack_delete_input) - output = self.ssh_connection.send(source_admin_openrc(cmd),get_pty=True) + output = self.ssh_connection.send(source_admin_openrc(cmd), get_pty=True) self.validate_success_return_code(self.ssh_connection) return output[0] def get_command(self, openstack_stack_delete_input: OpenstackStackDeleteInput) -> str: """ - Generates a string representing the 'openstack stack delete' command with parameters based on the values in - the 'openstack_stack_delete_input' argument. + Generates the 'openstack stack delete' command using values from the given input object. + Args: - openstack_stack_delete_input (OpenstackStackDeleteInput): an instance of OpenstackStackDeleteInput - configured with the parameters needed to execute the 'openstack stack delete' command properly. + openstack_stack_delete_input (OpenstackStackDeleteInput): Input parameters for the delete command Returns: - str: a string representing the 'openstack stack delete' command, configured according to the parameters - in the 'openstack_stack_delete_input' argument. - + str: a string representing the 'openstack stack delete' command, configured according to the parameters in the 'openstack_stack_delete_input' argument. """ - cmd = f'openstack stack delete -y {openstack_stack_delete_input.get_stack_name()}' + cmd = f"openstack stack delete -y {openstack_stack_delete_input.get_stack_name()}" return cmd diff --git a/keywords/openstack/openstack/stack/openstack_stack_list_keywords.py b/keywords/openstack/openstack/stack/openstack_stack_list_keywords.py index 101388ca..2da0af6a 100644 --- a/keywords/openstack/openstack/stack/openstack_stack_list_keywords.py +++ b/keywords/openstack/openstack/stack/openstack_stack_list_keywords.py @@ -1,4 +1,4 @@ -from framework.logging.automation_logger import get_logger +from framework.ssh.ssh_connection import SSHConnection from framework.validation.validation import validate_equals_with_retry from keywords.base_keyword import BaseKeyword from keywords.openstack.command_wrappers import source_admin_openrc @@ -6,15 +6,14 @@ from keywords.openstack.openstack.stack.object.openstack_stack_list_output impor class OpenstackStackListKeywords(BaseKeyword): - """ - Class for Openstack stack list keywords. - """ + """Class for Openstack stack list keywords""" - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ - Constructor + Constructor for OpenstackStackListKeywords class. + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection @@ -22,27 +21,23 @@ class OpenstackStackListKeywords(BaseKeyword): """ Gets a OpenstackStackListOutput object related to the execution of the 'openstack stack list' command. - Args: None - Returns: OpenstackStackListOutput: an instance of the OpenstackStackListOutput object representing the heat stacks on the host, as a result of the execution of the 'openstack stack list' command. """ - output = self.ssh_connection.send(source_admin_openrc('openstack stack list -f json'),get_pty=True) + output = self.ssh_connection.send(source_admin_openrc("openstack stack list -f json"), get_pty=True) self.validate_success_return_code(self.ssh_connection) openstack_stack_list_output = OpenstackStackListOutput(output) return openstack_stack_list_output - def validate_stack_status(self, stack_name: str, status: str): + def validate_stack_status(self, stack_name: str, status: str) -> None: """ This function will validate that the stack specified reaches the desired status. + Args: - stack_name: Name of the stack that we are waiting for. - status: Status in which we want to wait for the stack to reach. - - Returns: None - + stack_name (str): Name of the stack that we are waiting for. + status (str): Status in which we want to wait for the stack to reach. """ def get_stack_status(): @@ -52,4 +47,3 @@ class OpenstackStackListKeywords(BaseKeyword): message = f"Openstack stack {stack_name}'s status is {status}" validate_equals_with_retry(get_stack_status, status, message, timeout=300) - diff --git a/keywords/openstack/openstack/stack/openstack_stack_update_keywords.py b/keywords/openstack/openstack/stack/openstack_stack_update_keywords.py index cc7a961f..4d1ae6a2 100644 --- a/keywords/openstack/openstack/stack/openstack_stack_update_keywords.py +++ b/keywords/openstack/openstack/stack/openstack_stack_update_keywords.py @@ -1,35 +1,35 @@ from framework.logging.automation_logger import get_logger +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword from keywords.openstack.command_wrappers import source_admin_openrc -from keywords.openstack.openstack.stack.object.openstack_stack_update_input import OpenstackStackUpdateInput -from keywords.openstack.openstack.stack.openstack_stack_list_keywords import OpenstackStackListKeywords from keywords.openstack.openstack.stack.object.openstack_stack_output import OpenstackStackOutput from keywords.openstack.openstack.stack.object.openstack_stack_status_enum import OpenstackStackStatusEnum +from keywords.openstack.openstack.stack.object.openstack_stack_update_input import OpenstackStackUpdateInput +from keywords.openstack.openstack.stack.openstack_stack_list_keywords import OpenstackStackListKeywords from keywords.python.string import String class OpenstackStackUpdateKeywords(BaseKeyword): - """ - Class for Openstack Stack Update - """ + """Class for Openstack Stack Update Keywords""" - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ - Constructor + Constructor for OpenstackStackUpdateKeywords class. + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection to the active controller """ self.ssh_connection = ssh_connection def openstack_stack_update(self, openstack_stack_update_input: OpenstackStackUpdateInput) -> OpenstackStackOutput: """ Openstack stack update function, updates a given existing stack + Args: openstack_stack_update_input (OpenstackStackUpdateInput): an object with the required parameters Returns: - openstack_stack_output: An object with the openstack stack update output - + OpenstackStackOutput: the output of the openstack stack update command """ # Gets the command 'openstack stack update' with its parameters configured. cmd = self.get_command(openstack_stack_update_input) @@ -42,46 +42,39 @@ class OpenstackStackUpdateKeywords(BaseKeyword): # Tracks the execution of the command 'openstack stack update' until its completion or a timeout. openstack_stack_list_keywords = OpenstackStackListKeywords(self.ssh_connection) - openstack_stack_list_keywords.validate_stack_status(stack_name, 'UPDATE_COMPLETE') + openstack_stack_list_keywords.validate_stack_status(stack_name, "UPDATE_COMPLETE") # If the execution arrived here the status of the stack is 'updated'. - openstack_stack_output.get_openstack_stack_object().set_stack_status('update_complete') + openstack_stack_output.get_openstack_stack_object().set_stack_status("update_complete") return openstack_stack_output def is_already_updated(self, stack_name: str) -> bool: """ Verifies if the stack has already been updated. + Args: stack_name (str): a string representing the name of the stack. Returns: bool: True if the stack named 'stack_name' has already been updated; False otherwise. - """ openstack_stack_list_keywords = OpenstackStackListKeywords(self.ssh_connection) if openstack_stack_list_keywords.get_openstack_stack_list().is_in_stack_list(stack_name): stack = OpenstackStackListKeywords(self.ssh_connection).get_openstack_stack_list().get_stack(stack_name) - return ( - stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_IN_PROGRESS.value - or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_COMPLETE.value - ) + return stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_IN_PROGRESS.value or stack.get_stack_status() == OpenstackStackStatusEnum.UPDATE_COMPLETE.value return False def get_command(self, openstack_stack_update_input: OpenstackStackUpdateInput) -> str: """ - Generates a string representing the 'openstack stack update' command with parameters based on the values in - the 'openstack_stack_update_input' argument. + Generates the 'openstack stack update' command using values from the given input object. + Args: - openstack_stack_update_input (OpenstackStackUpdateInput): an instance of OpenstackStackUpdateInput - configured with the parameters needed to execute the 'openstack stack update' command properly. + openstack_stack_update_input (OpenstackStackUpdateInput): Input parameters for the update command Returns: - str: a string representing the 'openstack stack update' command, configured according to the parameters - in the 'openstack_stack_update_input' argument. - + str: a string representing the 'openstack stack update' command, configured according to the parameters in the 'openstack_stack_update_input' argument. """ - # 'template_file_path' and 'template_file_name' properties are required template_file_path = openstack_stack_update_input.get_template_file_path() template_file_name = openstack_stack_update_input.get_template_file_name() @@ -89,9 +82,7 @@ class OpenstackStackUpdateKeywords(BaseKeyword): error_message = "Template path and name must be specified" get_logger().log_exception(error_message) raise ValueError(error_message) - template_file_path_as_param = ( - f'--template={openstack_stack_update_input.get_template_file_path()}/{openstack_stack_update_input.get_template_file_name()}' - ) + template_file_path_as_param = f"--template={openstack_stack_update_input.get_template_file_path()}/{openstack_stack_update_input.get_template_file_name()}" # 'stack_name' is required stack_name = openstack_stack_update_input.get_stack_name() @@ -101,9 +92,9 @@ class OpenstackStackUpdateKeywords(BaseKeyword): raise ValueError(error_message) # 'return_format' property is optional. - return_format_as_param = f'-f {openstack_stack_update_input.get_return_format()}' + return_format_as_param = f"-f {openstack_stack_update_input.get_return_format()}" # Assembles the command. - cmd = f'openstack stack update {return_format_as_param} {template_file_path_as_param} {stack_name}' + cmd = f"openstack stack update {return_format_as_param} {template_file_path_as_param} {stack_name}" return cmd diff --git a/keywords/ptp/cat/cat_ptp_cgu_keywords.py b/keywords/ptp/cat/cat_ptp_cgu_keywords.py index a9a7e293..07b12a19 100644 --- a/keywords/ptp/cat/cat_ptp_cgu_keywords.py +++ b/keywords/ptp/cat/cat_ptp_cgu_keywords.py @@ -9,6 +9,12 @@ class CatPtpCguKeywords(BaseKeyword): """ def __init__(self, ssh_connection: SSHConnection): + """ + Constructor for CatPtpCguKeywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ self.ssh_connection = ssh_connection def cat_ptp_cgu(self, cgu_location: str) -> PtpCguComponentOutput: @@ -19,8 +25,7 @@ class CatPtpCguKeywords(BaseKeyword): cgu_location (str): the cgu location. Returns: - PtpCguComponentOutput - the PtpCguComponentOutput. - + PtpCguComponentOutput: the output of the cat ptp cgu command """ output = self.ssh_connection.send_as_sudo(f"cat {cgu_location}") cat_ptp_cgu_component_output = PtpCguComponentOutput(output) diff --git a/keywords/ptp/cat/cat_ptp_config_keywords.py b/keywords/ptp/cat/cat_ptp_config_keywords.py index c4759c27..fe2e80a9 100644 --- a/keywords/ptp/cat/cat_ptp_config_keywords.py +++ b/keywords/ptp/cat/cat_ptp_config_keywords.py @@ -4,24 +4,28 @@ from keywords.ptp.cat.objects.cat_ptp_config_output import CATPtpConfigOutput class CatPtpConfigKeywords(BaseKeyword): - """ - Class for Cat Ptp Config Keywords - """ + """Class for Cat Ptp Config Keywords""" def __init__(self, ssh_connection: SSHConnection): + """ + Constructor for CatPtpConfigKeywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ self.ssh_connection = ssh_connection def cat_ptp_config(self, config_file: str) -> CATPtpConfigOutput: """ - Run cat cpt config command + This command reads the contents of a PTP configuration file. + Args: - config_file (): the ptp config file - - Returns: CatPtpConfigOutput + config_file (str): the ptp config file + Returns: + CATPtpConfigOutput: the output of the cat ptp config command """ - output = self.ssh_connection.send(f'cat {config_file}') + output = self.ssh_connection.send(f"cat {config_file}") self.validate_success_return_code(self.ssh_connection) cat_ptp_config_output = CATPtpConfigOutput(output) return cat_ptp_config_output - diff --git a/keywords/ptp/cat/objects/clock_description_object.py b/keywords/ptp/cat/objects/clock_description_object.py index 2b90d601..40a6976a 100644 --- a/keywords/ptp/cat/objects/clock_description_object.py +++ b/keywords/ptp/cat/objects/clock_description_object.py @@ -4,104 +4,98 @@ class ClockDescriptionObject: """ def __init__(self): - self.product_description: str = '' - self.revision_data: str = '' - self.manufacturer_identity: str = '' - self.user_description: str = '' - self.time_source: str = '' + self.product_description: str = "" + self.revision_data: str = "" + self.manufacturer_identity: str = "" + self.user_description: str = "" + self.time_source: str = "" def get_product_description(self) -> str: """ Getter for product_description - Returns: product_description + Returns: + str: the product_description value """ return self.product_description - def set_product_description(self, product_description: str): + def set_product_description(self, product_description: str) -> None: """ Setter for product_description + Args: - product_description (): the product_description value - - Returns: - + product_description (str): the product_description value """ self.product_description = product_description def get_revision_data(self) -> str: """ Getter for revision_data - Returns: revision_data value + Returns: + str: the revision_data value """ return self.revision_data - def set_revision_data(self, revision_data: str): + def set_revision_data(self, revision_data: str) -> None: """ Setter for revision_data + Args: - revision_data (): revision_data value - - Returns: - + revision_data (str): revision_data value """ self.revision_data = revision_data def get_manufacturer_identity(self) -> str: """ Getter for manufacturer_identity - Returns: manufacturer_identity value + Returns: + str: the manufacturer_identity value """ return self.manufacturer_identity - def set_manufacturer_identity(self, manufacturer_identity: str): + def set_manufacturer_identity(self, manufacturer_identity: str) -> None: """ Setter for manufacturer_identity + Args: - manufacturer_identity (): manufacturer_identity value - - Returns: - + manufacturer_identity (str): manufacturer_identity value """ self.manufacturer_identity = manufacturer_identity def get_user_description(self) -> str: """ Getter for user_description - Returns: the user_description value + Returns: + str: the user_description value """ return self.user_description - def set_user_description(self, user_description: str): + def set_user_description(self, user_description: str) -> None: """ Setter for user_description + Args: - user_description (): the user_description value - - Returns: - + user_description (str): the user_description value """ self.user_description = user_description def get_time_source(self) -> str: """ Getter for time_source - Returns: time_source value + Returns: + str: the time_source value """ return self.time_source - def set_time_source(self, time_source: str): + def set_time_source(self, time_source: str) -> None: """ Setter for time_source + Args: - time_source (): the time_source value - - Returns: - + time_source (str): the time_source value """ self.time_source = time_source - diff --git a/keywords/ptp/cat/objects/clock_description_output.py b/keywords/ptp/cat/objects/clock_description_output.py index 1a89fbf8..c72bdd07 100644 --- a/keywords/ptp/cat/objects/clock_description_output.py +++ b/keywords/ptp/cat/objects/clock_description_output.py @@ -15,39 +15,37 @@ class ClockDescriptionOutput: """ - def __init__(self, clock_description_output: [str]): + def __init__(self, clock_description_output: list[str]): """ - Constructor. - Create an internal ClockDescriptionObject from the passed parameter. + Create an internal ClockDescriptionObject from the passed parameter. + Args: clock_description_output (list[str]): a list of strings representing the clock description output - """ cat_ptp_table_parser = CatPtpTableParser(clock_description_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.clock_description_object = ClockDescriptionObject() - if 'productDescription' in output_values: - self.clock_description_object.set_product_description(output_values['productDescription']) + if "productDescription" in output_values: + self.clock_description_object.set_product_description(output_values["productDescription"]) - if 'revisionData' in output_values: - self.clock_description_object.set_revision_data(output_values['revisionData']) + if "revisionData" in output_values: + self.clock_description_object.set_revision_data(output_values["revisionData"]) - if 'manufacturerIdentity' in output_values: - self.clock_description_object.set_manufacturer_identity(output_values['manufacturerIdentity']) + if "manufacturerIdentity" in output_values: + self.clock_description_object.set_manufacturer_identity(output_values["manufacturerIdentity"]) - if 'userDescription' in output_values: - self.clock_description_object.set_user_description(output_values['userDescription']) + if "userDescription" in output_values: + self.clock_description_object.set_user_description(output_values["userDescription"]) - if 'timeSource' in output_values: - self.clock_description_object.set_time_source(output_values['timeSource']) + if "timeSource" in output_values: + self.clock_description_object.set_time_source(output_values["timeSource"]) def get_clock_description_object(self) -> ClockDescriptionObject: """ Getter for ClockDescriptionObject object. Returns: - A ClockDescriptionObject - + ClockDescriptionObject: the clock_description_object """ return self.clock_description_object diff --git a/keywords/ptp/cat/objects/default_interface_options_object.py b/keywords/ptp/cat/objects/default_interface_options_object.py index 93333c4f..37716944 100644 --- a/keywords/ptp/cat/objects/default_interface_options_object.py +++ b/keywords/ptp/cat/objects/default_interface_options_object.py @@ -4,12 +4,12 @@ class DefaultInterfaceOptionsObject: """ def __init__(self): - self.clock_type: str = '' - self.network_transport: str = '' - self.delay_mechanism: str = '' - self.time_stamping: str = '' - self.tsproc_mode: str = '' - self.delay_filter: str = '' + self.clock_type: str = "" + self.network_transport: str = "" + self.delay_mechanism: str = "" + self.time_stamping: str = "" + self.tsproc_mode: str = "" + self.delay_filter: str = "" self.delay_filter_length: int = -1 self.egress_latency: int = -1 self.ingress_latency: int = -1 @@ -18,189 +18,179 @@ class DefaultInterfaceOptionsObject: def get_clock_type(self) -> str: """ Getter for clock_type - Returns: clock_type + Returns: + str: the clock_type value """ return self.clock_type - def set_clock_type(self, clock_type: str): + def set_clock_type(self, clock_type: str) -> None: """ Setter for clock_type + Args: - clock_type (): the clock_type value - - Returns: - + clock_type (str): the clock_type value """ self.clock_type = clock_type def get_network_transport(self) -> str: """ Getter for network_transport - Returns: network_transport value + Returns: + str: the network_transport value """ return self.network_transport - def set_network_transport(self, network_transport: str): + def set_network_transport(self, network_transport: str) -> None: """ Setter for network_transport + Args: - network_transport (): network_transport value - - Returns: - + network_transport (str): network_transport value """ self.network_transport = network_transport def get_delay_mechanism(self) -> str: """ Getter for delay_mechanism - Returns: delay_mechanism value + Returns: + str: the delay_mechanism value """ return self.delay_mechanism - def set_delay_mechanism(self, delay_mechanism: str): + def set_delay_mechanism(self, delay_mechanism: str) -> None: """ Setter for delay_mechanism + Args: - delay_mechanism (): delay_mechanism value - - Returns: - + delay_mechanism (str): delay_mechanism value """ self.delay_mechanism = delay_mechanism def get_time_stamping(self) -> str: """ Getter for time_stamping - Returns: the time_stamping value + Returns: + str: the time_stamping value """ return self.time_stamping - def set_time_stamping(self, time_stamping: str): + def set_time_stamping(self, time_stamping: str) -> None: """ Setter for time_stamping + Args: - time_stamping (): the time_stamping value - - Returns: - + time_stamping (str): the time_stamping value """ self.time_stamping = time_stamping def get_tsproc_mode(self) -> str: """ Getter for tsproc_mode - Returns: tsproc_mode value + Returns: + str: the tsproc_mode value """ return self.tsproc_mode - def set_tsproc_mode(self, tsproc_mode: str): + def set_tsproc_mode(self, tsproc_mode: str) -> None: """ Setter for tsproc_mode + Args: - tsproc_mode (): the tsproc_mode value - - Returns: - + tsproc_mode (str): the tsproc_mode value """ self.tsproc_mode = tsproc_mode def get_delay_filter(self) -> str: """ Getter for delay_filter - Returns: delay_filter value + Returns: + str: the delay_filter value """ return self.delay_filter - def set_delay_filter(self, delay_filter: str): + def set_delay_filter(self, delay_filter: str) -> None: """ Setter for delay_filter + Args: - delay_filter (): the delay_filter value - - Returns: - + delay_filter (str): the delay_filter value """ self.delay_filter = delay_filter def get_delay_filter_length(self) -> int: """ Getter for delay_filter_length - Returns: delay_filter_length value + Returns: + int: the delay_filter_length value """ return self.delay_filter_length - def set_delay_filter_length(self, delay_filter_length: int): + def set_delay_filter_length(self, delay_filter_length: int) -> None: """ Setter for delay_filter_length + Args: - delay_filter_length (): the delay_filter_length value - - Returns: - + delay_filter_length (int): the delay_filter_length value """ self.delay_filter_length = delay_filter_length def get_egress_latency(self) -> int: """ Getter for egress_latency - Returns: egress_latency value + Returns: + int: the egress_latency value """ return self.egress_latency - def set_egress_latency(self, egress_latency: int): + def set_egress_latency(self, egress_latency: int) -> None: """ Setter for egress_latency + Args: - egress_latency (): the egress_latency value - - Returns: - + egress_latency (int): the egress_latency value """ self.egress_latency = egress_latency def get_ingress_latency(self) -> int: """ Getter for ingress_latency - Returns: ingress_latency value + Returns: + int: the ingress_latency value """ return self.ingress_latency - def set_ingress_latency(self, ingress_latency: int): + def set_ingress_latency(self, ingress_latency: int) -> None: """ Setter for ingress_latency + Args: - ingress_latency (): the ingress_latency value - - Returns: - + ingress_latency (int): the ingress_latency value """ self.ingress_latency = ingress_latency def get_boundary_clock_jbod(self) -> int: """ Getter for boundary_clock_jbod - Returns: boundary_clock_jbod value + Returns: + int: the boundary_clock_jbod value """ return self.boundary_clock_jbod - def set_boundary_clock_jbod(self, boundary_clock_jbod: int): + def set_boundary_clock_jbod(self, boundary_clock_jbod: int) -> None: """ Setter for boundary_clock_jbod + Args: - boundary_clock_jbod (): the boundary_clock_jbod value - - Returns: - + boundary_clock_jbod (int): the boundary_clock_jbod value """ self.boundary_clock_jbod = boundary_clock_jbod diff --git a/keywords/ptp/cat/objects/default_interface_options_output.py b/keywords/ptp/cat/objects/default_interface_options_output.py index 507f0ace..de081b28 100644 --- a/keywords/ptp/cat/objects/default_interface_options_output.py +++ b/keywords/ptp/cat/objects/default_interface_options_output.py @@ -20,54 +20,52 @@ class DefaultInterfaceOptionsOutput: """ - def __init__(self, default_interface_options_output: [str]): + def __init__(self, default_interface_options_output: list[str]): """ - Constructor. - Create an internal DefaultInterfaceOptionsObject from the passed parameter. + Create an internal DefaultInterfaceOptionsObject from the passed parameter. + Args: default_interface_options_output (list[str]): a list of strings representing the default interface options output - """ cat_ptp_table_parser = CatPtpTableParser(default_interface_options_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.default_interface_options_object = DefaultInterfaceOptionsObject() - if 'clock_type' in output_values: - self.default_interface_options_object.set_clock_type(output_values['clock_type']) + if "clock_type" in output_values: + self.default_interface_options_object.set_clock_type(output_values["clock_type"]) - if 'network_transport' in output_values: - self.default_interface_options_object.set_network_transport(output_values['network_transport']) + if "network_transport" in output_values: + self.default_interface_options_object.set_network_transport(output_values["network_transport"]) - if 'delay_mechanism' in output_values: - self.default_interface_options_object.set_delay_mechanism(output_values['delay_mechanism']) + if "delay_mechanism" in output_values: + self.default_interface_options_object.set_delay_mechanism(output_values["delay_mechanism"]) - if 'time_stamping' in output_values: - self.default_interface_options_object.set_time_stamping(output_values['time_stamping']) + if "time_stamping" in output_values: + self.default_interface_options_object.set_time_stamping(output_values["time_stamping"]) - if 'tsproc_mode' in output_values: - self.default_interface_options_object.set_tsproc_mode(output_values['tsproc_mode']) + if "tsproc_mode" in output_values: + self.default_interface_options_object.set_tsproc_mode(output_values["tsproc_mode"]) - if 'delay_filter' in output_values: - self.default_interface_options_object.set_delay_filter(output_values['delay_filter']) + if "delay_filter" in output_values: + self.default_interface_options_object.set_delay_filter(output_values["delay_filter"]) - if 'delay_filter_length' in output_values: - self.default_interface_options_object.set_delay_filter_length(int(output_values['delay_filter_length'])) + if "delay_filter_length" in output_values: + self.default_interface_options_object.set_delay_filter_length(int(output_values["delay_filter_length"])) - if 'egressLatency' in output_values: - self.default_interface_options_object.set_egress_latency(int(output_values['egressLatency'])) + if "egressLatency" in output_values: + self.default_interface_options_object.set_egress_latency(int(output_values["egressLatency"])) - if 'ingressLatency' in output_values: - self.default_interface_options_object.set_ingress_latency(int(output_values['ingressLatency'])) + if "ingressLatency" in output_values: + self.default_interface_options_object.set_ingress_latency(int(output_values["ingressLatency"])) - if 'boundary_clock_jbod' in output_values: - self.default_interface_options_object.set_boundary_clock_jbod(int(output_values['boundary_clock_jbod'])) + if "boundary_clock_jbod" in output_values: + self.default_interface_options_object.set_boundary_clock_jbod(int(output_values["boundary_clock_jbod"])) def get_default_interface_options_object(self) -> DefaultInterfaceOptionsObject: """ Getter for DefaultInterfaceOptionsObject object. Returns: - A DefaultInterfaceOptionsObject - + DefaultInterfaceOptionsObject: The default interface options object containing parsed values. """ return self.default_interface_options_object diff --git a/keywords/ptp/cat/objects/port_data_set_object.py b/keywords/ptp/cat/objects/port_data_set_object.py index ed462986..5df2b88c 100644 --- a/keywords/ptp/cat/objects/port_data_set_object.py +++ b/keywords/ptp/cat/objects/port_data_set_object.py @@ -1,7 +1,5 @@ class PortDataSetObject: - """ - Object to hold the values of port data set - """ + """Object to hold the values of port data set""" def __init__(self): self.log_announce_interval: int = -1 @@ -16,336 +14,314 @@ class PortDataSetObject: self.fault_reset_interval: int = -1 self.neighbor_prop_delay_thresh: int = -1 self.master_only: int = -1 - self.as_capable: str = '' - self.bmca: str = '' + self.as_capable: str = "" + self.bmca: str = "" self.inhibit_announce: int = -1 - self.inhibit_delay_req: int = '' + self.inhibit_delay_req: int = "" self.ignore_source_id: int = -1 def get_log_announce_interval(self) -> int: """ Getter for log_announce_interval - Returns: log_announce_interval + Returns: + int: the log_announce_interval value """ return self.log_announce_interval - def set_log_announce_interval(self, log_announce_interval: int): + def set_log_announce_interval(self, log_announce_interval: int) -> None: """ Setter for two_step_flag + Args: - log_announce_interval (): the log_announce_interval value - - Returns: - + log_announce_interval (int): the log_announce_interval value """ self.log_announce_interval = log_announce_interval def get_log_sync_interval(self) -> int: """ Getter for log_sync_interval - Returns: log_sync_interval value + Returns: + int: the log_sync_interval value """ return self.log_sync_interval - def set_log_sync_interval(self, log_sync_interval: int): + def set_log_sync_interval(self, log_sync_interval: int) -> None: """ Setter for log_sync_interval + Args: - log_sync_interval (): log_sync_interval value - - Returns: - + log_sync_interval (int): log_sync_interval value """ self.log_sync_interval = log_sync_interval def get_oper_log_sync_interval(self) -> int: """ Getter for oper_log_sync_interval - Returns: oper_log_sync_interval value + Returns: + int: the oper_log_sync_interval value """ return self.oper_log_sync_interval - def set_oper_log_sync_interval(self, oper_log_sync_interval: int): + def set_oper_log_sync_interval(self, oper_log_sync_interval: int) -> None: """ Setter for oper_log_sync_interval + Args: - oper_log_sync_interval (): oper_log_sync_interval value - - Returns: - + oper_log_sync_interval (int): oper_log_sync_interval value """ self.oper_log_sync_interval = oper_log_sync_interval def get_log_min_delay_req_interval(self) -> int: """ Getter for log_min_delay_req_interval - Returns: the log_min_delay_req_interval value + Returns: + int: the log_min_delay_req_interval value """ return self.log_min_delay_req_interval - def set_log_min_delay_req_interval(self, log_min_delay_req_interval: int): + def set_log_min_delay_req_interval(self, log_min_delay_req_interval: int) -> None: """ Setter for log_min_delay_req_interval + Args: - log_min_delay_req_interval (): the log_min_delay_req_interval value - - Returns: - + log_min_delay_req_interval (int): the log_min_delay_req_interval value """ self.log_min_delay_req_interval = log_min_delay_req_interval def get_log_min_p_delay_req_interval(self) -> int: """ Getter for log_min_p_delay_req_interval - Returns: log_min_p_delay_req_interval value + Returns: + int: the log_min_p_delay_req_interval value """ return self.log_min_p_delay_req_interval - def set_log_min_p_delay_req_interval(self, log_min_p_delay_req_interval: int): + def set_log_min_p_delay_req_interval(self, log_min_p_delay_req_interval: int) -> None: """ Setter for log_min_p_delay_req_interval + Args: - log_min_p_delay_req_interval (): the log_min_p_delay_req_interval value - - Returns: - + log_min_p_delay_req_interval (int): the log_min_p_delay_req_interval value """ self.log_min_p_delay_req_interval = log_min_p_delay_req_interval def get_oper_log_p_delay_req_interval(self) -> int: """ Getter for oper_log_p_delay_req_interval - Returns: the oper_log_p_delay_req_interval value + Returns: + int: the oper_log_p_delay_req_interval value """ return self.oper_log_p_delay_req_interval - def set_oper_log_p_delay_req_interval(self, oper_log_p_delay_req_interval: int): + def set_oper_log_p_delay_req_interval(self, oper_log_p_delay_req_interval: int) -> None: """ Setter for oper_log_p_delay_req_interval + Args: - oper_log_p_delay_req_interval (): the oper_log_p_delay_req_interval value - - Returns: - + oper_log_p_delay_req_interval (int): the oper_log_p_delay_req_interval value """ self.oper_log_p_delay_req_interval = oper_log_p_delay_req_interval def get_announce_receipt_timeout(self) -> int: """ Getter for announce_receipt_timeout - Returns: the announce_receipt_timeout value + Returns: + int: the announce_receipt_timeout value """ return self.announce_receipt_timeout - def set_announce_receipt_timeout(self, announce_receipt_timeout: int): + def set_announce_receipt_timeout(self, announce_receipt_timeout: int) -> None: """ Setter for announce_receipt_timeout + Args: - announce_receipt_timeout (): the announce_receipt_timeout value - - Returns: - + announce_receipt_timeout (int): the announce_receipt_timeout value """ self.announce_receipt_timeout = announce_receipt_timeout def get_sync_receipt_timeout(self) -> int: """ Getter for sync_receipt_timeout - Returns: the sync_receipt_timeout value + Returns: + int: the sync_receipt_timeout value """ return self.sync_receipt_timeout - def set_sync_receipt_timeout(self, sync_receipt_timeout: int): + def set_sync_receipt_timeout(self, sync_receipt_timeout: int) -> None: """ Setter for sync_receipt_timeout + Args: - sync_receipt_timeout (): the sync_receipt_timeout value - - Returns: - + sync_receipt_timeout (int): the sync_receipt_timeout value """ self.sync_receipt_timeout = sync_receipt_timeout def get_delay_asymmetry(self) -> int: """ Getter for delay_asymmetry - Returns: the delay_asymmetry value + Returns: + int: the delay_asymmetry value """ return self.delay_asymmetry - def set_delay_asymmetry(self, delay_asymmetry: int): + def set_delay_asymmetry(self, delay_asymmetry: int) -> None: """ Setter for delay_asymmetry + Args: - delay_asymmetry (): the delay_asymmetry value - - Returns: - + delay_asymmetry (int): the delay_asymmetry value """ self.delay_asymmetry = delay_asymmetry def get_fault_reset_interval(self) -> int: """ Getter for fault_reset_interval - Returns: the fault_reset_interval value + Returns: + int: the fault_reset_interval value """ return self.fault_reset_interval - def set_fault_reset_interval(self, fault_reset_interval: int): + def set_fault_reset_interval(self, fault_reset_interval: int) -> None: """ Setter for fault_reset_interval + Args: - fault_reset_interval (): the fault_reset_interval value - - Returns: - + fault_reset_interval (int): the fault_reset_interval value """ self.fault_reset_interval = fault_reset_interval def get_neighbor_prop_delay_thresh(self) -> int: """ Getter for neighbor_prop_delay_thresh - Returns: the neighbor_prop_delay_thresh value + Returns: + int: the neighbor_prop_delay_thresh value """ return self.neighbor_prop_delay_thresh - def set_neighbor_prop_delay_thresh(self, neighbor_prop_delay_thresh: int): + def set_neighbor_prop_delay_thresh(self, neighbor_prop_delay_thresh: int) -> None: """ Setter for neighbor_prop_delay_thresh + Args: - neighbor_prop_delay_thresh (): the neighbor_prop_delay_thresh value - - Returns: - + neighbor_prop_delay_thresh (int): the neighbor_prop_delay_thresh value """ self.neighbor_prop_delay_thresh = neighbor_prop_delay_thresh def get_master_only(self) -> int: """ Getter for master_only - Returns: the master_only value + Returns: + int: the master_only value """ return self.master_only - def set_master_only(self, master_only: int): + def set_master_only(self, master_only: int) -> None: """ Setter for master_only + Args: - master_only (): the master_only value - - Returns: - + master_only (int): the master_only value """ self.master_only = master_only def get_as_capable(self) -> str: """ Getter for as_capable - Returns: the as_capable value + Returns: + str: the as_capable value """ return self.as_capable - def set_as_capable(self, as_capable: str): + def set_as_capable(self, as_capable: str) -> None: """ Setter for as_capable + Args: - as_capable (): the as_capable value - - Returns: - + as_capable (str): the as_capable value """ self.as_capable = as_capable def get_bmca(self) -> str: """ Getter for bmca - Returns: the bmca value + Returns: + str: the bmca value """ return self.bmca - def set_bmca(self, bmca: str): + def set_bmca(self, bmca: str) -> None: """ Setter for bmca + Args: - bmca (): the bmca value - - Returns: - + bmca (str): the bmca value """ self.bmca = bmca def get_inhibit_announce(self) -> int: """ Getter for inhibit_announce - Returns: the inhibit_announce value + Returns: + int: the inhibit_announce value """ return self.inhibit_announce - def set_inhibit_announce(self, inhibit_announce: int): + def set_inhibit_announce(self, inhibit_announce: int) -> None: """ Setter for inhibit_announce + Args: - inhibit_announce (): the inhibit_announce value - - Returns: - + inhibit_announce (int): the inhibit_announce value """ self.inhibit_announce = inhibit_announce def get_inhibit_delay_req(self) -> int: """ Getter for inhibit_delay_req - Returns: the inhibit_delay_req value + Returns: + int: the inhibit_delay_req value """ return self.inhibit_delay_req - def set_inhibit_delay_req(self, inhibit_delay_req: int): + def set_inhibit_delay_req(self, inhibit_delay_req: int) -> None: """ Setter for inhibit_delay_req + Args: - inhibit_delay_req (): the inhibit_delay_req value - - Returns: - + inhibit_delay_req (int): the inhibit_delay_req value """ self.inhibit_delay_req = inhibit_delay_req def get_ignore_source_id(self) -> int: """ Getter for ignore_source_id - Returns: the ignore_source_id value + Returns: + int: the ignore_source_id value """ return self.ignore_source_id - def set_ignore_source_id(self, ignore_source_id: int): + def set_ignore_source_id(self, ignore_source_id: int) -> None: """ Setter for ignore_source_id + Args: - ignore_source_id (): the ignore_source_id value - - Returns: - + ignore_source_id (int): the ignore_source_id value """ self.ignore_source_id = ignore_source_id - - - - - diff --git a/keywords/ptp/cat/objects/port_data_set_output.py b/keywords/ptp/cat/objects/port_data_set_output.py index 1984f2d1..ef9b2091 100644 --- a/keywords/ptp/cat/objects/port_data_set_output.py +++ b/keywords/ptp/cat/objects/port_data_set_output.py @@ -28,75 +28,73 @@ class PortDataSetOutput: """ - def __init__(self, port_data_set_output: [str]): + def __init__(self, port_data_set_output: list[str]): """ - Constructor. - Create an internal PortDataSetObject from the passed parameter. + Create an internal PortDataSetObject from the passed parameter. + Args: port_data_set_output (list[str]): a list of strings representing the port data set output - """ cat_ptp_table_parser = CatPtpTableParser(port_data_set_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.port_data_set_object = PortDataSetObject() - if 'logAnnounceInterval' in output_values: - self.port_data_set_object.set_log_announce_interval(int(output_values['logAnnounceInterval'])) + if "logAnnounceInterval" in output_values: + self.port_data_set_object.set_log_announce_interval(int(output_values["logAnnounceInterval"])) - if 'logSyncInterval' in output_values: - self.port_data_set_object.set_log_sync_interval(int(output_values['logSyncInterval'])) + if "logSyncInterval" in output_values: + self.port_data_set_object.set_log_sync_interval(int(output_values["logSyncInterval"])) - if 'operLogSyncInterval' in output_values: - self.port_data_set_object.set_oper_log_sync_interval(int(output_values['operLogSyncInterval'])) + if "operLogSyncInterval" in output_values: + self.port_data_set_object.set_oper_log_sync_interval(int(output_values["operLogSyncInterval"])) - if 'logMinDelayReqInterval' in output_values: - self.port_data_set_object.set_log_min_delay_req_interval(int(output_values['logMinDelayReqInterval'])) + if "logMinDelayReqInterval" in output_values: + self.port_data_set_object.set_log_min_delay_req_interval(int(output_values["logMinDelayReqInterval"])) - if 'logMinPdelayReqInterval' in output_values: - self.port_data_set_object.set_log_min_p_delay_req_interval(int(output_values['logMinPdelayReqInterval'])) + if "logMinPdelayReqInterval" in output_values: + self.port_data_set_object.set_log_min_p_delay_req_interval(int(output_values["logMinPdelayReqInterval"])) - if 'operLogPdelayReqInterval' in output_values: - self.port_data_set_object.set_oper_log_p_delay_req_interval(int(output_values['operLogPdelayReqInterval'])) + if "operLogPdelayReqInterval" in output_values: + self.port_data_set_object.set_oper_log_p_delay_req_interval(int(output_values["operLogPdelayReqInterval"])) - if 'announceReceiptTimeout' in output_values: - self.port_data_set_object.set_announce_receipt_timeout(int(output_values['announceReceiptTimeout'])) + if "announceReceiptTimeout" in output_values: + self.port_data_set_object.set_announce_receipt_timeout(int(output_values["announceReceiptTimeout"])) - if 'syncReceiptTimeout' in output_values: - self.port_data_set_object.set_sync_receipt_timeout(int(output_values['syncReceiptTimeout'])) + if "syncReceiptTimeout" in output_values: + self.port_data_set_object.set_sync_receipt_timeout(int(output_values["syncReceiptTimeout"])) - if 'delayAsymmetry' in output_values: - self.port_data_set_object.set_delay_asymmetry(int(output_values['delayAsymmetry'])) + if "delayAsymmetry" in output_values: + self.port_data_set_object.set_delay_asymmetry(int(output_values["delayAsymmetry"])) - if 'fault_reset_interval' in output_values: - self.port_data_set_object.set_fault_reset_interval(int(output_values['fault_reset_interval'])) + if "fault_reset_interval" in output_values: + self.port_data_set_object.set_fault_reset_interval(int(output_values["fault_reset_interval"])) - if 'neighborPropDelayThresh' in output_values: - self.port_data_set_object.set_neighbor_prop_delay_thresh(int(output_values['neighborPropDelayThresh'])) + if "neighborPropDelayThresh" in output_values: + self.port_data_set_object.set_neighbor_prop_delay_thresh(int(output_values["neighborPropDelayThresh"])) - if 'masterOnly' in output_values: - self.port_data_set_object.set_master_only(int(output_values['masterOnly'])) + if "masterOnly" in output_values: + self.port_data_set_object.set_master_only(int(output_values["masterOnly"])) - if 'asCapable' in output_values: - self.port_data_set_object.set_as_capable(output_values['asCapable']) + if "asCapable" in output_values: + self.port_data_set_object.set_as_capable(output_values["asCapable"]) - if 'BMCA' in output_values: - self.port_data_set_object.set_bmca(output_values['BMCA']) + if "BMCA" in output_values: + self.port_data_set_object.set_bmca(output_values["BMCA"]) - if 'inhibit_announce' in output_values: - self.port_data_set_object.set_inhibit_announce(int(output_values['inhibit_announce'])) + if "inhibit_announce" in output_values: + self.port_data_set_object.set_inhibit_announce(int(output_values["inhibit_announce"])) - if 'inhibit_delay_req' in output_values: - self.port_data_set_object.set_inhibit_delay_req(int(output_values['inhibit_delay_req'])) + if "inhibit_delay_req" in output_values: + self.port_data_set_object.set_inhibit_delay_req(int(output_values["inhibit_delay_req"])) - if 'ignore_source_id' in output_values: - self.port_data_set_object.set_ignore_source_id(int(output_values['ignore_source_id'])) + if "ignore_source_id" in output_values: + self.port_data_set_object.set_ignore_source_id(int(output_values["ignore_source_id"])) def get_port_data_set_object(self) -> PortDataSetObject: """ Getter for port_data_set_object object. Returns: - A PortDataSetObject - + PortDataSetObject: The port data set object containing parsed values. """ return self.port_data_set_object diff --git a/keywords/ptp/cat/objects/ptp_cgu_component_output.py b/keywords/ptp/cat/objects/ptp_cgu_component_output.py index 87dc1fde..afe76b3a 100644 --- a/keywords/ptp/cat/objects/ptp_cgu_component_output.py +++ b/keywords/ptp/cat/objects/ptp_cgu_component_output.py @@ -16,7 +16,6 @@ class PtpCguComponentOutput: Gets the cgu component. Returns: - PtpCguComponentObject - the PtpCguComponentObject - + PtpCguComponentObject: The cgu component object containing parsed values. """ return self.cgu_component diff --git a/keywords/ptp/cat/objects/run_time_options_object.py b/keywords/ptp/cat/objects/run_time_options_object.py index f2db2d05..a63b203c 100644 --- a/keywords/ptp/cat/objects/run_time_options_object.py +++ b/keywords/ptp/cat/objects/run_time_options_object.py @@ -19,334 +19,311 @@ class RunTimeOptionsObject: self.use_syslog: int = -1 self.verbose: int = -1 self.summary_interval: int = -1 - self.kernel_leap: int = '' + self.kernel_leap: int = "" self.check_fup_sync: int = -1 def get_assume_two_step(self) -> int: """ Getter for assume_two_step - Returns: assume_two_step + Returns: + int: the assume_two_step value """ return self.assume_two_step - def set_assume_two_step(self, assume_two_step: int): + def set_assume_two_step(self, assume_two_step: int) -> None: """ Setter for assume_two_step + Args: - assume_two_step (): the assume_two_step value - - Returns: - + assume_two_step (int): the assume_two_step value """ self.assume_two_step = assume_two_step def get_logging_level(self) -> int: """ Getter for logging_level - Returns: logging_level value + Returns: + int: the logging_level value """ return self.logging_level - def set_logging_level(self, logging_level: int): + def set_logging_level(self, logging_level: int) -> None: """ Setter for logging_level + Args: - log_sync_interval (): log_sync_interval value - - Returns: - + logging_level (int): the logging_level value """ self.logging_level = logging_level def get_path_trace_enabled(self) -> int: """ Getter for path_trace_enabled - Returns: path_trace_enabled value + Returns: + int: the path_trace_enabled value """ return self.path_trace_enabled - def set_path_trace_enabled(self, path_trace_enabled: int): + def set_path_trace_enabled(self, path_trace_enabled: int) -> None: """ Setter for path_trace_enabled + Args: - path_trace_enabled (): path_trace_enabled value - - Returns: - + path_trace_enabled (int): path_trace_enabled value """ self.path_trace_enabled = path_trace_enabled def get_follow_up_info(self) -> int: """ Getter for follow_up_info - Returns: the follow_up_info value + Returns: + int: the follow_up_info value """ return self.follow_up_info - def set_follow_up_info(self, follow_up_info: int): + def set_follow_up_info(self, follow_up_info: int) -> None: """ Setter for follow_up_info + Args: - follow_up_info (): the follow_up_info value - - Returns: - + follow_up_info (int): the follow_up_info value """ self.follow_up_info = follow_up_info def get_hybrid_e2e(self) -> int: """ Getter for hybrid_e2e - Returns: hybrid_e2e value + Returns: + int: the hybrid_e2e value """ return self.hybrid_e2e - def set_hybrid_e2e(self, hybrid_e2e: int): + def set_hybrid_e2e(self, hybrid_e2e: int) -> None: """ Setter for hybrid_e2e + Args: - hybrid_e2e (): the hybrid_e2e value - - Returns: - + hybrid_e2e (int): the hybrid_e2e value """ self.hybrid_e2e = hybrid_e2e def get_inhibit_multicast_service(self) -> int: """ Getter for inhibit_multicast_service - Returns: the inhibit_multicast_service value + Returns: + int: the inhibit_multicast_service value """ return self.inhibit_multicast_service - def set_inhibit_multicast_service(self, inhibit_multicast_service: int): + def set_inhibit_multicast_service(self, inhibit_multicast_service: int) -> None: """ Setter for inhibit_multicast_service + Args: - inhibit_multicast_service (): the inhibit_multicast_service value - - Returns: - + inhibit_multicast_service (int): the inhibit_multicast_service value """ self.inhibit_multicast_service = inhibit_multicast_service def get_net_sync_monitor(self) -> int: """ Getter for net_sync_monitor - Returns: the net_sync_monitor value + Returns: + int: the net_sync_monitor value """ return self.net_sync_monitor - def set_net_sync_monitor(self, net_sync_monitor: int): + def set_net_sync_monitor(self, net_sync_monitor: int) -> None: """ Setter for net_sync_monitor + Args: - net_sync_monitor (): the net_sync_monitor value - - Returns: - + net_sync_monitor (int): the net_sync_monitor value """ self.net_sync_monitor = net_sync_monitor def get_tc_spanning_tree(self) -> int: """ Getter for tc_spanning_tree - Returns: the tc_spanning_tree value + Returns: + int: the tc_spanning_tree value """ return self.tc_spanning_tree - def set_tc_spanning_tree(self, tc_spanning_tree: int): + def set_tc_spanning_tree(self, tc_spanning_tree: int) -> None: """ Setter for sync_receipt_timeout + Args: - tc_spanning_tree (): the tc_spanning_tree value - - Returns: - + tc_spanning_tree (int): the tc_spanning_tree value """ self.tc_spanning_tree = tc_spanning_tree def get_tx_timestamp_timeout(self) -> int: """ Getter for tx_timestamp_timeout - Returns: the tx_timestamp_timeout value + Returns: + int: the tx_timestamp_timeout value """ return self.tx_timestamp_timeout - def set_tx_timestamp_timeout(self, tx_timestamp_timeout: int): + def set_tx_timestamp_timeout(self, tx_timestamp_timeout: int) -> None: """ Setter for tx_timestamp_timeout + Args: - tx_timestamp_timeout (): the tx_timestamp_timeout value - - Returns: - + tx_timestamp_timeout (int): the tx_timestamp_timeout value """ self.tx_timestamp_timeout = tx_timestamp_timeout def get_unicast_listen(self) -> int: """ Getter for unicast_listen - Returns: the unicast_listen value + Returns: + int: the unicast_listen value """ return self.unicast_listen - def set_unicast_listen(self, unicast_listen: int): + def set_unicast_listen(self, unicast_listen: int) -> None: """ Setter for unicast_listen + Args: - unicast_listen (): the unicast_listen value - - Returns: - + unicast_listen (int): the unicast_listen value """ self.unicast_listen = unicast_listen def get_unicast_master_table(self) -> int: """ Getter for unicast_master_table - Returns: the unicast_master_table value + Returns: + int: the unicast_master_table value """ return self.unicast_master_table - def set_unicast_master_table(self, unicast_master_table: int): + def set_unicast_master_table(self, unicast_master_table: int) -> None: """ Setter for unicast_master_table + Args: - unicast_master_table (): the unicast_master_table value - - Returns: - + unicast_master_table (int): the unicast_master_table value """ self.unicast_master_table = unicast_master_table def get_unicast_req_duration(self) -> int: """ Getter for unicast_req_duration - Returns: the unicast_req_duration value + Returns: + int: the unicast_req_duration value """ return self.unicast_req_duration - def set_unicast_req_duration(self, unicast_req_duration: int): + def set_unicast_req_duration(self, unicast_req_duration: int) -> None: """ Setter for unicast_req_duration + Args: - unicast_req_duration (): the unicast_req_duration value - - Returns: - + unicast_req_duration (int): the unicast_req_duration value """ self.unicast_req_duration = unicast_req_duration def get_use_syslog(self) -> int: """ Getter for use_syslog - Returns: the use_syslog value + Returns: + int: the use_syslog value """ return self.use_syslog - def set_use_syslog(self, use_syslog: int): + def set_use_syslog(self, use_syslog: int) -> None: """ Setter for use_syslog + Args: - use_syslog (): the use_syslog value - - Returns: - + use_syslog (int): the use_syslog value """ self.use_syslog = use_syslog def get_verbose(self) -> int: """ Getter for verbose - Returns: the verbose value + Returns: + int: the verbose value """ return self.verbose - def set_verbose(self, verbose: int): + def set_verbose(self, verbose: int) -> None: """ Setter for verbose + Args: - verbose (): the verbose value - - Returns: - + verbose (int): the verbose value """ self.verbose = verbose def get_summary_interval(self) -> int: """ Getter for summary_interval - Returns: the summary_interval value + Returns: + int: the summary_interval value """ return self.summary_interval - def set_summary_interval(self, summary_interval: int): + def set_summary_interval(self, summary_interval: int) -> None: """ Setter for summary_interval + Args: - summary_interval (): the summary_interval value - - Returns: - + summary_interval (int): the summary_interval value """ self.summary_interval = summary_interval def get_kernel_leap(self) -> int: """ Getter for kernel_leap - Returns: the kernel_leap value + Returns: + int: the kernel_leap value """ return self.kernel_leap - def set_kernel_leap(self, kernel_leap: int): + def set_kernel_leap(self, kernel_leap: int) -> None: """ Setter for kernel_leap + Args: - kernel_leap (): the kernel_leap value - - Returns: - + kernel_leap (int): the kernel_leap value """ self.kernel_leap = kernel_leap def get_check_fup_sync(self) -> int: """ Getter for check_fup_sync - Returns: the check_fup_sync value + Returns: + int: the check_fup_sync value """ return self.check_fup_sync - def set_check_fup_sync(self, check_fup_sync: int): + def set_check_fup_sync(self, check_fup_sync: int) -> None: """ Setter for check_fup_sync + Args: - check_fup_sync (): the check_fup_sync value - - Returns: - + check_fup_sync (int): the check_fup_sync value """ self.check_fup_sync = check_fup_sync - - - - - - diff --git a/keywords/ptp/cat/objects/run_time_options_output.py b/keywords/ptp/cat/objects/run_time_options_output.py index 62ff9012..ccbfb982 100644 --- a/keywords/ptp/cat/objects/run_time_options_output.py +++ b/keywords/ptp/cat/objects/run_time_options_output.py @@ -27,75 +27,73 @@ class RunTimeOptionsOutput: """ - def __init__(self, run_time_options_output: [str]): + def __init__(self, run_time_options_output: list[str]): """ - Constructor. - Create an internal RunTimeOptionsObject from the passed parameter. + Create an internal RunTimeOptionsObject from the passed parameter. + Args: run_time_options_output (list[str]): a list of strings representing the run time options output - """ cat_ptp_table_parser = CatPtpTableParser(run_time_options_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.run_time_options_object = RunTimeOptionsObject() - if 'assume_two_step' in output_values: - self.run_time_options_object.set_assume_two_step(int(output_values['assume_two_step'])) + if "assume_two_step" in output_values: + self.run_time_options_object.set_assume_two_step(int(output_values["assume_two_step"])) - if 'logging_level' in output_values: - self.run_time_options_object.set_logging_level(int(output_values['logging_level'])) + if "logging_level" in output_values: + self.run_time_options_object.set_logging_level(int(output_values["logging_level"])) - if 'path_trace_enabled' in output_values: - self.run_time_options_object.set_path_trace_enabled(int(output_values['path_trace_enabled'])) + if "path_trace_enabled" in output_values: + self.run_time_options_object.set_path_trace_enabled(int(output_values["path_trace_enabled"])) - if 'follow_up_info' in output_values: - self.run_time_options_object.set_follow_up_info(int(output_values['follow_up_info'])) + if "follow_up_info" in output_values: + self.run_time_options_object.set_follow_up_info(int(output_values["follow_up_info"])) - if 'hybrid_e2e' in output_values: - self.run_time_options_object.set_hybrid_e2e(int(output_values['hybrid_e2e'])) + if "hybrid_e2e" in output_values: + self.run_time_options_object.set_hybrid_e2e(int(output_values["hybrid_e2e"])) - if 'inhibit_multicast_service' in output_values: - self.run_time_options_object.set_inhibit_multicast_service(int(output_values['inhibit_multicast_service'])) + if "inhibit_multicast_service" in output_values: + self.run_time_options_object.set_inhibit_multicast_service(int(output_values["inhibit_multicast_service"])) - if 'net_sync_monitor' in output_values: - self.run_time_options_object.set_net_sync_monitor(int(output_values['net_sync_monitor'])) + if "net_sync_monitor" in output_values: + self.run_time_options_object.set_net_sync_monitor(int(output_values["net_sync_monitor"])) - if 'tc_spanning_tree' in output_values: - self.run_time_options_object.set_tc_spanning_tree(int(output_values['tc_spanning_tree'])) + if "tc_spanning_tree" in output_values: + self.run_time_options_object.set_tc_spanning_tree(int(output_values["tc_spanning_tree"])) - if 'tx_timestamp_timeout' in output_values: - self.run_time_options_object.set_tx_timestamp_timeout(int(output_values['tx_timestamp_timeout'])) + if "tx_timestamp_timeout" in output_values: + self.run_time_options_object.set_tx_timestamp_timeout(int(output_values["tx_timestamp_timeout"])) - if 'unicast_listen' in output_values: - self.run_time_options_object.set_unicast_listen(int(output_values['unicast_listen'])) + if "unicast_listen" in output_values: + self.run_time_options_object.set_unicast_listen(int(output_values["unicast_listen"])) - if 'unicast_master_table' in output_values: - self.run_time_options_object.set_unicast_master_table(int(output_values['unicast_master_table'])) + if "unicast_master_table" in output_values: + self.run_time_options_object.set_unicast_master_table(int(output_values["unicast_master_table"])) - if 'unicast_req_duration' in output_values: - self.run_time_options_object.set_unicast_req_duration(int(output_values['unicast_req_duration'])) + if "unicast_req_duration" in output_values: + self.run_time_options_object.set_unicast_req_duration(int(output_values["unicast_req_duration"])) - if 'use_syslog' in output_values: - self.run_time_options_object.set_use_syslog(int(output_values['use_syslog'])) + if "use_syslog" in output_values: + self.run_time_options_object.set_use_syslog(int(output_values["use_syslog"])) - if 'verbose' in output_values: - self.run_time_options_object.set_verbose(int(output_values['verbose'])) + if "verbose" in output_values: + self.run_time_options_object.set_verbose(int(output_values["verbose"])) - if 'summary_interval' in output_values: - self.run_time_options_object.set_summary_interval(int(output_values['summary_interval'])) + if "summary_interval" in output_values: + self.run_time_options_object.set_summary_interval(int(output_values["summary_interval"])) - if 'kernel_leap' in output_values: - self.run_time_options_object.set_kernel_leap(int(output_values['kernel_leap'])) + if "kernel_leap" in output_values: + self.run_time_options_object.set_kernel_leap(int(output_values["kernel_leap"])) - if 'check_fup_sync' in output_values: - self.run_time_options_object.set_check_fup_sync(int(output_values['check_fup_sync'])) + if "check_fup_sync" in output_values: + self.run_time_options_object.set_check_fup_sync(int(output_values["check_fup_sync"])) def get_run_time_options_object(self) -> RunTimeOptionsObject: """ Getter for run_time_options_object object. Returns: - A RunTimeOptionsObject - + RunTimeOptionsObject: The run time options object containing parsed values. """ return self.run_time_options_object diff --git a/keywords/ptp/cat/objects/servo_options_object.py b/keywords/ptp/cat/objects/servo_options_object.py index 1c3e5652..03295dbf 100644 --- a/keywords/ptp/cat/objects/servo_options_object.py +++ b/keywords/ptp/cat/objects/servo_options_object.py @@ -1,7 +1,5 @@ class ServoOptionsObject: - """ - Object to hold the values of Servo Options Options - """ + """Object to hold the values of Servo Options Options""" def __init__(self): self.pi_proportional_const: float = None @@ -15,7 +13,7 @@ class ServoOptionsObject: self.step_threshold: float = None self.first_step_threshold: float = None self.max_frequency: int = None - self.clock_servo: str = '' + self.clock_servo: str = "" self.sanity_freq_limit: int = -1 self.ntpshm_segment: int = -1 self.msg_interval_request: int = -1 @@ -26,348 +24,323 @@ class ServoOptionsObject: def get_pi_proportional_const(self) -> float: """ Getter for pi_proportional - Returns: pi_proportional + Returns: + float: the pi_proportional_const value """ return self.pi_proportional_const - def set_pi_proportional_const(self, pi_proportional_const: float): + def set_pi_proportional_const(self, pi_proportional_const: float) -> None: """ Setter for pi_proportional_const + Args: - pi_proportional_const (): the pi_proportional_const value - - Returns: - + pi_proportional_const (float): the pi_proportional_const value """ self.pi_proportional_const = pi_proportional_const def get_pi_integral_const(self) -> float: """ Getter for pi_integral_const - Returns: pi_integral_const value + Returns: + float: the pi_integral_const value """ return self.pi_integral_const - def set_pi_integral_const(self, pi_integral_const: float): + def set_pi_integral_const(self, pi_integral_const: float) -> None: """ Setter for pi_integral_const + Args: - pi_integral_const (): pi_integral_const value - - Returns: - + pi_integral_const (float): pi_integral_const value """ self.pi_integral_const = pi_integral_const def get_pi_proportional_scale(self) -> float: """ Getter for pi_proportional_scale - Returns: pi_proportional_scale value + Returns: + float: the pi_proportional_scale value """ return self.pi_proportional_scale - def set_pi_proportional_scale(self, pi_proportional_scale: float): + def set_pi_proportional_scale(self, pi_proportional_scale: float) -> None: """ Setter for pi_proportional_scale + Args: - pi_proportional_scale (): pi_proportional_scale value - - Returns: - + pi_proportional_scale (float): pi_proportional_scale value """ self.pi_proportional_scale = pi_proportional_scale def get_pi_proportional_exponent(self) -> float: """ Getter for pi_proportional_exponent - Returns: the pi_proportional_exponent value + Returns: + float: the pi_proportional_exponent value """ return self.pi_proportional_exponent - def set_pi_proportional_exponent(self, pi_proportional_exponent: float): + def set_pi_proportional_exponent(self, pi_proportional_exponent: float) -> None: """ Setter for pi_proportional_exponent + Args: - pi_proportional_exponent (): the pi_proportional_exponent value - - Returns: - + pi_proportional_exponent (float): the pi_proportional_exponent value """ self.pi_proportional_exponent = pi_proportional_exponent def get_pi_proportional_norm_max(self) -> float: """ Getter for pi_proportional_norm_max - Returns: pi_proportional_norm_max value + Returns: + float: the pi_proportional_norm_max value """ return self.pi_proportional_norm_max - def set_pi_proportional_norm_max(self, pi_proportional_norm_max: float): + def set_pi_proportional_norm_max(self, pi_proportional_norm_max: float) -> None: """ Setter for pi_proportional_norm_max + Args: - pi_proportional_norm_max (): the pi_proportional_norm_max value - - Returns: - + pi_proportional_norm_max (float): the pi_proportional_norm_max value """ self.pi_proportional_norm_max = pi_proportional_norm_max def get_pi_integral_scale(self) -> float: """ Getter for pi_integral_scale - Returns: pi_integral_scale value + Returns: + float: the pi_integral_scale value """ return self.pi_integral_scale - def set_pi_integral_scale(self, pi_integral_scale: float): + def set_pi_integral_scale(self, pi_integral_scale: float) -> None: """ Setter for pi_integral_scale + Args: - pi_integral_scale (): the pi_integral_scale value - - Returns: - + pi_integral_scale (float): the pi_integral_scale value """ self.pi_integral_scale = pi_integral_scale def get_pi_integral_exponent(self) -> float: """ Getter for pi_integral_exponent - Returns: pi_integral_exponent value + Returns: + float: pi_integral_exponent value """ return self.pi_integral_exponent - def set_pi_integral_exponent(self, pi_integral_exponent: float): + def set_pi_integral_exponent(self, pi_integral_exponent: float) -> None: """ Setter for pi_integral_exponent + Args: - pi_integral_exponent (): the pi_integral_exponent value - - Returns: - + pi_integral_exponent (float): the pi_integral_exponent value """ self.pi_integral_exponent = pi_integral_exponent def get_pi_integral_norm_max(self) -> float: """ Getter for pi_integral_norm_max - Returns: pi_integral_norm_max value + Returns: + float: the pi_integral_norm_max value """ return self.pi_integral_norm_max - def set_pi_integral_norm_max(self, pi_integral_norm_max: float): + def set_pi_integral_norm_max(self, pi_integral_norm_max: float) -> None: """ Setter for pi_integral_norm_max + Args: - pi_integral_norm_max (): the pi_integral_norm_max value - - Returns: - + pi_integral_norm_max (float): the pi_integral_norm_max value """ self.pi_integral_norm_max = pi_integral_norm_max def get_step_threshold(self) -> float: """ Getter for step_threshold - Returns: the step_threshold value + Returns: + float: the step_threshold value """ return self.step_threshold - def set_step_threshold(self, step_threshold: float): + def set_step_threshold(self, step_threshold: float) -> None: """ Setter for step_threshold + Args: - step_threshold (): the step_threshold value - - Returns: - + step_threshold (float): the step_threshold value """ self.step_threshold = step_threshold def get_first_step_threshold(self) -> float: """ Getter for first_step_threshold - Returns: the first_step_threshold value + Returns: + float: the first_step_threshold value """ return self.first_step_threshold - def set_first_step_threshold(self, first_step_threshold: float): + def set_first_step_threshold(self, first_step_threshold: float) -> None: """ Setter for first_step_threshold + Args: - first_step_threshold (): the first_step_threshold value - - Returns: - + first_step_threshold (float): the first_step_threshold value """ self.first_step_threshold = first_step_threshold def get_max_frequency(self) -> int: """ Getter for max_frequency - Returns: the max_frequency value + Returns: + int: the max_frequency value """ return self.max_frequency - def set_max_frequency(self, max_frequency: int): + def set_max_frequency(self, max_frequency: int) -> None: """ Setter for max_frequency + Args: - max_frequency (): the max_frequency value - - Returns: - + max_frequency (int): the max_frequency value """ self.max_frequency = max_frequency def get_clock_servo(self) -> str: """ Getter for clock_servo - Returns: the clock_servo value + Returns: + str: the clock_servo value """ return self.clock_servo - def set_clock_servo(self, clock_servo: str): + def set_clock_servo(self, clock_servo: str) -> None: """ Setter for clock_servo + Args: - clock_servo (): the clock_servo value - - Returns: - + clock_servo (str): the clock_servo value """ self.clock_servo = clock_servo def get_sanity_freq_limit(self) -> int: """ Getter for sanity_freq_limit - Returns: the sanity_freq_limit value + Returns: + int: the sanity_freq_limit value """ return self.sanity_freq_limit - def set_sanity_freq_limit(self, sanity_freq_limit: int): + def set_sanity_freq_limit(self, sanity_freq_limit: int) -> None: """ Setter for sanity_freq_limit + Args: - sanity_freq_limit (): the sanity_freq_limit value - - Returns: - + sanity_freq_limit (int): the sanity_freq_limit value """ self.sanity_freq_limit = sanity_freq_limit def get_ntpshm_segment(self) -> int: """ Getter for ntpshm_segment - Returns: the ntpshm_segment value + Returns: + int: the ntpshm_segment value """ return self.ntpshm_segment - def set_ntpshm_segment(self, ntpshm_segment: int): + def set_ntpshm_segment(self, ntpshm_segment: int) -> None: """ Setter for ntpshm_segment + Args: - ntpshm_segment (): the ntpshm_segment value - - Returns: - + ntpshm_segment (int): the ntpshm_segment value """ self.ntpshm_segment = ntpshm_segment def get_msg_interval_request(self) -> int: """ Getter for msg_interval_request - Returns: the msg_interval_request value + Returns: + int: the msg_interval_request value """ return self.msg_interval_request - def set_msg_interval_request(self, msg_interval_request: int): + def set_msg_interval_request(self, msg_interval_request: int) -> None: """ Setter for msg_interval_request + Args: - msg_interval_request (): the msg_interval_request value - - Returns: - + msg_interval_request (int): the msg_interval_request value """ self.msg_interval_request = msg_interval_request def get_servo_num_offset_values(self) -> int: """ Getter for servo_num_offset_values - Returns: the servo_num_offset_values value + Returns: + int: the servo_num_offset_values value """ return self.servo_num_offset_values - def set_servo_num_offset_values(self, servo_num_offset_values: int): + def set_servo_num_offset_values(self, servo_num_offset_values: int) -> None: """ Setter for servo_num_offset_values + Args: - servo_num_offset_values (): the servo_num_offset_values value - - Returns: - + servo_num_offset_values (int): the servo_num_offset_values value """ self.servo_num_offset_values = servo_num_offset_values def get_servo_offset_threshold(self) -> int: """ Getter for servo_offset_threshold - Returns: the servo_offset_threshold value + Returns: + int: the servo_offset_threshold value """ return self.servo_offset_threshold - def set_servo_offset_threshold(self, servo_offset_threshold: int): + def set_servo_offset_threshold(self, servo_offset_threshold: int) -> None: """ Setter for servo_offset_threshold + Args: - servo_offset_threshold (): the servo_offset_threshold value - - Returns: - + servo_offset_threshold (int): the servo_offset_threshold value """ self.servo_offset_threshold = servo_offset_threshold def get_write_phase_mode(self) -> int: """ Getter for write_phase_mode - Returns: the write_phase_mode value + Returns: + int: the write_phase_mode value """ return self.write_phase_mode - def set_write_phase_mode(self, write_phase_mode: int): + def set_write_phase_mode(self, write_phase_mode: int) -> None: """ Setter for write_phase_mode + Args: - write_phase_mode (): the write_phase_mode value - - Returns: - + write_phase_mode (int): the write_phase_mode value """ self.write_phase_mode = write_phase_mode - - - - - - - diff --git a/keywords/ptp/cat/objects/servo_options_output.py b/keywords/ptp/cat/objects/servo_options_output.py index df38bd2d..3f27bea6 100644 --- a/keywords/ptp/cat/objects/servo_options_output.py +++ b/keywords/ptp/cat/objects/servo_options_output.py @@ -28,78 +28,76 @@ class ServoOptionsOutput: """ - def __init__(self, servo_options_output: [str]): + def __init__(self, servo_options_output: list[str]): """ - Constructor. - Create an internal ServoOptionsObject from the passed parameter. + Create an internal ServoOptionsObject from the passed parameter. + Args: servo_options_output (list[str]): a list of strings representing the servo options output - """ cat_ptp_table_parser = CatPtpTableParser(servo_options_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.servo_options_object = ServoOptionsObject() - if 'pi_proportional_const' in output_values: - self.servo_options_object.set_pi_proportional_const(float(output_values['pi_proportional_const'])) + if "pi_proportional_const" in output_values: + self.servo_options_object.set_pi_proportional_const(float(output_values["pi_proportional_const"])) - if 'pi_integral_const' in output_values: - self.servo_options_object.set_pi_integral_const(float(output_values['pi_integral_const'])) + if "pi_integral_const" in output_values: + self.servo_options_object.set_pi_integral_const(float(output_values["pi_integral_const"])) - if 'pi_proportional_scale' in output_values: - self.servo_options_object.set_pi_proportional_scale(float(output_values['pi_proportional_scale'])) + if "pi_proportional_scale" in output_values: + self.servo_options_object.set_pi_proportional_scale(float(output_values["pi_proportional_scale"])) - if 'pi_proportional_exponent' in output_values: - self.servo_options_object.set_pi_proportional_exponent(float(output_values['pi_proportional_exponent'])) + if "pi_proportional_exponent" in output_values: + self.servo_options_object.set_pi_proportional_exponent(float(output_values["pi_proportional_exponent"])) - if 'pi_proportional_norm_max' in output_values: - self.servo_options_object.set_pi_proportional_norm_max(float(output_values['pi_proportional_norm_max'])) + if "pi_proportional_norm_max" in output_values: + self.servo_options_object.set_pi_proportional_norm_max(float(output_values["pi_proportional_norm_max"])) - if 'pi_integral_scale' in output_values: - self.servo_options_object.set_pi_integral_scale(float(output_values['pi_integral_scale'])) + if "pi_integral_scale" in output_values: + self.servo_options_object.set_pi_integral_scale(float(output_values["pi_integral_scale"])) - if 'pi_integral_exponent' in output_values: - self.servo_options_object.set_pi_integral_exponent(float(output_values['pi_integral_exponent'])) + if "pi_integral_exponent" in output_values: + self.servo_options_object.set_pi_integral_exponent(float(output_values["pi_integral_exponent"])) - if 'pi_integral_norm_max' in output_values: - self.servo_options_object.set_pi_integral_norm_max(float(output_values['pi_integral_norm_max'])) + if "pi_integral_norm_max" in output_values: + self.servo_options_object.set_pi_integral_norm_max(float(output_values["pi_integral_norm_max"])) - if 'step_threshold' in output_values: - self.servo_options_object.set_step_threshold(float(output_values['step_threshold'])) + if "step_threshold" in output_values: + self.servo_options_object.set_step_threshold(float(output_values["step_threshold"])) - if 'first_step_threshold' in output_values: - self.servo_options_object.set_first_step_threshold(float(output_values['first_step_threshold'])) + if "first_step_threshold" in output_values: + self.servo_options_object.set_first_step_threshold(float(output_values["first_step_threshold"])) - if 'max_frequency' in output_values: - self.servo_options_object.set_max_frequency(int(output_values['max_frequency'])) + if "max_frequency" in output_values: + self.servo_options_object.set_max_frequency(int(output_values["max_frequency"])) - if 'clock_servo' in output_values: - self.servo_options_object.set_clock_servo(output_values['clock_servo']) + if "clock_servo" in output_values: + self.servo_options_object.set_clock_servo(output_values["clock_servo"]) - if 'sanity_freq_limit' in output_values: - self.servo_options_object.set_sanity_freq_limit(int(output_values['sanity_freq_limit'])) + if "sanity_freq_limit" in output_values: + self.servo_options_object.set_sanity_freq_limit(int(output_values["sanity_freq_limit"])) - if 'ntpshm_segment' in output_values: - self.servo_options_object.set_ntpshm_segment(int(output_values['ntpshm_segment'])) + if "ntpshm_segment" in output_values: + self.servo_options_object.set_ntpshm_segment(int(output_values["ntpshm_segment"])) - if 'msg_interval_request' in output_values: - self.servo_options_object.set_msg_interval_request(int(output_values['msg_interval_request'])) + if "msg_interval_request" in output_values: + self.servo_options_object.set_msg_interval_request(int(output_values["msg_interval_request"])) - if 'servo_num_offset_values' in output_values: - self.servo_options_object.set_servo_num_offset_values(int(output_values['servo_num_offset_values'])) + if "servo_num_offset_values" in output_values: + self.servo_options_object.set_servo_num_offset_values(int(output_values["servo_num_offset_values"])) - if 'servo_offset_threshold' in output_values: - self.servo_options_object.set_servo_offset_threshold(int(output_values['servo_offset_threshold'])) + if "servo_offset_threshold" in output_values: + self.servo_options_object.set_servo_offset_threshold(int(output_values["servo_offset_threshold"])) - if 'write_phase_mode' in output_values: - self.servo_options_object.set_write_phase_mode(int(output_values['write_phase_mode'])) + if "write_phase_mode" in output_values: + self.servo_options_object.set_write_phase_mode(int(output_values["write_phase_mode"])) def get_servo_options_object(self) -> ServoOptionsObject: """ Getter for servo_options_object object. Returns: - A ServoOptionsObject - + ServoOptionsObject: The servo options object containing parsed values. """ return self.servo_options_object diff --git a/keywords/ptp/cat/objects/transport_options_object.py b/keywords/ptp/cat/objects/transport_options_object.py index b2139ffe..f7fd1132 100644 --- a/keywords/ptp/cat/objects/transport_options_object.py +++ b/keywords/ptp/cat/objects/transport_options_object.py @@ -4,143 +4,136 @@ class TransportOptionsObject: """ def __init__(self): - self.transport_specific: str = '' - self.ptp_dst_mac: str = '' - self.p2p_dst_mac: str = '' + self.transport_specific: str = "" + self.ptp_dst_mac: str = "" + self.p2p_dst_mac: str = "" self.udp_ttl: int = -1 - self.udp6_scope: str = '' - self.uds_address: str = '' - self.uds_ro_address: str = '' + self.udp6_scope: str = "" + self.uds_address: str = "" + self.uds_ro_address: str = "" def get_transport_specific(self) -> str: """ Getter for transport_specific - Returns: transport_specific + Returns: + str: the transport_specific value """ return self.transport_specific - def set_transport_specific(self, transport_specific: str): + def set_transport_specific(self, transport_specific: str) -> None: """ Setter for transport_specific + Args: - transport_specific (): the transport_specific value - - Returns: - + transport_specific (str): the transport_specific value """ self.transport_specific = transport_specific def get_ptp_dst_mac(self) -> str: """ Getter for ptp_dst_mac - Returns: ptp_dst_mac value + Returns: + str: the ptp_dst_mac value """ return self.ptp_dst_mac - def set_ptp_dst_mac(self, ptp_dst_mac: str): + def set_ptp_dst_mac(self, ptp_dst_mac: str) -> None: """ Setter for ptp_dst_mac + Args: - ptp_dst_mac (): ptp_dst_mac value - - Returns: - + ptp_dst_mac (str): the ptp_dst_mac value """ self.ptp_dst_mac = ptp_dst_mac def get_p2p_dst_mac(self) -> str: """ Getter for p2p_dst_mac - Returns: p2p_dst_mac value + Returns: + str: the p2p_dst_mac value """ return self.p2p_dst_mac - def set_p2p_dst_mac(self, p2p_dst_mac: str): + def set_p2p_dst_mac(self, p2p_dst_mac: str) -> None: """ Setter for p2p_dst_mac + Args: - p2p_dst_mac (): p2p_dst_mac value - - Returns: - + p2p_dst_mac (str): p2p_dst_mac value """ self.p2p_dst_mac = p2p_dst_mac def get_udp_ttl(self) -> int: """ Getter for udp_ttl - Returns: the udp_ttl value + Returns: + int: the udp_ttl value """ return self.udp_ttl - def set_udp_ttl(self, udp_ttl: int): + def set_udp_ttl(self, udp_ttl: int) -> None: """ Setter for udp_ttl + Args: - udp_ttl (): the udp_ttl value - - Returns: - + udp_ttl (int): the udp_ttl value """ self.udp_ttl = udp_ttl def get_udp6_scope(self) -> str: """ Getter for udp6_scope - Returns: udp6_scope value + Returns: + str: the udp6_scope value """ return self.udp6_scope - def set_udp6_scope(self, udp6_scope: str): + def set_udp6_scope(self, udp6_scope: str) -> None: """ Setter for udp6_scope + Args: - udp6_scope (): the udp6_scope value - - Returns: - + udp6_scope (str): the udp6_scope value """ self.udp6_scope = udp6_scope def get_uds_address(self) -> str: """ Getter for uds_address - Returns: uds_address value + Returns: + str: uds_address value """ return self.uds_address - def set_uds_address(self, uds_address: str): + def set_uds_address(self, uds_address: str) -> None: """ Setter for uds_address + Args: - uds_address (): the uds_address value - - Returns: - + uds_address (str): the uds_address value """ self.uds_address = uds_address def get_uds_ro_address(self) -> str: """ Getter for uds_ro_address - Returns: uds_ro_address value + Returns: + str: uds_ro_address value """ return self.uds_ro_address - def set_uds_ro_address(self, uds_ro_address: str): + def set_uds_ro_address(self, uds_ro_address: str) -> None: """ Setter for uds_ro_address + Args: - uds_ro_address (): the uds_ro_address value - - Returns: - + uds_ro_address (str): the uds_ro_address value """ self.uds_ro_address = uds_ro_address diff --git a/keywords/ptp/cat/objects/transport_options_output.py b/keywords/ptp/cat/objects/transport_options_output.py index 2b0af4af..457e6ebc 100644 --- a/keywords/ptp/cat/objects/transport_options_output.py +++ b/keywords/ptp/cat/objects/transport_options_output.py @@ -17,45 +17,43 @@ class TransportOptionsOutput: """ - def __init__(self, transport_options_output: [str]): + def __init__(self, transport_options_output: list[str]): """ - Constructor. - Create an internal ServoOptionsObject from the passed parameter. + Create an internal ServoOptionsObject from the passed parameter. + Args: transport_options_output (list[str]): a list of strings representing the servo options output - """ cat_ptp_table_parser = CatPtpTableParser(transport_options_output) output_values = cat_ptp_table_parser.get_output_values_dict() self.transport_options_object = TransportOptionsObject() - if 'transportSpecific' in output_values: - self.transport_options_object.set_transport_specific(output_values['transportSpecific']) + if "transportSpecific" in output_values: + self.transport_options_object.set_transport_specific(output_values["transportSpecific"]) - if 'ptp_dst_mac' in output_values: - self.transport_options_object.set_ptp_dst_mac(output_values['ptp_dst_mac']) + if "ptp_dst_mac" in output_values: + self.transport_options_object.set_ptp_dst_mac(output_values["ptp_dst_mac"]) - if 'p2p_dst_mac' in output_values: - self.transport_options_object.set_p2p_dst_mac(output_values['p2p_dst_mac']) + if "p2p_dst_mac" in output_values: + self.transport_options_object.set_p2p_dst_mac(output_values["p2p_dst_mac"]) - if 'udp_ttl' in output_values: - self.transport_options_object.set_udp_ttl(int(output_values['udp_ttl'])) + if "udp_ttl" in output_values: + self.transport_options_object.set_udp_ttl(int(output_values["udp_ttl"])) - if 'udp6_scope' in output_values: - self.transport_options_object.set_udp6_scope(output_values['udp6_scope']) + if "udp6_scope" in output_values: + self.transport_options_object.set_udp6_scope(output_values["udp6_scope"]) - if 'uds_address' in output_values: - self.transport_options_object.set_uds_address(output_values['uds_address']) + if "uds_address" in output_values: + self.transport_options_object.set_uds_address(output_values["uds_address"]) - if 'uds_ro_address' in output_values: - self.transport_options_object.set_uds_ro_address(output_values['uds_ro_address']) + if "uds_ro_address" in output_values: + self.transport_options_object.set_uds_ro_address(output_values["uds_ro_address"]) def get_transport_options_object(self) -> TransportOptionsObject: """ Getter for TransportOptionsObject object. Returns: - A TransportOptionsObject - + TransportOptionsObject: The transport options object containing parsed values. """ return self.transport_options_object diff --git a/keywords/ptp/phc_ctl_keywords.py b/keywords/ptp/phc_ctl_keywords.py index ceed58e4..0e0210d6 100644 --- a/keywords/ptp/phc_ctl_keywords.py +++ b/keywords/ptp/phc_ctl_keywords.py @@ -4,6 +4,7 @@ from typing import List from config.configuration_manager import ConfigurationManager from framework.ssh.prompt_response import PromptResponse +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword from keywords.cloud_platform.fault_management.alarms.alarm_list_keywords import AlarmListKeywords from keywords.cloud_platform.fault_management.alarms.objects.alarm_list_object import AlarmListObject @@ -11,19 +12,14 @@ from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKey class PhcCtlKeywords(BaseKeyword): - """ - Directly control PHC device clock using given SSH connection. + """Directly control PHC device clock using given SSH connection.""" - Attributes: - ssh_connection: An instance of an SSH connection. - """ - - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ Initializes the PhcCtlKeywords with an SSH connection. Args: - ssh_connection: An instance of an SSH connection. + ssh_connection (SSHConnection): An instance of an SSH connection. """ self.ssh_connection = ssh_connection @@ -120,9 +116,7 @@ class PhcCtlKeywords(BaseKeyword): def wait_for_phc_ctl_adjustment_alarm(self, interface: str, alarms: List[AlarmListObject], timeout: int = 120, polling_interval: int = 10) -> None: """ - Run a remote phc_ctl adjustment loop on the controller as root, - and stop it once the specified PTP alarm(s) are detected or - a timeout occurs. + Run a remote phc_ctl adjustment loop on the controller as root, and stop it once the specified PTP alarm(s) are detected or a timeout occurs. Args: interface (str): The interface to apply phc_ctl adjustments to. @@ -130,8 +124,6 @@ class PhcCtlKeywords(BaseKeyword): timeout (int): Maximum wait time in seconds (default: 120). polling_interval (int): Interval in seconds between polling attempts (default: 10). - Returns: None - Raises: TimeoutError: If the expected alarms are not observed within the timeout period. """ @@ -145,9 +137,6 @@ class PhcCtlKeywords(BaseKeyword): Args: command (str): The shell command to be executed with root privileges. - - Returns: - None """ root_prompt = PromptResponse("#", command) self.ssh_connection.send_expect_prompts("sudo su", [password_prompt, root_prompt])