starlingx/test
Code Issues Proposed changes

Added keywords for power on and power off GNSS

Browse Source 
Change-Id: I2f1d55e507fab091fea6503f6fa5b714dcb29e41
Signed-off-by: Guntaka Umashankar Reddy <umashankarguntaka.reddy@windriver.com>
This commit is contained in:
Guntaka Umashankar Reddy
2025-03-21 10:38:42 -04:00
parent fb55a68f1d
commit d1bed50b29
4 changed files with 226 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
config/ptp/objects/ptp_config.py
View File

@@ -28,6 +28,33 @@ class PTPConfig:
# Extract the NIC Connections and Host information from the dictionary
self.ptp_hosts = self._extract_ptp_hosts(ptp_dict)
def get_gnss_server_host(self) -> str:
"""
Getter for the GNSS server host.
Returns:
str: gnss server host
"""
return self.gnss_server_host
def get_gnss_server_username(self) -> str:
"""
Getter for the GNSS server username.
Returns:
str: gnss server username
"""
return self.gnss_server_username
def get_gnss_server_password(self) -> str:
"""
Getter for the GNSS server password.
Returns:
str: gnss server password
"""
return self.gnss_server_password
def _extract_ptp_hosts(self, ptp_dict: Dict[str, str]) -> List[PTPHost]:
"""
Build the PTPHost objects from the dictionary

50
keywords/cloud_platform/ssh/lab_connection_keywords.py
View File

@@ -15,8 +15,9 @@ class LabConnectionKeywords(BaseKeyword):
def get_active_controller_ssh(self) -> SSHConnection:
"""
Gets the active controller ssh
Returns: the ssh for the active controller
Returns:
SSHConnection: the ssh for the active controller
"""
lab_config = ConfigurationManager.get_lab_config()
@@ -37,15 +38,16 @@ class LabConnectionKeywords(BaseKeyword):
def get_standby_controller_ssh(self) -> SSHConnection:
"""
Gets the standby controller ssh
Returns: the ssh for the standby controller
Returns:
SSHConnection: the ssh for the standby controller
"""
lab_config = ConfigurationManager.get_lab_config()
# get the standby controller
standby_controller = SystemHostListKeywords(self.get_active_controller_ssh()).get_standby_controller()
if not standby_controller:
raise KeywordException('System does not have a standby controller')
raise KeywordException("System does not have a standby controller")
standby_host_name = standby_controller.get_host_name()
@@ -65,13 +67,44 @@ class LabConnectionKeywords(BaseKeyword):
return connection
def get_ssh_for_hostname(self, hostname: str) -> SSHConnection:
"""
Gets the ssh connection for the hostname
Args:
hostname (str): The name of the host
Returns:
SSHConnection: the ssh for the hostname
"""
lab_config = ConfigurationManager.get_lab_config()
host_ip = lab_config.get_node(hostname).get_ip()
jump_host_config = None
if lab_config.is_use_jump_server():
jump_host_config = lab_config.get_jump_host_configuration()
connection = SSHConnectionManager.create_ssh_connection(
host_ip,
lab_config.get_admin_credentials().get_user_name(),
lab_config.get_admin_credentials().get_password(),
ssh_port=lab_config.get_ssh_port(),
jump_host=jump_host_config,
)
return connection
def get_compute_ssh(self, compute_name: str) -> SSHConnection:
"""
Gets an SSH connection to the 'Compute' node whose name is specified by the argument 'compute_name'.
Args:
compute_name (string): The name of the 'Compute' node.
Returns: the SSH connection to the 'Compute' node whose name is specified by the argument 'compute_name'.
Args:
compute_name (str): The name of the 'Compute' node.
Returns:
SSHConnection: the SSH connection to the 'Compute' node whose name is specified by the argument 'compute_name'.
NOTE: this 'ssh connection' actually uses ssh_pass to make a call from the active controller connection.
@@ -88,9 +121,10 @@ class LabConnectionKeywords(BaseKeyword):
"""
Gets an SSH connection to the 'Subcloud' node whose name is specified by the argument 'subcloud_name'.
Args:
subcloud_name (string): The name of the 'subcloud' node.
subcloud_name (str): The name of the 'subcloud' node.
Returns: the SSH connection to the 'subcloud' node whose name is specified by the argument 'subcloud_name'.
Returns:
SSHConnection: the SSH connection to the 'subcloud' node whose name is specified by the argument 'subcloud_name'.
"""
lab_config = ConfigurationManager.get_lab_config()

27
keywords/cloud_platform/ssh/ptp_connection_keywords.py Normal file
View File

