starlingx/test
Code Issues Proposed changes

Removed hardcoded hosts and pci-slot

Browse Source 
Implementation of ptp_parameters_parser.py

Change-Id: I12d6606870f0e85705f3264d054a79bf66e59523
Signed-off-by: Guntaka Umashankar Reddy <umashankarguntaka.reddy@windriver.com>
This commit is contained in:
Guntaka Umashankar Reddy
2025-05-08 04:57:01 -04:00
committed by Christian Roy
parent 7dcca5b7a7
commit ade9a3df1c
13 changed files with 109 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
config/ptp/files/default.json5
View File

@@ -42,7 +42,6 @@
// By default, we use the first port.
base_port: "enp81s0f0",
gpio_switch_port: "30",
pci_slot: "00:007.01",
nic_connection: {
to_host: "controller_1",

15
config/ptp/objects/ptp_nic.py
View File

@@ -21,7 +21,6 @@ class PTPNic:
"""
self.name: str = nic_name
self.gpio_switch_port = None
self.pci_slot = None
self.base_port = None
self.sma1 = None
self.sma2 = None
@@ -36,9 +35,6 @@ class PTPNic:
if "gpio_switch_port" in nic_dict and nic_dict["gpio_switch_port"]:
self.gpio_switch_port = nic_dict["gpio_switch_port"]
if "pci_slot" in nic_dict and nic_dict["pci_slot"]:
self.pci_slot = nic_dict["pci_slot"]
if "base_port" in nic_dict and nic_dict["base_port"]:
self.base_port = nic_dict["base_port"]
@@ -99,7 +95,6 @@ class PTPNic:
ptp_nic_dictionary = {
"name": self.name,
"gpio_switch_port": self.gpio_switch_port,
"pci_slot": self.pci_slot,
"base_port": self.base_port,
"sma1": sma1_dict,
"sma2": sma2_dict,
@@ -133,16 +128,6 @@ class PTPNic:
"""
return self.gpio_switch_port
def get_pci_slot(self) -> str:
"""
Gets the pci slot
Returns (str):
The pci slot
"""
return self.pci_slot
def get_base_port(self) -> str:
"""
Gets the base port.

5
keywords/cloud_platform/system/ptp/objects/system_ptp_instance_output.py
View File

@@ -1,5 +1,6 @@
from framework.exceptions.keyword_exception import KeywordException
from keywords.cloud_platform.system.ptp.objects.system_ptp_instance_object import SystemPTPInstanceObject
from keywords.cloud_platform.system.ptp.ptp_parameters_parser import PTPParametersParser
from keywords.cloud_platform.system.system_vertical_table_parser import SystemVerticalTableParser
from keywords.python.type_converter import TypeConverter
@@ -53,7 +54,5 @@ class SystemPTPInstanceOutput:
Returns:
str: ptp instance parameters
Example : "cmdline_opts='-s xxxx -O -37 -m' boundary_clock_jbod=1 domainNumber=24"
"""
return repr(" ".join(self.system_ptp_instance_object.get_parameters()))
return PTPParametersParser(self.system_ptp_instance_object.get_parameters()).process_cmdline_opts()

10
keywords/cloud_platform/system/ptp/objects/system_ptp_interface_list_output.py
View File

@@ -1,4 +1,5 @@
from keywords.cloud_platform.system.ptp.objects.system_ptp_interface_list_object import SystemPTPInterfaceListObject
from keywords.cloud_platform.system.ptp.ptp_parameters_parser import PTPParametersParser
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
@@ -51,12 +52,5 @@ class SystemPTPInterfaceListOutput:
Returns:
str: ptp interface parameters
Example : "cmdline_opts='-s xxxx -O -37 -m' boundary_clock_jbod=1 domainNumber=24"
"""
parameters = ptp_interface_obj.get_parameters()
if not parameters:
return ""
return repr(" ".join(parameters))
return PTPParametersParser(ptp_interface_obj.get_parameters()).process_cmdline_opts()

5
keywords/cloud_platform/system/ptp/objects/system_ptp_interface_output.py
View File

@@ -1,5 +1,6 @@
from framework.exceptions.keyword_exception import KeywordException
from keywords.cloud_platform.system.ptp.objects.system_ptp_interface_object import SystemPTPInterfaceObject
from keywords.cloud_platform.system.ptp.ptp_parameters_parser import PTPParametersParser
from keywords.cloud_platform.system.system_vertical_table_parser import SystemVerticalTableParser
from keywords.python.type_converter import TypeConverter
@@ -67,7 +68,5 @@ class SystemPTPInterfaceOutput:
Returns:
str: ptp interface parameters
Example : "cmdline_opts='-s xxxx -O -37 -m' boundary_clock_jbod=1 domainNumber=24"
"""
return repr(" ".join(self.system_ptp_interface_object.get_parameters()))
return PTPParametersParser(self.system_ptp_interface_object.get_parameters()).process_cmdline_opts()

49
keywords/cloud_platform/system/ptp/ptp_parameters_parser.py Normal file
View File

@@ -0,0 +1,49 @@
import re
from typing import Union
class PTPParametersParser:
"""
Class for ptp parameters parsing
Example:
['domainNumber=24', 'dataset_comparison=G.8275.x', 'priority2=110', 'boundary_clock_jbod=1']
cmdline_opts=-s xxxx -O -37 -m boundary_clock_jbod=1 domainNumber=24
"""
def __init__(self, parameters: Union[str, list]):
"""
Constructor
Args:
parameters Union[str, list]): The input data, which can be a string
containing cmdline_opts or a list of strings representing
ptp parameters.
"""
self.parameters = parameters
def process_cmdline_opts(self) -> str:
"""
Processes the cmdline_opts data, handling both string and list inputs,
to ensure the value is enclosed in single quotes.
Returns:
str: The modified string with the cmdline_opts value properly quoted,
or the original input string if cmdline_opts is not found.
"""
if isinstance(self.parameters, list):
parameters_str = " ".join(self.parameters) # Convert list to string
else:
parameters_str = self.parameters
match = re.search(r"(cmdline_opts=)(.*?)(?=\s+\w+=|$)", parameters_str)
if match:
prefix, value = match.group(1), match.group(2).strip()
if not (value.startswith("'") and value.endswith("'")):
value = f"'{value}'"
output_str = parameters_str.replace(match.group(0), f"{prefix}{value}")
else:
output_str = parameters_str
return output_str

74
keywords/cloud_platform/system/ptp/ptp_setup_executor_keywords.py
View File

@@ -1,9 +1,10 @@
from typing import Any
from framework.logging.automation_logger import get_logger
from framework.validation.validation import validate_equals, validate_str_contains
from framework.validation.validation import validate_equals, validate_equals_with_retry, validate_str_contains
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.system.host.system_host_list_keywords import SystemHostListKeywords
from keywords.cloud_platform.system.ptp.system_host_if_ptp_keywords import SystemHostIfPTPKeywords
from keywords.cloud_platform.system.ptp.system_host_ptp_instance_keywords import SystemHostPTPInstanceKeywords
from keywords.cloud_platform.system.ptp.system_ptp_instance_keywords import SystemPTPInstanceKeywords
@@ -11,6 +12,7 @@ from keywords.cloud_platform.system.ptp.system_ptp_instance_parameter_keywords i
from keywords.cloud_platform.system.ptp.system_ptp_interface_keywords import SystemPTPInterfaceKeywords
from keywords.ptp.setup.object.ptp_setup import PTPSetup
from keywords.ptp.setup.ptp_setup_reader import PTPSetupKeywords
from starlingx.keywords.linux.systemctl.systemctl_status_keywords import SystemCTLStatusKeywords
class PTPSetupExecutorKeywords(BaseKeyword):
@@ -42,9 +44,7 @@ class PTPSetupExecutorKeywords(BaseKeyword):
"""
Configure all ptp configurations
"""
lab_connect_keywords = LabConnectionKeywords()
ssh_connection = lab_connect_keywords.get_active_controller_ssh()
system_ptp_instance_keywords = SystemPTPInstanceKeywords(ssh_connection)
system_ptp_instance_keywords = SystemPTPInstanceKeywords(self.ssh_connection)
self.add_all_ptp_configurations_for_ptp4l_service()
@@ -57,6 +57,25 @@ class PTPSetupExecutorKeywords(BaseKeyword):
system_ptp_instance_apply_output = system_ptp_instance_keywords.system_ptp_instance_apply()
validate_equals(system_ptp_instance_apply_output, "Applying the PTP Instance configuration", "apply PTP instance configuration")
# After applying the instance configuration, it needs to check whether the services are available or not.
def check_ptp4l_status() -> bool:
"""
Checks if the PTP4L service is active and running.
Returns:
bool: True if the service is active and running, False otherwise.
"""
ptp4l_status_output = SystemCTLStatusKeywords(self.ssh_connection).get_status("ptp4l@*")
if not ptp4l_status_output:
return False # Handle the case of an empty list
for line in ptp4l_status_output:
if "Active: active (running)" in line:
return True
return False
validate_equals_with_retry(check_ptp4l_status, True, 600)
def add_all_ptp_configurations_for_ptp4l_service(self):
"""
Configure all ptp configurations for ptp4l service
@@ -164,49 +183,28 @@ class PTPSetupExecutorKeywords(BaseKeyword):
name = ptp_instance_obj.get_name()
ptp_host_ifs = ptp_instance_obj.get_ptp_interfaces()
hosts = SystemHostListKeywords(ssh_connection).get_system_host_list().get_controllers_and_computes()
for ptp_host_if in ptp_host_ifs:
interface_name = ptp_host_if.get_name()
system_ptp_interface_output = system_ptp_interface_keywords.system_ptp_interface_add(interface_name, name)
validate_equals(system_ptp_interface_output.get_ptp_interface().get_ptp_instance_name(), name, "PTP instance name of the PTP interface")
validate_str_contains(system_ptp_interface_output.get_ptp_interface().get_name(), interface_name, "add PTP interface")
controller_0_interfaces = ptp_host_if.get_controller_0_interfaces()
ctrl0_hostname = "controller-0"
for interface in controller_0_interfaces:
if not interface :
continue
system_host_if_ptp_keywords.system_host_if_ptp_assign(ctrl0_hostname, interface, interface_name)
system_ptp_interface_show_output = system_ptp_interface_keywords.get_system_ptp_interface_show(interface_name)
validate_str_contains(system_ptp_interface_show_output.get_ptp_interface().get_interface_names(), f"{ctrl0_hostname}/{interface}", f"assign ptp interface for {ctrl0_hostname}")
controller_1_interfaces = ptp_host_if.get_controller_1_interfaces()
ctrl1_hostname = "controller-1"
for interface in controller_1_interfaces:
if not interface :
continue
system_host_if_ptp_keywords.system_host_if_ptp_assign(ctrl1_hostname, interface, interface_name)
system_ptp_interface_show_output = system_ptp_interface_keywords.get_system_ptp_interface_show(interface_name)
validate_str_contains(system_ptp_interface_show_output.get_ptp_interface().get_interface_names(), f"{ctrl1_hostname}/{interface}", f"assign ptp interface for {ctrl1_hostname}")
compute_0_interfaces = ptp_host_if.get_compute_0_interfaces()
comp0_hostname = "compute-0"
for interface in compute_0_interfaces:
if not interface :
continue
system_host_if_ptp_keywords.system_host_if_ptp_assign(comp0_hostname, interface, interface_name)
system_ptp_interface_show_output = system_ptp_interface_keywords.get_system_ptp_interface_show(interface_name)
validate_str_contains(system_ptp_interface_show_output.get_ptp_interface().get_interface_names(), f"{comp0_hostname}/{interface}", f"assign ptp interface for {comp0_hostname}")
for host in hosts:
hostname = host.get_host_name()
interfaces = ptp_host_if.get_interfaces_for_hostname(hostname)
for interface in filter(None, interfaces):
system_host_if_ptp_keywords.system_host_if_ptp_assign(hostname, interface, interface_name)
system_ptp_interface_show_output = system_ptp_interface_keywords.get_system_ptp_interface_show(interface_name)
validate_str_contains(system_ptp_interface_show_output.get_ptp_interface().get_interface_names(), f"{hostname}/{interface}", f"assign ptp interface for {hostname}")
ptp_interface_parameters = ptp_host_if.get_ptp_interface_parameter()
if ptp_interface_parameters :
if ptp_interface_parameters:
system_ptp_interface_parameter_add_output = system_ptp_interface_keywords.system_ptp_interface_parameter_add(interface_name, ptp_interface_parameters)
self.validate_parameters(system_ptp_interface_parameter_add_output.get_ptp_interface_parameters(), ptp_interface_parameters, "add PTP interface parameters")
def validate_parameters(self, observed_value: str, expected_value: str, validation_description: str) -> None :
def validate_parameters(self, observed_value: str, expected_value: str, validation_description: str) -> None:
"""
This function will validate if the observed value matches the expected value with associated logging.
@@ -220,4 +218,4 @@ class PTPSetupExecutorKeywords(BaseKeyword):
Raises:
Exception: raised when validate fails
"""
validate_equals(set(observed_value.split()), set(expected_value.split()), validation_description)
validate_equals(set(observed_value.split()), set(expected_value.split()), validation_description)

2
keywords/cloud_platform/system/ptp/ptp_verify_config.py → keywords/cloud_platform/system/ptp/ptp_verify_config_keywords.py
View File

@@ -97,7 +97,7 @@ class PTPVerifyConfigKeywords(BaseKeyword):
verify SMA status
Args:
hostnames (list): list of controllers and computes
hosts (list): list of controllers and computes
Returns: None
"""

2
keywords/ptp/cat/cat_ptp_cgu_parser.py
View File

@@ -65,7 +65,7 @@ class CatPtpCguParser:
"""
cgu: PtpCguComponentObject = None
match = re.match(r"Found (\S+) CGU", self.cat_ptp_cgu_output[0]) # Ask about this
match = re.search(r"Found (\S+) CGU", self.cat_ptp_cgu_output[0])
if match:
chip_model = match.group(1)
config_version_match = re.search(r"DPLL Config ver: (.*)", self.cat_ptp_cgu_output[1])

24
keywords/ptp/gnss_keywords.py
View File

@@ -20,7 +20,7 @@ class GnssKeywords(BaseKeyword):
Initializes the GnssKeywords.
"""
def get_pci_slot_name(self, hostname: str, interface: str) -> str:
def get_pci_slot_name(self, hostname: str, interface: str) -> str:
"""
Retrieves the PCI_SLOT_NAME from the uevent file for a given PTP interface.
@@ -30,7 +30,7 @@ class GnssKeywords(BaseKeyword):
Returns:
str: The PCI slot name if found, otherwise None.
Raises:
Exception: raised when PCI_SLOT_NAME not found
"""
@@ -50,14 +50,14 @@ class GnssKeywords(BaseKeyword):
else:
raise Exception(f"PCI_SLOT_NAME not found in {uevent_path}")
def get_gnss_serial_port_from_gnss_directory(self, hostname: str, interface: str) -> str:
def get_gnss_serial_port_from_gnss_directory(self, hostname: str, interface: str) -> str:
"""
Get GNSS serial port from the specified gnss directory.
Args:
hostname (str) : The name of the host
interface (str): The name of the PTP interface (e.g., "enp138s0f0").
Returns:
str: The GNSS serial port value (e.g., "gnss0") if found, otherwise None.
"""
@@ -72,8 +72,8 @@ class GnssKeywords(BaseKeyword):
if not contents:
get_logger().log_info(f"The directory {gnss_dir} is empty.")
return None
return " ".join(contents).strip() # Return the captured value in str, removing leading/trailing spaces
return " ".join(contents).strip() # Return the captured value in str, removing leading/trailing spaces
def extract_gnss_port(self, instance_parameters: str) -> str:
"""
@@ -104,8 +104,9 @@ class GnssKeywords(BaseKeyword):
host_name = hostname.replace("-", "_")
ptp_config = ConfigurationManager.get_ptp_config()
pci_slot = ptp_config.get_host(host_name).get_nic(nic).get_pci_slot()
cgu_location = f"/sys/kernel/debug/ice/{pci_slot}/cgu"
interface = ptp_config.get_host(host_name).get_nic(nic).get_base_port()
pci_address = self.get_pci_slot_name(hostname, interface)
cgu_location = f"/sys/kernel/debug/ice/{pci_address}/cgu"
gpio_switch_port = ptp_config.get_host(host_name).get_nic(nic).get_gpio_switch_port()
command = f"echo 1 > /sys/class/gpio/gpio{gpio_switch_port}/value"
@@ -129,8 +130,9 @@ class GnssKeywords(BaseKeyword):
host_name = hostname.replace("-", "_")
ptp_config = ConfigurationManager.get_ptp_config()
pci_slot = ptp_config.get_host(host_name).get_nic(nic).get_pci_slot()
cgu_location = f"/sys/kernel/debug/ice/{pci_slot}/cgu"
interface = ptp_config.get_host(host_name).get_nic(nic).get_base_port()
pci_address = self.get_pci_slot_name(hostname, interface)
cgu_location = f"/sys/kernel/debug/ice/{pci_address}/cgu"
gpio_switch_port = ptp_config.get_host(host_name).get_nic(nic).get_gpio_switch_port()
command = f"echo 0 > /sys/class/gpio/gpio{gpio_switch_port}/value"
@@ -160,7 +162,7 @@ class GnssKeywords(BaseKeyword):
expected_pps_dpll_status (list): expected list of PPS DPLL status values.
timeout (int): The maximum time (in seconds) to wait for the match.
polling_sleep_time (int): The time period to wait to receive the expected output.
Returns: None
Raises:

27
keywords/ptp/setup/object/ptp_host_interface_setup.py
View File

@@ -62,33 +62,6 @@ class PTPHostInterfaceSetup:
"""
return self.ptp_interface_parameter
def get_controller_0_interfaces(self) -> List[str]:
"""
Gets the controller_0_interfaces of this ptp host interface setup.
Returns:
List[str]: The controller_0_interfaces of this ptp host interface setup.
"""
return self.controller_0_interfaces
def get_controller_1_interfaces(self) -> List[str]:
"""
Gets the controller_1_interfaces of this ptp host interface setup.
Returns:
List[str]: The controller_1_interfaces of this ptp host interface setup.
"""
return self.controller_1_interfaces
def get_compute_0_interfaces(self) -> List[str]:
"""
Gets the compute_0_interfaces of this ptp host interface setup.
Returns:
List[str]: The compute_0_interfaces of this ptp host interface setup.
"""
return self.compute_0_interfaces
def get_interfaces_for_hostname(self, hostname: str) -> List[str]:
"""
Gets the interfaces for the given hostname in this PTP host interface setup.

2
testcases/cloud_platform/regression/ptp/test_ptp.py
View File

@@ -7,7 +7,7 @@ from framework.resources.resource_finder import get_stx_resource_path
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.system.ptp.ptp_setup_executor_keywords import PTPSetupExecutorKeywords
from keywords.cloud_platform.system.ptp.ptp_teardown_executor_keywords import PTPTeardownExecutorKeywords
from keywords.cloud_platform.system.ptp.ptp_verify_config import PTPVerifyConfigKeywords
from keywords.cloud_platform.system.ptp.ptp_verify_config_keywords import PTPVerifyConfigKeywords
from keywords.files.file_keywords import FileKeywords

4
unit_tests/keyword/ptp/ptp_setup_reader_test.py
View File

@@ -21,9 +21,9 @@ def test_generate_ptp_setup_from_template():
assert len(ptp4l_setup_list) == 4
ptp1 = ptp_setup.get_ptp4l_setup("ptp1")
ptp1if1 = ptp1.get_ptp_interface("ptp1if1")
assert ptp1if1.get_controller_0_interfaces() == ["enp81s0f1"]
assert ptp1if1.get_interfaces_for_hostname("controller-0") == ["enp81s0f1"]
ptp1if2 = ptp1.get_ptp_interface("ptp1if2")
assert ptp1if2.get_controller_0_interfaces() == ["conn_spirent_placeholder"]
assert ptp1if2.get_interfaces_for_hostname("controller-0") == ["conn_spirent_placeholder"]
# phc2sys Validations
phc2sys_setup_list = ptp_setup.get_phc2sys_setup_list()