starlingx/test
Code Issues Proposed changes

Merge "Added pmc, phc_ctl and ip link set keywords"

Browse Source
This commit is contained in:
Zuul
2025-02-21 16:27:07 +00:00
committed by Gerrit Code Review
parent 49fa8730eb 3758f1758b
commit d0c1fc2904
13 changed files with 1159 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
keywords/linux/ip/ip_keywords.py
View File

@@ -116,3 +116,16 @@ class IPKeywords(BaseKeyword):
ip_br_addr_output = IPBrAddrOutput(output)
return ip_br_addr_output
def set_ip_port_state(self, port: str, state: str = 'up'):
"""
sets a ip specific port state UP or DOWN as specified via ip link cmd
Args:
port : port to set
state: state to set port (up or down)
Returns:
"""
self.ssh_connection.send_as_sudo(f"ip link set dev {port} {state}")
self.validate_success_return_code(self.ssh_connection)

106
keywords/ptp/phc_ctl_keywords.py Normal file
View File

@@ -0,0 +1,106 @@
from keywords.base_keyword import BaseKeyword
class PhcCtlKeywords(BaseKeyword):
"""
Directly control PHC device clock using given SSH connection.
Attributes:
ssh_connection: An instance of an SSH connection.
"""
def __init__(self, ssh_connection):
"""
Initializes the PhcCtlKeywords with an SSH connection.
Args:
ssh_connection: An instance of an SSH connection.
"""
self.ssh_connection = ssh_connection
def phc_ctl_get(self, device: str) -> str:
"""
Get the current time of the PHC clock device
Args:
device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl.
Example:
phc_ctl[643764.828]: clock time is 1739856255.215802036 or Tue Feb 18 05:24:15 2025
Returns:
"""
output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} get")
self.validate_success_return_code(self.ssh_connection)
output_str = ''.join(output).replace('\n', '')
if output_str and len(output_str.split()) > 4:
return output_str.split()[4]
else:
raise "output_str.split() is expected to be a List with four elements."
def phc_ctl_cmp(self, device: str) -> str:
"""
Compare the PHC clock device to CLOCK_REALTIME
Args:
device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl.
Example:
phc_ctl[645639.878]: offset from CLOCK_REALTIME is -37000000008ns
Returns:
"""
output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} cmp")
self.validate_success_return_code(self.ssh_connection)
output_str = ''.join(output)
if output_str and len(output_str.split()) > 5:
return output_str.split()[5]
else:
raise "output_str.split() is expected to be a List with five elements."
def phc_ctl_adj(self, device: str, seconds: str) -> str:
"""
Adjust the PHC clock by an amount of seconds provided
Args:
device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl.
seconds :
Example:
phc_ctl[646368.470]: adjusted clock by 0.000001 seconds
Returns:
"""
output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} adj {seconds}")
self.validate_success_return_code(self.ssh_connection)
output_str = ''.join(output).replace('\n', '')
if output_str and len(output_str.split()) > 5:
return output_str.split()[4]
else:
raise "output_str.split() is expected to be a List with five elements."
def phc_ctl_set(self, device: str, seconds: str = None) -> str:
"""
Set the PHC clock time to the value specified in seconds. Defaults to reading CLOCK_REALTIME if no value is provided.
Args:
device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl.
seconds :
Example :
phc_ctl[647759.416]: set clock time to 1739860212.789318498 or Tue Feb 18 06:30:12 2025
Returns:
"""
if seconds:
cmd = f"phc_ctl {device} set {seconds}"
else:
cmd = f"phc_ctl {device} set"
output = self.ssh_connection.send_as_sudo(cmd)
self.validate_success_return_code(self.ssh_connection)
output_str = ''.join(output).replace('\n', '')
if output_str and len(output_str.split()) > 5:
return output_str.split()[5]
else:
raise "output_str.split() is expected to be a List with five elements."

66
keywords/ptp/pmc/objects/pmc_get_current_data_set_object.py Normal file
View File

