starlingx/test
Code Issues Proposed changes

Merge "pylint fix 01"

Browse Source
This commit is contained in:
Zuul
2025-07-30 16:25:41 +00:00
committed by Gerrit Code Review
parent ab511ed620 65d935a2b8
commit 3bebef55a9
29 changed files with 793 additions and 910 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
keywords/linux/ip/object/interface_object.py
View File

@@ -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

18
keywords/linux/ip/object/ip_br_addr_output.py
View File

@@ -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

14
keywords/linux/ip/object/ip_link_show_output.py
View File

@@ -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: <BROADCAST,MULTICAST,UP,LOWER_UP> 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<interface_name>\S+):\s+<(?P<mode>\S+)>' r'\s+mtu\s+(?P<mtu>\d+)[\s\S]+state\s+(?P<state>\w+)'
regex = r"\d+:\s+(?P<interface_name>\S+):\s+<(?P<mode>\S+)>" r"\s+mtu\s+(?P<mtu>\d+)[\s\S]+state\s+(?P<state>\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)

48
keywords/linux/ip/object/ip_object.py
View File

@@ -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 /<prefix length>.
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 /<prefix length>.
"""
# 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

5
keywords/linux/keyring/keyring_keywords.py
View File

@@ -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()
return keyring_value[0].strip()

15
keywords/linux/process_status/process_status_args_keywords.py
View File

@@ -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

17
keywords/linux/systemctl/systemctl_is_active_keywords.py
View File

@@ -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 <service_name>"
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"

66
keywords/openstack/openstack/stack/object/openstack_stack_list_output.py
View File

@@ -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

48
keywords/openstack/openstack/stack/object/openstack_stack_output.py
View File

@@ -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

57
keywords/openstack/openstack/stack/openstack_stack_create_keywords.py
View File

@@ -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

28
keywords/openstack/openstack/stack/openstack_stack_delete_keywords.py
View File

@@ -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

28
keywords/openstack/openstack/stack/openstack_stack_list_keywords.py
View File

@@ -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)

51
keywords/openstack/openstack/stack/openstack_stack_update_keywords.py
View File

@@ -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

9
keywords/ptp/cat/cat_ptp_cgu_keywords.py
View File

@@ -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)

22
keywords/ptp/cat/cat_ptp_config_keywords.py
View File

@@ -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

66
keywords/ptp/cat/objects/clock_description_object.py
View File

@@ -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

30
keywords/ptp/cat/objects/clock_description_output.py
View File

@@ -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

122
keywords/ptp/cat/objects/default_interface_options_object.py
View File

@@ -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

50
keywords/ptp/cat/objects/default_interface_options_output.py
View File

@@ -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

202
keywords/ptp/cat/objects/port_data_set_object.py
View File

@@ -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

78
keywords/ptp/cat/objects/port_data_set_output.py
View File

@@ -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

3
keywords/ptp/cat/objects/ptp_cgu_component_output.py
View File

@@ -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

195
keywords/ptp/cat/objects/run_time_options_object.py
View File

@@ -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

78
keywords/ptp/cat/objects/run_time_options_output.py
View File

@@ -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

211
keywords/ptp/cat/objects/servo_options_object.py
View File

@@ -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

82
keywords/ptp/cat/objects/servo_options_output.py
View File

@@ -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

89
keywords/ptp/cat/objects/transport_options_object.py
View File

@@ -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

38
keywords/ptp/cat/objects/transport_options_output.py
View File

@@ -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

21
keywords/ptp/phc_ctl_keywords.py
View File

@@ -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])