starlingx/test
Code Issues Proposed changes

Merge "Added PROXMOX keywords to config, verify and operate on PTP"

Browse Source
This commit is contained in:
Zuul
2025-07-07 20:17:31 +00:00
committed by Gerrit Code Review
parent 444301fcc4 19135c0b69
commit 162152c153
6 changed files with 344 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,6 @@
from typing import Dict
from config.host.objects.host_configuration import HostConfiguration
from config.ptp.objects.ptp_nic_connection import PTPNicConnection
from config.ptp.objects.sma_connector import SMAConnector
from config.ptp.objects.ufl_connector import UFLConnector
@@ -31,6 +32,7 @@ class PTPNic:
self.spirent_port = None
self.conn_to_proxmox = None
self.proxmox_port = None
self.proxmox_ptp_vm_config = None
if "gpio_switch_port" in nic_dict and nic_dict["gpio_switch_port"]:
self.gpio_switch_port = nic_dict["gpio_switch_port"]
@@ -53,6 +55,9 @@ class PTPNic:
if "proxmox_port" in nic_dict and nic_dict["proxmox_port"]:
self.proxmox_port = nic_dict["proxmox_port"]
if "proxmox_ptp_vm_config" in nic_dict and nic_dict["proxmox_ptp_vm_config"]:
self.proxmox_ptp_vm_config = HostConfiguration(nic_dict["proxmox_ptp_vm_config"])
def __str__(self):
"""
String representation of this object.
@@ -105,6 +110,7 @@ class PTPNic:
"spirent_port": self.spirent_port,
"conn_to_proxmox": self.conn_to_proxmox,
"proxmox_port": self.proxmox_port,
"proxmox_ptp_vm_config": self.proxmox_ptp_vm_config,
}
return ptp_nic_dictionary
@@ -265,3 +271,12 @@ class PTPNic:
The proxmox port.
"""
return self.proxmox_port
def get_proxmox_ptp_vm_config(self) -> HostConfiguration:
"""
Gets the proxmox PTP VM config.
Returns (HostConfiguration):
The proxmox PTP VM configuration object.
"""
return self.proxmox_ptp_vm_config

2
keywords/cloud_platform/system/ptp/ptp_readiness_keywords.py
View File

@@ -52,7 +52,7 @@ class PTPReadinessKeywords:
return observed_states
validate_equals_with_retry(lambda: check_port_state_in_port_data_set(name), expected_port_states, "port state in port data set", 120, 30)
validate_equals_with_retry(lambda: check_port_state_in_port_data_set(name), expected_port_states, "port state in port data set", 180, 30)
def wait_for_clock_class_appear_in_grandmaster_settings_np(self, name: str, expected_clock_class: Union[int, list]) -> None:
"""

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

@@ -128,7 +128,7 @@ class PMCKeywords(BaseKeyword):
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:
def pmc_get_time_properties_data_set(self, config_file: str, socket_file: str = None, unicast: bool = True, boundry_clock: int = 0) -> PMCGetTimePropertiesDataSetOutput:
"""
Gets the time_properties_data_set_object
@@ -144,7 +144,10 @@ class PMCKeywords(BaseKeyword):
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'"
if socket_file:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET TIME_PROPERTIES_DATA_SET'"
else:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} 'GET TIME_PROPERTIES_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_time_properties_data_set_output = PMCGetTimePropertiesDataSetOutput(output)
@@ -153,13 +156,13 @@ class PMCKeywords(BaseKeyword):
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 set
Args:
config_file (str): the config file
socket_file (str): the socket file
unicast (bool): true to use unicast
boundry_clock (int): the boundry clock
Returns:
PMCGetDefaultDataSetOutput: the default data set output
@@ -171,22 +174,25 @@ class PMCKeywords(BaseKeyword):
pmc_get_default_data_set_output = PMCGetDefaultDataSetOutput(output)
return pmc_get_default_data_set_output
def pmc_get_port_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetPortDataSetOutput:
def pmc_get_port_data_set(self, config_file: str, socket_file: str = None, unicast: bool = True, boundry_clock: int = 0) -> PMCGetPortDataSetOutput:
"""
Gets the port data set
Args:
config_file (str): the config file
socket_file (str): the socket file
unicast (bool): true to use unicast
boundry_clock (int): the boundry clock
Returns:
PMCGetPortDataSetOutput: the port data set output
Example: PMCKeywords(ssh_connection).pmc_get_port_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 PORT_DATA_SET'"
if socket_file:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET PORT_DATA_SET'"
else:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} 'GET PORT_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_port_data_set_output = PMCGetPortDataSetOutput(output)
@@ -214,7 +220,7 @@ class PMCKeywords(BaseKeyword):
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:
def pmc_get_parent_data_set(self, config_file: str, socket_file: str = None, unicast: bool = True, boundry_clock: int = 0) -> PMCGetParentDataSetOutput:
"""
Gets the parent data set
@@ -230,7 +236,10 @@ class PMCKeywords(BaseKeyword):
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'"
if socket_file:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET PARENT_DATA_SET'"
else:
cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} 'GET PARENT_DATA_SET'"
output = self.ssh_connection.send_as_sudo(cmd)
pmc_get_parent_data_set_output = PMCGetParentDataSetOutput(output)

