github/krkn
1
0
Fork 0
mirror of https://github.com/krkn-chaos/krkn.git synced 2026-08-25 09:27:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: fail node actions when actions list is empty (#1273)

Browse Source 
Signed-off-by: 1PoPTRoN <vrxn.arp1traj@gmail.com>
This commit is contained in:
Arpit Raj
2026-05-08 09:06:05 -04:00
committed by GitHub
parent 458f89f30e
commit 4e87841aca
2 changed files with 88 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
krkn/scenario_plugins/node_actions/node_actions_scenario_plugin.py
+19 -13
View File
@@ -58,23 +58,29 @@ class NodeActionsScenarioPlugin(AbstractScenarioPlugin):
) -> int:
with open(scenario, "r") as f:
node_scenario_config = yaml.safe_load(f)
for node_scenario in node_scenario_config["node_scenarios"]:
for index, node_scenario in enumerate(node_scenario_config["node_scenarios"]):
try:
actions = node_scenario.get("actions")
if not actions:
logging.error(
"NodeActionsScenarioPlugin: 'actions' must be defined and non-empty in %s node_scenarios[%s]"
% (scenario, index)
)
return 1
node_scenario_object = self.get_node_scenario_object(
node_scenario, lib_telemetry.get_lib_kubernetes()
)
if node_scenario["actions"]:
for action in node_scenario["actions"]:
start_time = int(time.time())
self.inject_node_scenario(
action,
node_scenario,
node_scenario_object,
lib_telemetry.get_lib_kubernetes(),
scenario_telemetry,
)
end_time = int(time.time())
cerberus.get_status(start_time, end_time)
for action in actions:
start_time = int(time.time())
self.inject_node_scenario(
action,
node_scenario,
node_scenario_object,
lib_telemetry.get_lib_kubernetes(),
scenario_telemetry,
)
end_time = int(time.time())
cerberus.get_status(start_time, end_time)
except (RuntimeError, Exception) as e:
logging.error("Node Actions exiting due to Exception %s" % e)
return 1
tests/test_node_actions_scenario_plugin.py
+69 -1
View File
@@ -687,7 +687,8 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase):
scenario_yaml = {
"node_scenarios": [
{
"cloud_type": "unsupported"
"cloud_type": "unsupported",
"actions": ["node_stop_scenario"]
}
]
}
@@ -703,6 +704,73 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase):
self.assertEqual(result, 1)
mock_logging.assert_called()
def _assert_run_returns_error_for_invalid_actions(self, node_scenario):
scenario_yaml = {
"node_scenarios": [
node_scenario
]
}
with patch('yaml.safe_load', return_value=scenario_yaml):
with self.assertLogs('root', level='ERROR') as log_ctx:
result = self.plugin.run(
"test-uuid",
"/path/to/scenario.yaml",
self.mock_lib_telemetry,
self.mock_scenario_telemetry
)
self.assertEqual(result, 1)
self.assertTrue(
any("actions" in msg for msg in log_ctx.output),
f"Expected 'actions' in error log, got: {log_ctx.output}",
)
self.assertTrue(
any("/path/to/scenario.yaml" in msg for msg in log_ctx.output),
f"Expected scenario file path in error log, got: {log_ctx.output}",
)
@patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios')
@patch('builtins.open', new_callable=mock_open)
def test_run_returns_error_when_actions_empty(self, mock_file, mock_general_scenarios):
"""
Test run returns 1 when actions is an empty list
"""
self._assert_run_returns_error_for_invalid_actions(
{
"cloud_type": "generic",
"actions": []
}
)
mock_general_scenarios.assert_not_called()
@patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios')
@patch('builtins.open', new_callable=mock_open)
def test_run_returns_error_when_actions_missing(self, mock_file, mock_general_scenarios):
"""
Test run returns 1 when actions is missing
"""
self._assert_run_returns_error_for_invalid_actions(
{
"cloud_type": "generic"
}
)
mock_general_scenarios.assert_not_called()
@patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios')
@patch('builtins.open', new_callable=mock_open)
def test_run_returns_error_when_actions_none(self, mock_file, mock_general_scenarios):
"""
Test run returns 1 when actions is None
"""
self._assert_run_returns_error_for_invalid_actions(
{
"cloud_type": "generic",
"actions": None
}
)
mock_general_scenarios.assert_not_called()
@patch('logging.info')
def test_multiprocess_nodes(self, mock_logging):
"""