Adding ptp keywords for pmc command

Adding keywords and objects for both pmc
get_default_data_set and get_parent_data_set

Change-Id: I0ed90d6f50f253280ad63feaedbfc88c2033a402
This commit is contained in:
jpike
2025-01-30 15:07:45 -05:00
parent a5d75dd54f
commit daf491c666
7 changed files with 782 additions and 0 deletions

View File

@@ -0,0 +1,206 @@
class PMCGetDefaultDataSetObject:
"""
Object to hold the values of default data set
"""
def __init__(self):
self.two_step_flag: int = -1
self.slave_only: int = -1
self.number_ports: int = -1
self.priority1: int = -1
self.clock_class: int = -1
self.clock_accuracy: str = ''
self.offset_scaled_log_variance: str = ''
self.priority2: int = -1
self.clock_identity: str = ''
self.domain_number: str = ''
def get_two_step_flag(self) -> int:
"""
Getter for two_step_flag
Returns: two_step_flag
"""
return self.two_step_flag
def set_two_step_flag(self, two_step_flag: int):
"""
Setter for two_step_flag
Args:
two_step_flag (): the two_step_flag value
Returns:
"""
self.two_step_flag = two_step_flag
def get_slave_only(self) -> int:
"""
Getter for slave_only
Returns: slave_only value
"""
return self.slave_only
def set_slave_only(self, slave_only: int):
"""
Setter for slave_only
Args:
slave_only (): slave_only value
Returns:
"""
self.slave_only = slave_only
def get_number_ports(self) -> int:
"""
Getter for number_ports
Returns: the number_ports value
"""
return self.number_ports
def set_number_ports(self, number_ports: int):
"""
Setter for number_ports
Args:
number_ports (): the number_ports value
Returns:
"""
self.number_ports = number_ports
def get_priority1(self) -> int:
"""
Getter for priority1
Returns: priority1 value
"""
return self.priority1
def set_priority1(self, priority1: int):
"""
Setter for priority1
Args:
priority1 (): the priority1 value
Returns:
"""
self.priority1 = priority1
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_priority2(self) -> int:
"""
Getter for priority2
Returns: the priority2 value
"""
return self.priority2
def set_priority2(self, priority2: int):
"""
Setter for priority2
Args:
priority2 (): the priority2 value
Returns:
"""
self.priority2 = priority2
def get_clock_identity(self) -> str:
"""
Getter for clock_identity
Returns: the clock_identity value
"""
return self.clock_identity
def set_clock_identity(self, clock_identity: str):
"""
Setter for clock_identity
Args:
clock_identity (): the clock_identity value
Returns:
"""
self.clock_identity = clock_identity
def get_domain_number(self) -> str:
"""
Getter for domain_number
Returns: the domain_number value
"""
return self.domain_number
def set_domain_number(self, domain_number: str):
"""
Setter for domain_number
Args:
domain_number (): the domain_number value
Returns:
"""
self.domain_number = domain_number

View File