@@ -0,0 +1,66 @@
class PMCGetCurrentDataSetObject:
"""
Object to hold the values of Current data set
"""
def __init__(self):
self.steps_removed: int = -1
self.offset_from_master: float = 0.0
self.mean_path_delay: float = 0.0
def get_steps_removed(self) -> int:
"""
Getter for steps_removed
Returns: the steps_removed value
"""
return self.steps_removed
def set_steps_removed(self, steps_removed: int):
"""
Setter for steps_removed
Args:
steps_removed (): the steps_removed value
Returns:
"""
self.steps_removed = steps_removed
def get_offset_from_master(self) -> float:
"""
Getter for offset_from_master
Returns: offset_from_master
"""
return self.offset_from_master
def set_offset_from_master(self, offset_from_master: float):
"""
Setter for offset_from_master
Args:
offset_from_master (): the offset_from_master value
Returns:
"""
self.offset_from_master = offset_from_master
def get_mean_path_delay(self) -> float:
"""
Getter for mean_path_delay
Returns: mean_path_delay value
"""
return self.mean_path_delay
def set_mean_path_delay(self, mean_path_delay: float):
"""
Setter for mean_path_delay
Args:
mean_path_delay (): mean_path_delay value
Returns:
"""
self.mean_path_delay = mean_path_delay

46
keywords/ptp/pmc/objects/pmc_get_current_data_set_output.py Normal file
View File

@@ -0,0 +1,46 @@
from keywords.ptp.pmc.objects.pmc_get_current_data_set_object import PMCGetCurrentDataSetObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetCurrentDataSetOutput:
"""
This class parses the output of commands such as 'pmc CURRENT_DATA_SET'
Example:
sending: GET CURRENT_DATA_SET
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT CURRENT_DATA_SET
stepsRemoved 0
offsetFromMaster 0.0
meanPathDelay 0.0
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal PMCGetCurrentDataSet from the passed parameter.
Args:
pmc_output (list[str]): a list of strings representing the output of the pmc command
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_values = pmc_table_parser.get_output_values_dict()
self.pmc_get_current_data_set_object = PMCGetCurrentDataSetObject()
if 'stepsRemoved' in output_values:
self.pmc_get_current_data_set_object.set_steps_removed(int(output_values['stepsRemoved']))
if 'offsetFromMaster' in output_values:
self.pmc_get_current_data_set_object.set_offset_from_master(float(output_values['offsetFromMaster']))
if 'meanPathDelay' in output_values:
self.pmc_get_current_data_set_object.set_mean_path_delay(float(output_values['meanPathDelay']))
def get_pmc_get_current_data_set_object(self) -> PMCGetCurrentDataSetObject:
"""
Getter for pmc_get_current_data_set_object object.
Returns:
A PMCGetCurrentDataSetObject
"""
return self.pmc_get_current_data_set_object

29
keywords/ptp/pmc/objects/pmc_get_domain_object.py Normal file
View File

@@ -0,0 +1,29 @@
class PMCGetDomainObject:
"""
Object to hold the values of DOMAIN
"""
def __init__(self):
self.domain_number: int = -1
def get_domain_number(self) -> int:
"""
Getter for domain_number
Returns: the domain_number value
"""
return self.domain_number
def set_domain_number(self, domain_number: int):
"""
Setter for domain_number
Args:
domain_number : the domain_number value
Returns:
"""
self.domain_number = domain_number

39
keywords/ptp/pmc/objects/pmc_get_domain_output.py Normal file
View File

@@ -0,0 +1,39 @@
from keywords.ptp.pmc.objects.pmc_get_domain_object import PMCGetDomainObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetDomainOutput:
"""
This class parses the output of commands such as 'pmc GET DOMAIN'
Example:
sending: GET DOMAIN
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT DOMAIN
domainNumber 24
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal DOMAIN from the passed parameter.
Args:
pmc_output (list[str]): a list of strings representing the output of the pmc command
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_values = pmc_table_parser.get_output_values_dict()
self.pmc_get_domain_object = PMCGetDomainObject()
if 'domainNumber' in output_values:
self.pmc_get_domain_object.set_domain_number(int(output_values['domainNumber']))
def get_pmc_get_domain_object(self) -> PMCGetDomainObject:
"""
Getter for pmc_get_domain_object object.
Returns:
A PMCGetDomainObject
"""
return self.pmc_get_domain_object

