diff --git a/krkn/scenario_plugins/managed_cluster/managed_cluster_scenario_plugin.py b/krkn/scenario_plugins/managed_cluster/managed_cluster_scenario_plugin.py index 666c0f13..fc5f91ed 100644 --- a/krkn/scenario_plugins/managed_cluster/managed_cluster_scenario_plugin.py +++ b/krkn/scenario_plugins/managed_cluster/managed_cluster_scenario_plugin.py @@ -26,22 +26,26 @@ class ManagedClusterScenarioPlugin(AbstractScenarioPlugin): lib_telemetry.get_lib_kubernetes() ) if managedcluster_scenario["actions"]: + for action in managedcluster_scenario["actions"]: - try: - self.inject_managedcluster_scenario( - action, - managedcluster_scenario, - managedcluster_scenario_object, - lib_telemetry.get_lib_kubernetes(), - ) - except Exception as e: - logging.error( - "ManagedClusterScenarioPlugin exiting due to Exception %s" - % e - ) - return 1 - else: - return 0 + try: + self.inject_managedcluster_scenario( + action, + managedcluster_scenario, + managedcluster_scenario_object, + lib_telemetry.get_lib_kubernetes(), + ) + except Exception as e: + logging.error( + "ManagedClusterScenarioPlugin exiting due to Exception %s" + % e + ) + return 1 + else: + logging.error( + "ManagedClusterScenarioPlugin: 'actions' must be defined and non-empty in the scenario config" + ) + return 1 return 0 def inject_managedcluster_scenario( diff --git a/krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py b/krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py index 68c91202..cd26f697 100644 --- a/krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py +++ b/krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py @@ -40,7 +40,9 @@ class ZoneOutageScenarioPlugin(AbstractScenarioPlugin): start_time = int(time.time()) if cloud_type.lower() == "aws": self.cloud_object = AWS() - self.network_based_zone(scenario_config) + result = self.network_based_zone(scenario_config) + if result != 0: + return 1 else: kubecli = lib_telemetry.get_lib_kubernetes() if cloud_type.lower() == "gcp": @@ -171,70 +173,77 @@ class ZoneOutageScenarioPlugin(AbstractScenarioPlugin): raise def network_based_zone(self, scenario_config: dict[str, any]): - - vpc_id = scenario_config["vpc_id"] - subnet_ids = scenario_config["subnet_id"] - duration = scenario_config["duration"] - # Add support for user-provided default network ACL - default_acl_id = scenario_config.get("default_acl_id") - ids = {} - acl_ids_created = [] - for subnet_id in subnet_ids: - logging.info("Targeting subnet_id") - network_association_ids = [] - associations, original_acl_id = self.cloud_object.describe_network_acls( - vpc_id, subnet_id - ) - for entry in associations: - if entry["SubnetId"] == subnet_id: - network_association_ids.append( - entry["NetworkAclAssociationId"] - ) - logging.info( - "Network association ids associated with " - "the subnet %s: %s" % (subnet_id, network_association_ids) - ) - - # Use provided default ACL if available, otherwise create a new one - if default_acl_id: - acl_id = default_acl_id + try: + vpc_id = scenario_config["vpc_id"] + subnet_ids = scenario_config["subnet_id"] + duration = scenario_config["duration"] + # Add support for user-provided default network ACL + default_acl_id = scenario_config.get("default_acl_id") + ids = {} + acl_ids_created = [] + for subnet_id in subnet_ids: + logging.info("Targeting subnet_id") + network_association_ids = [] + associations, original_acl_id = self.cloud_object.describe_network_acls( + vpc_id, subnet_id + ) + for entry in associations: + if entry["SubnetId"] == subnet_id: + network_association_ids.append( + entry["NetworkAclAssociationId"] + ) logging.info( - "Using provided default ACL ID %s - this ACL will not be deleted after the scenario", - default_acl_id + "Network association ids associated with " + "the subnet %s: %s" % (subnet_id, network_association_ids) ) - # Don't add to acl_ids_created since we don't want to delete user-provided ACLs at cleanup - else: - acl_id = self.cloud_object.create_default_network_acl(vpc_id) - logging.info("Created new default ACL %s", acl_id) - acl_ids_created.append(acl_id) - new_association_id = self.cloud_object.replace_network_acl_association( - network_association_ids[0], acl_id + # Use provided default ACL if available, otherwise create a new one + if default_acl_id: + acl_id = default_acl_id + logging.info( + "Using provided default ACL ID %s - this ACL will not be deleted after the scenario", + default_acl_id + ) + # Don't add to acl_ids_created since we don't want to delete user-provided ACLs at cleanup + else: + acl_id = self.cloud_object.create_default_network_acl(vpc_id) + logging.info("Created new default ACL %s", acl_id) + acl_ids_created.append(acl_id) + + new_association_id = self.cloud_object.replace_network_acl_association( + network_association_ids[0], acl_id + ) + + # capture the orginal_acl_id, created_acl_id and + # new association_id to use during the recovery + ids[new_association_id] = original_acl_id + + # wait for the specified duration + logging.info( + "Waiting for the specified duration " "in the config: %s" % duration ) + time.sleep(duration) - # capture the original_acl_id, created_acl_id and - # new association_id to use during the recovery - ids[new_association_id] = original_acl_id - - # wait for the specified duration - logging.info( - "Waiting for the specified duration " "in the config: %s" % duration - ) - time.sleep(duration) - - # replace the applied acl with the previous acl in use - for new_association_id, original_acl_id in ids.items(): - self.cloud_object.replace_network_acl_association( - new_association_id, original_acl_id + # replace the applied acl with the previous acl in use + for new_association_id, original_acl_id in ids.items(): + self.cloud_object.replace_network_acl_association( + new_association_id, original_acl_id + ) + logging.info( + "Wating for 60 seconds to make sure " "the changes are in place" ) - logging.info( - "Waiting for 60 seconds to make sure " "the changes are in place" - ) - time.sleep(60) + time.sleep(60) - # delete the network acl created for the run - for acl_id in acl_ids_created: - self.cloud_object.delete_network_acl(acl_id) + # delete the network acl created for the run + for acl_id in acl_ids_created: + self.cloud_object.delete_network_acl(acl_id) + except Exception as e: + logging.error( + f"Network based zone outage scenario failed with exception: {e}" + ) + return 1 + + return 0 def get_scenario_types(self) -> list[str]: return ["zone_outages_scenarios"] diff --git a/tests/test_application_outage_scenario_plugin.py b/tests/test_application_outage_scenario_plugin.py index 9137d65f..bde43bbf 100644 --- a/tests/test_application_outage_scenario_plugin.py +++ b/tests/test_application_outage_scenario_plugin.py @@ -26,6 +26,10 @@ class TestApplicationOutageScenarioPlugin(unittest.TestCase): """ self.plugin = ApplicationOutageScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type diff --git a/tests/test_container_scenario_plugin.py b/tests/test_container_scenario_plugin.py index 8ca40365..fb1171cd 100644 --- a/tests/test_container_scenario_plugin.py +++ b/tests/test_container_scenario_plugin.py @@ -26,6 +26,10 @@ class TestContainerScenarioPlugin(unittest.TestCase): """ self.plugin = ContainerScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type diff --git a/tests/test_managed_cluster_scenario_plugin.py b/tests/test_managed_cluster_scenario_plugin.py index 028b42ce..80f7d972 100644 --- a/tests/test_managed_cluster_scenario_plugin.py +++ b/tests/test_managed_cluster_scenario_plugin.py @@ -36,6 +36,187 @@ class TestManagedClusterScenarioPlugin(unittest.TestCase): self.assertEqual(result, ["managedcluster_scenarios"]) self.assertEqual(len(result), 1) + @patch('time.time') + @patch('builtins.open', create=True) + @patch('yaml.full_load') + @patch('krkn.cerberus.get_status') + def test_run_multiple_actions_executes_all(self, mock_cerberus, mock_yaml, mock_open, mock_time): + """ + Test that run() executes all actions, not just the first one + This tests the fix for the early return bug + """ + mock_time.return_value = 1234567890 + + # Setup mock scenario config with multiple actions + mock_yaml.return_value = { + "managedcluster_scenarios": [ + { + "actions": [ + "managedcluster_start_scenario", + "managedcluster_stop_scenario", + "managedcluster_reboot_scenario" + ], + "managedcluster_name": "test-cluster", + "runs": 1, + "instance_count": 1, + "timeout": 120 + } + ] + } + + mock_lib_telemetry = Mock(spec=KrknTelemetryOpenshift) + mock_lib_telemetry.get_lib_kubernetes.return_value = self.mock_kubecli + + mock_scenario_telemetry = Mock() + + # Mock inject_managedcluster_scenario to track calls + call_tracker = [] + original_inject = self.plugin.inject_managedcluster_scenario + def track_inject(action, *args, **kwargs): + call_tracker.append(action) + + with patch.object(self.plugin, 'inject_managedcluster_scenario', side_effect=track_inject): + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert all three actions were called + self.assertEqual(result, 0) + self.assertEqual(len(call_tracker), 3) + self.assertIn("managedcluster_start_scenario", call_tracker) + self.assertIn("managedcluster_stop_scenario", call_tracker) + self.assertIn("managedcluster_reboot_scenario", call_tracker) + + + @patch('time.time') + @patch('builtins.open', create=True) + @patch('yaml.full_load') + def test_run_stops_on_first_error(self, mock_yaml, mock_open, mock_time): + """ + Test that run() returns 1 and stops executing on first error + """ + mock_time.return_value = 1234567890 + + # Setup mock scenario config with multiple actions + mock_yaml.return_value = { + "managedcluster_scenarios": [ + { + "actions": [ + "managedcluster_start_scenario", + "managedcluster_stop_scenario", + "managedcluster_reboot_scenario" + ], + "managedcluster_name": "test-cluster", + "runs": 1, + "instance_count": 1, + "timeout": 120 + } + ] + } + + mock_lib_telemetry = Mock(spec=KrknTelemetryOpenshift) + mock_lib_telemetry.get_lib_kubernetes.return_value = self.mock_kubecli + + mock_scenario_telemetry = Mock() + + # Mock inject_managedcluster_scenario to raise exception on first call + call_tracker = [] + def track_inject_with_error(action, *args, **kwargs): + call_tracker.append(action) + if action == "managedcluster_start_scenario": + raise Exception("Test failure") + + with patch.object(self.plugin, 'inject_managedcluster_scenario', side_effect=track_inject_with_error): + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert failure and only first action was attempted + self.assertEqual(result, 1) + self.assertEqual(len(call_tracker), 1) + self.assertEqual(call_tracker[0], "managedcluster_start_scenario") + + + @patch('builtins.open', create=True) + @patch('yaml.full_load') + def test_run_returns_error_when_actions_empty(self, mock_yaml, _mock_open): + """ + Test that run() returns 1 and logs an error when actions is an empty list + """ + mock_yaml.return_value = { + "managedcluster_scenarios": [ + { + "actions": [], + "managedcluster_name": "test-cluster", + "runs": 1, + "instance_count": 1, + "timeout": 120 + } + ] + } + + mock_lib_telemetry = Mock(spec=KrknTelemetryOpenshift) + mock_lib_telemetry.get_lib_kubernetes.return_value = self.mock_kubecli + mock_scenario_telemetry = Mock() + + with self.assertLogs('root', level='ERROR') as log_ctx: + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 1) + self.assertTrue( + any("actions" in msg for msg in log_ctx.output), + f"Expected 'actions' in error log, got: {log_ctx.output}", + ) + + @patch('builtins.open', create=True) + @patch('yaml.full_load') + def test_run_returns_error_when_actions_none(self, mock_yaml, _mock_open): + """ + Test that run() returns 1 and logs an error when actions is None + """ + mock_yaml.return_value = { + "managedcluster_scenarios": [ + { + "actions": None, + "managedcluster_name": "test-cluster", + "runs": 1, + "instance_count": 1, + "timeout": 120 + } + ] + } + + mock_lib_telemetry = Mock(spec=KrknTelemetryOpenshift) + mock_lib_telemetry.get_lib_kubernetes.return_value = self.mock_kubecli + mock_scenario_telemetry = Mock() + + with self.assertLogs('root', level='ERROR') as log_ctx: + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry, + ) + + self.assertEqual(result, 1) + self.assertTrue( + any("actions" in msg for msg in log_ctx.output), + f"Expected 'actions' in error log, got: {log_ctx.output}", + ) + class TestCommonFunctions(unittest.TestCase): """ diff --git a/tests/test_native_scenario_plugin.py b/tests/test_native_scenario_plugin.py index 5d73059b..ebda42c1 100644 --- a/tests/test_native_scenario_plugin.py +++ b/tests/test_native_scenario_plugin.py @@ -26,6 +26,10 @@ class TestNativeScenarioPlugin(unittest.TestCase): """ self.plugin = NativeScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario types diff --git a/tests/test_openstack_node_scenarios.py b/tests/test_openstack_node_scenarios.py index 6a906134..842d7043 100644 --- a/tests/test_openstack_node_scenarios.py +++ b/tests/test_openstack_node_scenarios.py @@ -30,6 +30,10 @@ class TestOPENSTACKCLOUD(unittest.TestCase): """Set up test fixtures""" self.openstack = OPENSTACKCLOUD() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.openstack = None + def test_openstackcloud_init(self): """Test OPENSTACKCLOUD class initialization""" self.assertEqual(self.openstack.Wait, 30) @@ -276,6 +280,13 @@ class TestOpenstackNodeScenarios(unittest.TestCase): affected_nodes_status=self.affected_nodes_status ) + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.scenario = None + self.kubecli = None + self.mock_openstack = None + self.affected_nodes_status = None + @patch('krkn.scenario_plugins.node_actions.common_node_functions.wait_for_ready_status') def test_node_start_scenario_success(self, mock_wait_ready): """Test node start scenario successfully""" diff --git a/tests/test_pod_disruption_scenario_plugin.py b/tests/test_pod_disruption_scenario_plugin.py index 6c069f16..30a664f8 100644 --- a/tests/test_pod_disruption_scenario_plugin.py +++ b/tests/test_pod_disruption_scenario_plugin.py @@ -26,6 +26,10 @@ class TestPodDisruptionScenarioPlugin(unittest.TestCase): """ self.plugin = PodDisruptionScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type diff --git a/tests/test_pvc_scenario_plugin.py b/tests/test_pvc_scenario_plugin.py index 18495f3a..d4c6b9ec 100644 --- a/tests/test_pvc_scenario_plugin.py +++ b/tests/test_pvc_scenario_plugin.py @@ -32,6 +32,10 @@ class TestPvcScenarioPlugin(unittest.TestCase): """ self.plugin = PvcScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type @@ -49,6 +53,10 @@ class TestToKbytes(unittest.TestCase): """Set up test fixtures""" self.plugin = PvcScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_to_kbytes_1ki(self): """Test to_kbytes with 1Ki""" self.assertEqual(self.plugin.to_kbytes("1Ki"), 1) @@ -120,6 +128,10 @@ class TestRemoveTempFile(unittest.TestCase): """Set up test fixtures""" self.plugin = PvcScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_remove_temp_file_success(self): """Test successful removal of temp file""" mock_kubecli = MagicMock(spec=KrknKubernetes) @@ -231,6 +243,10 @@ class TestPvcScenarioPluginRun(unittest.TestCase): """Set up test fixtures""" self.plugin = PvcScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def create_scenario_file(self, config: dict, temp_dir: str) -> str: """Helper to create a temporary scenario YAML file in the given directory""" path = os.path.join(temp_dir, "scenario.yaml") diff --git a/tests/test_syn_flood_scenario_plugin.py b/tests/test_syn_flood_scenario_plugin.py index cae8b48a..ec86628f 100644 --- a/tests/test_syn_flood_scenario_plugin.py +++ b/tests/test_syn_flood_scenario_plugin.py @@ -27,6 +27,10 @@ class TestSynFloodScenarioPlugin(unittest.TestCase): """ self.plugin = SynFloodScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type @@ -63,6 +67,10 @@ class TestIsNodeAffinityCorrect(unittest.TestCase): def setUp(self): self.plugin = SynFloodScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_valid_node_affinity(self): """Test valid node affinity configuration""" valid_affinity = { @@ -110,6 +118,10 @@ class TestParseConfig(unittest.TestCase): def setUp(self): self.plugin = SynFloodScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def _create_scenario_file(self, tmp_path, config=None): """Helper to create a temporary scenario YAML file""" import yaml diff --git a/tests/test_time_actions_scenario_plugin.py b/tests/test_time_actions_scenario_plugin.py index 524e9b3f..a2dc0538 100644 --- a/tests/test_time_actions_scenario_plugin.py +++ b/tests/test_time_actions_scenario_plugin.py @@ -26,6 +26,10 @@ class TestTimeActionsScenarioPlugin(unittest.TestCase): """ self.plugin = TimeActionsScenarioPlugin() + def tearDown(self): + """Clean up after each test to prevent state leakage""" + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type @@ -50,6 +54,72 @@ class TestTimeActionsScenarioPlugin(unittest.TestCase): logged_msg = mock_logging.error.call_args[0][0] self.assertIn("disk quota exceeded", logged_msg) self.assertNotIn("NameError", logged_msg) + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + @unittest.mock.patch('logging.error') + def test_run_exception_handling_with_variable(self, mock_logging_error, mock_yaml, mock_open): + """ + Test that run() properly captures exception variable and logs it + This tests the fix for the undefined variable 'e' bug + """ + # Setup mock to raise exception + mock_yaml.side_effect = RuntimeError("Test exception message") + + mock_lib_telemetry = MagicMock() + mock_scenario_telemetry = MagicMock() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert failure is returned + self.assertEqual(result, 1) + + # Assert logging.error was called with the exception message + mock_logging_error.assert_called_once() + error_call_args = str(mock_logging_error.call_args) + self.assertIn("Test exception message", error_call_args) + self.assertIn("TimeActionsScenarioPlugin", error_call_args) + + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + def test_run_with_skew_time_exception(self, mock_yaml, mock_open): + """ + Test that run() handles exceptions from skew_time method + """ + # Setup mock scenario config + mock_yaml.return_value = { + "time_scenarios": [ + { + "action": "skew_time", + "object_type": "node", + "object_name": ["test-node"] + } + ] + } + + mock_lib_telemetry = MagicMock() + mock_kubecli = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_kubecli + + # Make skew_time raise an exception + with unittest.mock.patch.object(self.plugin, 'skew_time', side_effect=Exception("Skew failed")): + mock_scenario_telemetry = MagicMock() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert failure is returned + self.assertEqual(result, 1) if __name__ == "__main__": diff --git a/tests/test_zone_outage_scenario_plugin.py b/tests/test_zone_outage_scenario_plugin.py index 10143c9a..d684d310 100644 --- a/tests/test_zone_outage_scenario_plugin.py +++ b/tests/test_zone_outage_scenario_plugin.py @@ -31,9 +31,20 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase): def setUp(self): """ Set up test fixtures for ZoneOutageScenarioPlugin + Creates a fresh plugin instance for each test to avoid state pollution """ self.plugin = ZoneOutageScenarioPlugin() + def tearDown(self): + """ + Clean up after each test to prevent state leakage between tests + """ + # Clear any cloud_object that might have been set + if hasattr(self.plugin, 'cloud_object'): + delattr(self.plugin, 'cloud_object') + # Create a completely fresh instance for the next test + self.plugin = None + def test_get_scenario_types(self): """ Test get_scenario_types returns correct scenario type @@ -43,6 +54,175 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase): self.assertEqual(result, ["zone_outages_scenarios"]) self.assertEqual(len(result), 1) + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.gcp_node_scenarios') + @unittest.mock.patch('krkn.cerberus.publish_kraken_status') + def test_run_propagates_node_based_zone_failure(self, mock_cerberus, mock_gcp_scenarios, mock_yaml, mock_open): + """ + Test that run() properly propagates failure from node_based_zone method + This tests the fix for the ignored return value bug + """ + # Setup mock scenario config for GCP + mock_yaml.return_value = { + "zone_outage": { + "cloud_type": "gcp", + "zone": "us-central1-a", + "duration": 60, + "timeout": 180, + "kube_check": True + } + } + + mock_lib_telemetry = MagicMock() + mock_kubecli = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_kubecli + + # Mock GCP scenarios + mock_gcp_instance = MagicMock() + mock_affected_nodes_status = MagicMock() + mock_affected_nodes_status.affected_nodes = [] + mock_gcp_instance.affected_nodes_status = mock_affected_nodes_status + mock_gcp_scenarios.return_value = mock_gcp_instance + + # Mock node_based_zone to return failure + with unittest.mock.patch.object(self.plugin, 'node_based_zone', return_value=1): + mock_scenario_telemetry = MagicMock() + mock_scenario_telemetry.affected_nodes = [] # Must be a list for .extend() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert failure is properly propagated + self.assertEqual(result, 1) + + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.gcp_node_scenarios') + @unittest.mock.patch('krkn.cerberus.publish_kraken_status') + def test_run_succeeds_when_node_based_zone_succeeds(self, mock_cerberus, mock_gcp_scenarios, mock_yaml, mock_open): + """ + Test that run() returns 0 when node_based_zone succeeds + """ + # Setup mock scenario config for GCP + mock_yaml.return_value = { + "zone_outage": { + "cloud_type": "gcp", + "zone": "us-central1-a", + "duration": 60, + "timeout": 180, + "kube_check": True + } + } + + mock_lib_telemetry = MagicMock() + mock_kubecli = MagicMock() + mock_lib_telemetry.get_lib_kubernetes.return_value = mock_kubecli + + # Mock GCP scenarios + mock_gcp_instance = MagicMock() + mock_affected_nodes_status = MagicMock() + mock_affected_nodes_status.affected_nodes = [] + mock_gcp_instance.affected_nodes_status = mock_affected_nodes_status + mock_gcp_scenarios.return_value = mock_gcp_instance + + # Mock node_based_zone to return success + with unittest.mock.patch.object(self.plugin, 'node_based_zone', return_value=0): + mock_scenario_telemetry = MagicMock() + mock_scenario_telemetry.affected_nodes = [] # Must be a list for .extend() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert success + self.assertEqual(result, 0) + + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.AWS') + @unittest.mock.patch('krkn.cerberus.publish_kraken_status') + def test_run_aws_network_based_zone(self, mock_cerberus, mock_aws_class, mock_yaml, mock_open): + """ + Test that run() handles AWS network-based zone outage correctly + """ + # Setup mock scenario config for AWS + mock_yaml.return_value = { + "zone_outage": { + "cloud_type": "aws", + "vpc_id": "vpc-12345", + "subnet_id": ["subnet-1", "subnet-2"], + "duration": 60 + } + } + + mock_aws_instance = MagicMock() + mock_aws_class.return_value = mock_aws_instance + + # Mock the network_based_zone method to return success + with unittest.mock.patch.object(self.plugin, 'network_based_zone', return_value=0): + mock_lib_telemetry = MagicMock() + mock_scenario_telemetry = MagicMock() + mock_scenario_telemetry.affected_nodes = [] # Must be a list for .extend() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert success + self.assertEqual(result, 0) + + @unittest.mock.patch('builtins.open', create=True) + @unittest.mock.patch('yaml.full_load') + @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.AWS') + @unittest.mock.patch('krkn.cerberus.publish_kraken_status') + def test_run_aws_network_based_zone_failure(self, mock_cerberus, mock_aws_class, mock_yaml, mock_open): + """ + Test that run() properly propagates failure from network_based_zone method + """ + # Setup mock scenario config for AWS + mock_yaml.return_value = { + "zone_outage": { + "cloud_type": "aws", + "vpc_id": "vpc-12345", + "subnet_id": ["subnet-1", "subnet-2"], + "duration": 60 + } + } + + mock_aws_instance = MagicMock() + mock_aws_class.return_value = mock_aws_instance + + # Mock the network_based_zone method to return failure + with unittest.mock.patch.object(self.plugin, 'network_based_zone', return_value=1): + mock_lib_telemetry = MagicMock() + mock_scenario_telemetry = MagicMock() + mock_scenario_telemetry.affected_nodes = [] # Must be a list for .extend() + + # Execute the run method + result = self.plugin.run( + run_uuid="test-uuid", + scenario="test_scenario.yaml", + lib_telemetry=mock_lib_telemetry, + scenario_telemetry=mock_scenario_telemetry + ) + + # Assert failure is properly propagated + self.assertEqual(result, 1) + class TestRollbackGcpZoneOutage(unittest.TestCase): """Tests for the GCP zone outage rollback functionality"""