@@ -0,0 +1,75 @@
from keywords.ptp.pmc.objects.pmc_get_default_data_set_object import PMCGetDefaultDataSetObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetDefaultDataSetOutput:
"""
This class parses the output of commands such as 'pmc GET_DEFAULT_DATASET'
Example:
sending: GET DEFAULT_DATA_SET
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT DEFAULT_DATA_SET
twoStepFlag 1
slaveOnly 0
numberPorts 1
priority1 128
clockClass 248
clockAccuracy 0xfe
offsetScaledLogVariance 0xffff
priority2 128
clockIdentity 507c6f.fffe.0b5a4d
domainNumber 0.
"""
def __init__(self, pmc_output: [str]):
"""
Constructor.
Create an internal PMCGetDefaultDataSet 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_default_data_set_object = PMCGetDefaultDataSetObject()
if 'twoStepFlag' in output_values:
self.pmc_get_default_data_set_object.set_two_step_flag(int(output_values['twoStepFlag']))
if 'slaveOnly' in output_values:
self.pmc_get_default_data_set_object.set_slave_only(int(output_values['slaveOnly']))
if 'numberPorts' in output_values:
self.pmc_get_default_data_set_object.set_number_ports(int(output_values['numberPorts']))
if 'priority1' in output_values:
self.pmc_get_default_data_set_object.set_priority1(int(output_values['priority1']))
if 'clockClass' in output_values:
self.pmc_get_default_data_set_object.set_clock_class(int(output_values['clockClass']))
if 'clockAccuracy' in output_values:
self.pmc_get_default_data_set_object.set_clock_accuracy(output_values['clockAccuracy'])
if 'offsetScaledLogVariance' in output_values:
self.pmc_get_default_data_set_object.set_offset_scaled_log_variance(output_values['offsetScaledLogVariance'])
if 'priority2' in output_values:
self.pmc_get_default_data_set_object.set_priority2(int(output_values['priority2']))
if 'clockIdentity' in output_values:
self.pmc_get_default_data_set_object.set_clock_identity(output_values['clockIdentity'])
if 'domainNumber' in output_values:
self.pmc_get_default_data_set_object.set_domain_number(output_values['domainNumber'])
def get_pmc_get_default_data_set_object(self) -> PMCGetDefaultDataSetObject:
"""
Getter for pmc_get_default_data_set_object object.
Returns:
A PMCGetDefaultDataSetObject
"""
return self.pmc_get_default_data_set_object

View File

@@ -0,0 +1,206 @@
class PMCGetParentDataSetObject:
"""
Object to hold attributes of Parent set object
"""
def __init__(self):
self.parent_port_identity: str = ''
self.parent_stats: int = -1
self.observed_parent_offset_scaled_log_variance: str = ''
self.observed_parent_clock_phase_change_rate: str = ''
self.grandmaster_priority1: int = -1
self.gm_clock_class: int = -1
self.gm_clock_accuracy: str = ''
self.gm_offset_scaled_log_variance: str = ''
self.grandmaster_priority2: int = -1
self.grandmaster_identity: str = ''
def get_parent_port_identity(self) -> str:
"""
Getter for parent_port_identity
Returns: parent_port_identity
"""
return self.parent_port_identity
def set_parent_port_identity(self, parent_port_identity: str):
"""
Setter for parent_port_identity
Args:
parent_port_identity (): the parent_port_identity value
Returns:
"""
self.parent_port_identity = parent_port_identity
def get_parent_stats(self) -> int:
"""
Getter for parent_stats
Returns: parent_stats value
"""
return self.parent_stats
def set_parent_stats(self, parent_stats: int):
"""
Setter for parent_stats
Args:
parent_stats (): parent_stats value
Returns:
"""
self.parent_stats = parent_stats
def get_observed_parent_offset_scaled_log_variance(self) -> str:
"""
Getter for observed_parent_offset_scaled_log_variance
Returns: the observed_parent_offset_scaled_log_variance value
"""
return self.observed_parent_offset_scaled_log_variance
def set_observed_parent_offset_scaled_log_variance(self, observed_parent_offset_scaled_log_variance: str):
"""
Setter for observed_parent_offset_scaled_log_variance
Args:
observed_parent_offset_scaled_log_variance (): the observed_parent_offset_scaled_log_variance value
Returns:
"""
self.observed_parent_offset_scaled_log_variance = observed_parent_offset_scaled_log_variance
def get_observed_parent_clock_phase_change_rate(self) -> str:
"""
Getter for observed_parent_clock_phase_change_rate
Returns: observed_parent_clock_phase_change_rate value
"""
return self.observed_parent_clock_phase_change_rate
def set_observed_parent_clock_phase_change_rate(self, observed_parent_clock_phase_change_rate: str):
"""
Setter for observed_parent_clock_phase_change_rate
Args:
observed_parent_clock_phase_change_rate (): the observed_parent_clock_phase_change_rate value
Returns:
"""
self.observed_parent_clock_phase_change_rate = observed_parent_clock_phase_change_rate
def get_grandmaster_priority1(self) -> int:
"""
Getter for grandmaster_priority1
Returns: the grandmaster_priority1 value
"""
return self.grandmaster_priority1
def set_grandmaster_priority1(self, grandmaster_priority1: int):
"""
Setter for grandmaster_priority1
Args:
grandmaster_priority1 (): the grandmaster_priority1 value
Returns:
"""
self.grandmaster_priority1 = grandmaster_priority1
def get_gm_clock_class(self) -> int:
"""
Getter for gm_clock_class
Returns: the gm_clock_class value
"""
return self.gm_clock_class
def set_gm_clock_class(self, gm_clock_class: int):
"""
Setter for gm_clock_class
Args:
gm_clock_class (): the gm_clock_class value
Returns:
"""
self.gm_clock_class = gm_clock_class
def get_gm_clock_accuracy(self) -> str:
"""
Getter for gm_clock_accuracy
Returns: the gm_clock_accuracy value
"""
return self.gm_clock_accuracy
def set_gm_clock_accuracy(self, gm_clock_accuracy: str):
"""
Setter for gm_clock_accuracy
Args:
gm_clock_accuracy (): the gm_clock_accuracy value
Returns:
"""
self.gm_clock_accuracy = gm_clock_accuracy
def get_gm_offset_scaled_log_variance(self) -> str:
"""
Getter for gm_offset_scaled_log_variance
Returns: the gm_offset_scaled_log_variance value
"""
return self.gm_offset_scaled_log_variance
def set_gm_offset_scaled_log_variance(self, gm_offset_scaled_log_variance: str):
"""
Setter for gm_offset_scaled_log_variance
Args:
gm_offset_scaled_log_variance (): the gm_offset_scaled_log_variance value
Returns:
"""
self.gm_offset_scaled_log_variance = gm_offset_scaled_log_variance
def get_grandmaster_priority2(self) -> int:
"""
Getter for grandmaster_priority2
Returns: the grandmaster_priority2 value
"""
return self.grandmaster_priority2
def set_grandmaster_priority2(self, grandmaster_priority2: int):
"""
Setter for grandmaster_priority2
Args:
grandmaster_priority2 (): the grandmaster_priority2 value
Returns:
"""
self.grandmaster_priority2 = grandmaster_priority2
def get_grandmaster_identity(self) -> str:
"""
Getter for grandmaster_identity
Returns: the grandmaster_identity value
"""
return self.grandmaster_identity
def set_grandmaster_identity(self, grandmaster_identity: str):
"""
Setter for grandmaster_identity
Args:
grandmaster_identity (): the grandmaster_identity value
Returns:
"""
self.grandmaster_identity = grandmaster_identity

View File

@@ -0,0 +1,74 @@
from keywords.ptp.pmc.objects.pmc_get_parent_data_set_object import PMCGetParentDataSetObject
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
class PMCGetParentDataSetOutput:
"""
This class parses the output of commands such as 'pmc GET_PARENT_DATASET'
Example:
sending: GET PARENT_DATA_SET
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT PARENT_DATA_SET
parentPortIdentity 507c6f.fffe.0b5a4d-0
parentStats 0
observedParentOffsetScaledLogVariance 0xffff
observedParentClockPhaseChangeRate 0x7fffffff
grandmasterPriority1 128
gm.ClockClass 248
gm.ClockAccuracy 0xfe
gm.OffsetScaledLogVariance 0xffff
grandmasterPriority2 128
grandmasterIdentity 507c6f.fffe.0b5a4d
"""
def __init__(self, pmc_output):
"""
Constructor.
Create an internal PMCGetDefaultDataSet 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_parent_data_set_object = PMCGetParentDataSetObject()
if 'parentPortIdentity' in output_values:
self.pmc_get_parent_data_set_object.set_parent_port_identity(output_values['parentPortIdentity'])
if 'parentStats' in output_values:
self.pmc_get_parent_data_set_object.set_parent_stats(output_values['parentStats'])
if 'observedParentOffsetScaledLogVariance' in output_values:
self.pmc_get_parent_data_set_object.set_observed_parent_offset_scaled_log_variance(output_values['observedParentOffsetScaledLogVariance'])
if 'observedParentClockPhaseChangeRate' in output_values:
self.pmc_get_parent_data_set_object.set_observed_parent_clock_phase_change_rate(output_values['observedParentClockPhaseChangeRate'])
if 'grandmasterPriority1' in output_values:
self.pmc_get_parent_data_set_object.set_grandmaster_priority1(int(output_values['grandmasterPriority1']))
if 'gm.ClockClass' in output_values:
self.pmc_get_parent_data_set_object.set_gm_clock_class(int(output_values['gm.ClockClass']))
if 'gm.ClockAccuracy' in output_values:
self.pmc_get_parent_data_set_object.set_gm_clock_accuracy(output_values['gm.ClockAccuracy'])
if 'gm.OffsetScaledLogVariance' in output_values:
self.pmc_get_parent_data_set_object.set_gm_offset_scaled_log_variance(output_values['gm.OffsetScaledLogVariance'])
if 'grandmasterPriority2' in output_values:
self.pmc_get_parent_data_set_object.set_grandmaster_priority2(int(output_values['grandmasterPriority2']))
if 'grandmasterIdentity' in output_values:
self.pmc_get_parent_data_set_object.set_grandmaster_identity(output_values['grandmasterIdentity'])
def get_pmc_get_parent_data_set_object(self) -> PMCGetParentDataSetObject:
"""
Getter for pmc_get_parent_data_set_object object.
Returns:
A PMCGetParentDataSetObject
"""
return self.pmc_get_parent_data_set_object