157
keywords/ptp/proxmox_keywords.py Normal file
View File

@@ -0,0 +1,157 @@
import time
from config.host.objects.host_configuration import HostConfiguration
from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from framework.ssh.ssh_connection_manager import SSHConnectionManager
from keywords.base_keyword import BaseKeyword
class ProxmoxKeywords(BaseKeyword):
"""
Class for Proxmox PTP VM Keywords
"""
def __init__(self, proxmox_vm_config: HostConfiguration):
"""Initializes the ProxmoxKeywords.
Args:
proxmox_vm_config (HostConfiguration): Proxmox VM configuration
"""
self.proxmox_vm_config = proxmox_vm_config
self.proxmox_vm_connection = None
def get_proxmox_vm_ssh_connection(self) -> SSHConnection:
"""
Gets the PTP VM SSH connection using configuration from proxmox_ptp_vm_config
Returns:
SSHConnection: the SSH connection to the PTP VM
"""
if self.proxmox_vm_connection is None:
if not self.proxmox_vm_config:
raise ValueError("No proxmox_ptp_vm_config found in PTP NIC configuration")
self.proxmox_vm_connection = SSHConnectionManager.create_ssh_connection(
self.proxmox_vm_config.get_host(),
self.proxmox_vm_config.get_credentials().get_user_name(),
self.proxmox_vm_config.get_credentials().get_password(),
)
get_logger().log_info(f"Connected to proxmox VM at {self.proxmox_vm_config.get_host()}")
return self.proxmox_vm_connection
def start_ptp_service(self) -> str:
"""
Starts the PTP service by running runptp.sh script in background
Creates the runptp.sh script if it doesn't exist
Returns:
str: the command output
"""
ssh_connection = self.get_proxmox_vm_ssh_connection()
get_logger().log_info("Starting PTP service with runptp.sh")
# Check if runptp.sh exists
check_cmd = "test -f ./runptp.sh && echo 'exists' || echo 'not found'"
check_output = ssh_connection.send(check_cmd)
if "not found" in str(check_output):
get_logger().log_info("runptp.sh not found, creating it")
# Create runptp.sh with the required content
create_script_cmd = """cat > runptp.sh << 'EOF'
#!/bin/bash
ptp4l -2 -S -m -A -f /etc/ptp4l.conf
EOF"""
ssh_connection.send(create_script_cmd)
# Make it executable
chmod_cmd = "chmod +x runptp.sh"
ssh_connection.send(chmod_cmd)
get_logger().log_info("runptp.sh created and made executable")
else:
get_logger().log_info("runptp.sh already exists")
# Start the runptp.sh script in background
cmd = "nohup ./runptp.sh > /dev/null 2>&1"
output = ssh_connection.send_as_sudo(cmd)
time.sleep(10) # Wait longer for service to stabilize
service_running = self._verify_ptp_service_running()
if service_running:
get_logger().log_info("PTP service auto-recovery completed successfully")
else:
raise Exception("Failed to start PTP service")
get_logger().log_info("PTP service started")
return output
def _verify_ptp_service_running(self) -> bool:
"""
Verifies that the PTP service (runptp.sh) is running
Returns:
bool: True if the service is running, False otherwise
"""
ssh_connection = self.get_proxmox_vm_ssh_connection()
get_logger().log_info("Checking if PTP service is running")
cmd = "ps aux | grep runptp"
output = ssh_connection.send(cmd)
# Handle both string and list outputs
if isinstance(output, list):
output_lines = output
else:
output_lines = output.split("\n")
# Check if runptp processes are found (excluding the grep command itself)
running_processes = [line for line in output_lines if "runptp" in line and "grep" not in line]
is_running = len(running_processes) > 0
if is_running:
get_logger().log_info(f"PTP service is running. Found {len(running_processes)} processes:")
for process in running_processes:
get_logger().log_info(f" {process.strip()}")
else:
get_logger().log_warning("PTP service is not running")
return is_running
def verify_ptp_service_with_auto_recovery(self):
"""
Verifies PTP service is running and automatically starts it if not running
Includes automated recovery mechanism for PTP service management
"""
get_logger().log_info("Verifying PTP service status with auto-recovery")
# Check if service is running, start if needed
service_running = self._verify_ptp_service_running()
if not service_running:
get_logger().log_warning("PTP service is not running - initiating auto-recovery")
self.start_ptp_service()
def stop_ptp_service(self) -> str:
"""
Stops the PTP service by killing runptp.sh processes
Returns:
str: the command output
"""
ssh_connection = self.get_proxmox_vm_ssh_connection()
get_logger().log_info("Stopping PTP service")
# Kill all runptp processes
cmd = "pkill -f runptp.sh"
output = ssh_connection.send_as_sudo(cmd)
get_logger().log_info("PTP service stopped")
return output

