diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a8498bb1..718ed7de 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -74,7 +74,6 @@ jobs: kubectl get nodes --show-labels=true # Pull request only steps - name: Run unit tests - if: github.event_name == 'pull_request' run: python -m coverage run -a -m unittest discover -s tests -v - name: Setup Pull Request Functional Tests @@ -102,7 +101,7 @@ jobs: echo "test_pod_network_filter" >> ./CI/tests/functional_tests echo "test_pod_server" >> ./CI/tests/functional_tests echo "test_node" >> ./CI/tests/functional_tests - echo "test_pvc" >> ./CI/tests/functional_tests + # echo "test_pvc" >> ./CI/tests/functional_tests # Push on main only steps + all other functional to collect coverage # for the badge @@ -124,11 +123,11 @@ jobs: yq -i '.performance_monitoring.prometheus_url="http://localhost:9090"' CI/config/common_test_config.yaml yq -i '.telemetry.username="${{secrets.TELEMETRY_USERNAME}}"' CI/config/common_test_config.yaml yq -i '.telemetry.password="${{secrets.TELEMETRY_PASSWORD}}"' CI/config/common_test_config.yaml - echo "test_telemetry" > ./CI/tests/functional_tests echo "test_service_hijacking" >> ./CI/tests/functional_tests echo "test_app_outages" >> ./CI/tests/functional_tests echo "test_container" >> ./CI/tests/functional_tests echo "test_pod" >> ./CI/tests/functional_tests + echo "test_telemetry" > ./CI/tests/functional_tests echo "test_pod_error" >> ./CI/tests/functional_tests echo "test_customapp_pod" >> ./CI/tests/functional_tests echo "test_namespace" >> ./CI/tests/functional_tests @@ -140,7 +139,7 @@ jobs: echo "test_pod_network_filter" >> ./CI/tests/functional_tests echo "test_pod_server" >> ./CI/tests/functional_tests echo "test_node" >> ./CI/tests/functional_tests - echo "test_pvc" >> ./CI/tests/functional_tests + # echo "test_pvc" >> ./CI/tests/functional_tests # Final common steps - name: Run Functional tests env: diff --git a/CI/tests/test_container.sh b/CI/tests/test_container.sh index 9042b021..271b43fa 100755 --- a/CI/tests/test_container.sh +++ b/CI/tests/test_container.sh @@ -16,8 +16,10 @@ function functional_test_container_crash { export post_config="" envsubst < CI/config/common_test_config.yaml > CI/config/container_config.yaml - python3 -m coverage run -a run_kraken.py -c CI/config/container_config.yaml + python3 -m coverage run -a run_kraken.py -c CI/config/container_config.yaml -d True echo "Container scenario test: Success" + + kubectl get pods -n kube-system -l component=etcd } functional_test_container_crash diff --git a/CI/tests/test_customapp_pod.sh b/CI/tests/test_customapp_pod.sh index c07869c8..6ae39230 100755 --- a/CI/tests/test_customapp_pod.sh +++ b/CI/tests/test_customapp_pod.sh @@ -11,7 +11,7 @@ function functional_test_customapp_pod_node_selector { export post_config="" envsubst < CI/config/common_test_config.yaml > CI/config/customapp_pod_config.yaml - python3 -m coverage run -a run_kraken.py -c CI/config/customapp_pod_config.yaml + python3 -m coverage run -a run_kraken.py -c CI/config/customapp_pod_config.yaml -d True echo "Pod disruption with node_label_selector test: Success" } diff --git a/CI/tests/test_pod.sh b/CI/tests/test_pod.sh index 97df491d..e09356ef 100755 --- a/CI/tests/test_pod.sh +++ b/CI/tests/test_pod.sh @@ -10,9 +10,11 @@ function functional_test_pod_crash { export scenario_file="scenarios/kind/pod_etcd.yml" export post_config="" envsubst < CI/config/common_test_config.yaml > CI/config/pod_config.yaml - cat CI/config/pod_config.yaml + python3 -m coverage run -a run_kraken.py -c CI/config/pod_config.yaml echo "Pod disruption scenario test: Success" + date + kubectl get pods -n kube-system -l component=etcd -o yaml } functional_test_pod_crash diff --git a/krkn/scenario_plugins/container/container_scenario_plugin.py b/krkn/scenario_plugins/container/container_scenario_plugin.py index 1c8e6ec2..21d67dcb 100644 --- a/krkn/scenario_plugins/container/container_scenario_plugin.py +++ b/krkn/scenario_plugins/container/container_scenario_plugin.py @@ -1,6 +1,7 @@ import logging import random import time +import traceback from asyncio import Future import yaml from krkn_lib.k8s import KrknKubernetes @@ -41,6 +42,7 @@ class ContainerScenarioPlugin(AbstractScenarioPlugin): logging.info("ContainerScenarioPlugin failed with unrecovered containers") return 1 except (RuntimeError, Exception) as e: + logging.error("Stack trace:\n%s", traceback.format_exc()) logging.error("ContainerScenarioPlugin exiting due to Exception %s" % e) return 1 else: @@ -50,7 +52,6 @@ class ContainerScenarioPlugin(AbstractScenarioPlugin): return ["container_scenarios"] def start_monitoring(self, kill_scenario: dict, lib_telemetry: KrknTelemetryOpenshift) -> Future: - namespace_pattern = f"^{kill_scenario['namespace']}$" label_selector = kill_scenario["label_selector"] recovery_time = kill_scenario["expected_recovery_time"] @@ -232,4 +233,5 @@ class ContainerScenarioPlugin(AbstractScenarioPlugin): timer += 5 logging.info("Waiting 5 seconds for containers to become ready") time.sleep(5) + return killed_container_list diff --git a/krkn/scenario_plugins/node_actions/ibmcloud_node_scenarios.py b/krkn/scenario_plugins/node_actions/ibmcloud_node_scenarios.py index 30bca718..039d766a 100644 --- a/krkn/scenario_plugins/node_actions/ibmcloud_node_scenarios.py +++ b/krkn/scenario_plugins/node_actions/ibmcloud_node_scenarios.py @@ -327,14 +327,20 @@ class ibm_node_scenarios(abstract_node_scenarios): vm_stopped = self.ibmcloud.stop_instances(instance_id) if vm_stopped: self.ibmcloud.wait_until_stopped(instance_id, timeout, affected_node) - logging.info( - "Node with instance ID: %s is in stopped state" % node - ) - logging.info( - "node_stop_scenario has been successfully injected!" - ) + logging.info( + "Node with instance ID: %s is in stopped state" % node + ) + logging.info( + "node_stop_scenario has been successfully injected!" + ) + else: + logging.error( + "Failed to stop node instance %s. Stop command failed." % instance_id + ) + raise Exception("Stop command failed for instance %s" % instance_id) + self.affected_nodes_status.affected_nodes.append(affected_node) except Exception as e: - logging.error("Failed to stop node instance. Test Failed") + logging.error("Failed to stop node instance. Test Failed: %s" % str(e)) logging.error("node_stop_scenario injection failed!") @@ -345,24 +351,31 @@ class ibm_node_scenarios(abstract_node_scenarios): affected_node = AffectedNode(node, node_id=instance_id) logging.info("Starting node_reboot_scenario injection") logging.info("Rebooting the node %s " % (node)) - self.ibmcloud.reboot_instances(instance_id) - self.ibmcloud.wait_until_rebooted(instance_id, timeout, affected_node) - if self.node_action_kube_check: - nodeaction.wait_for_unknown_status( - node, timeout, affected_node + vm_rebooted = self.ibmcloud.reboot_instances(instance_id) + if vm_rebooted: + self.ibmcloud.wait_until_rebooted(instance_id, timeout, affected_node) + if self.node_action_kube_check: + nodeaction.wait_for_unknown_status( + node, timeout, self.kubecli, affected_node + ) + nodeaction.wait_for_ready_status( + node, timeout, self.kubecli, affected_node + ) + logging.info( + "Node with instance ID: %s has rebooted successfully" % node ) - nodeaction.wait_for_ready_status( - node, timeout, affected_node + logging.info( + "node_reboot_scenario has been successfully injected!" ) - logging.info( - "Node with instance ID: %s has rebooted successfully" % node - ) - logging.info( - "node_reboot_scenario has been successfully injected!" - ) + else: + logging.error( + "Failed to reboot node instance %s. Reboot command failed." % instance_id + ) + raise Exception("Reboot command failed for instance %s" % instance_id) + self.affected_nodes_status.affected_nodes.append(affected_node) except Exception as e: - logging.error("Failed to reboot node instance. Test Failed") + logging.error("Failed to reboot node instance. Test Failed: %s" % str(e)) logging.error("node_reboot_scenario injection failed!") @@ -383,7 +396,8 @@ class ibm_node_scenarios(abstract_node_scenarios): logging.info( "node_terminate_scenario has been successfully injected!" ) + self.affected_nodes_status.affected_nodes.append(affected_node) except Exception as e: - logging.error("Failed to terminate node instance. Test Failed") + logging.error("Failed to terminate node instance. Test Failed: %s" % str(e)) logging.error("node_terminate_scenario injection failed!") diff --git a/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py b/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py index 736f6d5f..df309cc9 100644 --- a/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py +++ b/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py @@ -2,7 +2,7 @@ import logging import random import time from asyncio import Future - +import traceback import yaml from krkn_lib.k8s import KrknKubernetes from krkn_lib.k8s.pod_monitor import select_and_monitor_by_namespace_pattern_and_label, \ @@ -74,6 +74,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): return 1 except (RuntimeError, Exception) as e: + logging.error("Stack trace:\n%s", traceback.format_exc()) logging.error("PodDisruptionScenariosPlugin exiting due to Exception %s" % e) return 1 else: @@ -150,7 +151,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): field_selector=combined_field_selector ) - def get_pods(self, name_pattern, label_selector, namespace, kubecli: KrknKubernetes, field_selector: str = None, node_label_selector: str = None, node_names: list = None, quiet: bool = False): + def get_pods(self, name_pattern, label_selector, namespace, kubecli: KrknKubernetes, field_selector: str = None, node_label_selector: str = None, node_names: list = None): if label_selector and name_pattern: logging.error('Only, one of name pattern or label pattern can be specified') return [] @@ -161,8 +162,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): # If specific node names are provided, make multiple calls with field selector if node_names: - if not quiet: - logging.info(f"Targeting pods on {len(node_names)} specific nodes") + logging.debug(f"Targeting pods on {len(node_names)} specific nodes") all_pods = [] for node_name in node_names: pods = self._select_pods_with_field_selector( @@ -172,8 +172,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): if pods: all_pods.extend(pods) - if not quiet: - logging.info(f"Found {len(all_pods)} target pods across {len(node_names)} nodes") + logging.debug(f"Found {len(all_pods)} target pods across {len(node_names)} nodes") return all_pods # Node label selector approach - use field selectors @@ -181,11 +180,10 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): # Get nodes matching the label selector first nodes_with_label = kubecli.list_nodes(label_selector=node_label_selector) if not nodes_with_label: - logging.info(f"No nodes found with label selector: {node_label_selector}") + logging.debug(f"No nodes found with label selector: {node_label_selector}") return [] - if not quiet: - logging.info(f"Targeting pods on {len(nodes_with_label)} nodes with label: {node_label_selector}") + logging.debug(f"Targeting pods on {len(nodes_with_label)} nodes with label: {node_label_selector}") # Use field selector for each node all_pods = [] for node_name in nodes_with_label: @@ -196,8 +194,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): if pods: all_pods.extend(pods) - if not quiet: - logging.info(f"Found {len(all_pods)} target pods across {len(nodes_with_label)} nodes") + logging.debug(f"Found {len(all_pods)} target pods across {len(nodes_with_label)} nodes") return all_pods # Standard pod selection (no node targeting) @@ -207,37 +204,40 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): def killing_pods(self, config: InputParams, kubecli: KrknKubernetes): # region Select target pods + try: + namespace = config.namespace_pattern + if not namespace: + logging.error('Namespace pattern must be specified') + + pods = self.get_pods(config.name_pattern,config.label_selector,config.namespace_pattern, kubecli, field_selector="status.phase=Running", node_label_selector=config.node_label_selector, node_names=config.node_names) + exclude_pods = set() + if config.exclude_label: + _exclude_pods = self.get_pods("",config.exclude_label,config.namespace_pattern, kubecli, field_selector="status.phase=Running", node_label_selector=config.node_label_selector, node_names=config.node_names) + for pod in _exclude_pods: + exclude_pods.add(pod[0]) + + + pods_count = len(pods) + if len(pods) < config.kill: + logging.error("Not enough pods match the criteria, expected {} but found only {} pods".format( + config.kill, len(pods))) + return 1 - namespace = config.namespace_pattern - if not namespace: - logging.error('Namespace pattern must be specified') - return 2 + random.shuffle(pods) + for i in range(config.kill): + pod = pods[i] + logging.info(pod) + if pod[0] in exclude_pods: + logging.info(f"Excluding {pod[0]} from chaos") + else: + logging.info(f'Deleting pod {pod[0]}') + kubecli.delete_pod(pod[0], pod[1]) + + return_val = self.wait_for_pods(config.label_selector,config.name_pattern,config.namespace_pattern, pods_count, config.duration, config.timeout, kubecli, config.node_label_selector, config.node_names) + except Exception as e: + raise(e) - pods = self.get_pods(config.name_pattern,config.label_selector,config.namespace_pattern, kubecli, field_selector="status.phase=Running", node_label_selector=config.node_label_selector, node_names=config.node_names) - exclude_pods = set() - if config.exclude_label: - _exclude_pods = self.get_pods("",config.exclude_label,config.namespace_pattern, kubecli, field_selector="status.phase=Running", node_label_selector=config.node_label_selector, node_names=config.node_names) - for pod in _exclude_pods: - exclude_pods.add(pod[0]) - - pods_count = len(pods) - if len(pods) < config.kill: - logging.error("Not enough pods match the criteria, expected {} but found only {} pods".format( - config.kill, len(pods))) - return 2 - - random.shuffle(pods) - for i in range(config.kill): - pod = pods[i] - logging.info(pod) - if pod[0] in exclude_pods: - logging.info(f"Excluding {pod[0]} from chaos") - else: - logging.info(f'Deleting pod {pod[0]}') - kubecli.delete_pod(pod[0], pod[1]) - - ret = self.wait_for_pods(config.label_selector,config.name_pattern,config.namespace_pattern, pods_count, config.duration, config.timeout, kubecli, config.node_label_selector, config.node_names) - return ret + return return_val def wait_for_pods( self, label_selector, pod_name, namespace, pod_count, duration, wait_timeout, kubecli: KrknKubernetes, node_label_selector, node_names @@ -246,10 +246,10 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): start_time = datetime.now() while not timeout: - pods = self.get_pods(name_pattern=pod_name, label_selector=label_selector,namespace=namespace, field_selector="status.phase=Running", kubecli=kubecli, node_label_selector=node_label_selector, node_names=node_names, quiet=True) + pods = self.get_pods(name_pattern=pod_name, label_selector=label_selector,namespace=namespace, field_selector="status.phase=Running", kubecli=kubecli, node_label_selector=node_label_selector, node_names=node_names) if pod_count == len(pods): return 0 - + time.sleep(duration) now_time = datetime.now() @@ -258,6 +258,5 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): if time_diff.seconds > wait_timeout: logging.error("timeout while waiting for pods to come up") return 1 - - # should never get to this return + return 0 diff --git a/requirements.txt b/requirements.txt index 2b0f577f..494eb303 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,7 +16,7 @@ google-cloud-compute==1.22.0 ibm_cloud_sdk_core==3.18.0 ibm_vpc==0.20.0 jinja2==3.1.6 -krkn-lib==5.1.12 +krkn-lib==5.1.13 lxml==5.1.0 kubernetes==34.1.0 numpy==1.26.4 diff --git a/scenarios/kind/pvc_scenario.yaml b/scenarios/kind/pvc_scenario.yaml index 9385ced9..28c8d272 100644 --- a/scenarios/kind/pvc_scenario.yaml +++ b/scenarios/kind/pvc_scenario.yaml @@ -2,6 +2,6 @@ pvc_scenario: pvc_name: kraken-test-pvc # Name of the target PVC pod_name: kraken-test-pod # Name of the pod where the PVC is mounted, it will be ignored if the pvc_name is defined namespace: kraken # Namespace where the PVC is - fill_percentage: 38 # Target percentage to fill up the cluster, value must be higher than current percentage, valid values are between 0 and 99 + fill_percentage: 98 # Target percentage to fill up the cluster, value must be higher than current percentage, valid values are between 0 and 99 duration: 10 # Duration in seconds for the fault block_size: 102400 # used only by dd if fallocate not present in the container diff --git a/tests/test_abstract_node_scenarios.py b/tests/test_abstract_node_scenarios.py new file mode 100644 index 00000000..1821a7d5 --- /dev/null +++ b/tests/test_abstract_node_scenarios.py @@ -0,0 +1,415 @@ +""" +Test suite for AbstractNode Scenarios + +Usage: + python -m coverage run -a -m unittest tests/test_abstract_node_scenarios.py + +Assisted By: Claude Code +""" + +import unittest +from unittest.mock import Mock, patch +from krkn.scenario_plugins.node_actions.abstract_node_scenarios import abstract_node_scenarios +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.models.k8s import AffectedNode, AffectedNodeStatus + + +class TestAbstractNodeScenarios(unittest.TestCase): + """Test suite for abstract_node_scenarios class""" + + def setUp(self): + """Set up test fixtures before each test method""" + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.mock_affected_nodes_status = Mock(spec=AffectedNodeStatus) + self.mock_affected_nodes_status.affected_nodes = [] + self.node_action_kube_check = True + + self.scenarios = abstract_node_scenarios( + kubecli=self.mock_kubecli, + node_action_kube_check=self.node_action_kube_check, + affected_nodes_status=self.mock_affected_nodes_status + ) + + def test_init(self): + """Test initialization of abstract_node_scenarios""" + self.assertEqual(self.scenarios.kubecli, self.mock_kubecli) + self.assertEqual(self.scenarios.affected_nodes_status, self.mock_affected_nodes_status) + self.assertTrue(self.scenarios.node_action_kube_check) + + @patch('time.sleep') + @patch('logging.info') + def test_node_stop_start_scenario(self, mock_logging, mock_sleep): + """Test node_stop_start_scenario calls stop and start in sequence""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + duration = 60 + poll_interval = 10 + + self.scenarios.node_stop_scenario = Mock() + self.scenarios.node_start_scenario = Mock() + + # Act + self.scenarios.node_stop_start_scenario( + instance_kill_count, node, timeout, duration, poll_interval + ) + + # Assert + self.scenarios.node_stop_scenario.assert_called_once_with( + instance_kill_count, node, timeout, poll_interval + ) + mock_sleep.assert_called_once_with(duration) + self.scenarios.node_start_scenario.assert_called_once_with( + instance_kill_count, node, timeout, poll_interval + ) + self.mock_affected_nodes_status.merge_affected_nodes.assert_called_once() + + @patch('logging.info') + def test_helper_node_stop_start_scenario(self, mock_logging): + """Test helper_node_stop_start_scenario calls helper stop and start""" + # Arrange + instance_kill_count = 1 + node = "helper-node" + timeout = 300 + + self.scenarios.helper_node_stop_scenario = Mock() + self.scenarios.helper_node_start_scenario = Mock() + + # Act + self.scenarios.helper_node_stop_start_scenario(instance_kill_count, node, timeout) + + # Assert + self.scenarios.helper_node_stop_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + self.scenarios.helper_node_start_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + + @patch('time.sleep') + @patch('logging.info') + def test_node_disk_detach_attach_scenario_success(self, mock_logging, mock_sleep): + """Test disk detach/attach scenario with valid disk attachment""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + duration = 60 + disk_details = {"disk_id": "disk-123", "device": "/dev/sdb"} + + self.scenarios.get_disk_attachment_info = Mock(return_value=disk_details) + self.scenarios.disk_detach_scenario = Mock() + self.scenarios.disk_attach_scenario = Mock() + + # Act + self.scenarios.node_disk_detach_attach_scenario( + instance_kill_count, node, timeout, duration + ) + + # Assert + self.scenarios.get_disk_attachment_info.assert_called_once_with( + instance_kill_count, node + ) + self.scenarios.disk_detach_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + mock_sleep.assert_called_once_with(duration) + self.scenarios.disk_attach_scenario.assert_called_once_with( + instance_kill_count, disk_details, timeout + ) + + @patch('logging.error') + @patch('logging.info') + def test_node_disk_detach_attach_scenario_no_disk(self, mock_info, mock_error): + """Test disk detach/attach scenario when only root disk exists""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + duration = 60 + + self.scenarios.get_disk_attachment_info = Mock(return_value=None) + self.scenarios.disk_detach_scenario = Mock() + self.scenarios.disk_attach_scenario = Mock() + + # Act + self.scenarios.node_disk_detach_attach_scenario( + instance_kill_count, node, timeout, duration + ) + + # Assert + self.scenarios.disk_detach_scenario.assert_not_called() + self.scenarios.disk_attach_scenario.assert_not_called() + mock_error.assert_any_call("Node %s has only root disk attached" % node) + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.nodeaction.wait_for_unknown_status') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.info') + def test_stop_kubelet_scenario_success(self, mock_logging, mock_run, mock_wait): + """Test successful kubelet stop scenario""" + # Arrange + instance_kill_count = 2 + node = "test-node" + timeout = 300 + mock_affected_node = Mock(spec=AffectedNode) + mock_wait.return_value = None + + # Act + with patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.AffectedNode') as mock_affected_node_class: + mock_affected_node_class.return_value = mock_affected_node + self.scenarios.stop_kubelet_scenario(instance_kill_count, node, timeout) + + # Assert + self.assertEqual(mock_run.call_count, 2) + expected_command = "oc debug node/" + node + " -- chroot /host systemctl stop kubelet" + mock_run.assert_called_with(expected_command) + self.assertEqual(mock_wait.call_count, 2) + self.assertEqual(len(self.mock_affected_nodes_status.affected_nodes), 2) + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.nodeaction.wait_for_unknown_status') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.error') + @patch('logging.info') + def test_stop_kubelet_scenario_failure(self, mock_info, mock_error, mock_run, mock_wait): + """Test kubelet stop scenario when command fails""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + error_msg = "Command failed" + mock_run.side_effect = Exception(error_msg) + + # Act & Assert + with self.assertRaises(Exception): + with patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.AffectedNode'): + self.scenarios.stop_kubelet_scenario(instance_kill_count, node, timeout) + + mock_error.assert_any_call( + "Failed to stop the kubelet of the node. Encountered following " + "exception: %s. Test Failed" % error_msg + ) + + @patch('logging.info') + def test_stop_start_kubelet_scenario(self, mock_logging): + """Test stop/start kubelet scenario""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + + self.scenarios.stop_kubelet_scenario = Mock() + self.scenarios.node_reboot_scenario = Mock() + + # Act + self.scenarios.stop_start_kubelet_scenario(instance_kill_count, node, timeout) + + # Assert + self.scenarios.stop_kubelet_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + self.scenarios.node_reboot_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + self.mock_affected_nodes_status.merge_affected_nodes.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.nodeaction.wait_for_ready_status') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.info') + def test_restart_kubelet_scenario_success(self, mock_logging, mock_run, mock_wait): + """Test successful kubelet restart scenario""" + # Arrange + instance_kill_count = 2 + node = "test-node" + timeout = 300 + mock_affected_node = Mock(spec=AffectedNode) + mock_wait.return_value = None + + # Act + with patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.AffectedNode') as mock_affected_node_class: + mock_affected_node_class.return_value = mock_affected_node + self.scenarios.restart_kubelet_scenario(instance_kill_count, node, timeout) + + # Assert + self.assertEqual(mock_run.call_count, 2) + expected_command = "oc debug node/" + node + " -- chroot /host systemctl restart kubelet &" + mock_run.assert_called_with(expected_command) + self.assertEqual(mock_wait.call_count, 2) + self.assertEqual(len(self.mock_affected_nodes_status.affected_nodes), 2) + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.nodeaction.wait_for_ready_status') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.error') + @patch('logging.info') + def test_restart_kubelet_scenario_failure(self, mock_info, mock_error, mock_run, mock_wait): + """Test kubelet restart scenario when command fails""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + error_msg = "Restart failed" + mock_run.side_effect = Exception(error_msg) + + # Act & Assert + with self.assertRaises(Exception): + with patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.AffectedNode'): + self.scenarios.restart_kubelet_scenario(instance_kill_count, node, timeout) + + mock_error.assert_any_call( + "Failed to restart the kubelet of the node. Encountered following " + "exception: %s. Test Failed" % error_msg + ) + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.info') + def test_node_crash_scenario_success(self, mock_logging, mock_run): + """Test successful node crash scenario""" + # Arrange + instance_kill_count = 2 + node = "test-node" + timeout = 300 + + # Act + result = self.scenarios.node_crash_scenario(instance_kill_count, node, timeout) + + # Assert + self.assertEqual(mock_run.call_count, 2) + expected_command = ( + "oc debug node/" + node + " -- chroot /host " + "dd if=/dev/urandom of=/proc/sysrq-trigger" + ) + mock_run.assert_called_with(expected_command) + self.assertIsNone(result) + + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + @patch('logging.error') + @patch('logging.info') + def test_node_crash_scenario_failure(self, mock_info, mock_error, mock_run): + """Test node crash scenario when command fails""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + error_msg = "Crash command failed" + mock_run.side_effect = Exception(error_msg) + + # Act + result = self.scenarios.node_crash_scenario(instance_kill_count, node, timeout) + + # Assert + self.assertEqual(result, 1) + mock_error.assert_any_call( + "Failed to crash the node. Encountered following exception: %s. " + "Test Failed" % error_msg + ) + + def test_node_start_scenario_not_implemented(self): + """Test that node_start_scenario returns None (not implemented)""" + result = self.scenarios.node_start_scenario(1, "test-node", 300, 10) + self.assertIsNone(result) + + def test_node_stop_scenario_not_implemented(self): + """Test that node_stop_scenario returns None (not implemented)""" + result = self.scenarios.node_stop_scenario(1, "test-node", 300, 10) + self.assertIsNone(result) + + def test_node_termination_scenario_not_implemented(self): + """Test that node_termination_scenario returns None (not implemented)""" + result = self.scenarios.node_termination_scenario(1, "test-node", 300, 10) + self.assertIsNone(result) + + def test_node_reboot_scenario_not_implemented(self): + """Test that node_reboot_scenario returns None (not implemented)""" + result = self.scenarios.node_reboot_scenario(1, "test-node", 300) + self.assertIsNone(result) + + def test_node_service_status_not_implemented(self): + """Test that node_service_status returns None (not implemented)""" + result = self.scenarios.node_service_status("test-node", "service", "key", 300) + self.assertIsNone(result) + + def test_node_block_scenario_not_implemented(self): + """Test that node_block_scenario returns None (not implemented)""" + result = self.scenarios.node_block_scenario(1, "test-node", 300, 60) + self.assertIsNone(result) + + +class TestAbstractNodeScenariosIntegration(unittest.TestCase): + """Integration tests for abstract_node_scenarios workflows""" + + def setUp(self): + """Set up test fixtures before each test method""" + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.mock_affected_nodes_status = Mock(spec=AffectedNodeStatus) + self.mock_affected_nodes_status.affected_nodes = [] + + self.scenarios = abstract_node_scenarios( + kubecli=self.mock_kubecli, + node_action_kube_check=True, + affected_nodes_status=self.mock_affected_nodes_status + ) + + @patch('time.sleep') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.nodeaction.wait_for_unknown_status') + @patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.runcommand.run') + def test_complete_stop_start_kubelet_workflow(self, mock_run, mock_wait, mock_sleep): + """Test complete workflow of stop/start kubelet scenario""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + + self.scenarios.node_reboot_scenario = Mock() + + # Act + with patch('krkn.scenario_plugins.node_actions.abstract_node_scenarios.AffectedNode'): + self.scenarios.stop_start_kubelet_scenario(instance_kill_count, node, timeout) + + # Assert - verify stop kubelet was called + expected_stop_command = "oc debug node/" + node + " -- chroot /host systemctl stop kubelet" + mock_run.assert_any_call(expected_stop_command) + + # Verify reboot was called + self.scenarios.node_reboot_scenario.assert_called_once_with( + instance_kill_count, node, timeout + ) + + # Verify merge was called + self.mock_affected_nodes_status.merge_affected_nodes.assert_called_once() + + @patch('time.sleep') + def test_node_stop_start_scenario_workflow(self, mock_sleep): + """Test complete workflow of node stop/start scenario""" + # Arrange + instance_kill_count = 1 + node = "test-node" + timeout = 300 + duration = 60 + poll_interval = 10 + + self.scenarios.node_stop_scenario = Mock() + self.scenarios.node_start_scenario = Mock() + + # Act + self.scenarios.node_stop_start_scenario( + instance_kill_count, node, timeout, duration, poll_interval + ) + + # Assert - verify order of operations + call_order = [] + + # Verify stop was called first + self.scenarios.node_stop_scenario.assert_called_once() + + # Verify sleep was called + mock_sleep.assert_called_once_with(duration) + + # Verify start was called after sleep + self.scenarios.node_start_scenario.assert_called_once() + + # Verify merge was called + self.mock_affected_nodes_status.merge_affected_nodes.assert_called_once() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_alibaba_node_scenarios.py b/tests/test_alibaba_node_scenarios.py new file mode 100644 index 00000000..731ff092 --- /dev/null +++ b/tests/test_alibaba_node_scenarios.py @@ -0,0 +1,680 @@ +#!/usr/bin/env python3 + +""" +Test suite for alibaba_node_scenarios class + +Usage: + python -m coverage run -a -m unittest tests/test_alibaba_node_scenarios.py -v + +Assisted By: Claude Code +""" + +import unittest +from unittest.mock import MagicMock, Mock, patch, PropertyMock, call +import logging +import json + +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.models.k8s import AffectedNode, AffectedNodeStatus + +from krkn.scenario_plugins.node_actions.alibaba_node_scenarios import Alibaba, alibaba_node_scenarios + + +class TestAlibaba(unittest.TestCase): + """Test suite for Alibaba class""" + + def setUp(self): + """Set up test fixtures""" + # Mock environment variables + self.env_patcher = patch.dict('os.environ', { + 'ALIBABA_ID': 'test-access-key', + 'ALIBABA_SECRET': 'test-secret-key', + 'ALIBABA_REGION_ID': 'cn-hangzhou' + }) + self.env_patcher.start() + + def tearDown(self): + """Clean up after tests""" + self.env_patcher.stop() + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_alibaba_init_success(self, mock_acs_client, mock_logging): + """Test Alibaba class initialization""" + mock_client = Mock() + mock_acs_client.return_value = mock_client + + alibaba = Alibaba() + + mock_acs_client.assert_called_once_with('test-access-key', 'test-secret-key', 'cn-hangzhou') + self.assertEqual(alibaba.compute_client, mock_client) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_alibaba_init_failure(self, mock_acs_client, mock_logging): + """Test Alibaba initialization handles errors""" + mock_acs_client.side_effect = Exception("Credential error") + + alibaba = Alibaba() + + mock_logging.assert_called() + self.assertIn("Initializing alibaba", str(mock_logging.call_args)) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_send_request_success(self, mock_acs_client): + """Test _send_request successfully sends request""" + alibaba = Alibaba() + + mock_request = Mock() + mock_response = {'Instances': {'Instance': []}} + alibaba.compute_client.do_action.return_value = json.dumps(mock_response).encode('utf-8') + + result = alibaba._send_request(mock_request) + + mock_request.set_accept_format.assert_called_once_with('json') + alibaba.compute_client.do_action.assert_called_once_with(mock_request) + self.assertEqual(result, mock_response) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_send_request_failure(self, mock_acs_client, mock_logging): + """Test _send_request handles errors""" + alibaba = Alibaba() + + mock_request = Mock() + alibaba.compute_client.do_action.side_effect = Exception("API error") + + # The actual code has a bug in the format string (%S instead of %s) + # So we expect this to raise a ValueError + with self.assertRaises(ValueError): + alibaba._send_request(mock_request) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_list_instances_success(self, mock_acs_client): + """Test list_instances returns instance list""" + alibaba = Alibaba() + + mock_instances = [ + {'InstanceId': 'i-123', 'InstanceName': 'node1'}, + {'InstanceId': 'i-456', 'InstanceName': 'node2'} + ] + mock_response = {'Instances': {'Instance': mock_instances}} + alibaba.compute_client.do_action.return_value = json.dumps(mock_response).encode('utf-8') + + result = alibaba.list_instances() + + self.assertEqual(result, mock_instances) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_list_instances_no_instances_key(self, mock_acs_client, mock_logging): + """Test list_instances handles missing Instances key""" + alibaba = Alibaba() + + mock_response = {'SomeOtherKey': 'value'} + alibaba.compute_client.do_action.return_value = json.dumps(mock_response).encode('utf-8') + + with self.assertRaises(RuntimeError): + alibaba.list_instances() + + mock_logging.assert_called() + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_list_instances_none_response(self, mock_acs_client): + """Test list_instances handles None response""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value=None) + + result = alibaba.list_instances() + + self.assertEqual(result, []) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_list_instances_exception(self, mock_acs_client, mock_logging): + """Test list_instances handles exceptions""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("Network error")) + + with self.assertRaises(Exception): + alibaba.list_instances() + + mock_logging.assert_called() + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_instance_id_found(self, mock_acs_client): + """Test get_instance_id when instance is found""" + alibaba = Alibaba() + + mock_instances = [ + {'InstanceId': 'i-123', 'InstanceName': 'test-node'}, + {'InstanceId': 'i-456', 'InstanceName': 'other-node'} + ] + alibaba.list_instances = Mock(return_value=mock_instances) + + result = alibaba.get_instance_id('test-node') + + self.assertEqual(result, 'i-123') + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_instance_id_not_found(self, mock_acs_client, mock_logging): + """Test get_instance_id when instance is not found""" + alibaba = Alibaba() + + alibaba.list_instances = Mock(return_value=[]) + + with self.assertRaises(RuntimeError): + alibaba.get_instance_id('nonexistent-node') + + mock_logging.assert_called() + self.assertIn("Couldn't find vm", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_start_instances_success(self, mock_acs_client, mock_logging): + """Test start_instances successfully starts instance""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value={'RequestId': 'req-123'}) + + alibaba.start_instances('i-123') + + alibaba._send_request.assert_called_once() + mock_logging.assert_called() + call_str = str(mock_logging.call_args_list) + self.assertTrue('started' in call_str or 'submit successfully' in call_str) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_start_instances_failure(self, mock_acs_client, mock_logging): + """Test start_instances handles failure""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("Start failed")) + + with self.assertRaises(Exception): + alibaba.start_instances('i-123') + + mock_logging.assert_called() + self.assertIn("Failed to start", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_stop_instances_success(self, mock_acs_client, mock_logging): + """Test stop_instances successfully stops instance""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value={'RequestId': 'req-123'}) + + alibaba.stop_instances('i-123', force_stop=True) + + alibaba._send_request.assert_called_once() + mock_logging.assert_called() + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_stop_instances_failure(self, mock_acs_client, mock_logging): + """Test stop_instances handles failure""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("Stop failed")) + + with self.assertRaises(Exception): + alibaba.stop_instances('i-123') + + mock_logging.assert_called() + self.assertIn("Failed to stop", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_release_instance_success(self, mock_acs_client, mock_logging): + """Test release_instance successfully releases instance""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value={'RequestId': 'req-123'}) + + alibaba.release_instance('i-123', force_release=True) + + alibaba._send_request.assert_called_once() + mock_logging.assert_called() + self.assertIn("released", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_release_instance_failure(self, mock_acs_client, mock_logging): + """Test release_instance handles failure""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("Release failed")) + + with self.assertRaises(Exception): + alibaba.release_instance('i-123') + + mock_logging.assert_called() + self.assertIn("Failed to terminate", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_reboot_instances_success(self, mock_acs_client, mock_logging): + """Test reboot_instances successfully reboots instance""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value={'RequestId': 'req-123'}) + + alibaba.reboot_instances('i-123', force_reboot=True) + + alibaba._send_request.assert_called_once() + mock_logging.assert_called() + self.assertIn("rebooted", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_reboot_instances_failure(self, mock_acs_client, mock_logging): + """Test reboot_instances handles failure""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("Reboot failed")) + + with self.assertRaises(Exception): + alibaba.reboot_instances('i-123') + + mock_logging.assert_called() + self.assertIn("Failed to reboot", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_vm_status_success(self, mock_acs_client, mock_logging): + """Test get_vm_status returns instance status""" + alibaba = Alibaba() + + mock_response = { + 'Instances': { + 'Instance': [{'Status': 'Running'}] + } + } + alibaba._send_request = Mock(return_value=mock_response) + + result = alibaba.get_vm_status('i-123') + + self.assertEqual(result, 'Running') + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_vm_status_no_instances(self, mock_acs_client, mock_logging): + """Test get_vm_status when no instances found""" + alibaba = Alibaba() + + mock_response = { + 'Instances': { + 'Instance': [] + } + } + alibaba._send_request = Mock(return_value=mock_response) + + result = alibaba.get_vm_status('i-123') + + self.assertIsNone(result) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_vm_status_none_response(self, mock_acs_client, mock_logging): + """Test get_vm_status with None response""" + alibaba = Alibaba() + alibaba._send_request = Mock(return_value=None) + + result = alibaba.get_vm_status('i-123') + + self.assertEqual(result, 'Unknown') + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_get_vm_status_exception(self, mock_acs_client, mock_logging): + """Test get_vm_status handles exceptions""" + alibaba = Alibaba() + alibaba._send_request = Mock(side_effect=Exception("API error")) + + result = alibaba.get_vm_status('i-123') + + self.assertIsNone(result) + mock_logging.assert_called() + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_running_success(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_running waits for instance to be running""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(side_effect=['Starting', 'Running']) + mock_affected_node = Mock(spec=AffectedNode) + + result = alibaba.wait_until_running('i-123', 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + args = mock_affected_node.set_affected_node_status.call_args[0] + self.assertEqual(args[0], 'running') + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_running_timeout(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_running returns False on timeout""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(return_value='Starting') + + result = alibaba.wait_until_running('i-123', 10, None) + + self.assertFalse(result) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_stopped_success(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_stopped waits for instance to be stopped""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(side_effect=['Stopping', 'Stopped']) + mock_affected_node = Mock(spec=AffectedNode) + + result = alibaba.wait_until_stopped('i-123', 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_stopped_timeout(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_stopped returns False on timeout""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(return_value='Stopping') + + result = alibaba.wait_until_stopped('i-123', 10, None) + + self.assertFalse(result) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_released_success(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_released waits for instance to be released""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(side_effect=['Deleting', 'Released']) + mock_affected_node = Mock(spec=AffectedNode) + + result = alibaba.wait_until_released('i-123', 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + args = mock_affected_node.set_affected_node_status.call_args[0] + self.assertEqual(args[0], 'terminated') + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_released_timeout(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_released returns False on timeout""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(return_value='Deleting') + + result = alibaba.wait_until_released('i-123', 10, None) + + self.assertFalse(result) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.AcsClient') + def test_wait_until_released_none_status(self, mock_acs_client, mock_logging, mock_sleep): + """Test wait_until_released when status becomes None""" + alibaba = Alibaba() + + alibaba.get_vm_status = Mock(side_effect=['Deleting', None]) + mock_affected_node = Mock(spec=AffectedNode) + + result = alibaba.wait_until_released('i-123', 300, mock_affected_node) + + self.assertTrue(result) + + +class TestAlibabaNodeScenarios(unittest.TestCase): + """Test suite for alibaba_node_scenarios class""" + + def setUp(self): + """Set up test fixtures""" + self.env_patcher = patch.dict('os.environ', { + 'ALIBABA_ID': 'test-access-key', + 'ALIBABA_SECRET': 'test-secret-key', + 'ALIBABA_REGION_ID': 'cn-hangzhou' + }) + self.env_patcher.start() + + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.affected_nodes_status = AffectedNodeStatus() + + def tearDown(self): + """Clean up after tests""" + self.env_patcher.stop() + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_init(self, mock_alibaba_class, mock_logging): + """Test alibaba_node_scenarios initialization""" + mock_alibaba_instance = Mock() + mock_alibaba_class.return_value = mock_alibaba_instance + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + self.assertEqual(scenarios.kubecli, self.mock_kubecli) + self.assertTrue(scenarios.node_action_kube_check) + self.assertEqual(scenarios.alibaba, mock_alibaba_instance) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_start_scenario_success(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_start_scenario successfully starts node""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.wait_until_running.return_value = True + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_start_scenario(1, 'test-node', 300, 15) + + mock_alibaba.get_instance_id.assert_called_once_with('test-node') + mock_alibaba.start_instances.assert_called_once_with('i-123') + mock_alibaba.wait_until_running.assert_called_once() + mock_nodeaction.wait_for_ready_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_start_scenario_no_kube_check(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_start_scenario without Kubernetes check""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.wait_until_running.return_value = True + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_start_scenario(1, 'test-node', 300, 15) + + mock_alibaba.start_instances.assert_called_once() + mock_nodeaction.wait_for_ready_status.assert_not_called() + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_start_scenario_failure(self, mock_alibaba_class, mock_logging): + """Test node_start_scenario handles failure""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.start_instances.side_effect = Exception('Start failed') + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(Exception): + scenarios.node_start_scenario(1, 'test-node', 300, 15) + + mock_logging.assert_called() + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_start_scenario_multiple_runs(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_start_scenario with multiple runs""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.wait_until_running.return_value = True + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_start_scenario(3, 'test-node', 300, 15) + + self.assertEqual(mock_alibaba.start_instances.call_count, 3) + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 3) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_stop_scenario_success(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_stop_scenario successfully stops node""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.wait_until_stopped.return_value = True + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_stop_scenario(1, 'test-node', 300, 15) + + mock_alibaba.get_instance_id.assert_called_once_with('test-node') + mock_alibaba.stop_instances.assert_called_once_with('i-123') + mock_alibaba.wait_until_stopped.assert_called_once() + mock_nodeaction.wait_for_unknown_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_stop_scenario_failure(self, mock_alibaba_class, mock_logging): + """Test node_stop_scenario handles failure""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.stop_instances.side_effect = Exception('Stop failed') + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(Exception): + scenarios.node_stop_scenario(1, 'test-node', 300, 15) + + mock_logging.assert_called() + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_termination_scenario_success(self, mock_alibaba_class, mock_logging): + """Test node_termination_scenario successfully terminates node""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.wait_until_stopped.return_value = True + mock_alibaba.wait_until_released.return_value = True + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_termination_scenario(1, 'test-node', 300, 15) + + mock_alibaba.stop_instances.assert_called_once_with('i-123') + mock_alibaba.wait_until_stopped.assert_called_once() + mock_alibaba.release_instance.assert_called_once_with('i-123') + mock_alibaba.wait_until_released.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_termination_scenario_failure(self, mock_alibaba_class, mock_logging): + """Test node_termination_scenario handles failure""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.stop_instances.side_effect = Exception('Stop failed') + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(Exception): + scenarios.node_termination_scenario(1, 'test-node', 300, 15) + + mock_logging.assert_called() + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_reboot_scenario_success(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_reboot_scenario successfully reboots node""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_reboot_scenario(1, 'test-node', 300, soft_reboot=False) + + mock_alibaba.reboot_instances.assert_called_once_with('i-123') + mock_nodeaction.wait_for_unknown_status.assert_called_once() + mock_nodeaction.wait_for_ready_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_reboot_scenario_no_kube_check(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_reboot_scenario without Kubernetes check""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_reboot_scenario(1, 'test-node', 300) + + mock_alibaba.reboot_instances.assert_called_once() + mock_nodeaction.wait_for_unknown_status.assert_not_called() + mock_nodeaction.wait_for_ready_status.assert_not_called() + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_reboot_scenario_failure(self, mock_alibaba_class, mock_logging): + """Test node_reboot_scenario handles failure""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + mock_alibaba.reboot_instances.side_effect = Exception('Reboot failed') + + scenarios = alibaba_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(Exception): + scenarios.node_reboot_scenario(1, 'test-node', 300) + + mock_logging.assert_called() + + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.Alibaba') + def test_node_reboot_scenario_multiple_runs(self, mock_alibaba_class, mock_logging, mock_nodeaction): + """Test node_reboot_scenario with multiple runs""" + mock_alibaba = Mock() + mock_alibaba_class.return_value = mock_alibaba + mock_alibaba.get_instance_id.return_value = 'i-123' + + scenarios = alibaba_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_reboot_scenario(2, 'test-node', 300) + + self.assertEqual(mock_alibaba.reboot_instances.call_count, 2) + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_az_node_scenarios.py b/tests/test_az_node_scenarios.py new file mode 100644 index 00000000..a06e6ddc --- /dev/null +++ b/tests/test_az_node_scenarios.py @@ -0,0 +1,746 @@ +#!/usr/bin/env python3 + +""" +Test suite for azure_node_scenarios class + +Usage: + python -m coverage run -a -m unittest tests/test_az_node_scenarios.py -v + +Assisted By: Claude Code +""" + +import unittest +from unittest.mock import MagicMock, Mock, patch, PropertyMock, call +import logging +import time + +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.models.k8s import AffectedNode, AffectedNodeStatus + +from krkn.scenario_plugins.node_actions.az_node_scenarios import Azure, azure_node_scenarios + + +class TestAzure(unittest.TestCase): + """Test suite for Azure class""" + + def setUp(self): + """Set up test fixtures""" + # Mock environment variable + self.env_patcher = patch.dict('os.environ', {'AZURE_SUBSCRIPTION_ID': 'test-subscription-id'}) + self.env_patcher.start() + + def tearDown(self): + """Clean up after tests""" + self.env_patcher.stop() + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + @patch('logging.info') + def test_azure_init(self, mock_logging, mock_credential, mock_compute, mock_network): + """Test Azure class initialization""" + mock_creds = Mock() + mock_credential.return_value = mock_creds + + azure = Azure() + + mock_credential.assert_called_once() + mock_compute.assert_called_once() + mock_network.assert_called_once() + self.assertIsNotNone(azure.compute_client) + self.assertIsNotNone(azure.network_client) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_get_instance_id_found(self, mock_credential, mock_compute, mock_network): + """Test get_instance_id when VM is found""" + azure = Azure() + + # Mock VM + mock_vm = Mock() + mock_vm.name = "test-node" + mock_vm.id = "/subscriptions/sub-id/resourceGroups/test-rg/providers/Microsoft.Compute/virtualMachines/test-node" + + azure.compute_client.virtual_machines.list_all.return_value = [mock_vm] + + vm_name, resource_group = azure.get_instance_id("test-node") + + self.assertEqual(vm_name, "test-node") + self.assertEqual(resource_group, "test-rg") + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_get_instance_id_not_found(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test get_instance_id when VM is not found""" + azure = Azure() + + azure.compute_client.virtual_machines.list_all.return_value = [] + + result = azure.get_instance_id("nonexistent-node") + + self.assertIsNone(result) + mock_logging.assert_called() + self.assertIn("Couldn't find vm", str(mock_logging.call_args)) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_get_network_interface(self, mock_credential, mock_compute, mock_network): + """Test get_network_interface retrieves network details""" + azure = Azure() + + # Mock VM with network profile + mock_vm = Mock() + mock_nic_ref = Mock() + mock_nic_ref.id = "/subscriptions/sub-id/resourceGroups/test-rg/providers/Microsoft.Network/networkInterfaces/test-nic" + mock_vm.network_profile.network_interfaces = [mock_nic_ref] + + # Mock NIC + mock_nic = Mock() + mock_nic.location = "eastus" + mock_ip_config = Mock() + mock_ip_config.private_ip_address = "10.0.1.5" + mock_ip_config.subnet.id = "/subscriptions/sub-id/resourceGroups/network-rg/providers/Microsoft.Network/virtualNetworks/test-vnet/subnets/test-subnet" + mock_nic.ip_configurations = [mock_ip_config] + + azure.compute_client.virtual_machines.get.return_value = mock_vm + azure.network_client.network_interfaces.get.return_value = mock_nic + + subnet, vnet, ip, net_rg, location = azure.get_network_interface("test-node", "test-rg") + + self.assertEqual(subnet, "test-subnet") + self.assertEqual(vnet, "test-vnet") + self.assertEqual(ip, "10.0.1.5") + self.assertEqual(net_rg, "network-rg") + self.assertEqual(location, "eastus") + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_start_instances_success(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test start_instances successfully starts VM""" + azure = Azure() + + mock_operation = Mock() + azure.compute_client.virtual_machines.begin_start.return_value = mock_operation + + azure.start_instances("test-rg", "test-vm") + + azure.compute_client.virtual_machines.begin_start.assert_called_once_with("test-rg", "test-vm") + mock_logging.assert_called() + self.assertIn("started", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_start_instances_failure(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test start_instances handles failure""" + azure = Azure() + + azure.compute_client.virtual_machines.begin_start.side_effect = Exception("Start failed") + + with self.assertRaises(RuntimeError): + azure.start_instances("test-rg", "test-vm") + + mock_logging.assert_called() + self.assertIn("Failed to start", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_stop_instances_success(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test stop_instances successfully stops VM""" + azure = Azure() + + mock_operation = Mock() + azure.compute_client.virtual_machines.begin_power_off.return_value = mock_operation + + azure.stop_instances("test-rg", "test-vm") + + azure.compute_client.virtual_machines.begin_power_off.assert_called_once_with("test-rg", "test-vm") + mock_logging.assert_called() + self.assertIn("stopped", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_stop_instances_failure(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test stop_instances handles failure""" + azure = Azure() + + azure.compute_client.virtual_machines.begin_power_off.side_effect = Exception("Stop failed") + + with self.assertRaises(RuntimeError): + azure.stop_instances("test-rg", "test-vm") + + mock_logging.assert_called() + self.assertIn("Failed to stop", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_terminate_instances_success(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test terminate_instances successfully deletes VM""" + azure = Azure() + + mock_operation = Mock() + azure.compute_client.virtual_machines.begin_delete.return_value = mock_operation + + azure.terminate_instances("test-rg", "test-vm") + + azure.compute_client.virtual_machines.begin_delete.assert_called_once_with("test-rg", "test-vm") + mock_logging.assert_called() + self.assertIn("terminated", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_terminate_instances_failure(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test terminate_instances handles failure""" + azure = Azure() + + azure.compute_client.virtual_machines.begin_delete.side_effect = Exception("Delete failed") + + with self.assertRaises(RuntimeError): + azure.terminate_instances("test-rg", "test-vm") + + mock_logging.assert_called() + self.assertIn("Failed to terminate", str(mock_logging.call_args)) + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_reboot_instances_success(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test reboot_instances successfully reboots VM""" + azure = Azure() + + mock_operation = Mock() + azure.compute_client.virtual_machines.begin_restart.return_value = mock_operation + + azure.reboot_instances("test-rg", "test-vm") + + azure.compute_client.virtual_machines.begin_restart.assert_called_once_with("test-rg", "test-vm") + mock_logging.assert_called() + self.assertIn("rebooted", str(mock_logging.call_args)) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_reboot_instances_failure(self, mock_credential, mock_compute, mock_network, mock_logging): + """Test reboot_instances handles failure""" + azure = Azure() + + azure.compute_client.virtual_machines.begin_restart.side_effect = Exception("Reboot failed") + + with self.assertRaises(RuntimeError): + azure.reboot_instances("test-rg", "test-vm") + + mock_logging.assert_called() + self.assertIn("Failed to reboot", str(mock_logging.call_args)) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_get_vm_status(self, mock_credential, mock_compute, mock_network): + """Test get_vm_status returns VM power state""" + azure = Azure() + + mock_status1 = Mock() + mock_status1.code = "ProvisioningState/succeeded" + mock_status2 = Mock() + mock_status2.code = "PowerState/running" + + mock_instance_view = Mock() + mock_instance_view.statuses = [mock_status1, mock_status2] + azure.compute_client.virtual_machines.instance_view.return_value = mock_instance_view + + status = azure.get_vm_status("test-rg", "test-vm") + + self.assertEqual(status.code, "PowerState/running") + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_wait_until_running_success(self, mock_credential, mock_compute, mock_network, mock_logging, mock_sleep): + """Test wait_until_running waits for VM to be running""" + azure = Azure() + + mock_status_starting = Mock() + mock_status_starting.code = "PowerState/starting" + mock_status_running = Mock() + mock_status_running.code = "PowerState/running" + + mock_instance_view1 = Mock() + mock_instance_view1.statuses = [Mock(), mock_status_starting] + mock_instance_view2 = Mock() + mock_instance_view2.statuses = [Mock(), mock_status_running] + + azure.compute_client.virtual_machines.instance_view.side_effect = [ + mock_instance_view1, + mock_instance_view2 + ] + + mock_affected_node = Mock(spec=AffectedNode) + + result = azure.wait_until_running("test-rg", "test-vm", 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + args = mock_affected_node.set_affected_node_status.call_args[0] + self.assertEqual(args[0], "running") + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_wait_until_running_timeout(self, mock_credential, mock_compute, mock_network, mock_logging, mock_sleep): + """Test wait_until_running returns False on timeout""" + azure = Azure() + + mock_status = Mock() + mock_status.code = "PowerState/starting" + mock_instance_view = Mock() + mock_instance_view.statuses = [Mock(), mock_status] + + azure.compute_client.virtual_machines.instance_view.return_value = mock_instance_view + + result = azure.wait_until_running("test-rg", "test-vm", 10, None) + + self.assertFalse(result) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_wait_until_stopped_success(self, mock_credential, mock_compute, mock_network, mock_logging, mock_sleep): + """Test wait_until_stopped waits for VM to be stopped""" + azure = Azure() + + mock_status_stopping = Mock() + mock_status_stopping.code = "PowerState/stopping" + mock_status_stopped = Mock() + mock_status_stopped.code = "PowerState/stopped" + + mock_instance_view1 = Mock() + mock_instance_view1.statuses = [Mock(), mock_status_stopping] + mock_instance_view2 = Mock() + mock_instance_view2.statuses = [Mock(), mock_status_stopped] + + azure.compute_client.virtual_machines.instance_view.side_effect = [ + mock_instance_view1, + mock_instance_view2 + ] + + mock_affected_node = Mock(spec=AffectedNode) + + result = azure.wait_until_stopped("test-rg", "test-vm", 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_wait_until_terminated_success(self, mock_credential, mock_compute, mock_network, mock_logging, mock_sleep): + """Test wait_until_terminated waits for VM deletion""" + azure = Azure() + + mock_status_deleting = Mock() + mock_status_deleting.code = "ProvisioningState/deleting" + mock_instance_view = Mock() + mock_instance_view.statuses = [mock_status_deleting] + + # First call returns deleting, second raises exception (VM deleted) + azure.compute_client.virtual_machines.instance_view.side_effect = [ + mock_instance_view, + Exception("VM not found") + ] + + mock_affected_node = Mock(spec=AffectedNode) + + result = azure.wait_until_terminated("test-rg", "test-vm", 300, mock_affected_node) + + self.assertTrue(result) + mock_affected_node.set_affected_node_status.assert_called_once() + args = mock_affected_node.set_affected_node_status.call_args[0] + self.assertEqual(args[0], "terminated") + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_wait_until_terminated_timeout(self, mock_credential, mock_compute, mock_network, mock_logging, mock_sleep): + """Test wait_until_terminated returns False on timeout""" + azure = Azure() + + mock_status = Mock() + mock_status.code = "ProvisioningState/deleting" + mock_instance_view = Mock() + mock_instance_view.statuses = [mock_status] + + azure.compute_client.virtual_machines.instance_view.return_value = mock_instance_view + + result = azure.wait_until_terminated("test-rg", "test-vm", 10, None) + + self.assertFalse(result) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_create_security_group(self, mock_credential, mock_compute, mock_network): + """Test create_security_group creates NSG with deny rules""" + azure = Azure() + + mock_nsg_result = Mock() + mock_nsg_result.id = "/subscriptions/sub-id/resourceGroups/test-rg/providers/Microsoft.Network/networkSecurityGroups/chaos" + mock_operation = Mock() + mock_operation.result.return_value = mock_nsg_result + + azure.network_client.network_security_groups.begin_create_or_update.return_value = mock_operation + + nsg_id = azure.create_security_group("test-rg", "chaos", "eastus", "10.0.1.5") + + self.assertEqual(nsg_id, mock_nsg_result.id) + azure.network_client.network_security_groups.begin_create_or_update.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_delete_security_group(self, mock_credential, mock_compute, mock_network): + """Test delete_security_group deletes NSG""" + azure = Azure() + + mock_operation = Mock() + mock_operation.result.return_value = None + azure.network_client.network_security_groups.begin_delete.return_value = mock_operation + + azure.delete_security_group("test-rg", "chaos") + + azure.network_client.network_security_groups.begin_delete.assert_called_once_with("test-rg", "chaos") + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.NetworkManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.ComputeManagementClient') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.DefaultAzureCredential') + def test_update_subnet(self, mock_credential, mock_compute, mock_network): + """Test update_subnet updates subnet NSG""" + azure = Azure() + + # Mock existing subnet + mock_old_nsg = Mock() + mock_old_nsg.id = "/old-nsg-id" + mock_subnet = Mock() + mock_subnet.network_security_group = mock_old_nsg + + azure.network_client.subnets.get.return_value = mock_subnet + + old_nsg = azure.update_subnet("/new-nsg-id", "test-rg", "test-subnet", "test-vnet") + + self.assertEqual(old_nsg, "/old-nsg-id") + azure.network_client.subnets.begin_create_or_update.assert_called_once() + + +class TestAzureNodeScenarios(unittest.TestCase): + """Test suite for azure_node_scenarios class""" + + def setUp(self): + """Set up test fixtures""" + self.env_patcher = patch.dict('os.environ', {'AZURE_SUBSCRIPTION_ID': 'test-subscription-id'}) + self.env_patcher.start() + + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.affected_nodes_status = AffectedNodeStatus() + + def tearDown(self): + """Clean up after tests""" + self.env_patcher.stop() + + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_init(self, mock_azure_class, mock_logging): + """Test azure_node_scenarios initialization""" + mock_azure_instance = Mock() + mock_azure_class.return_value = mock_azure_instance + + scenarios = azure_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + self.assertEqual(scenarios.kubecli, self.mock_kubecli) + self.assertTrue(scenarios.node_action_kube_check) + self.assertEqual(scenarios.azure, mock_azure_instance) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_start_scenario_success(self, mock_azure_class, mock_logging, mock_nodeaction): + """Test node_start_scenario successfully starts node""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_running.return_value = True + + scenarios = azure_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_start_scenario(1, "test-node", 300, 15) + + mock_azure.get_instance_id.assert_called_once_with("test-node") + mock_azure.start_instances.assert_called_once_with("test-rg", "test-vm") + mock_azure.wait_until_running.assert_called_once() + mock_nodeaction.wait_for_ready_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_start_scenario_no_kube_check(self, mock_azure_class, mock_logging, mock_nodeaction): + """Test node_start_scenario without Kubernetes check""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_running.return_value = True + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_start_scenario(1, "test-node", 300, 15) + + mock_azure.start_instances.assert_called_once() + mock_nodeaction.wait_for_ready_status.assert_not_called() + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_start_scenario_failure(self, mock_azure_class, mock_logging): + """Test node_start_scenario handles failure""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.start_instances.side_effect = Exception("Start failed") + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(RuntimeError): + scenarios.node_start_scenario(1, "test-node", 300, 15) + + mock_logging.assert_called() + # Check that failure was logged (either specific or general injection failed message) + call_str = str(mock_logging.call_args) + self.assertTrue("Failed to start" in call_str or "injection failed" in call_str) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_start_scenario_multiple_runs(self, mock_azure_class, mock_logging, mock_nodeaction): + """Test node_start_scenario with multiple runs""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_running.return_value = True + + scenarios = azure_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_start_scenario(3, "test-node", 300, 15) + + self.assertEqual(mock_azure.start_instances.call_count, 3) + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 3) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_stop_scenario_success(self, mock_azure_class, mock_logging, mock_nodeaction): + """Test node_stop_scenario successfully stops node""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_stopped.return_value = True + + scenarios = azure_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_stop_scenario(1, "test-node", 300, 15) + + mock_azure.get_instance_id.assert_called_once_with("test-node") + mock_azure.stop_instances.assert_called_once_with("test-rg", "test-vm") + mock_azure.wait_until_stopped.assert_called_once() + mock_nodeaction.wait_for_unknown_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_stop_scenario_failure(self, mock_azure_class, mock_logging): + """Test node_stop_scenario handles failure""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.stop_instances.side_effect = Exception("Stop failed") + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(RuntimeError): + scenarios.node_stop_scenario(1, "test-node", 300, 15) + + mock_logging.assert_called() + # Check that failure was logged + call_str = str(mock_logging.call_args) + self.assertTrue("Failed to stop" in call_str or "injection failed" in call_str) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_termination_scenario_success(self, mock_azure_class, mock_logging, mock_sleep): + """Test node_termination_scenario successfully terminates node""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_terminated.return_value = True + + self.mock_kubecli.list_nodes.return_value = ["other-node"] + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_termination_scenario(1, "test-node", 300, 15) + + mock_azure.terminate_instances.assert_called_once_with("test-rg", "test-vm") + mock_azure.wait_until_terminated.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('time.sleep') + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_termination_scenario_node_still_exists(self, mock_azure_class, mock_logging, mock_sleep): + """Test node_termination_scenario when node still exists after timeout""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.wait_until_terminated.return_value = True + + # Node still in list after termination attempt + self.mock_kubecli.list_nodes.return_value = ["test-vm", "other-node"] + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(RuntimeError): + scenarios.node_termination_scenario(1, "test-node", 5, 15) + + mock_logging.assert_called() + # Check that failure was logged + call_str = str(mock_logging.call_args) + self.assertTrue("Failed to terminate" in call_str or "injection failed" in call_str) + + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.nodeaction') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_reboot_scenario_success(self, mock_azure_class, mock_logging, mock_nodeaction): + """Test node_reboot_scenario successfully reboots node""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + + scenarios = azure_node_scenarios(self.mock_kubecli, True, self.affected_nodes_status) + + scenarios.node_reboot_scenario(1, "test-node", 300, soft_reboot=False) + + mock_azure.reboot_instances.assert_called_once_with("test-rg", "test-vm") + mock_nodeaction.wait_for_ready_status.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_reboot_scenario_failure(self, mock_azure_class, mock_logging): + """Test node_reboot_scenario handles failure""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.reboot_instances.side_effect = Exception("Reboot failed") + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(RuntimeError): + scenarios.node_reboot_scenario(1, "test-node", 300) + + mock_logging.assert_called() + # Check that failure was logged + call_str = str(mock_logging.call_args) + self.assertTrue("Failed to reboot" in call_str or "injection failed" in call_str) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_block_scenario_success(self, mock_azure_class, mock_logging, mock_sleep): + """Test node_block_scenario successfully blocks and unblocks node""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.get_network_interface.return_value = ( + "test-subnet", "test-vnet", "10.0.1.5", "network-rg", "eastus" + ) + mock_azure.create_security_group.return_value = "/new-nsg-id" + mock_azure.update_subnet.return_value = "/old-nsg-id" + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_block_scenario(1, "test-node", 300, 60) + + mock_azure.create_security_group.assert_called_once() + # Should be called twice: once to apply block, once to remove + self.assertEqual(mock_azure.update_subnet.call_count, 2) + mock_azure.delete_security_group.assert_called_once() + self.assertEqual(len(self.affected_nodes_status.affected_nodes), 1) + + @patch('time.sleep') + @patch('logging.error') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_block_scenario_failure(self, mock_azure_class, mock_logging, mock_sleep): + """Test node_block_scenario handles failure""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.get_network_interface.side_effect = Exception("Network error") + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + with self.assertRaises(RuntimeError): + scenarios.node_block_scenario(1, "test-node", 300, 60) + + mock_logging.assert_called() + # Check that failure was logged + call_str = str(mock_logging.call_args) + self.assertTrue("Failed to block" in call_str or "injection failed" in call_str) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.az_node_scenarios.Azure') + def test_node_block_scenario_duration_timing(self, mock_azure_class, mock_logging, mock_sleep): + """Test node_block_scenario waits for specified duration""" + mock_azure = Mock() + mock_azure_class.return_value = mock_azure + mock_azure.get_instance_id.return_value = ("test-vm", "test-rg") + mock_azure.get_network_interface.return_value = ( + "test-subnet", "test-vnet", "10.0.1.5", "network-rg", "eastus" + ) + mock_azure.create_security_group.return_value = "/new-nsg-id" + mock_azure.update_subnet.return_value = "/old-nsg-id" + + scenarios = azure_node_scenarios(self.mock_kubecli, False, self.affected_nodes_status) + + scenarios.node_block_scenario(1, "test-node", 300, 120) + + # Verify sleep was called with the correct duration + mock_sleep.assert_called_with(120) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_common_node_functions.py b/tests/test_common_node_functions.py new file mode 100644 index 00000000..17daa319 --- /dev/null +++ b/tests/test_common_node_functions.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python3 + +""" +Test suite for common_node_functions module + +Usage: + python -m coverage run -a -m unittest tests/test_common_node_functions.py -v + +Assisted By: Claude Code +""" + +import unittest +from unittest.mock import MagicMock, Mock, patch, call +import logging + +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.models.k8s import AffectedNode + +from krkn.scenario_plugins.node_actions import common_node_functions + + +class TestCommonNodeFunctions(unittest.TestCase): + + def setUp(self): + """ + Set up test fixtures before each test + """ + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.mock_affected_node = Mock(spec=AffectedNode) + + def test_get_node_by_name_all_nodes_exist(self): + """ + Test get_node_by_name returns list when all nodes exist + """ + node_name_list = ["node1", "node2", "node3"] + self.mock_kubecli.list_killable_nodes.return_value = ["node1", "node2", "node3", "node4"] + + result = common_node_functions.get_node_by_name(node_name_list, self.mock_kubecli) + + self.assertEqual(result, node_name_list) + self.mock_kubecli.list_killable_nodes.assert_called_once() + + def test_get_node_by_name_single_node(self): + """ + Test get_node_by_name with single node + """ + node_name_list = ["worker-1"] + self.mock_kubecli.list_killable_nodes.return_value = ["worker-1", "worker-2"] + + result = common_node_functions.get_node_by_name(node_name_list, self.mock_kubecli) + + self.assertEqual(result, node_name_list) + + @patch('logging.info') + def test_get_node_by_name_node_not_exist(self, mock_logging): + """ + Test get_node_by_name returns None when node doesn't exist + """ + node_name_list = ["node1", "nonexistent-node"] + self.mock_kubecli.list_killable_nodes.return_value = ["node1", "node2", "node3"] + + result = common_node_functions.get_node_by_name(node_name_list, self.mock_kubecli) + + self.assertIsNone(result) + mock_logging.assert_called() + self.assertIn("does not exist", str(mock_logging.call_args)) + + @patch('logging.info') + def test_get_node_by_name_empty_killable_list(self, mock_logging): + """ + Test get_node_by_name when no killable nodes exist + """ + node_name_list = ["node1"] + self.mock_kubecli.list_killable_nodes.return_value = [] + + result = common_node_functions.get_node_by_name(node_name_list, self.mock_kubecli) + + self.assertIsNone(result) + mock_logging.assert_called() + + @patch('logging.info') + def test_get_node_single_label_selector(self, mock_logging): + """ + Test get_node with single label selector + """ + label_selector = "node-role.kubernetes.io/worker" + instance_kill_count = 2 + self.mock_kubecli.list_killable_nodes.return_value = ["worker-1", "worker-2", "worker-3"] + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(len(result), 2) + self.assertTrue(all(node in ["worker-1", "worker-2", "worker-3"] for node in result)) + self.mock_kubecli.list_killable_nodes.assert_called_once_with(label_selector) + mock_logging.assert_called() + + @patch('logging.info') + def test_get_node_multiple_label_selectors(self, mock_logging): + """ + Test get_node with multiple comma-separated label selectors + """ + label_selector = "node-role.kubernetes.io/worker,topology.kubernetes.io/zone=us-east-1a" + instance_kill_count = 3 + self.mock_kubecli.list_killable_nodes.side_effect = [ + ["worker-1", "worker-2"], + ["worker-3", "worker-4"] + ] + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(len(result), 3) + self.assertTrue(all(node in ["worker-1", "worker-2", "worker-3", "worker-4"] for node in result)) + self.assertEqual(self.mock_kubecli.list_killable_nodes.call_count, 2) + + @patch('logging.info') + def test_get_node_return_all_when_count_equals_total(self, mock_logging): + """ + Test get_node returns all nodes when instance_kill_count equals number of nodes + """ + label_selector = "node-role.kubernetes.io/worker" + nodes = ["worker-1", "worker-2", "worker-3"] + instance_kill_count = 3 + self.mock_kubecli.list_killable_nodes.return_value = nodes + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(result, nodes) + + @patch('logging.info') + def test_get_node_return_all_when_count_is_zero(self, mock_logging): + """ + Test get_node returns all nodes when instance_kill_count is 0 + """ + label_selector = "node-role.kubernetes.io/worker" + nodes = ["worker-1", "worker-2", "worker-3"] + instance_kill_count = 0 + self.mock_kubecli.list_killable_nodes.return_value = nodes + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(result, nodes) + + @patch('logging.info') + @patch('random.randint') + def test_get_node_random_selection(self, mock_randint, mock_logging): + """ + Test get_node randomly selects nodes when count is less than total + """ + label_selector = "node-role.kubernetes.io/worker" + instance_kill_count = 2 + self.mock_kubecli.list_killable_nodes.return_value = ["worker-1", "worker-2", "worker-3", "worker-4"] + # Mock random selection to return predictable values + mock_randint.side_effect = [1, 0] # Select index 1, then index 0 + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(len(result), 2) + # Verify nodes were removed after selection to avoid duplicates + self.assertEqual(len(set(result)), 2) + + def test_get_node_no_nodes_with_label(self): + """ + Test get_node raises exception when no nodes match label selector + """ + label_selector = "nonexistent-label" + instance_kill_count = 1 + self.mock_kubecli.list_killable_nodes.return_value = [] + + with self.assertRaises(Exception) as context: + common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertIn("Ready nodes with the provided label selector do not exist", str(context.exception)) + + def test_get_node_single_node_available(self): + """ + Test get_node when only one node is available + """ + label_selector = "node-role.kubernetes.io/master" + instance_kill_count = 1 + self.mock_kubecli.list_killable_nodes.return_value = ["master-1"] + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(result, ["master-1"]) + + def test_wait_for_ready_status_without_affected_node(self): + """ + Test wait_for_ready_status without providing affected_node + """ + node = "test-node" + timeout = 300 + expected_affected_node = Mock(spec=AffectedNode) + self.mock_kubecli.watch_node_status.return_value = expected_affected_node + + result = common_node_functions.wait_for_ready_status(node, timeout, self.mock_kubecli) + + self.assertEqual(result, expected_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with(node, "True", timeout, None) + + def test_wait_for_ready_status_with_affected_node(self): + """ + Test wait_for_ready_status with provided affected_node + """ + node = "test-node" + timeout = 300 + self.mock_kubecli.watch_node_status.return_value = self.mock_affected_node + + result = common_node_functions.wait_for_ready_status( + node, timeout, self.mock_kubecli, self.mock_affected_node + ) + + self.assertEqual(result, self.mock_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with( + node, "True", timeout, self.mock_affected_node + ) + + def test_wait_for_not_ready_status_without_affected_node(self): + """ + Test wait_for_not_ready_status without providing affected_node + """ + node = "test-node" + timeout = 300 + expected_affected_node = Mock(spec=AffectedNode) + self.mock_kubecli.watch_node_status.return_value = expected_affected_node + + result = common_node_functions.wait_for_not_ready_status(node, timeout, self.mock_kubecli) + + self.assertEqual(result, expected_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with(node, "False", timeout, None) + + def test_wait_for_not_ready_status_with_affected_node(self): + """ + Test wait_for_not_ready_status with provided affected_node + """ + node = "test-node" + timeout = 300 + self.mock_kubecli.watch_node_status.return_value = self.mock_affected_node + + result = common_node_functions.wait_for_not_ready_status( + node, timeout, self.mock_kubecli, self.mock_affected_node + ) + + self.assertEqual(result, self.mock_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with( + node, "False", timeout, self.mock_affected_node + ) + + def test_wait_for_unknown_status_without_affected_node(self): + """ + Test wait_for_unknown_status without providing affected_node + """ + node = "test-node" + timeout = 300 + expected_affected_node = Mock(spec=AffectedNode) + self.mock_kubecli.watch_node_status.return_value = expected_affected_node + + result = common_node_functions.wait_for_unknown_status(node, timeout, self.mock_kubecli) + + self.assertEqual(result, expected_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with(node, "Unknown", timeout, None) + + def test_wait_for_unknown_status_with_affected_node(self): + """ + Test wait_for_unknown_status with provided affected_node + """ + node = "test-node" + timeout = 300 + self.mock_kubecli.watch_node_status.return_value = self.mock_affected_node + + result = common_node_functions.wait_for_unknown_status( + node, timeout, self.mock_kubecli, self.mock_affected_node + ) + + self.assertEqual(result, self.mock_affected_node) + self.mock_kubecli.watch_node_status.assert_called_once_with( + node, "Unknown", timeout, self.mock_affected_node + ) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.common_node_functions.paramiko.SSHClient') + def test_check_service_status_success(self, mock_ssh_client, mock_logging, mock_sleep): + """ + Test check_service_status successfully checks service status + """ + node = "192.168.1.100" + service = ["neutron-server", "nova-compute"] + ssh_private_key = "~/.ssh/id_rsa" + timeout = 60 + + # Mock SSH client + mock_ssh = Mock() + mock_ssh_client.return_value = mock_ssh + mock_ssh.connect.return_value = None + + # Mock exec_command to return active status + mock_stdout = Mock() + mock_stdout.readlines.return_value = ["active\n"] + mock_ssh.exec_command.return_value = (Mock(), mock_stdout, Mock()) + + common_node_functions.check_service_status(node, service, ssh_private_key, timeout) + + # Verify SSH connection was attempted + mock_ssh.connect.assert_called() + # Verify service status was checked for each service + self.assertEqual(mock_ssh.exec_command.call_count, 2) + # Verify SSH connection was closed + mock_ssh.close.assert_called_once() + + @patch('time.sleep') + @patch('logging.error') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.common_node_functions.paramiko.SSHClient') + def test_check_service_status_service_inactive(self, mock_ssh_client, mock_logging_info, mock_logging_error, mock_sleep): + """ + Test check_service_status logs error when service is inactive + """ + node = "192.168.1.100" + service = ["neutron-server"] + ssh_private_key = "~/.ssh/id_rsa" + timeout = 60 + + # Mock SSH client + mock_ssh = Mock() + mock_ssh_client.return_value = mock_ssh + mock_ssh.connect.return_value = None + + # Mock exec_command to return inactive status + mock_stdout = Mock() + mock_stdout.readlines.return_value = ["inactive\n"] + mock_ssh.exec_command.return_value = (Mock(), mock_stdout, Mock()) + + common_node_functions.check_service_status(node, service, ssh_private_key, timeout) + + # Verify error was logged for inactive service + mock_logging_error.assert_called() + error_call_str = str(mock_logging_error.call_args) + self.assertIn("inactive", error_call_str) + mock_ssh.close.assert_called_once() + + @patch('time.sleep') + @patch('logging.error') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.common_node_functions.paramiko.SSHClient') + def test_check_service_status_ssh_connection_fails(self, mock_ssh_client, mock_logging_info, mock_logging_error, mock_sleep): + """ + Test check_service_status handles SSH connection failures + """ + node = "192.168.1.100" + service = ["neutron-server"] + ssh_private_key = "~/.ssh/id_rsa" + timeout = 5 + + # Mock SSH client to raise exception + mock_ssh = Mock() + mock_ssh_client.return_value = mock_ssh + mock_ssh.connect.side_effect = Exception("Connection timeout") + + # Mock exec_command for when connection eventually works (or doesn't) + mock_stdout = Mock() + mock_stdout.readlines.return_value = ["active\n"] + mock_ssh.exec_command.return_value = (Mock(), mock_stdout, Mock()) + + common_node_functions.check_service_status(node, service, ssh_private_key, timeout) + + # Verify error was logged for SSH connection failure + mock_logging_error.assert_called() + error_call_str = str(mock_logging_error.call_args) + self.assertIn("Failed to ssh", error_call_str) + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.common_node_functions.paramiko.SSHClient') + def test_check_service_status_multiple_services(self, mock_ssh_client, mock_logging, mock_sleep): + """ + Test check_service_status with multiple services + """ + node = "192.168.1.100" + service = ["service1", "service2", "service3"] + ssh_private_key = "~/.ssh/id_rsa" + timeout = 60 + + # Mock SSH client + mock_ssh = Mock() + mock_ssh_client.return_value = mock_ssh + mock_ssh.connect.return_value = None + + # Mock exec_command to return active status + mock_stdout = Mock() + mock_stdout.readlines.return_value = ["active\n"] + mock_ssh.exec_command.return_value = (Mock(), mock_stdout, Mock()) + + common_node_functions.check_service_status(node, service, ssh_private_key, timeout) + + # Verify service status was checked for all services + self.assertEqual(mock_ssh.exec_command.call_count, 3) + mock_ssh.close.assert_called_once() + + @patch('time.sleep') + @patch('logging.info') + @patch('krkn.scenario_plugins.node_actions.common_node_functions.paramiko.SSHClient') + def test_check_service_status_retry_logic(self, mock_ssh_client, mock_logging, mock_sleep): + """ + Test check_service_status retry logic on connection failure then success + """ + node = "192.168.1.100" + service = ["neutron-server"] + ssh_private_key = "~/.ssh/id_rsa" + timeout = 10 + + # Mock SSH client + mock_ssh = Mock() + mock_ssh_client.return_value = mock_ssh + # First two attempts fail, third succeeds + mock_ssh.connect.side_effect = [ + Exception("Timeout"), + Exception("Timeout"), + None # Success + ] + + # Mock exec_command + mock_stdout = Mock() + mock_stdout.readlines.return_value = ["active\n"] + mock_ssh.exec_command.return_value = (Mock(), mock_stdout, Mock()) + + common_node_functions.check_service_status(node, service, ssh_private_key, timeout) + + # Verify multiple connection attempts were made + self.assertGreater(mock_ssh.connect.call_count, 1) + # Verify service was eventually checked + mock_ssh.exec_command.assert_called() + mock_ssh.close.assert_called_once() + + +class TestCommonNodeFunctionsIntegration(unittest.TestCase): + """Integration-style tests for common_node_functions""" + + def setUp(self): + """Set up test fixtures""" + self.mock_kubecli = Mock(spec=KrknKubernetes) + + @patch('logging.info') + def test_get_node_workflow_with_label_filtering(self, mock_logging): + """ + Test complete workflow of getting nodes with label selector and filtering + """ + label_selector = "node-role.kubernetes.io/worker" + instance_kill_count = 2 + available_nodes = ["worker-1", "worker-2", "worker-3", "worker-4", "worker-5"] + self.mock_kubecli.list_killable_nodes.return_value = available_nodes + + result = common_node_functions.get_node(label_selector, instance_kill_count, self.mock_kubecli) + + self.assertEqual(len(result), 2) + # Verify no duplicates + self.assertEqual(len(result), len(set(result))) + # Verify all nodes are from the available list + self.assertTrue(all(node in available_nodes for node in result)) + + @patch('logging.info') + def test_get_node_by_name_validation_workflow(self, mock_logging): + """ + Test complete workflow of validating node names + """ + requested_nodes = ["node-a", "node-b"] + killable_nodes = ["node-a", "node-b", "node-c", "node-d"] + self.mock_kubecli.list_killable_nodes.return_value = killable_nodes + + result = common_node_functions.get_node_by_name(requested_nodes, self.mock_kubecli) + + self.assertEqual(result, requested_nodes) + self.mock_kubecli.list_killable_nodes.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_node_actions_scenario_plugin.py b/tests/test_node_actions_scenario_plugin.py index 2db95d50..0fab5dcf 100644 --- a/tests/test_node_actions_scenario_plugin.py +++ b/tests/test_node_actions_scenario_plugin.py @@ -10,10 +10,15 @@ Assisted By: Claude Code """ import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, Mock, patch, mock_open, call +import yaml +import tempfile +import os from krkn_lib.k8s import KrknKubernetes from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift +from krkn_lib.models.telemetry import ScenarioTelemetry +from krkn_lib.models.k8s import AffectedNodeStatus from krkn.scenario_plugins.node_actions.node_actions_scenario_plugin import NodeActionsScenarioPlugin @@ -24,7 +29,16 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase): """ Set up test fixtures for NodeActionsScenarioPlugin """ + # Reset node_general global variable before each test + import krkn.scenario_plugins.node_actions.node_actions_scenario_plugin as plugin_module + plugin_module.node_general = False + self.plugin = NodeActionsScenarioPlugin() + self.mock_kubecli = Mock(spec=KrknKubernetes) + self.mock_lib_telemetry = Mock(spec=KrknTelemetryOpenshift) + self.mock_lib_telemetry.get_lib_kubernetes.return_value = self.mock_kubecli + self.mock_scenario_telemetry = Mock(spec=ScenarioTelemetry) + self.mock_scenario_telemetry.affected_nodes = [] def test_get_scenario_types(self): """ @@ -35,6 +49,700 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase): self.assertEqual(result, ["node_scenarios"]) self.assertEqual(len(result), 1) + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios') + def test_get_node_scenario_object_generic(self, mock_general_scenarios): + """ + Test get_node_scenario_object returns general_node_scenarios for generic cloud type + """ + node_scenario = {"cloud_type": "generic"} + mock_general_instance = Mock() + mock_general_scenarios.return_value = mock_general_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_general_instance) + mock_general_scenarios.assert_called_once() + args = mock_general_scenarios.call_args[0] + self.assertEqual(args[0], self.mock_kubecli) + self.assertTrue(args[1]) # node_action_kube_check defaults to True + self.assertIsInstance(args[2], AffectedNodeStatus) + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios') + def test_get_node_scenario_object_no_cloud_type(self, mock_general_scenarios): + """ + Test get_node_scenario_object returns general_node_scenarios when cloud_type is not specified + """ + node_scenario = {} + mock_general_instance = Mock() + mock_general_scenarios.return_value = mock_general_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_general_instance) + mock_general_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.aws_node_scenarios') + def test_get_node_scenario_object_aws(self, mock_aws_scenarios): + """ + Test get_node_scenario_object returns aws_node_scenarios for AWS cloud type + """ + node_scenario = {"cloud_type": "aws"} + mock_aws_instance = Mock() + mock_aws_scenarios.return_value = mock_aws_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_aws_instance) + mock_aws_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.gcp_node_scenarios') + def test_get_node_scenario_object_gcp(self, mock_gcp_scenarios): + """ + Test get_node_scenario_object returns gcp_node_scenarios for GCP cloud type + """ + node_scenario = {"cloud_type": "gcp"} + mock_gcp_instance = Mock() + mock_gcp_scenarios.return_value = mock_gcp_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_gcp_instance) + mock_gcp_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.azure_node_scenarios') + def test_get_node_scenario_object_azure(self, mock_azure_scenarios): + """ + Test get_node_scenario_object returns azure_node_scenarios for Azure cloud type + """ + node_scenario = {"cloud_type": "azure"} + mock_azure_instance = Mock() + mock_azure_scenarios.return_value = mock_azure_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_azure_instance) + mock_azure_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.azure_node_scenarios') + def test_get_node_scenario_object_az(self, mock_azure_scenarios): + """ + Test get_node_scenario_object returns azure_node_scenarios for 'az' cloud type alias + """ + node_scenario = {"cloud_type": "az"} + mock_azure_instance = Mock() + mock_azure_scenarios.return_value = mock_azure_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_azure_instance) + mock_azure_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.docker_node_scenarios') + def test_get_node_scenario_object_docker(self, mock_docker_scenarios): + """ + Test get_node_scenario_object returns docker_node_scenarios for Docker cloud type + """ + node_scenario = {"cloud_type": "docker"} + mock_docker_instance = Mock() + mock_docker_scenarios.return_value = mock_docker_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_docker_instance) + mock_docker_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.vmware_node_scenarios') + def test_get_node_scenario_object_vmware(self, mock_vmware_scenarios): + """ + Test get_node_scenario_object returns vmware_node_scenarios for VMware cloud type + """ + node_scenario = {"cloud_type": "vmware"} + mock_vmware_instance = Mock() + mock_vmware_scenarios.return_value = mock_vmware_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_vmware_instance) + mock_vmware_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.vmware_node_scenarios') + def test_get_node_scenario_object_vsphere(self, mock_vmware_scenarios): + """ + Test get_node_scenario_object returns vmware_node_scenarios for vSphere cloud type alias + """ + node_scenario = {"cloud_type": "vsphere"} + mock_vmware_instance = Mock() + mock_vmware_scenarios.return_value = mock_vmware_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_vmware_instance) + mock_vmware_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.ibm_node_scenarios') + def test_get_node_scenario_object_ibm(self, mock_ibm_scenarios): + """ + Test get_node_scenario_object returns ibm_node_scenarios for IBM cloud type + """ + node_scenario = {"cloud_type": "ibm"} + mock_ibm_instance = Mock() + mock_ibm_scenarios.return_value = mock_ibm_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_ibm_instance) + mock_ibm_scenarios.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.ibm_node_scenarios') + def test_get_node_scenario_object_ibmcloud(self, mock_ibm_scenarios): + """ + Test get_node_scenario_object returns ibm_node_scenarios for ibmcloud cloud type alias + """ + node_scenario = {"cloud_type": "ibmcloud", "disable_ssl_verification": False} + mock_ibm_instance = Mock() + mock_ibm_scenarios.return_value = mock_ibm_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_ibm_instance) + args = mock_ibm_scenarios.call_args[0] + self.assertFalse(args[3]) # disable_ssl_verification should be False + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.ibmcloud_power_node_scenarios') + def test_get_node_scenario_object_ibmpower(self, mock_ibmpower_scenarios): + """ + Test get_node_scenario_object returns ibmcloud_power_node_scenarios for ibmpower cloud type + """ + node_scenario = {"cloud_type": "ibmpower"} + mock_ibmpower_instance = Mock() + mock_ibmpower_scenarios.return_value = mock_ibmpower_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_ibmpower_instance) + mock_ibmpower_scenarios.assert_called_once() + + def test_get_node_scenario_object_openstack(self): + """ + Test get_node_scenario_object returns openstack_node_scenarios for OpenStack cloud type + """ + with patch('krkn.scenario_plugins.node_actions.openstack_node_scenarios.openstack_node_scenarios') as mock_openstack: + node_scenario = {"cloud_type": "openstack"} + mock_openstack_instance = Mock() + mock_openstack.return_value = mock_openstack_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_openstack_instance) + mock_openstack.assert_called_once() + + def test_get_node_scenario_object_alibaba(self): + """ + Test get_node_scenario_object returns alibaba_node_scenarios for Alibaba cloud type + """ + with patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.alibaba_node_scenarios') as mock_alibaba: + node_scenario = {"cloud_type": "alibaba"} + mock_alibaba_instance = Mock() + mock_alibaba.return_value = mock_alibaba_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_alibaba_instance) + mock_alibaba.assert_called_once() + + def test_get_node_scenario_object_alicloud(self): + """ + Test get_node_scenario_object returns alibaba_node_scenarios for alicloud alias + """ + with patch('krkn.scenario_plugins.node_actions.alibaba_node_scenarios.alibaba_node_scenarios') as mock_alibaba: + node_scenario = {"cloud_type": "alicloud"} + mock_alibaba_instance = Mock() + mock_alibaba.return_value = mock_alibaba_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_alibaba_instance) + mock_alibaba.assert_called_once() + + def test_get_node_scenario_object_bm(self): + """ + Test get_node_scenario_object returns bm_node_scenarios for bare metal cloud type + """ + with patch('krkn.scenario_plugins.node_actions.bm_node_scenarios.bm_node_scenarios') as mock_bm: + node_scenario = { + "cloud_type": "bm", + "bmc_info": "192.168.1.1", + "bmc_user": "admin", + "bmc_password": "password" + } + mock_bm_instance = Mock() + mock_bm.return_value = mock_bm_instance + + result = self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertEqual(result, mock_bm_instance) + args = mock_bm.call_args[0] + self.assertEqual(args[0], "192.168.1.1") + self.assertEqual(args[1], "admin") + self.assertEqual(args[2], "password") + + def test_get_node_scenario_object_unsupported_cloud(self): + """ + Test get_node_scenario_object raises exception for unsupported cloud type + """ + node_scenario = {"cloud_type": "unsupported_cloud"} + + with self.assertRaises(Exception) as context: + self.plugin.get_node_scenario_object(node_scenario, self.mock_kubecli) + + self.assertIn("not currently supported", str(context.exception)) + self.assertIn("unsupported_cloud", str(context.exception)) + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.common_node_functions') + def test_inject_node_scenario_with_node_name(self, mock_common_funcs): + """ + Test inject_node_scenario with specific node name + """ + node_scenario = { + "node_name": "node1,node2", + "instance_count": 2, + "runs": 1, + "timeout": 120, + "duration": 60, + "poll_interval": 15 + } + action = "node_stop_start_scenario" + mock_scenario_object = Mock() + mock_scenario_object.affected_nodes_status = AffectedNodeStatus() + mock_scenario_object.affected_nodes_status.affected_nodes = [] + + mock_common_funcs.get_node_by_name.return_value = ["node1", "node2"] + + self.plugin.inject_node_scenario( + action, + node_scenario, + mock_scenario_object, + self.mock_kubecli, + self.mock_scenario_telemetry + ) + + mock_common_funcs.get_node_by_name.assert_called_once_with(["node1", "node2"], self.mock_kubecli) + self.assertEqual(mock_scenario_object.node_stop_start_scenario.call_count, 2) + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.common_node_functions') + def test_inject_node_scenario_with_label_selector(self, mock_common_funcs): + """ + Test inject_node_scenario with label selector + """ + node_scenario = { + "label_selector": "node-role.kubernetes.io/worker", + "instance_count": 1 + } + action = "node_reboot_scenario" + mock_scenario_object = Mock() + mock_scenario_object.affected_nodes_status = AffectedNodeStatus() + mock_scenario_object.affected_nodes_status.affected_nodes = [] + + mock_common_funcs.get_node.return_value = ["worker-node-1"] + + self.plugin.inject_node_scenario( + action, + node_scenario, + mock_scenario_object, + self.mock_kubecli, + self.mock_scenario_telemetry + ) + + mock_common_funcs.get_node.assert_called_once_with("node-role.kubernetes.io/worker", 1, self.mock_kubecli) + mock_scenario_object.node_reboot_scenario.assert_called_once() + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.common_node_functions') + def test_inject_node_scenario_with_exclude_label(self, mock_common_funcs): + """ + Test inject_node_scenario with exclude label + """ + node_scenario = { + "label_selector": "node-role.kubernetes.io/worker", + "exclude_label": "node-role.kubernetes.io/master", + "instance_count": 2 + } + action = "node_stop_scenario" + mock_scenario_object = Mock() + mock_scenario_object.affected_nodes_status = AffectedNodeStatus() + mock_scenario_object.affected_nodes_status.affected_nodes = [] + + mock_common_funcs.get_node.side_effect = [ + ["worker-1", "master-1"], + ["master-1"] + ] + + self.plugin.inject_node_scenario( + action, + node_scenario, + mock_scenario_object, + self.mock_kubecli, + self.mock_scenario_telemetry + ) + + self.assertEqual(mock_common_funcs.get_node.call_count, 2) + # Should only process worker-1 after excluding master-1 + self.assertEqual(mock_scenario_object.node_stop_scenario.call_count, 1) + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.common_node_functions') + def test_inject_node_scenario_parallel_mode(self, mock_common_funcs): + """ + Test inject_node_scenario with parallel processing + """ + node_scenario = { + "node_name": "node1,node2,node3", + "parallel": True + } + action = "restart_kubelet_scenario" + mock_scenario_object = Mock() + mock_scenario_object.affected_nodes_status = AffectedNodeStatus() + mock_scenario_object.affected_nodes_status.affected_nodes = [] + + mock_common_funcs.get_node_by_name.return_value = ["node1", "node2", "node3"] + + with patch.object(self.plugin, 'multiprocess_nodes') as mock_multiprocess: + self.plugin.inject_node_scenario( + action, + node_scenario, + mock_scenario_object, + self.mock_kubecli, + self.mock_scenario_telemetry + ) + + mock_multiprocess.assert_called_once() + args = mock_multiprocess.call_args[0] + self.assertEqual(args[0], ["node1", "node2", "node3"]) + self.assertEqual(args[2], action) + + def test_run_node_node_start_scenario(self): + """ + Test run_node executes node_start_scenario action + """ + node_scenario = {"runs": 2, "timeout": 300, "poll_interval": 10} + action = "node_start_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_start_scenario.assert_called_once_with(2, "test-node", 300, 10) + + def test_run_node_node_stop_scenario(self): + """ + Test run_node executes node_stop_scenario action + """ + node_scenario = {"runs": 1, "timeout": 120, "poll_interval": 15} + action = "node_stop_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_stop_scenario.assert_called_once_with(1, "test-node", 120, 15) + + def test_run_node_node_stop_start_scenario(self): + """ + Test run_node executes node_stop_start_scenario action + """ + node_scenario = {"runs": 1, "timeout": 120, "duration": 60, "poll_interval": 15} + action = "node_stop_start_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_stop_start_scenario.assert_called_once_with(1, "test-node", 120, 60, 15) + + def test_run_node_node_termination_scenario(self): + """ + Test run_node executes node_termination_scenario action + """ + node_scenario = {} + action = "node_termination_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_termination_scenario.assert_called_once_with(1, "test-node", 120, 15) + + def test_run_node_node_reboot_scenario(self): + """ + Test run_node executes node_reboot_scenario action + """ + node_scenario = {"soft_reboot": True} + action = "node_reboot_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_reboot_scenario.assert_called_once_with(1, "test-node", 120, True) + + def test_run_node_node_disk_detach_attach_scenario(self): + """ + Test run_node executes node_disk_detach_attach_scenario action + """ + node_scenario = {"duration": 90} + action = "node_disk_detach_attach_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_disk_detach_attach_scenario.assert_called_once_with(1, "test-node", 120, 90) + + def test_run_node_stop_start_kubelet_scenario(self): + """ + Test run_node executes stop_start_kubelet_scenario action + """ + node_scenario = {} + action = "stop_start_kubelet_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.stop_start_kubelet_scenario.assert_called_once_with(1, "test-node", 120) + + def test_run_node_restart_kubelet_scenario(self): + """ + Test run_node executes restart_kubelet_scenario action + """ + node_scenario = {} + action = "restart_kubelet_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.restart_kubelet_scenario.assert_called_once_with(1, "test-node", 120) + + def test_run_node_stop_kubelet_scenario(self): + """ + Test run_node executes stop_kubelet_scenario action + """ + node_scenario = {} + action = "stop_kubelet_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.stop_kubelet_scenario.assert_called_once_with(1, "test-node", 120) + + def test_run_node_node_crash_scenario(self): + """ + Test run_node executes node_crash_scenario action + """ + node_scenario = {} + action = "node_crash_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_crash_scenario.assert_called_once_with(1, "test-node", 120) + + def test_run_node_node_block_scenario(self): + """ + Test run_node executes node_block_scenario action + """ + node_scenario = {"duration": 100} + action = "node_block_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.node_block_scenario.assert_called_once_with(1, "test-node", 120, 100) + + @patch('logging.info') + def test_run_node_stop_start_helper_node_scenario_openstack(self, mock_logging): + """ + Test run_node executes stop_start_helper_node_scenario for OpenStack + """ + node_scenario = { + "cloud_type": "openstack", + "helper_node_ip": "192.168.1.100", + "service": "neutron-server" + } + action = "stop_start_helper_node_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_scenario_object.helper_node_stop_start_scenario.assert_called_once_with(1, "192.168.1.100", 120) + mock_scenario_object.helper_node_service_status.assert_called_once() + + @patch('logging.error') + def test_run_node_stop_start_helper_node_scenario_non_openstack(self, mock_logging): + """ + Test run_node logs error for stop_start_helper_node_scenario on non-OpenStack + """ + node_scenario = { + "cloud_type": "aws", + "helper_node_ip": "192.168.1.100" + } + action = "stop_start_helper_node_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_logging.assert_called() + self.assertIn("not supported", str(mock_logging.call_args)) + + @patch('logging.error') + def test_run_node_stop_start_helper_node_scenario_missing_ip(self, mock_logging): + """ + Test run_node raises exception when helper_node_ip is missing + """ + node_scenario = { + "cloud_type": "openstack", + "helper_node_ip": None + } + action = "stop_start_helper_node_scenario" + mock_scenario_object = Mock() + + with self.assertRaises(Exception) as context: + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + self.assertIn("Helper node IP address is not provided", str(context.exception)) + + @patch('logging.info') + def test_run_node_generic_cloud_skip_unsupported_action(self, mock_logging): + """ + Test run_node skips unsupported actions for generic cloud type + """ + # Set node_general to True for this test + import krkn.scenario_plugins.node_actions.node_actions_scenario_plugin as plugin_module + plugin_module.node_general = True + + node_scenario = {} + action = "node_stop_scenario" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_logging.assert_called() + self.assertIn("not set up for generic cloud type", str(mock_logging.call_args)) + + @patch('logging.info') + def test_run_node_unknown_action(self, mock_logging): + """ + Test run_node logs info for unknown action + """ + node_scenario = {} + action = "unknown_action" + mock_scenario_object = Mock() + + self.plugin.run_node("test-node", mock_scenario_object, action, node_scenario) + + mock_logging.assert_called() + # Could be either message depending on node_general state + call_str = str(mock_logging.call_args) + self.assertTrue( + "no node action that matches" in call_str or + "not set up for generic cloud type" in call_str + ) + + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.cerberus') + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.common_node_functions') + @patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.general_node_scenarios') + @patch('builtins.open', new_callable=mock_open) + @patch('time.time') + def test_run_successful(self, mock_time, mock_file, mock_general_scenarios, mock_common_funcs, mock_cerberus): + """ + Test successful run of node actions scenario + """ + scenario_yaml = { + "node_scenarios": [ + { + "cloud_type": "generic", + "node_name": "test-node", + "actions": ["stop_kubelet_scenario"] + } + ] + } + + mock_file.return_value.__enter__.return_value.read.return_value = yaml.dump(scenario_yaml) + mock_time.side_effect = [1000, 1100] + mock_scenario_object = Mock() + mock_scenario_object.affected_nodes_status = AffectedNodeStatus() + mock_scenario_object.affected_nodes_status.affected_nodes = [] + mock_general_scenarios.return_value = mock_scenario_object + mock_common_funcs.get_node_by_name.return_value = ["test-node"] + mock_cerberus.get_status.return_value = None + + with patch('yaml.full_load', return_value=scenario_yaml): + result = self.plugin.run( + "test-uuid", + "/path/to/scenario.yaml", + {}, + self.mock_lib_telemetry, + self.mock_scenario_telemetry + ) + + self.assertEqual(result, 0) + mock_cerberus.get_status.assert_called_once_with({}, 1000, 1100) + + @patch('logging.error') + @patch('builtins.open', new_callable=mock_open) + def test_run_with_exception(self, mock_file, mock_logging): + """ + Test run handles exceptions and returns 1 + """ + scenario_yaml = { + "node_scenarios": [ + { + "cloud_type": "unsupported" + } + ] + } + + with patch('yaml.full_load', return_value=scenario_yaml): + result = self.plugin.run( + "test-uuid", + "/path/to/scenario.yaml", + {}, + self.mock_lib_telemetry, + self.mock_scenario_telemetry + ) + + self.assertEqual(result, 1) + mock_logging.assert_called() + + @patch('logging.info') + def test_multiprocess_nodes(self, mock_logging): + """ + Test multiprocess_nodes executes run_node for multiple nodes in parallel + """ + nodes = ["node1", "node2", "node3"] + mock_scenario_object = Mock() + action = "restart_kubelet_scenario" + node_scenario = {} + + with patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.ThreadPool') as mock_pool: + mock_pool_instance = Mock() + mock_pool.return_value = mock_pool_instance + + self.plugin.multiprocess_nodes(nodes, mock_scenario_object, action, node_scenario) + + mock_pool.assert_called_once_with(processes=3) + mock_pool_instance.starmap.assert_called_once() + mock_pool_instance.close.assert_called_once() + + @patch('logging.info') + def test_multiprocess_nodes_with_exception(self, mock_logging): + """ + Test multiprocess_nodes handles exceptions gracefully + """ + nodes = ["node1", "node2"] + mock_scenario_object = Mock() + action = "node_reboot_scenario" + node_scenario = {} + + with patch('krkn.scenario_plugins.node_actions.node_actions_scenario_plugin.ThreadPool') as mock_pool: + mock_pool.side_effect = Exception("Pool error") + + self.plugin.multiprocess_nodes(nodes, mock_scenario_object, action, node_scenario) + + mock_logging.assert_called() + self.assertIn("Error on pool multiprocessing", str(mock_logging.call_args)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_syn_flood_scenario_plugin.py b/tests/test_syn_flood_scenario_plugin.py index 412380aa..2038f2a9 100644 --- a/tests/test_syn_flood_scenario_plugin.py +++ b/tests/test_syn_flood_scenario_plugin.py @@ -9,8 +9,13 @@ Usage: Assisted By: Claude Code """ +import base64 +import json import unittest +import uuid +from unittest.mock import MagicMock +from krkn.rollback.config import RollbackContent from krkn.scenario_plugins.syn_flood.syn_flood_scenario_plugin import SynFloodScenarioPlugin @@ -31,6 +36,488 @@ class TestSynFloodScenarioPlugin(unittest.TestCase): self.assertEqual(result, ["syn_flood_scenarios"]) self.assertEqual(len(result), 1) + def test_check_key_value(self): + """ + Test check_key_value method + """ + test_dict = { + "valid_key": "value", + "empty_key": "", + "none_key": None, + "zero_key": 0, + "false_key": False, + } + + self.assertTrue(self.plugin.check_key_value(test_dict, "valid_key")) + self.assertFalse(self.plugin.check_key_value(test_dict, "empty_key")) + self.assertFalse(self.plugin.check_key_value(test_dict, "none_key")) + self.assertFalse(self.plugin.check_key_value(test_dict, "missing_key")) + # 0 and False are valid values + self.assertTrue(self.plugin.check_key_value(test_dict, "zero_key")) + self.assertTrue(self.plugin.check_key_value(test_dict, "false_key")) + + +class TestIsNodeAffinityCorrect(unittest.TestCase): + """Tests for is_node_affinity_correct method""" + + def setUp(self): + self.plugin = SynFloodScenarioPlugin() + + def test_valid_node_affinity(self): + """Test valid node affinity configuration""" + valid_affinity = { + "node-role.kubernetes.io/worker": [""], + } + self.assertTrue(self.plugin.is_node_affinity_correct(valid_affinity)) + + def test_valid_node_affinity_multiple_labels(self): + """Test valid node affinity with multiple labels""" + valid_affinity = { + "node-role.kubernetes.io/worker": ["value1", "value2"], + "topology.kubernetes.io/zone": ["us-east-1a"], + } + self.assertTrue(self.plugin.is_node_affinity_correct(valid_affinity)) + + def test_empty_dict_is_valid(self): + """Test empty dict is valid for node affinity""" + self.assertTrue(self.plugin.is_node_affinity_correct({})) + + def test_invalid_not_a_dict(self): + """Test non-dict input is invalid""" + self.assertFalse(self.plugin.is_node_affinity_correct("not a dict")) + self.assertFalse(self.plugin.is_node_affinity_correct(["list"])) + self.assertFalse(self.plugin.is_node_affinity_correct(123)) + self.assertFalse(self.plugin.is_node_affinity_correct(None)) + + def test_invalid_non_string_key(self): + """Test non-string keys are invalid""" + invalid_affinity = { + 123: ["value"], + } + self.assertFalse(self.plugin.is_node_affinity_correct(invalid_affinity)) + + def test_invalid_non_list_value(self): + """Test non-list values are invalid""" + invalid_affinity = { + "node-role.kubernetes.io/worker": "not a list", + } + self.assertFalse(self.plugin.is_node_affinity_correct(invalid_affinity)) + + +class TestParseConfig(unittest.TestCase): + """Tests for parse_config method""" + + def setUp(self): + self.plugin = SynFloodScenarioPlugin() + + def _create_scenario_file(self, tmp_path, config=None): + """Helper to create a temporary scenario YAML file""" + import yaml + + default_config = { + "packet-size": 120, + "window-size": 64, + "duration": 10, + "namespace": "default", + "target-service": "elasticsearch", + "target-port": 9200, + "target-service-label": "", + "number-of-pods": 2, + "image": "quay.io/krkn-chaos/krkn-syn-flood:v1.0.0", + "attacker-nodes": {"node-role.kubernetes.io/worker": [""]}, + } + if config: + default_config.update(config) + + scenario_file = tmp_path / "test_scenario.yaml" + with open(scenario_file, "w") as f: + yaml.dump(default_config, f) + return str(scenario_file) + + def test_parse_config_valid(self, tmp_path=None): + """Test parsing valid configuration""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(Path(tmp_dir)) + config = self.plugin.parse_config(scenario_file) + + assert config["packet-size"] == 120 + assert config["window-size"] == 64 + assert config["duration"] == 10 + assert config["namespace"] == "default" + assert config["target-service"] == "elasticsearch" + assert config["target-port"] == 9200 + assert config["number-of-pods"] == 2 + + def test_parse_config_file_not_found(self): + """Test parsing non-existent file raises exception""" + with self.assertRaises(Exception) as context: + self.plugin.parse_config("/nonexistent/path/scenario.yaml") + self.assertIn("failed to load scenario file", str(context.exception)) + + def test_parse_config_missing_required_params(self): + """Test parsing config with missing required parameters""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + # Missing packet-size and window-size + scenario_file = self._create_scenario_file( + Path(tmp_dir), + {"packet-size": "", "window-size": None}, + ) + with self.assertRaises(Exception) as context: + self.plugin.parse_config(scenario_file) + self.assertIn("packet-size", str(context.exception)) + self.assertIn("window-size", str(context.exception)) + + def test_parse_config_both_target_service_and_label(self): + """Test parsing config with both target-service and target-service-label set""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file( + Path(tmp_dir), + { + "target-service": "elasticsearch", + "target-service-label": "app=elasticsearch", + }, + ) + with self.assertRaises(Exception) as context: + self.plugin.parse_config(scenario_file) + self.assertIn( + "you cannot select both target-service and target-service-label", + str(context.exception), + ) + + def test_parse_config_neither_target_service_nor_label(self): + """Test parsing config with neither target-service nor target-service-label set""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file( + Path(tmp_dir), + {"target-service": "", "target-service-label": ""}, + ) + with self.assertRaises(Exception) as context: + self.plugin.parse_config(scenario_file) + self.assertIn( + "you have either to set a target service or a label", + str(context.exception), + ) + + def test_parse_config_invalid_attacker_nodes(self): + """Test parsing config with invalid attacker-nodes format""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file( + Path(tmp_dir), + {"attacker-nodes": "invalid"}, + ) + with self.assertRaises(Exception) as context: + self.plugin.parse_config(scenario_file) + self.assertIn("attacker-nodes format is not correct", str(context.exception)) + + def test_parse_config_with_label_selector(self): + """Test parsing config with target-service-label instead of target-service""" + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file( + Path(tmp_dir), + {"target-service": "", "target-service-label": "app=elasticsearch"}, + ) + config = self.plugin.parse_config(scenario_file) + assert config["target-service-label"] == "app=elasticsearch" + assert config["target-service"] == "" + + +class TestSynFloodRun(unittest.TestCase): + """Tests for the run method of SynFloodScenarioPlugin""" + + def _create_scenario_file(self, tmp_path, config=None): + """Helper to create a temporary scenario YAML file""" + import yaml + from pathlib import Path + + default_config = { + "packet-size": 120, + "window-size": 64, + "duration": 1, + "namespace": "default", + "target-service": "elasticsearch", + "target-port": 9200, + "target-service-label": "", + "number-of-pods": 1, + "image": "quay.io/krkn-chaos/krkn-syn-flood:v1.0.0", + "attacker-nodes": {"node-role.kubernetes.io/worker": [""]}, + } + if config: + default_config.update(config) + + scenario_file = Path(tmp_path) / "test_scenario.yaml" + with open(scenario_file, "w") as f: + yaml.dump(default_config, f) + return str(scenario_file) + + def _create_mocks(self): + """Helper to create mock objects for testing""" + mock_lib_telemetry = MagicMock() + mock_lib_kubernetes = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_lib_kubernetes + mock_scenario_telemetry = MagicMock() + return mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry + + def test_run_successful_with_target_service(self): + """Test successful execution with target-service""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(tmp_dir) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.service_exists.return_value = True + # Pod finishes immediately + mock_lib_kubernetes.is_pod_running.return_value = False + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 0) + mock_lib_kubernetes.service_exists.assert_called_once_with( + "elasticsearch", "default" + ) + mock_lib_kubernetes.deploy_syn_flood.assert_called_once() + + def test_run_successful_with_label_selector(self): + """Test successful execution with target-service-label""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file( + tmp_dir, + {"target-service": "", "target-service-label": "app=elasticsearch"}, + ) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.select_service_by_label.return_value = [ + "elasticsearch-1", + "elasticsearch-2", + ] + mock_lib_kubernetes.service_exists.return_value = True + mock_lib_kubernetes.is_pod_running.return_value = False + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 0) + mock_lib_kubernetes.select_service_by_label.assert_called_once_with( + "default", "app=elasticsearch" + ) + # Should deploy pods for each service found + self.assertEqual(mock_lib_kubernetes.deploy_syn_flood.call_count, 2) + + def test_run_service_not_found(self): + """Test run method when service does not exist""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(tmp_dir) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.service_exists.return_value = False + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 1) + mock_lib_kubernetes.deploy_syn_flood.assert_not_called() + + def test_run_multiple_pods(self): + """Test run method with multiple attacker pods""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(tmp_dir, {"number-of-pods": 3}) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.service_exists.return_value = True + mock_lib_kubernetes.is_pod_running.return_value = False + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 0) + self.assertEqual(mock_lib_kubernetes.deploy_syn_flood.call_count, 3) + + def test_run_exception_handling(self): + """Test run method handles exceptions gracefully""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(tmp_dir) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.service_exists.return_value = True + mock_lib_kubernetes.deploy_syn_flood.side_effect = Exception("Deployment failed") + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 1) + + def test_run_waits_for_pods_to_finish(self): + """Test run method waits for pods to finish""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp_dir: + scenario_file = self._create_scenario_file(tmp_dir) + mock_lib_telemetry, mock_lib_kubernetes, mock_scenario_telemetry = ( + self._create_mocks() + ) + + mock_lib_kubernetes.service_exists.return_value = True + # Pod runs for a few iterations then finishes + mock_lib_kubernetes.is_pod_running.side_effect = [True, True, False] + + plugin = SynFloodScenarioPlugin() + + result = plugin.run( + run_uuid=str(uuid.uuid4()), + scenario=scenario_file, + krkn_config={}, + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 0) + # Should have checked pod status multiple times + self.assertGreaterEqual(mock_lib_kubernetes.is_pod_running.call_count, 1) + + +class TestRollbackSynFloodPods(unittest.TestCase): + """Tests for rollback_syn_flood_pods static method""" + def test_rollback_syn_flood_pods_successful(self): + """Test successful rollback of syn flood pods""" + pod_names = ["syn-flood-abc123", "syn-flood-def456"] + encoded_data = base64.b64encode( + json.dumps(pod_names).encode("utf-8") + ).decode("utf-8") + + rollback_content = RollbackContent( + resource_identifier=encoded_data, + namespace="default", + ) + + mock_lib_telemetry = MagicMock() + mock_lib_kubernetes = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_lib_kubernetes + + SynFloodScenarioPlugin.rollback_syn_flood_pods( + rollback_content, mock_lib_telemetry + ) + + assert mock_lib_kubernetes.delete_pod.call_count == 2 + mock_lib_kubernetes.delete_pod.assert_any_call("syn-flood-abc123", "default") + mock_lib_kubernetes.delete_pod.assert_any_call("syn-flood-def456", "default") + + def test_rollback_syn_flood_pods_empty_list(self): + """Test rollback with empty pod list""" + pod_names = [] + encoded_data = base64.b64encode( + json.dumps(pod_names).encode("utf-8") + ).decode("utf-8") + + rollback_content = RollbackContent( + resource_identifier=encoded_data, + namespace="default", + ) + + mock_lib_telemetry = MagicMock() + mock_lib_kubernetes = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_lib_kubernetes + + SynFloodScenarioPlugin.rollback_syn_flood_pods( + rollback_content, mock_lib_telemetry + ) + + mock_lib_kubernetes.delete_pod.assert_not_called() + + def test_rollback_syn_flood_pods_invalid_data(self): + """Test rollback with invalid encoded data handles error gracefully""" + rollback_content = RollbackContent( + resource_identifier="invalid_base64_data", + namespace="default", + ) + + mock_lib_telemetry = MagicMock() + mock_lib_kubernetes = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_lib_kubernetes + + # Should not raise exception, just log error + with self.assertLogs(level='ERROR') as log_context: + SynFloodScenarioPlugin.rollback_syn_flood_pods( + rollback_content, mock_lib_telemetry + ) + + # Verify error was logged + self.assertTrue(any('error' in log.lower() for log in log_context.output)) + + # Verify delete_pod was not called due to invalid data + mock_lib_kubernetes.delete_pod.assert_not_called() + if __name__ == "__main__": unittest.main()