229
keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_object.py Normal file
View File

@@ -0,0 +1,229 @@
class PMCGetGrandmasterSettingsNpObject:
"""
Object to hold the values of GRANDMASTER_SETTINGS_NP
"""
def __init__(self):
self.clock_class: int = -1
self.clock_accuracy: str = ''
self.offset_scaled_log_variance: str = ''
self.current_utc_offset: int = -1
self.leap61: int = -1
self.leap59: int = -1
self.current_utc_off_set_valid: int = -1
self.ptp_time_scale: int = -1
self.time_traceable: int = -1
self.frequency_traceable: int = -1
self.time_source: str = ''
def get_clock_class(self) -> int:
"""
Getter for clock_class
Returns: the clock_class value
"""
return self.clock_class
def set_clock_class(self, clock_class: int):
"""
Setter for clock_class
Args:
clock_class : the clock_class value
Returns:
"""
self.clock_class = clock_class
def get_clock_accuracy(self) -> str:
"""
Getter for clock_accuracy
Returns: the clock_accuracy value
"""
return self.clock_accuracy
def set_clock_accuracy(self, clock_accuracy: str):
"""
Setter for clock_accuracy
Args:
clock_accuracy : the clock_accuracy value
Returns:
"""
self.clock_accuracy = clock_accuracy
def get_offset_scaled_log_variance(self) -> str:
"""
Getter for offset_scaled_log_variance
Returns: the offset_scaled_log_variance value
"""
return self.offset_scaled_log_variance
def set_offset_scaled_log_variance(self, offset_scaled_log_variance: str):
"""
Setter for offset_scaled_log_variance
Args:
offset_scaled_log_variance : the offset_scaled_log_variance value
Returns:
"""
self.offset_scaled_log_variance = offset_scaled_log_variance
def get_current_utc_offset(self) -> int:
"""
Getter for current_utc_offset
Returns: the current_utc_offset value
"""
return self.current_utc_offset
def set_current_utc_offset(self, current_utc_offset: int):
"""
Setter for current_utc_offset
Args:
current_utc_offset : the current_utc_offset value
Returns:
"""
self.current_utc_offset = current_utc_offset
def get_leap61(self) -> int:
"""
Getter for leap61
Returns: the leap61 value
"""
return self.leap61
def set_leap61(self, leap61: int):
"""
Setter for leap61
Args:
leap61 : the leap61 value
Returns:
"""
self.leap61 = leap61
def get_leap59(self) -> int:
"""
Getter for leap59
Returns: the leap59 value
"""
return self.leap59
def set_leap59(self, leap59: int):
"""
Setter for leap59
Args:
leap59 : the leap59 value
Returns:
"""
self.leap59 = leap59
def get_current_utc_off_set_valid(self) -> int:
"""
Getter for current_utc_off_set_valid
Returns: the current_utc_off_set_valid value
"""
return self.current_utc_off_set_valid
def set_current_utc_off_set_valid(self, current_utc_off_set_valid: int):
"""
Setter for current_utc_off_set_valid
Args:
current_utc_off_set_valid : the current_utc_off_set_valid value
Returns:
"""
self.current_utc_off_set_valid = current_utc_off_set_valid
def get_ptp_time_scale(self) -> int:
"""
Getter for ptp_time_scale
Returns: the ptp_time_scale value
"""
return self.ptp_time_scale
def set_ptp_time_scale(self, ptp_time_scale: int):
"""
Setter for ptp_time_scale
Args:
ptp_time_scale : the ptp_time_scale value
Returns:
"""
self.ptp_time_scale = ptp_time_scale
def get_time_traceable(self) -> int:
"""
Getter for time_traceable
Returns: the time_traceable value
"""
return self.time_traceable
def set_time_traceable(self, time_traceable: int):
"""
Setter for time_traceable
Args:
time_traceable : the time_traceable value
Returns:
"""
self.time_traceable = time_traceable
def get_frequency_traceable(self) -> int:
"""
Getter for frequency_traceable
Returns: the frequency_traceable value
"""
return self.frequency_traceable
def set_frequency_traceable(self, frequency_traceable: int):
"""
Setter for frequency_traceable
Args:
frequency_traceable : the frequency_traceable value
Returns:
"""
self.frequency_traceable = frequency_traceable
def get_time_source(self) -> str:
"""
Getter for time_source
Returns: the time_source value
"""
return self.time_source
def set_time_source(self, time_source: str):
"""
Setter for time_source
Args:
time_source : the time_source value
Returns:
"""
self.time_source = time_source