View File

@@ -0,0 +1,49 @@
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
from keywords.ptp.pmc.objects.pmc_get_default_data_set_output import PMCGetDefaultDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_parent_data_set_output import PMCGetParentDataSetOutput
class PMCKeywords(BaseKeyword):
"""
Class for PMC Keywords
"""
def __init__(self, ssh_connection: SSHConnection):
self.ssh_connection = ssh_connection
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
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_default_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 DEFAULT_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_default_data_set_output = PMCGetDefaultDataSetOutput(output)
return pmc_get_default_data_set_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
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_parent_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 PARENT_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_parent_data_set_output = PMCGetParentDataSetOutput(output)
return pmc_get_parent_data_set_output

View File

@@ -0,0 +1,53 @@
import re
from framework.exceptions.keyword_exception import KeywordException
class PMCTableParser:
"""
Class for PMC table parsing
Example from get DEFAULT_DATA_SET below
sending: GET DEFAULT_DATA_SET
507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT DEFAULT_DATA_SET
twoStepFlag 1
slaveOnly 0
numberPorts 1
priority1 128
clockClass 248
clockAccuracy 0xfe
offsetScaledLogVariance 0xffff
priority2 128
clockIdentity 507c6f.fffe.0b5a4d
domainNumber 0.
"""
def __init__(self, pmc_output):
"""
Constructor
Args:
pmc_output (list[str]): a list of strings representing the output of a 'pmc' command.
"""
self.pmc_output = pmc_output
def get_output_values_dict(
self,
):
"""
Getter for output values dict
Returns: the output values dict
"""
output_values_dict = {}
total_rows = len(self.pmc_output)
for row in self.pmc_output[2:total_rows - 1]: # Ignore the first 2 rows and the last row (prompt)
values = row.split(None, 1) # split once
if len(values) == 2:
key, value = values
output_values_dict[key.strip()] = value.strip()
else:
raise KeywordException(f"Line with values: {row} was not in the expected format")
return output_values_dict