149
testcases/cloud_platform/regression/ptp/test_proxmox_ptp_verification.py Normal file
View File

@@ -0,0 +1,149 @@
from pytest import mark
from config.configuration_manager import ConfigurationManager
from framework.logging.automation_logger import get_logger
from framework.resources.resource_finder import get_stx_resource_path
from framework.validation.validation import validate_equals, validate_list_contains
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.ptp.pmc.pmc_keywords import PMCKeywords
from keywords.ptp.proxmox_keywords import ProxmoxKeywords
from keywords.ptp.setup.ptp_setup_reader import PTPSetupKeywords
@mark.p2
@mark.lab_has_compute
@mark.lab_has_ptp_configuration_compute
def test_proxmox_ptp_vm_verification(request):
"""
Test PTP VM verification with automatic service recovery.
This test verifies PTP (Precision Time Protocol) functionality in a Proxmox VM environment
by checking service status, retrieving PTP data sets, and validating against expected values.
Test Steps:
- Connect to PTP VM and setup test environment
- Verify PTP service is running (auto-start if needed)
- Validate PORT_DATA_SET - Check port state is SLAVE
- Validate PARENT_DATA_SET - Verify GM clock properties
- Validate TIME_PROPERTIES_DATA_SET - Check UTC offset and traceability
- Cross-validate parent port identity with master configuration
Expected Results:
- PTP service runs successfully with auto-recovery capability
- Port operates in SLAVE state as expected
- Parent data set matches expected GM clock configuration
- Time properties align with system UTC settings
- Parent port identity correctly maps to master port
Preconditions:
- System is set up with valid PTP configuration as defined in ptp_configuration_expectation_compute.json5.
"""
def cleanup_ptp_service():
"""Cleanup function to stop PTP service after test completion"""
get_logger().log_info("Test cleanup: Stopping PTP service")
proxmox_keywords.stop_ptp_service()
request.addfinalizer(cleanup_ptp_service)
lab_connect_keywords = LabConnectionKeywords()
controller_0_ssh_connection = lab_connect_keywords.get_ssh_for_hostname("controller-0")
# Get Proxmox VM configuration for PTP testing
ptp_config = ConfigurationManager.get_ptp_config()
proxmox_vm_config = ptp_config.get_host("controller_0").get_nic("nic1").get_proxmox_ptp_vm_config()
proxmox_keywords = ProxmoxKeywords(proxmox_vm_config)
# PTP configuration file path in the VM
ptp_config_file = "/etc/ptp4l.conf"
# Load expected PTP configuration from template
ptp_setup_keywords = PTPSetupKeywords()
expected_config_template_path = get_stx_resource_path("resources/ptp/setup/ptp_configuration_expectation_compute.json5")
# Define PTP instances and interfaces for both controllers
ptp_instance_selection = [("ptp1", "controller-0", ["ptp1if1", "ptp1if2"]), ("ptp1", "controller-1", ["ptp1if1", "ptp1if2"])]
ptp_expected_setup = ptp_setup_keywords.filter_and_render_ptp_config(expected_config_template_path, ptp_instance_selection)
# Get expected configuration for ptp1 instance
ptp1_expected_config = ptp_expected_setup.get_ptp4l_expected_by_name("ptp1")
get_logger().log_info("Starting PTP VM verification with auto-recovery capability")
# Verify PTP service is running, auto-start if needed
get_logger().log_info("Verifying PTP service status and enabling auto-recovery")
proxmox_keywords.verify_ptp_service_with_auto_recovery()
# Initialize PMC (PTP Management Client) for data retrieval
proxmox_vm_ssh_connection = proxmox_keywords.get_proxmox_vm_ssh_connection()
pmc_keywords = PMCKeywords(proxmox_vm_ssh_connection)
get_logger().log_info("Validating PORT_DATA_SET - checking port state")
# Retrieve current port data set from PTP service
port_data_set_response = pmc_keywords.pmc_get_port_data_set(ptp_config_file)
observed_port_data_sets = port_data_set_response.get_pmc_get_port_data_set_objects()
# Ensure we have at least one port data set object
if len(observed_port_data_sets) < 1:
raise Exception(f"Expected at least 1 port data set object, but found {len(observed_port_data_sets)}")
# Use the first port data set for validation
current_port_data_set = observed_port_data_sets[0]
# Validate port is operating in SLAVE state (receiving time from master)
validate_equals(current_port_data_set.get_port_state(), "SLAVE", "Port state should be SLAVE (receiving time from master)")
get_logger().log_info("Validating PARENT_DATA_SET - checking GM clock properties")
# Retrieve parent data set (information about the master clock)
parent_data_set_response = pmc_keywords.pmc_get_parent_data_set(ptp_config_file)
current_parent_data_set = parent_data_set_response.get_pmc_get_parent_data_set_object()
# Get expected parent data set configuration for controller-1
expected_parent_data_set = ptp1_expected_config.get_parent_data_set_for_hostname("controller-1")
validate_list_contains(current_parent_data_set.get_gm_clock_class(), expected_parent_data_set.get_gm_clock_class(), "gm.ClockClass value within GET PARENT_DATA_SET")
validate_equals(current_parent_data_set.get_gm_clock_accuracy(), expected_parent_data_set.get_gm_clock_accuracy(), "gm.ClockAccuracy value within GET PARENT_DATA_SET")
validate_equals(current_parent_data_set.get_gm_offset_scaled_log_variance(), expected_parent_data_set.get_gm_offset_scaled_log_variance(), "gm.OffsetScaledLogVariance value within GET PARENT_DATA_SET")
get_logger().log_info("Cross-validating parent port identity with master port data set")
# Get master port data set from PTP configuration
master_port_response = PMCKeywords(controller_0_ssh_connection).pmc_get_port_data_set("/etc/linuxptp/ptpinstance/ptp4l-ptp1.conf", "/var/run/ptp4l-ptp1")
master_port_objects = master_port_response.get_pmc_get_port_data_set_objects()
if len(master_port_objects) < 1:
raise Exception(f"Expected at least 1 port data set object, but found {len(master_port_objects)}")
# Extract master port identity (use first available port)
expected_master_port_identity = master_port_objects[0].get_port_identity()
# Compare parent port identity with master port identity (clock ID portion)
current_parent_port_identity = current_parent_data_set.get_parent_port_identity()
expected_port_identity = expected_master_port_identity.split("-")[0]
current_port_identity = current_parent_port_identity.split("-")[0]
validate_equals(current_port_identity, expected_port_identity, "Parent port identity matches the master port identity")
get_logger().log_info("Validating TIME_PROPERTIES_DATA_SET - checking UTC offset and traceability")
# Retrieve time properties data set (UTC and traceability information)
time_properties_response = pmc_keywords.pmc_get_time_properties_data_set(ptp_config_file)
current_time_properties = time_properties_response.get_pmc_get_time_properties_data_set_object()
# Extract current time properties
current_utc_offset = current_time_properties.get_current_utc_offset()
current_utc_offset_valid = current_time_properties.get_current_utc_off_set_valid()
current_time_traceable = current_time_properties.get_time_traceable()
current_frequency_traceable = current_time_properties.get_frequency_traceable()
# Get expected time properties configuration
expected_time_properties = ptp1_expected_config.get_time_properties_data_set_for_hostname("controller-1")
expected_utc_offset = expected_time_properties.get_current_utc_offset()
expected_utc_offset_valid = expected_time_properties.get_current_utc_offset_valid()
expected_time_traceable = expected_time_properties.get_time_traceable()
expected_frequency_traceable = expected_time_properties.get_frequency_traceable()
# Validate all time properties
validate_equals(current_utc_offset, expected_utc_offset, "currentUtcOffset value within GET TIME_PROPERTIES_DATA_SET")
validate_equals(current_utc_offset_valid, expected_utc_offset_valid, "currentUtcOffsetValid value within GET TIME_PROPERTIES_DATA_SET")
validate_equals(current_time_traceable, expected_time_traceable, "timeTraceable value within GET TIME_PROPERTIES_DATA_SET")
validate_equals(current_frequency_traceable, expected_frequency_traceable, "frequencyTraceable value within GET TIME_PROPERTIES_DATA_SET")
get_logger().log_info("TIME_PROPERTIES_DATA_SET validation completed - all time properties verified")

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