79
keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_output.py Normal file
View File

@@ -0,0 +1,79 @@
from keywords.ptp.pmc.objects.pmc_get_grandmaster_settings_np_object import PMCGetGrandmasterSettingsNpObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetGrandmasterSettingsNpOutput:
"""
This class parses the output of commands such as 'pmc GRANDMASTER_SETTINGS_NP'
Example:
sending: GET GRANDMASTER_SETTINGS_NP
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT GRANDMASTER_SETTINGS_NP
clockClass 6
clockAccuracy 0x20
offsetScaledLogVariance 0x4e5d
currentUtcOffset 37
leap61 0
leap59 0
currentUtcOffsetValid 0
ptpTimescale 1
timeTraceable 1
frequencyTraceable 1
timeSource 0x20
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal GRANDMASTER_SETTINGS_NP from the passed parameter.
Args:
pmc_output (list[str]): a list of strings representing the output of the pmc command
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_values = pmc_table_parser.get_output_values_dict()
self.pmc_get_grandmaster_settings_np_object = PMCGetGrandmasterSettingsNpObject()
if 'clockClass' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_clock_class(int(output_values['clockClass']))
if 'clockAccuracy' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_clock_accuracy(output_values['clockAccuracy'])
if 'offsetScaledLogVariance' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_offset_scaled_log_variance(output_values['offsetScaledLogVariance'])
if 'currentUtcOffset' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_current_utc_offset(int(output_values['currentUtcOffset']))
if 'leap61' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_leap61(int(output_values['leap61']))
if 'leap59' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_leap59(int(output_values['leap59']))
if 'currentUtcOffsetValid' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_current_utc_off_set_valid(int(output_values['currentUtcOffsetValid']))
if 'ptpTimescale' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_ptp_time_scale(int(output_values['ptpTimescale']))
if 'timeTraceable' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_time_traceable(int(output_values['timeTraceable']))
if 'frequencyTraceable' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_frequency_traceable(int(output_values['frequencyTraceable']))
if 'timeSource' in output_values:
self.pmc_get_grandmaster_settings_np_object.set_time_source(output_values['timeSource'])
def get_pmc_get_grandmaster_settings_np_object(self) -> PMCGetGrandmasterSettingsNpObject:
"""
Getter for pmc_get_grandmaster_settings_np_object object.
Returns:
A PMCGetGrandmasterSettingsNpObject
"""
return self.pmc_get_grandmaster_settings_np_object

169
keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_object.py Normal file
View File

@@ -0,0 +1,169 @@
class PMCGetTimePropertiesDataSetObject:
"""
Object to hold the values of TIME_PROPERTIES_DATA_SET
"""
def __init__(self):
self.current_utc_offset: int = -1
self.leap61: int = -1
self.leap59: int = -1
self.current_utc_off_set_valid: int = -1
self.ptp_time_scale: int = -1
self.time_traceable: int = -1
self.frequency_traceable: int = -1
self.time_source: str = ''
def get_current_utc_offset(self) -> int:
"""
Getter for current_utc_offset
Returns: the current_utc_offset value
"""
return self.current_utc_offset
def set_current_utc_offset(self, current_utc_offset: int):
"""
Setter for current_utc_offset
Args:
current_utc_offset : the current_utc_offset value
Returns:
"""
self.current_utc_offset = current_utc_offset
def get_leap61(self) -> int:
"""
Getter for leap61
Returns: the leap61 value
"""
return self.leap61
def set_leap61(self, leap61: int):
"""
Setter for leap61
Args:
leap61 : the leap61 value
Returns:
"""
self.leap61 = leap61
def get_leap59(self) -> int:
"""
Getter for leap59
Returns: the leap59 value
"""
return self.leap59
def set_leap59(self, leap59: int):
"""
Setter for leap59
Args:
leap59 : the leap59 value
Returns:
"""
self.leap59 = leap59
def get_current_utc_off_set_valid(self) -> int:
"""
Getter for current_utc_off_set_valid
Returns: the current_utc_off_set_valid value
"""
return self.current_utc_off_set_valid
def set_current_utc_off_set_valid(self, current_utc_off_set_valid: int):
"""
Setter for current_utc_off_set_valid
Args:
current_utc_off_set_valid : the current_utc_off_set_valid value
Returns:
"""
self.current_utc_off_set_valid = current_utc_off_set_valid
def get_ptp_time_scale(self) -> int:
"""
Getter for ptp_time_scale
Returns: the ptp_time_scale value
"""
return self.ptp_time_scale
def set_ptp_time_scale(self, ptp_time_scale: int):
"""
Setter for ptp_time_scale
Args:
ptp_time_scale : the ptp_time_scale value
Returns:
"""
self.ptp_time_scale = ptp_time_scale
def get_time_traceable(self) -> int:
"""
Getter for time_traceable
Returns: the time_traceable value
"""
return self.time_traceable
def set_time_traceable(self, time_traceable: int):
"""
Setter for time_traceable
Args:
time_traceable : the time_traceable value
Returns:
"""
self.time_traceable = time_traceable
def get_frequency_traceable(self) -> int:
"""
Getter for frequency_traceable
Returns: the frequency_traceable value
"""
return self.frequency_traceable
def set_frequency_traceable(self, frequency_traceable: int):
"""
Setter for frequency_traceable
Args:
frequency_traceable : the frequency_traceable value
Returns:
"""
self.frequency_traceable = frequency_traceable
def get_time_source(self) -> str:
"""
Getter for time_source
Returns: the time_source value
"""
return self.time_source
def set_time_source(self, time_source: str):
"""
Setter for time_source
Args:
time_source : the time_source value
Returns:
"""
self.time_source = time_source

67
keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_output.py Normal file
View File

@@ -0,0 +1,67 @@
from keywords.ptp.pmc.objects.pmc_get_time_properties_data_set_object import PMCGetTimePropertiesDataSetObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetTimePropertiesDataSetOutput:
"""
This class parses the output of commands such as 'pmc TIME_PROPERTIES_DATA_SET'
Example:
sending: GET TIME_PROPERTIES_DATA_SET
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT TIME_PROPERTIES_DATA_SET
currentUtcOffset 37
leap61 0
leap59 0
currentUtcOffsetValid 0
ptpTimescale 1
timeTraceable 1
frequencyTraceable 0
timeSource 0x20
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal TIME_PROPERTIES_DATA_SET from the passed parameter.
Args:
pmc_output (list[str]): a list of strings representing the output of the pmc command
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_values = pmc_table_parser.get_output_values_dict()
self.pmc_get_time_properties_data_set_object = PMCGetTimePropertiesDataSetObject()
if 'currentUtcOffset' in output_values:
self.pmc_get_time_properties_data_set_object.set_current_utc_offset(int(output_values['currentUtcOffset']))
if 'leap61' in output_values:
self.pmc_get_time_properties_data_set_object.set_leap61(int(output_values['leap61']))
if 'leap59' in output_values:
self.pmc_get_time_properties_data_set_object.set_leap59(int(output_values['leap59']))
if 'currentUtcOffsetValid' in output_values:
self.pmc_get_time_properties_data_set_object.set_current_utc_off_set_valid(int(output_values['currentUtcOffsetValid']))
if 'ptpTimescale' in output_values:
self.pmc_get_time_properties_data_set_object.set_ptp_time_scale(int(output_values['ptpTimescale']))
if 'timeTraceable' in output_values:
self.pmc_get_time_properties_data_set_object.set_time_traceable(int(output_values['timeTraceable']))
if 'frequencyTraceable' in output_values:
self.pmc_get_time_properties_data_set_object.set_frequency_traceable(int(output_values['frequencyTraceable']))
if 'timeSource' in output_values:
self.pmc_get_time_properties_data_set_object.set_time_source(output_values['timeSource'])
def get_pmc_get_time_properties_data_set_object(self) -> PMCGetTimePropertiesDataSetObject:
"""
Getter for pmc_get_time_properties_data_set_object object.
Returns:
A PMCGetTimePropertiesDataSetObject
"""
return self.pmc_get_time_properties_data_set_object

