Add host_swact_force keyword and fix minor issue in GrandmasterSettings.clock_class.

Change-Id: Iaf7de63f303931bc754f0b4206b88b791046c603
Signed-off-by: Guntaka Umashankar Reddy <umashankarguntaka.reddy@windriver.com>
This commit is contained in:
Guntaka Umashankar Reddy
2025-06-12 11:10:23 -04:00
parent 971942681f
commit 4bb9012f23
6 changed files with 36 additions and 16 deletions

View File

@@ -21,23 +21,16 @@ class SystemHostSwactKeywords(BaseKeyword):
"""
self.ssh_connection = ssh_connection
def host_swact(self, force: bool = False):
def host_swact(self) -> bool:
"""
Performs a controller switchover action (swact).
Args:
force: If True, forces the swact operation even if there are warnings.
If False, performs a normal swact without force option.
Returns:
bool: True if swact was successful, raises exception otherwise.
"""
active_controller = SystemHostListKeywords(self.ssh_connection).get_active_controller()
standby_controller = SystemHostListKeywords(self.ssh_connection).get_standby_controller()
if force:
self.ssh_connection.send(source_openrc(f"system host-swact --force {active_controller.get_host_name()}"))
else:
self.ssh_connection.send(source_openrc(f"system host-swact {active_controller.get_host_name()}"))
self.ssh_connection.send(source_openrc(f"system host-swact {active_controller.get_host_name()}"))
self.validate_success_return_code(self.ssh_connection)
is_swact_success = self.wait_for_swact(active_controller, standby_controller)
if not is_swact_success:
@@ -46,6 +39,24 @@ class SystemHostSwactKeywords(BaseKeyword):
raise KeywordException(f"Swact was not successful. Current active controller is {active_controller} " f"and standby controller is {standby_controller}")
return True
def host_swact_force(self) -> bool:
"""
Performs a controller force switchover action (swact).
Returns:
bool: True if swact --force was successful, raises exception otherwise.
"""
active_controller = SystemHostListKeywords(self.ssh_connection).get_active_controller()
standby_controller = SystemHostListKeywords(self.ssh_connection).get_standby_controller()
self.ssh_connection.send(source_openrc(f"system host-swact --force {active_controller.get_host_name()}"))
self.validate_success_return_code(self.ssh_connection)
is_swact_success = self.wait_for_swact(active_controller, standby_controller)
if not is_swact_success:
active_controller = SystemHostListKeywords(self.ssh_connection).get_active_controller()
standby_controller = SystemHostListKeywords(self.ssh_connection).get_standby_controller()
raise KeywordException(f"Force swact was not successful. Current active controller is {active_controller} " f"and standby controller is {standby_controller}")
return True
def wait_for_swact(
self,
current_active_controller: SystemHostObject,

View File

@@ -16,7 +16,8 @@ class GrandmasterSettings:
"""
if "clock_class" not in expected_dict:
raise Exception("Every expected dict should have a clock_class.")
self.clock_class = expected_dict["clock_class"]
clock_class = expected_dict["clock_class"]
self.clock_class = clock_class if isinstance(clock_class, list) else [clock_class]
if "clock_accuracy" not in expected_dict:
raise Exception("Every expected dict should have a clock_accuracy.")
@@ -42,12 +43,12 @@ class GrandmasterSettings:
raise Exception("Every expected dict should have a current_utc_offset_valid.")
self.current_utc_offset_valid = expected_dict["current_utc_offset_valid"]
def get_clock_class(self) -> int:
def get_clock_class(self) -> list:
"""
Gets the clock class.
Returns:
int: The clock class.
list: The clock class.
"""
return self.clock_class

View File

@@ -3,7 +3,7 @@ from typing import Any, Dict, List
from keywords.ptp.setup.object.grandmaster_settings import GrandmasterSettings
from keywords.ptp.setup.object.parent_data_set import ParentDataSet
from keywords.ptp.setup.object.port_data_set import PortDataSet
from keywords.ptp.setup.time_properties_data_set import TimePropertiesDataSet
from keywords.ptp.setup.object.time_properties_data_set import TimePropertiesDataSet
class PTP4LExpectedDict:

View File

@@ -291,7 +291,7 @@
"parent_data_set" : {{ parent_data_set_tgm_default }},
"time_properties_data_set": {{ time_properties_data_set_tgm_default }},
"grandmaster_settings": {
"clock_class": 165,
"clock_class": [165, 248],
"clock_accuracy": "0xfe",
"offset_scaled_log_variance": "0xffff",
"time_traceable": 0,
@@ -364,7 +364,15 @@
"controller-1" : {
"parent_data_set" : {{ parent_data_set_tgm_default }},
"time_properties_data_set": {{ time_properties_data_set_tgm_default }},
"grandmaster_settings": {{ grandmaster_settings.grandmaster_settings_tbc_default }},
"grandmaster_settings": {
"clock_class": [165, 248],
"clock_accuracy": "0xfe",
"offset_scaled_log_variance": "0xffff",
"time_traceable": 0,
"frequency_traceable": 0,
"time_source": "0xa0",
"current_utc_offset_valid": 0
},
"port_data_set": [
{
"interface": "{{ controller_1.nic2.nic_connection.interface }}",

View File

@@ -1126,7 +1126,7 @@ def test_ptp_host_operation_force_switchover(request):
get_logger().log_info("Performing controller force switchover operation")
system_host_swact_keywords = SystemHostSwactKeywords(ssh_connection)
system_host_swact_keywords.host_swact(force=True)
system_host_swact_keywords.host_swact_force()
swact_success = system_host_swact_keywords.wait_for_swact(active_controller, standby_controller)
validate_equals(swact_success, True, "Host swact")