@@ -404,7 +404,7 @@ def test_ptp_operation_sma_disabled_and_enable():
"frequency_traceable": 1 // Frequency of the clock is traceable to a stable
},
"grandmaster_settings": {
"clock_class": 165, // The GM clock loses its connection to the Primary Reference Time Clock
"clock_class": [165, 248], // The GM clock loses its connection to the Primary Reference Time Clock
"clock_accuracy": "0xfe", // Unknown
"offset_scaled_log_variance": "0xffff", // Unknown or unspecified stability
"time_traceable": 0, // Time is not traceable — the clock may be in holdover, unsynchronized, or degraded.
@@ -465,8 +465,7 @@ def test_ptp_operation_sma_disabled_and_enable():
get_logger().log_info("Verifying PTP operation and corresponding status changes when SMA is enabled.")
ctrl0_nic2_sma1_enable_ptp_selection = [("ptp3", "controller-0", []), ("ptp4", "controller-1", [])]
ctrl0_nic2_sma1_enable_exp_dict_overrides = {"ptp4l": [{"name": "ptp4", "controller-1": {"grandmaster_settings": {"clock_class": 165}}}]}
ctrl0_nic2_sma1_enable_exp_ptp_setup = ptp_setup_keywords.filter_and_render_ptp_config(ptp_setup_template_path, ctrl0_nic2_sma1_enable_ptp_selection, expected_dict_overrides=ctrl0_nic2_sma1_enable_exp_dict_overrides)
ctrl0_nic2_sma1_enable_exp_ptp_setup = ptp_setup_keywords.filter_and_render_ptp_config(ptp_setup_template_path, ctrl0_nic2_sma1_enable_ptp_selection)
sma_keywords.enable_sma("controller-0", "nic2")
get_logger().log_info("Waiting for 100.119 alarm to clear after SMA1 is enabled")
@@ -991,7 +990,7 @@ def test_ptp_operation_service_stop_start_restart():
get_logger().log_info("Verifying PMC configuration and clock class after service restart...")
ptp_readiness_keywords = PTPReadinessKeywords(LabConnectionKeywords().get_ssh_for_hostname("controller-0"))
ptp_readiness_keywords.wait_for_clock_class_appear_in_grandmaster_settings_np("ptp1", "controller-0", 6)
ptp_readiness_keywords.wait_for_clock_class_appear_in_grandmaster_settings_np("ptp1", 6)
ptp_readiness_keywords.wait_for_gm_clock_class_appear_in_parent_data_set("ptp1", 6)
ptp_verify_config_keywords.verify_ptp_pmc_values(check_domain=False)