134
keywords/ptp/pmc/objects/pmc_get_time_status_np_object.py Normal file
View File

@@ -0,0 +1,134 @@
class PMCGetTimeStatusNpObject:
"""
Object to hold the values of TIME_STATUS_NP
"""
def __init__(self):
self.master_offset = -1
self.ingress_time = -1
self.cumulative_scaled_rate_offset = ''
self.scaled_last_gm_phase_change = -1
self.gm_time_base_indicator = -1
self.last_gm_phase_change = ''
self.gm_present = ''
self.gm_identity = ''
def get_master_offset(self) -> int:
"""Getter for master offset
Returns: the master offset
"""
return self.master_offset
def set_master_offset(self, master_offset : int):
"""Setter for master offset
Args:
master_offset : the master offset value
"""
self.master_offset = master_offset
def get_ingress_time(self) -> int:
"""Getter for ingress time
Returns: the ingress time
"""
return self.ingress_time
def set_ingress_time(self, ingress_time : int):
"""Setter for ingress time
Args:
ingress_time : the ingress time value
"""
self.ingress_time = ingress_time
def get_cumulative_scaled_rate_offset(self) -> str:
"""Getter for cumulative scaled rate offset
Returns: the cumulative scaled rate offset
"""
return self.cumulative_scaled_rate_offset
def set_cumulative_scaled_rate_offset(self, cumulative_scaled_rate_offset : str):
"""Setter for cumulative scaled rate offset
Args:
cumulative_scaled_rate_offset : the cumulative scaled rate offset value
"""
self.cumulative_scaled_rate_offset = cumulative_scaled_rate_offset
def get_scaled_last_gm_phase_change(self) -> int:
"""Getter for scaled last GM phase change
Returns: the scaled last GM phase change
"""
return self.scaled_last_gm_phase_change
def set_scaled_last_gm_phase_change(self, scaled_last_gm_phase_change : int):
"""Setter for scaled last GM phase change
Args:
scaled_last_gm_phase_change : the scaled last GM phase change value
"""
self.scaled_last_gm_phase_change = scaled_last_gm_phase_change
def get_gm_time_base_indicator(self) -> int:
"""Getter for GM time base indicator
Returns: the GM time base indicator value
"""
return self.gm_time_base_indicator
def set_gm_time_base_indicator(self, gm_time_base_indicator : int):
"""Setter for GM time base indicator
Args:
gm_time_base_indicator : the GM time base indicator value
"""
self.gm_time_base_indicator = gm_time_base_indicator
def get_last_gm_phase_change(self) -> str:
"""Getter for last GM phase change
Returns: the last GM phase change value
"""
return self.last_gm_phase_change
def set_last_gm_phase_change(self, last_gm_phase_change : str):
"""Setter for last GM phase change
Args:
last_gm_phase_change : the last GM phase change value
"""
self.last_gm_phase_change = last_gm_phase_change
def get_gm_present(self) -> str:
"""Getter for GM presence
Returns: the GM present
"""
return self.gm_present
def set_gm_present(self, gm_present : str):
"""Setter for GM presence.
Args:
gm_present: the GM present value
"""
self.gm_present = gm_present
def get_gm_identity(self) -> str:
"""Getter for GM identity
Returns: the GM identity
"""
return self.gm_identity
def set_gm_identity(self, gm_identity : str):
"""Setter for GM identity.
Args:
gm_identity: the GM identity value
"""
self.gm_identity = gm_identity