View File

@@ -0,0 +1,119 @@
from keywords.ptp.pmc.objects.pmc_get_default_data_set_output import PMCGetDefaultDataSetOutput
from keywords.ptp.pmc.objects.pmc_get_parent_data_set_output import PMCGetParentDataSetOutput
from keywords.ptp.pmc.pmc_table_parser import PMCTableParser
pmc_output = [
'sending: GET DEFAULT_DATA_SET\n',
' 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT DEFAULT_DATA_SET\n',
' twoStepFlag 1\n',
' slaveOnly 0\n',
' numberPorts 1\n',
' priority1 128\n',
' clockClass 248\n',
' clockAccuracy 0xfe\n',
' offsetScaledLogVariance 0xffff\n',
' priority2 128\n',
' clockIdentity 507c6f.fffe.0b5a4d\n',
' domainNumber 0.\n',
'sysadmin@controller-0:~$\n',
]
pmc_parent_data_set_output = [
'sending: GET PARENT_DATA_SET\n',
' 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT PARENT_DATA_SET\n',
' parentPortIdentity 507c6f.fffe.0b5a4d-0\n',
' parentStats 0\n',
' observedParentOffsetScaledLogVariance 0xffff\n',
' observedParentClockPhaseChangeRate 0x7fffffff\n',
' grandmasterPriority1 128\n',
' gm.ClockClass 248\n',
' gm.ClockAccuracy 0xfe\n',
' gm.OffsetScaledLogVariance 0xffff\n',
' grandmasterPriority2 128\n',
' grandmasterIdentity 507c6f.fffe.0b5a4d\n',
'sysadmin@controller-0:~$\n',
]
def test_pmc_get_default_data_set_table_parser():
"""
Tests the pmc table parser for get_default_data_set output parser
Returns:
"""
pmc_table_parser = PMCTableParser(pmc_output)
output_dict = pmc_table_parser.get_output_values_dict()
assert output_dict['twoStepFlag'] == '1'
assert output_dict['slaveOnly'] == '0'
assert output_dict['numberPorts'] == '1'
assert output_dict['priority1'] == '128'
assert output_dict['clockClass'] == '248'
assert output_dict['clockAccuracy'] == '0xfe'
assert output_dict['offsetScaledLogVariance'] == '0xffff'
assert output_dict['priority2'] == '128'
assert output_dict['clockIdentity'] == '507c6f.fffe.0b5a4d'
assert output_dict['domainNumber'] == '0.'
def test_pmc_get_default_data_set_output():
"""
Tests pmc get_default_data_set output
Returns:
"""
pmc_get_default_data_set_output = PMCGetDefaultDataSetOutput(pmc_output)
pmc_get_default_data_set_object = pmc_get_default_data_set_output.get_pmc_get_default_data_set_object()
assert pmc_get_default_data_set_object.get_two_step_flag() == 1
assert pmc_get_default_data_set_object.get_slave_only() == 0
assert pmc_get_default_data_set_object.get_number_ports() == 1
assert pmc_get_default_data_set_object.get_priority1() == 128
assert pmc_get_default_data_set_object.get_clock_class() == 248
assert pmc_get_default_data_set_object.get_clock_accuracy() == '0xfe'
assert pmc_get_default_data_set_object.get_offset_scaled_log_variance() == '0xffff'
assert pmc_get_default_data_set_object.get_priority2() == 128
assert pmc_get_default_data_set_object.get_clock_identity() == '507c6f.fffe.0b5a4d'
assert pmc_get_default_data_set_object.get_domain_number() == '0.'
def test_pmc_get_parent_data_set_table_parser():
"""
Tests the pmc table parser for get_parent_data_set output parser
Returns:
"""
pmc_table_parser = PMCTableParser(pmc_parent_data_set_output)
output_dict = pmc_table_parser.get_output_values_dict()
assert output_dict['parentPortIdentity'] == '507c6f.fffe.0b5a4d-0'
assert output_dict['gm.ClockClass'] == '248'
assert output_dict['gm.ClockAccuracy'] == '0xfe'
assert output_dict['gm.OffsetScaledLogVariance'] == '0xffff'
assert output_dict['grandmasterIdentity'] == '507c6f.fffe.0b5a4d'
assert output_dict['grandmasterPriority1'] == '128'
assert output_dict['grandmasterPriority2'] == '128'
assert output_dict['observedParentClockPhaseChangeRate'] == '0x7fffffff'
assert output_dict['observedParentOffsetScaledLogVariance'] == '0xffff'
assert output_dict['parentStats'] == '0'
def test_pmc_get_parent_data_set_output():
"""
Tests pmc get_parent_data_set output
Returns:
"""
pmc_get_parent_data_set_output = PMCGetParentDataSetOutput(pmc_parent_data_set_output)
pmc_get_parent_data_set_object = pmc_get_parent_data_set_output.get_pmc_get_parent_data_set_object()
assert pmc_get_parent_data_set_object.get_parent_port_identity() == '507c6f.fffe.0b5a4d-0'
assert pmc_get_parent_data_set_object.get_gm_clock_class() == 248
assert pmc_get_parent_data_set_object.get_gm_clock_accuracy() == '0xfe'
assert pmc_get_parent_data_set_object.get_gm_offset_scaled_log_variance() == '0xffff'
assert pmc_get_parent_data_set_object.get_grandmaster_identity() == '507c6f.fffe.0b5a4d'
assert pmc_get_parent_data_set_object.get_grandmaster_priority1() == 128
assert pmc_get_parent_data_set_object.get_grandmaster_priority2() == 128
assert pmc_get_parent_data_set_object.get_observed_parent_clock_phase_change_rate() == '0x7fffffff'
assert pmc_get_parent_data_set_object.get_observed_parent_offset_scaled_log_variance() == '0xffff'
assert pmc_get_parent_data_set_object.get_parent_stats() == '0'