@@ -0,0 +1,27 @@
from config.configuration_manager import ConfigurationManager
from framework.ssh.ssh_connection import SSHConnection
from framework.ssh.ssh_connection_manager import SSHConnectionManager
from keywords.base_keyword import BaseKeyword
class PTPConnectionKeywords(BaseKeyword):
"""
Class to hold PTP connection keywords
"""
def get_gnss_server_ssh(self) -> SSHConnection:
"""
Gets the gnss server ssh
Returns:
SSHConnection: the ssh for the gnss server
"""
ptp_config = ConfigurationManager.get_ptp_config()
connection = SSHConnectionManager.create_ssh_connection(
ptp_config.get_gnss_server_host(),
ptp_config.get_gnss_server_username(),
ptp_config.get_gnss_server_password(),
)
return connection

130
keywords/ptp/gnss_keywords.py Normal file
View File

@@ -0,0 +1,130 @@
import time
from multiprocessing import get_logger
from time import sleep
from config.configuration_manager import ConfigurationManager
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.ssh.ptp_connection_keywords import PTPConnectionKeywords
from keywords.ptp.cat.cat_ptp_cgu_keywords import CatPtpCguKeywords
class GnssKeywords(BaseKeyword):
"""
Gnss power on and off using GNSS SSH connection.
"""
def __init__(self):
"""
Initializes the GnssKeywords.
"""
def gnss_power_on(self, hostname: str, nic: str) -> None:
"""
Power on gnss
Args:
hostname (str) : The name of the host
nic (str) : The name of the nic
"""
ptp_connect_keywords = PTPConnectionKeywords()
gnss_ssh_connection = ptp_connect_keywords.get_gnss_server_ssh()
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"
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"
# power on gnss
gnss_ssh_connection.send_as_sudo(command)
expected_gnss_1pps_state = "valid"
expected_pps_dpll_status = ["locked_ho_acq"]
self.validate_gnss_1pps_state_and_pps_dpll_status(hostname, cgu_location, expected_gnss_1pps_state, expected_pps_dpll_status)
def gnss_power_off(self, hostname: str, nic: str) -> None:
"""
Power off gnss
Args:
hostname (str) : The name of the host
nic (str) : The name of the nic
"""
ptp_connect_keywords = PTPConnectionKeywords()
gnss_ssh_connection = ptp_connect_keywords.get_gnss_server_ssh()
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"
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"
# power off gnss
gnss_ssh_connection.send_as_sudo(command)
expected_gnss_1pps_state = "invalid"
expected_pps_dpll_status = ["holdover", "freerun"]
self.validate_gnss_1pps_state_and_pps_dpll_status(hostname, cgu_location, expected_gnss_1pps_state, expected_pps_dpll_status)
def validate_gnss_1pps_state_and_pps_dpll_status(
self,
hostname: str,
cgu_location: str,
expected_gnss_1pps_state: str,
expected_pps_dpll_status: list,
timeout: int = 800,
polling_sleep_time: int = 60,
) -> None:
"""
Validates the GNSS-1PPS state and PPS DPLL status within the specified time.
Args:
hostname (str): The name of the host.
cgu_location (str): the cgu location.
expected_gnss_1pps_state (str): The expected gnss 1pss state value.
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.
Raises:
TimeoutError: raised when validate does not equal in the required time
"""
get_logger().log_info("Attempting Validation - GNSS-1PPS state and PPS DPLL status")
end_time = time.time() + timeout
lab_connect_keywords = LabConnectionKeywords()
ssh_connection = lab_connect_keywords.get_ssh_for_hostname(hostname)
cat_ptp_cgu_keywords = CatPtpCguKeywords(ssh_connection)
# Attempt the validation
while True:
# Compute the actual status and state that we are trying to validate.
ptp_cgu_output = cat_ptp_cgu_keywords.cat_ptp_cgu(cgu_location)
ptp_cgu_component = ptp_cgu_output.get_cgu_component()
pps_dpll_object = ptp_cgu_component.get_pps_dpll()
status = pps_dpll_object.get_status()
input_object = ptp_cgu_component.get_cgu_input("GNSS-1PPS")
state = input_object.get_state()
if status in expected_pps_dpll_status and state == expected_gnss_1pps_state:
get_logger().log_info("Validation Successful - GNSS-1PPS state and PPS DPLL status")
return
else:
get_logger().log_info("Validation Failed")
get_logger().log_info(f"Expected GNSS-1PPS state: {expected_gnss_1pps_state}")
get_logger().log_info(f"Observed GNSS-1PPS state: {state}")
get_logger().log_info(f"Expected PPS DPLL status: {expected_pps_dpll_status}")
get_logger().log_info(f"Observed PPS DPLL status: {status}")
if time.time() < end_time:
get_logger().log_info(f"Retrying in {polling_sleep_time}s")
sleep(polling_sleep_time)
# Move on to the next iteration
else:
raise TimeoutError("Timeout performing validation - GNSS-1PPS state and PPS DPLL status")