66
keywords/ptp/pmc/objects/pmc_get_time_status_np_output.py Normal file
View File

@@ -0,0 +1,66 @@
from keywords.ptp.pmc.objects.pmc_get_time_status_np_object import PMCGetTimeStatusNpObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetTimeStatusNpOutput:
"""
This class parses the output of commands such as 'pmc TIME_STATUS_NP'
Example:
sending: GET TIME_STATUS_NP
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT TIME_STATUS_NP
master_offset -62
ingress_time 0
cumulativeScaledRateOffset +0.000000000
scaledLastGmPhaseChange 0
gmTimeBaseIndicator 0
lastGmPhaseChange 0x0000'0000000000000000.0000
gmPresent false
gmIdentity 507c6f.fffe.0b5a4d
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal TIME_STATUS_NP from the passed parameter.
Args:
pmc_output (list[str]): a list of strings representing the output of the pmc command
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_values = pmc_table_parser.get_output_values_dict()
self.pmc_get_time_status_np_object = PMCGetTimeStatusNpObject()
if 'master_offset' in output_values:
self.pmc_get_time_status_np_object.set_master_offset(int(output_values['master_offset']))
if 'ingress_time' in output_values:
self.pmc_get_time_status_np_object.set_ingress_time(int(output_values['ingress_time']))
if 'cumulativeScaledRateOffset' in output_values:
self.pmc_get_time_status_np_object.set_cumulative_scaled_rate_offset(output_values['cumulativeScaledRateOffset'])
if 'scaledLastGmPhaseChange' in output_values:
self.pmc_get_time_status_np_object.set_scaled_last_gm_phase_change(int(output_values['scaledLastGmPhaseChange']))
if 'gmTimeBaseIndicator' in output_values:
self.pmc_get_time_status_np_object.set_gm_time_base_indicator(int(output_values['gmTimeBaseIndicator']))
if 'lastGmPhaseChange' in output_values:
self.pmc_get_time_status_np_object.set_last_gm_phase_change(output_values['lastGmPhaseChange'])
if 'gmPresent' in output_values:
self.pmc_get_time_status_np_object.set_gm_present(output_values['gmPresent'])
if 'gmIdentity' in output_values:
self.pmc_get_time_status_np_object.set_gm_identity(output_values['gmIdentity'])
def get_pmc_get_time_status_np_object(self) -> PMCGetTimeStatusNpObject:
"""
Getter for pmc_get_time_status_np_object object.
Returns:
A PMCGetTimeStatusNpObject
"""
return self.pmc_get_time_status_np_object

116
keywords/ptp/pmc/pmc_keywords.py
View File

@@ -1,7 +1,12 @@
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
from keywords.ptp.pmc.objects.pmc_get_current_data_set_output import PMCGetCurrentDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_default_data_set_output import PMCGetDefaultDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_domain_output import PMCGetDomainOutput
from keywords.ptp.pmc.objects.pmc_get_grandmaster_settings_np_output import PMCGetGrandmasterSettingsNpOutput
from keywords.ptp.pmc.objects.pmc_get_parent_data_set_output import PMCGetParentDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_time_properties_data_set_output import PMCGetTimePropertiesDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_time_status_np_output import PMCGetTimeStatusNpOutput
class PMCKeywords(BaseKeyword):
@@ -12,6 +17,99 @@ class PMCKeywords(BaseKeyword):
def __init__(self, ssh_connection: SSHConnection):
self.ssh_connection = ssh_connection
def pmc_get_time_status_np(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetTimeStatusNpOutput:
"""
Gets the TIME_STATUS_NP
Args:
config_file (): the config file
socket_file (): the socket file
unicast (): true to use unicast
boundry_clock (): the boundry clock
Example: PMCKeywords(ssh_connection).pmc_get_time_status_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5')
"""
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET TIME_STATUS_NP'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_time_status_np_output = PMCGetTimeStatusNpOutput(output)
return pmc_get_time_status_np_output
def pmc_get_current_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetCurrentDataSetOutput:
"""
Gets the CURRENT_DATA_SET
Args:
config_file (): the config file
socket_file (): the socket file
unicast (): true to use unicast
boundry_clock (): the boundry clock
Example: PMCKeywords(ssh_connection).pmc_get_current_data_set('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5')
"""
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET CURRENT_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_current_data_set_output = PMCGetCurrentDataSetOutput(output)
return pmc_get_current_data_set_output
def pmc_set_grandmaster_settings_np(self, config_file: str, socket_file: str, clock_class: int, time_traceable: int) -> PMCGetGrandmasterSettingsNpOutput:
"""
SET GRANDMASTER_SETTINGS_NP
Args:
config_file : the config file
socket_file : the socket file
clock_class : clockClass of ptp
time_traceable : timeTraceable of ptp
Example: PMCKeywords(ssh_connection).pmc_set_grandmaster_settings_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', '/var/run/ptp4l-ptp5', 7, 1)
"""
cmd = f"pmc -u -b 0 -f {config_file} -s {socket_file} 'SET GRANDMASTER_SETTINGS_NP clockClass {clock_class} clockAccuracy 0xfe offsetScaledLogVariance 0xffff currentUtcOffset 37 leap61 0 leap59 0 currentUtcOffsetValid 0 ptpTimescale 1 timeTraceable {time_traceable} frequencyTraceable 0 timeSource 0xa0'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_set_grandmaster_settings_np_output = PMCGetGrandmasterSettingsNpOutput(output)
return pmc_set_grandmaster_settings_np_output
def pmc_get_grandmaster_settings_np(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetGrandmasterSettingsNpOutput:
"""
Gets the grandmaster_settings_np
Args:
config_file : the config file
socket_file : the socket file
unicast : true to use unicast
boundry_clock : the boundry clock
Example: PMCKeywords(ssh_connection).pmc_get_grandmaster_settings_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5')
"""
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET GRANDMASTER_SETTINGS_NP'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_grandmaster_settings_np_output = PMCGetGrandmasterSettingsNpOutput(output)
return pmc_get_grandmaster_settings_np_output
def pmc_get_time_properties_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetTimePropertiesDataSetOutput:
"""
Gets the time_properties_data_set_object
Args:
config_file : the config file
socket_file : the socket file
unicast : true to use unicast
boundry_clock : the boundry clock
Example: PMCKeywords(ssh_connection).pmc_get_time_properties_data_set('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5')
"""
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET TIME_PROPERTIES_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_time_properties_data_set_output = PMCGetTimePropertiesDataSetOutput(output)
return pmc_get_time_properties_data_set_output
def pmc_get_default_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetDefaultDataSetOutput:
"""
Gets the default data et
@@ -30,6 +128,24 @@ class PMCKeywords(BaseKeyword):
pmc_get_default_data_set_output = PMCGetDefaultDataSetOutput(output)
return pmc_get_default_data_set_output
def pmc_get_domain(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetDomainOutput:
"""
Gets the domain
Args:
config_file (): the config file
socket_file (): the socket file
unicast (): true to use unicast
boundry_clock (): the boundry clock
Example: PMCKeywords(ssh_connection).pmc_get_domain('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5')
"""
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET DOMAIN'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_domain_output = PMCGetDomainOutput(output)
return pmc_get_domain_output
def pmc_get_parent_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetParentDataSetOutput:
"""
Gets the parent data set