diff --git a/tests/kubevirt_vm_outage/test_kubevirt_vm_outage.py b/tests/kubevirt_vm_outage/test_kubevirt_vm_outage.py deleted file mode 100644 index e78aad3c..00000000 --- a/tests/kubevirt_vm_outage/test_kubevirt_vm_outage.py +++ /dev/null @@ -1,395 +0,0 @@ -import unittest -import itertools -import time -from datetime import datetime, timedelta, timezone -from unittest.mock import MagicMock, patch - -import yaml -from kubernetes.client.rest import ApiException -from krkn_lib.k8s import KrknKubernetes -from krkn_lib.models.k8s import AffectedPod, PodsStatus -from krkn_lib.models.telemetry import ScenarioTelemetry -from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift -import tempfile -import os - -from krkn.scenario_plugins.kubevirt_vm_outage.kubevirt_vm_outage_scenario_plugin import KubevirtVmOutageScenarioPlugin - - -class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): - - def setUp(self): - """ - Set up test fixtures for KubevirtVmOutageScenarioPlugin - """ - self.plugin = KubevirtVmOutageScenarioPlugin() - - # Create mock k8s client - self.k8s_client = MagicMock() - self.custom_object_client = MagicMock() - self.k8s_client.custom_object_client = self.custom_object_client - self.plugin.k8s_client = self.k8s_client - self.plugin.custom_object_client = self.custom_object_client - - # Mock methods needed for KubeVirt operations - self.k8s_client.list_custom_resource_definition = MagicMock() - - # Mock custom resource definition list with KubeVirt CRDs - crd_list = MagicMock() - crd_item = MagicMock() - crd_item.spec = MagicMock() - crd_item.spec.group = "kubevirt.io" - crd_list.items = [crd_item] - self.k8s_client.list_custom_resource_definition.return_value = crd_list - - # Mock VMI data - base_time = datetime.now(timezone.utc) - self.mock_vmi = { - "metadata": { - "name": "test-vm", - "namespace": "default", - "creationTimestamp": base_time.isoformat() + "Z" - }, - "status": { - "phase": "Running" - } - } - - # Mock VMI with new creation timestamp (after recreation) - self.mock_vmi_recreated = { - "metadata": { - "name": "test-vm", - "namespace": "default", - "creationTimestamp": (base_time + timedelta(minutes=1)).isoformat() + "Z" - }, - "status": { - "phase": "Running" - } - } - - # Create test config - self.config = { - "scenarios": [ - { - "name": "kubevirt outage test", - "scenario": "kubevirt_vm_outage", - "parameters": { - "vm_name": "test-vm", - "namespace": "default", - "duration": 0 - } - } - ] - } - - # Create a temporary config file - temp_dir = tempfile.gettempdir() - self.scenario_file = os.path.join(temp_dir, "test_kubevirt_scenario.yaml") - with open(self.scenario_file, "w") as f: - yaml.dump(self.config, f) - - # Mock dependencies - self.telemetry = MagicMock(spec=KrknTelemetryOpenshift) - self.scenario_telemetry = MagicMock(spec=ScenarioTelemetry) - self.telemetry.get_lib_kubernetes.return_value = self.k8s_client - - def create_incrementing_time_function(self): - """ - Create an incrementing time function that returns sequential float values. - Returns a callable that can be used with patch('time.time', side_effect=...) - """ - counter = itertools.count(1) - def mock_time(): - return float(next(counter)) - return mock_time - - def test_successful_injection_and_recovery(self): - """ - Test successful deletion and recovery of a VMI - """ - # Mock list_namespaces_by_regex to return a single namespace - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - - # Mock list_namespaced_custom_object to return our VMI - self.custom_object_client.list_namespaced_custom_object = MagicMock( - side_effect=[ - {"items": [self.mock_vmi]}, # For get_vmis - {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment - ] - ) - - # Mock get_namespaced_custom_object with a sequence that handles multiple calls - # Call sequence: - # 1. validate_environment: get original VMI - # 2. execute_scenario: get VMI before deletion - # 3. delete_vmi: loop checking if timestamp changed (returns recreated VMI on first check) - # 4+. wait_for_running: loop until phase is Running (may call multiple times) - get_vmi_responses = [ - self.mock_vmi, # Initial get in validate_environment - self.mock_vmi, # Get before delete - self.mock_vmi_recreated, # After delete (recreated with new timestamp) - self.mock_vmi_recreated, # Check if running - ] - - class GetVmiSideEffect: - """ - Callable helper that returns a predefined sequence of VMIs. - If called more times than there are responses, it fails the test - to surface unexpected additional calls instead of silently - masking them. - """ - def __init__(self, responses): - self._responses = responses - self._call_iter = itertools.count() - self.call_count = 0 - - def __call__(self, *args, **kwargs): - call_num = next(self._call_iter) - self.call_count = call_num + 1 - - if call_num >= len(self._responses): - raise AssertionError( - f"get_vmi_side_effect called more times ({call_num + 1}) " - f"than expected ({len(self._responses)})." - ) - return self._responses[call_num] - - get_vmi_side_effect = GetVmiSideEffect(get_vmi_responses) - self.custom_object_client.get_namespaced_custom_object = MagicMock( - side_effect=get_vmi_side_effect - ) - - # Mock delete operation - self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) - - with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) - - self.assertEqual(result, 0) - # Verify get_namespaced_custom_object was called exactly as many times as - # there are predefined responses; get_vmi_side_effect will raise if any - # additional unexpected calls are made. - self.assertEqual( - self.custom_object_client.get_namespaced_custom_object.call_count, - len(get_vmi_responses), - ) - - # Verify that the VMI delete operation was performed once with expected parameters. - self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( - group="kubevirt.io", - version="v1", - namespace="default", - plural="virtualmachineinstances", - name="test-vm", - ) - - def test_injection_failure(self): - """ - Test failure during VMI deletion - """ - # Mock list_namespaces_by_regex - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - - # Mock list to return VMI - self.custom_object_client.list_namespaced_custom_object = MagicMock( - side_effect=[ - {"items": [self.mock_vmi]}, # For get_vmis - {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment - ] - ) - - # Mock get_vmi - self.custom_object_client.get_namespaced_custom_object = MagicMock( - side_effect=[ - self.mock_vmi, # validate_environment - self.mock_vmi, # get before delete - ] - ) - - # Mock delete to raise an error - self.custom_object_client.delete_namespaced_custom_object = MagicMock( - side_effect=ApiException(status=500, reason="Internal Server Error") - ) - - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) - - self.assertEqual(result, 1) - # Verify delete was attempted before the error occurred - self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( - group="kubevirt.io", - version="v1", - namespace="default", - plural="virtualmachineinstances", - name="test-vm" - ) - - def test_disable_auto_restart(self): - """ - Test VM auto-restart can be disabled - """ - # Configure test with disable_auto_restart=True - self.config["scenarios"][0]["parameters"]["disable_auto_restart"] = True - - # Mock list_namespaces_by_regex - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - - # Mock VM object for patching - mock_vm = { - "metadata": {"name": "test-vm", "namespace": "default"}, - "spec": {"running": True} - } - - # Mock list to return VMI - self.custom_object_client.list_namespaced_custom_object = MagicMock( - side_effect=[ - {"items": [self.mock_vmi]}, # For get_vmis - {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment - ] - ) - - # Mock get_namespaced_custom_object with detailed call sequence. - # NOTE: validate_environment uses list_namespaced_custom_object earlier in the flow; - # this mock only covers subsequent get_namespaced_custom_object calls. - # It handles both VMI (virtualmachineinstances) and VM (virtualmachines) resources. - # Call sequence for get_namespaced_custom_object: - # 1. execute_scenario: get VMI before deletion (plural=virtualmachineinstances) - # 2. patch_vm_spec: get VM for patching (plural=virtualmachines) - # 3. delete_vmi: loop checking if VMI timestamp changed (plural=virtualmachineinstances) - # 4+. wait_for_running: loop until VMI phase is Running (plural=virtualmachineinstances) - self.custom_object_client.get_namespaced_custom_object = MagicMock( - side_effect=[ - self.mock_vmi, # Call 1: get VMI before delete - mock_vm, # Call 2: get VM for patching (different resource type) - self.mock_vmi_recreated, # Call 3: delete_vmi detects new timestamp - self.mock_vmi_recreated, # Call 4: wait_for_running checks phase - ] - ) - - # Mock patch and delete operations - self.custom_object_client.patch_namespaced_custom_object = MagicMock(return_value=mock_vm) - self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) - - with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) - - self.assertEqual(result, 0) - # Verify patch was called to disable auto-restart - self.custom_object_client.patch_namespaced_custom_object.assert_called() - - def test_recovery_when_vmi_does_not_exist(self): - """ - Test recovery logic when VMI does not exist after deletion - """ - # Store the original VMI in the plugin for recovery - self.plugin.original_vmi = self.mock_vmi.copy() - - # Initialize affected_pod which is used by wait_for_running - self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") - - # Set up running VMI data for after recovery - running_vmi = { - "metadata": { - "name": "test-vm", - "namespace": "default", - "creationTimestamp": (datetime.now(timezone.utc) + timedelta(minutes=2)).isoformat() + "Z" - }, - "status": {"phase": "Running"} - } - - # Mock get_namespaced_custom_object call sequence triggered during recovery. - # Call sequence: - # 1. wait_for_running (line 391): first loop iteration - VMI creation has been - # requested by recover(), but the VMI may not yet be visible in the API. - # In this case, get_vmi/get_namespaced_custom_object can legitimately return 404. - # 2. wait_for_running: subsequent loop iterations - VMI exists and is running. - # Note: recover() itself does NOT call get_vmi before creating the VMI. It only - # checks if self.original_vmi exists, then directly calls create_namespaced_custom_object. - # All get_vmi calls (and thus the simulated 404 followed by a Running VMI) happen - # within wait_for_running *after* the create call has been issued. - self.custom_object_client.get_namespaced_custom_object = MagicMock( - side_effect=[ - ApiException(status=404, reason="Not Found"), # wait_for_running: VMI not visible yet after create - running_vmi, # wait_for_running: VMI now exists and is running - ] - ) - - # Mock the create API to return success - self.custom_object_client.create_namespaced_custom_object = MagicMock(return_value=running_vmi) - - # Run recovery with mocked time - with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): - result = self.plugin.recover("test-vm", "default", False) - - self.assertEqual(result, 0) - # Verify create was called with the right arguments - self.custom_object_client.create_namespaced_custom_object.assert_called_once() - - def test_validation_failure(self): - """ - Test validation failure when KubeVirt is not installed - """ - # Mock list_namespaces_by_regex - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - - # Mock list to return VMI for get_vmis - self.custom_object_client.list_namespaced_custom_object = MagicMock( - side_effect=[ - {"items": [self.mock_vmi]}, # For get_vmis - {"items": []}, # For validate_environment - empty result simulating no KubeVirt CRDs - ] - ) - - # Mock get_vmi to return VMI - self.custom_object_client.get_namespaced_custom_object = MagicMock( - return_value=self.mock_vmi - ) - - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) - - self.assertEqual(result, 1) - - def test_delete_vmi_timeout(self): - """ - Test timeout during VMI deletion - """ - # Store original VMI - self.plugin.original_vmi = self.mock_vmi - - # Initialize required attributes - self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") - self.plugin.pods_status = PodsStatus() - - # Mock successful delete operation - self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) - - # Mock get_vmi to always return the same VMI with unchanged creationTimestamp. - # This simulates a scenario where the VMI has NOT been recreated after deletion - # (i.e., still has the original creationTimestamp from before deletion). - # delete_vmi (lines 315-320) loops checking if creationTimestamp changed to detect - # successful recreation. Since it never changes here, the loop continues until - # the mocked time exceeds the timeout value, exercising the timeout path. - - self.custom_object_client.get_namespaced_custom_object = MagicMock( - return_value=self.mock_vmi - ) - - # Simulate timeout by making time.time return values that exceed the timeout - with patch('time.sleep'), patch('time.time', side_effect=[0, 10, 20, 130, 140]): - result = self.plugin.delete_vmi("test-vm", "default", False, timeout=120) - - self.assertEqual(result, 1) - self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( - group="kubevirt.io", - version="v1", - namespace="default", - plural="virtualmachineinstances", - name="test-vm" - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/ingress_network/test_ingress_network_plugin.py b/tests/test_ingress_network_plugin.py similarity index 100% rename from tests/ingress_network/test_ingress_network_plugin.py rename to tests/test_ingress_network_plugin.py diff --git a/tests/test_kubevirt_vm_outage.py b/tests/test_kubevirt_vm_outage.py index 4d884a14..0bc689b4 100644 --- a/tests/test_kubevirt_vm_outage.py +++ b/tests/test_kubevirt_vm_outage.py @@ -1,11 +1,27 @@ #!/usr/bin/env python3 """ -Test suite for KubeVirt VM Outage Scenario Plugin class +Test suite for KubeVirt VM Outage Scenario Plugin -Note: This test file uses mocks extensively to avoid needing actual Kubernetes/KubeVirt infrastructure. +This comprehensive test suite covers the KubevirtVmOutageScenarioPlugin class +using extensive mocks to avoid needing actual Kubernetes/KubeVirt infrastructure. + +Test Coverage: +- Core scenario flows: injection, recovery, deletion, waiting +- Edge cases: timeouts, missing parameters, validation failures +- API exceptions: 404, 500, and general exceptions +- Helper methods: get_vmi, get_vmis, patch_vm_spec, validate_environment +- Multiple VMI scenarios with kill_count +- Auto-restart disable functionality + +IMPORTANT: These tests use comprehensive mocking and do NOT require any Kubernetes +cluster or KubeVirt installation. All API calls are mocked. Usage: + # Run all tests + python -m unittest tests.test_kubevirt_vm_outage -v + + # Run with coverage python -m coverage run -a -m unittest tests/test_kubevirt_vm_outage.py -v Assisted By: Claude Code @@ -16,6 +32,7 @@ import itertools import os import tempfile import unittest +from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch import yaml @@ -27,8 +44,9 @@ from kubernetes.client.rest import ApiException from krkn.scenario_plugins.kubevirt_vm_outage.kubevirt_vm_outage_scenario_plugin import KubevirtVmOutageScenarioPlugin + class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): - + def setUp(self): """ Set up test fixtures for KubevirtVmOutageScenarioPlugin @@ -52,18 +70,32 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): crd_item.spec.group = "kubevirt.io" crd_list.items = [crd_item] self.k8s_client.list_custom_resource_definition.return_value = crd_list - - # Mock VMI data + + # Mock VMI data with timezone-aware timestamps + base_time = datetime.now(timezone.utc) self.mock_vmi = { "metadata": { "name": "test-vm", - "namespace": "default" + "namespace": "default", + "creationTimestamp": base_time.isoformat() + "Z" }, "status": { "phase": "Running" } } - + + # Mock VMI with new creation timestamp (after recreation) + self.mock_vmi_recreated = { + "metadata": { + "name": "test-vm", + "namespace": "default", + "creationTimestamp": (base_time + timedelta(minutes=1)).isoformat() + "Z" + }, + "status": { + "phase": "Running" + } + } + # Create test config self.config = { "scenarios": [ @@ -73,18 +105,18 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): "parameters": { "vm_name": "test-vm", "namespace": "default", - "duration": 0 + "duration": 0 } } ] } - + # Create a temporary config file temp_dir = tempfile.gettempdir() self.scenario_file = os.path.join(temp_dir, "test_kubevirt_scenario.yaml") with open(self.scenario_file, "w") as f: yaml.dump(self.config, f) - + # Mock dependencies self.telemetry = MagicMock(spec=KrknTelemetryOpenshift) self.scenario_telemetry = MagicMock(spec=ScenarioTelemetry) @@ -94,63 +126,152 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): self.delete_count = 0 self.wait_count = 0 + def create_incrementing_time_function(self): + """ + Create an incrementing time function that returns sequential float values. + Returns a callable that can be used with patch('time.time', side_effect=...) + """ + counter = itertools.count(1) + def mock_time(): + return float(next(counter)) + return mock_time + def mock_delete(self, *args, **kwargs): + """Reusable mock for delete_vmi that tracks calls and sets up affected_pod""" self.delete_count += 1 self.plugin.affected_pod = AffectedPod(pod_name=f"test-vm-{self.delete_count}", namespace="default") self.plugin.affected_pod.pod_rescheduling_time = 5.0 return 0 def mock_wait(self, *args, **kwargs): + """Reusable mock for wait_for_running that tracks calls and sets pod_readiness_time""" self.wait_count += 1 self.plugin.affected_pod.pod_readiness_time = 3.0 return 0 - + + # ==================== Core Scenario Tests ==================== + def test_successful_injection_and_recovery(self): """ - Test successful deletion and recovery of a VMI + Test successful deletion and recovery of a VMI using detailed mocking """ - # Populate vmis_list to avoid randrange error - self.plugin.vmis_list = [self.mock_vmi] + # Mock list_namespaces_by_regex to return a single namespace + self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - # Mock get_vmis to not clear the list - with patch.object(self.plugin, 'get_vmis'): - # Mock get_vmi to return our mock VMI - with patch.object(self.plugin, 'get_vmi', return_value=self.mock_vmi): - # Mock validate_environment to return True - with patch.object(self.plugin, 'validate_environment', return_value=True): - # Mock delete_vmi and wait_for_running to simulate success - with patch.object(self.plugin, 'delete_vmi', side_effect=self.mock_delete) as mock_delete: - with patch.object(self.plugin, 'wait_for_running', side_effect=self.mock_wait) as mock_wait: - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) + # Mock list_namespaced_custom_object to return our VMI + self.custom_object_client.list_namespaced_custom_object = MagicMock( + side_effect=[ + {"items": [self.mock_vmi]}, # For get_vmis + {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment + ] + ) + + # Mock get_namespaced_custom_object with a sequence that handles multiple calls + # Call sequence: + # 1. validate_environment: get original VMI + # 2. execute_scenario: get VMI before deletion + # 3. delete_vmi: loop checking if timestamp changed (returns recreated VMI on first check) + # 4+. wait_for_running: loop until phase is Running (may call multiple times) + get_vmi_responses = [ + self.mock_vmi, # Initial get in validate_environment + self.mock_vmi, # Get before delete + self.mock_vmi_recreated, # After delete (recreated with new timestamp) + self.mock_vmi_recreated, # Check if running + ] + + class GetVmiSideEffect: + """ + Callable helper that returns a predefined sequence of VMIs. + If called more times than there are responses, it fails the test + to surface unexpected additional calls instead of silently + masking them. + """ + def __init__(self, responses): + self._responses = responses + self._call_iter = itertools.count() + self.call_count = 0 + + def __call__(self, *args, **kwargs): + call_num = next(self._call_iter) + self.call_count = call_num + 1 + + if call_num >= len(self._responses): + raise AssertionError( + f"get_vmi_side_effect called more times ({call_num + 1}) " + f"than expected ({len(self._responses)})." + ) + return self._responses[call_num] + + get_vmi_side_effect = GetVmiSideEffect(get_vmi_responses) + self.custom_object_client.get_namespaced_custom_object = MagicMock( + side_effect=get_vmi_side_effect + ) + + # Mock delete operation + self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) + + with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): + with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): + result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) self.assertEqual(result, 0) - mock_delete.assert_called_once_with("test-vm", "default", False) - mock_wait.assert_called_once_with("test-vm", "default", 60) - + # Verify get_namespaced_custom_object was called exactly as many times as + # there are predefined responses + self.assertEqual( + self.custom_object_client.get_namespaced_custom_object.call_count, + len(get_vmi_responses), + ) + + # Verify that the VMI delete operation was performed once with expected parameters + self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( + group="kubevirt.io", + version="v1", + namespace="default", + plural="virtualmachineinstances", + name="test-vm", + ) + def test_injection_failure(self): """ Test failure during VMI deletion """ - # Populate vmis_list to avoid randrange error - self.plugin.vmis_list = [self.mock_vmi] + # Mock list_namespaces_by_regex + self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - # Mock get_vmis to not clear the list - with patch.object(self.plugin, 'get_vmis'): - # Mock get_vmi to return our mock VMI - with patch.object(self.plugin, 'get_vmi', return_value=self.mock_vmi): - # Mock validate_environment to return True - with patch.object(self.plugin, 'validate_environment', return_value=True): - # Mock delete_vmi to simulate failure - with patch.object(self.plugin, 'delete_vmi', return_value=1) as mock_delete: - with patch.object(self.plugin, 'wait_for_running', return_value=0) as mock_wait: - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) + # Mock list to return VMI + self.custom_object_client.list_namespaced_custom_object = MagicMock( + side_effect=[ + {"items": [self.mock_vmi]}, # For get_vmis + {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment + ] + ) + + # Mock get_vmi + self.custom_object_client.get_namespaced_custom_object = MagicMock( + side_effect=[ + self.mock_vmi, # validate_environment + self.mock_vmi, # get before delete + ] + ) + + # Mock delete to raise an error + self.custom_object_client.delete_namespaced_custom_object = MagicMock( + side_effect=ApiException(status=500, reason="Internal Server Error") + ) + + with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): + result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) self.assertEqual(result, 1) - mock_delete.assert_called_once_with("test-vm", "default", False) - mock_wait.assert_not_called() - + # Verify delete was attempted before the error occurred + self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( + group="kubevirt.io", + version="v1", + namespace="default", + plural="virtualmachineinstances", + name="test-vm" + ) + def test_disable_auto_restart(self): """ Test VM auto-restart can be disabled @@ -158,66 +279,92 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): # Configure test with disable_auto_restart=True self.config["scenarios"][0]["parameters"]["disable_auto_restart"] = True - # Populate vmis_list to avoid randrange error - self.plugin.vmis_list = [self.mock_vmi] + # Mock list_namespaces_by_regex + self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) - # Mock get_vmis to not clear the list - with patch.object(self.plugin, 'get_vmis'): - # Mock get_vmi to return our mock VMI - with patch.object(self.plugin, 'get_vmi', return_value=self.mock_vmi): - # Mock validate_environment to return True - with patch.object(self.plugin, 'validate_environment', return_value=True): - # Mock delete_vmi and wait_for_running - with patch.object(self.plugin, 'delete_vmi', side_effect=self.mock_delete) as mock_delete: - with patch.object(self.plugin, 'wait_for_running', side_effect=self.mock_wait) as mock_wait: - with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): - result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) + # Mock VM object for patching + mock_vm = { + "metadata": {"name": "test-vm", "namespace": "default"}, + "spec": {"running": True} + } + + # Mock list to return VMI + self.custom_object_client.list_namespaced_custom_object = MagicMock( + side_effect=[ + {"items": [self.mock_vmi]}, # For get_vmis + {"items": [{"metadata": {"name": "test-vm"}}]}, # For validate_environment + ] + ) + + # Mock get_namespaced_custom_object with detailed call sequence + # Call sequence: + # 1. execute_scenario: get VMI before deletion + # 2. patch_vm_spec: get VM for patching + # 3. delete_vmi: loop checking if VMI timestamp changed + # 4+. wait_for_running: loop until VMI phase is Running + self.custom_object_client.get_namespaced_custom_object = MagicMock( + side_effect=[ + self.mock_vmi, # Call 1: get VMI before delete + mock_vm, # Call 2: get VM for patching (different resource type) + self.mock_vmi_recreated, # Call 3: delete_vmi detects new timestamp + self.mock_vmi_recreated, # Call 4: wait_for_running checks phase + ] + ) + + # Mock patch and delete operations + self.custom_object_client.patch_namespaced_custom_object = MagicMock(return_value=mock_vm) + self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) + + with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): + with patch("builtins.open", unittest.mock.mock_open(read_data=yaml.dump(self.config))): + result = self.plugin.run("test-uuid", self.scenario_file, {}, self.telemetry, self.scenario_telemetry) self.assertEqual(result, 0) - # delete_vmi should be called with disable_auto_restart=True - mock_delete.assert_called_once_with("test-vm", "default", True) - mock_wait.assert_called_once_with("test-vm", "default", 60) - + # Verify patch was called to disable auto-restart + self.custom_object_client.patch_namespaced_custom_object.assert_called() + def test_recovery_when_vmi_does_not_exist(self): """ Test recovery logic when VMI does not exist after deletion """ - # Initialize the plugin's custom_object_client - self.plugin.custom_object_client = self.custom_object_client - # Store the original VMI in the plugin for recovery self.plugin.original_vmi = self.mock_vmi.copy() - # Create a cleaned vmi_dict as the plugin would - vmi_dict = self.mock_vmi.copy() + # Initialize affected_pod which is used by wait_for_running + self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") # Set up running VMI data for after recovery running_vmi = { - "metadata": {"name": "test-vm", "namespace": "default"}, + "metadata": { + "name": "test-vm", + "namespace": "default", + "creationTimestamp": (datetime.now(timezone.utc) + timedelta(minutes=2)).isoformat() + "Z" + }, "status": {"phase": "Running"} } - # Set up time.time to immediately exceed the timeout for auto-recovery - with patch('time.time', side_effect=[0, 301, 301, 301, 301, 310, 320]): - # Mock get_vmi to always return None (not auto-recovered) - with patch.object(self.plugin, 'get_vmi', side_effect=[None, None, running_vmi]): - # Mock the custom object API to return success - self.custom_object_client.create_namespaced_custom_object = MagicMock(return_value=running_vmi) + # Mock get_namespaced_custom_object call sequence triggered during recovery + # Call sequence: + # 1. wait_for_running: first loop iteration - VMI creation requested but not visible yet + # 2. wait_for_running: subsequent iterations - VMI exists and is running + self.custom_object_client.get_namespaced_custom_object = MagicMock( + side_effect=[ + ApiException(status=404, reason="Not Found"), # VMI not visible yet after create + running_vmi, # VMI now exists and is running + ] + ) - # Run recovery with mocked time.sleep - with patch('time.sleep'): - result = self.plugin.recover("test-vm", "default", False) + # Mock the create API to return success + self.custom_object_client.create_namespaced_custom_object = MagicMock(return_value=running_vmi) + + # Run recovery with mocked time + with patch('time.time', side_effect=self.create_incrementing_time_function()), patch('time.sleep'): + result = self.plugin.recover("test-vm", "default", False) self.assertEqual(result, 0) - # Verify create was called with the right arguments for our API version and kind - self.custom_object_client.create_namespaced_custom_object.assert_called_once_with( - group="kubevirt.io", - version="v1", - namespace="default", - plural="virtualmachineinstances", - body=vmi_dict - ) - + # Verify create was called with the right arguments + self.custom_object_client.create_namespaced_custom_object.assert_called_once() + def test_validation_failure(self): """ Test validation failure when KubeVirt is not installed @@ -236,34 +383,32 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): # When validation fails, run() returns 1 due to exception handling self.assertEqual(result, 1) - + + # ==================== Timeout Tests ==================== + def test_delete_vmi_timeout(self): """ Test timeout during VMI deletion """ - # Initialize the plugin's custom_object_client and required attributes - self.plugin.custom_object_client = self.custom_object_client + # Store original VMI + self.plugin.original_vmi = self.mock_vmi - # Initialize original_vmi which is required by delete_vmi - self.plugin.original_vmi = self.mock_vmi.copy() - self.plugin.original_vmi['metadata']['creationTimestamp'] = '2023-01-01T00:00:00Z' - - # Initialize pods_status which delete_vmi needs - from krkn_lib.models.k8s import PodsStatus, AffectedPod - self.plugin.pods_status = PodsStatus() + # Initialize required attributes self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") + self.plugin.pods_status = PodsStatus() # Mock successful delete operation self.custom_object_client.delete_namespaced_custom_object = MagicMock(return_value={}) - # Mock that get_vmi always returns VMI with same creationTimestamp (never gets recreated) - mock_vmi_with_time = self.mock_vmi.copy() - mock_vmi_with_time['metadata']['creationTimestamp'] = '2023-01-01T00:00:00Z' + # Mock get_vmi to always return the same VMI with unchanged creationTimestamp + # This simulates that the VMI has NOT been recreated after deletion + self.custom_object_client.get_namespaced_custom_object = MagicMock( + return_value=self.mock_vmi + ) - with patch.object(self.plugin, 'get_vmi', return_value=mock_vmi_with_time): - # Simulate timeout by making time.time return values that exceed the timeout - with patch('time.sleep'), patch('time.time', side_effect=[0, 10, 20, 130, 130, 130, 130, 140]): - result = self.plugin.delete_vmi("test-vm", "default", False) + # Simulate timeout by making time.time return values that exceed the timeout + with patch('time.sleep'), patch('time.time', side_effect=[0, 10, 20, 130, 140]): + result = self.plugin.delete_vmi("test-vm", "default", False, timeout=120) self.assertEqual(result, 1) self.custom_object_client.delete_namespaced_custom_object.assert_called_once_with( @@ -274,12 +419,29 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): name="test-vm" ) + def test_wait_for_running_timeout(self): + """ + Test wait_for_running times out when VMI doesn't reach Running state + """ + self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") + + # Mock VMI in Pending state + pending_vmi = self.mock_vmi.copy() + pending_vmi['status']['phase'] = 'Pending' + + with patch.object(self.plugin, 'get_vmi', return_value=pending_vmi): + with patch('time.sleep'): + with patch('time.time', side_effect=[0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 121]): + result = self.plugin.wait_for_running("test-vm", "default", 120) + + self.assertEqual(result, 1) + + # ==================== API Exception Tests ==================== def test_get_vmi_api_exception_non_404(self): """ Test get_vmi raises ApiException for non-404 errors """ - # Mock API exception with non-404 status api_error = ApiException(status=500, reason="Internal Server Error") self.custom_object_client.get_namespaced_custom_object = MagicMock(side_effect=api_error) @@ -299,37 +461,10 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): with self.assertRaises(Exception): self.plugin.get_vmi("test-vm", "default") - def test_get_vmis_with_regex_matching(self): - """ - Test get_vmis successfully filters VMIs by regex pattern - """ - # Mock namespace list - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default", "test-ns"]) - - # Mock VMI list with multiple VMIs - vmi_list = { - "items": [ - {"metadata": {"name": "test-vm-1"}, "status": {"phase": "Running"}}, - {"metadata": {"name": "test-vm-2"}, "status": {"phase": "Running"}}, - {"metadata": {"name": "other-vm"}, "status": {"phase": "Running"}}, - ] - } - self.custom_object_client.list_namespaced_custom_object = MagicMock(return_value=vmi_list) - - # Test with regex pattern that matches test-vm-* - self.plugin.get_vmis("test-vm-.*", "default") - - # Should have 4 VMs (2 per namespace * 2 namespaces) - self.assertEqual(len(self.plugin.vmis_list), 4) - # Verify only test-vm-* were added - for vmi in self.plugin.vmis_list: - self.assertTrue(vmi["metadata"]["name"].startswith("test-vm-")) - def test_get_vmis_api_exception_404(self): """ Test get_vmis handles 404 ApiException gracefully """ - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) api_error = ApiException(status=404, reason="Not Found") self.custom_object_client.list_namespaced_custom_object = MagicMock(side_effect=api_error) @@ -342,7 +477,6 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): """ Test get_vmis raises ApiException for non-404 errors """ - self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default"]) api_error = ApiException(status=500, reason="Internal Server Error") self.custom_object_client.list_namespaced_custom_object = MagicMock(side_effect=api_error) @@ -350,52 +484,10 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): with self.assertRaises(ApiException): self.plugin.get_vmis("test-vm", "default") - def test_patch_vm_spec_success(self): - """ - Test patch_vm_spec successfully patches VM - """ - mock_vm = { - "metadata": {"name": "test-vm", "namespace": "default"}, - "spec": {"running": True} - } - - self.custom_object_client.get_namespaced_custom_object = MagicMock(return_value=mock_vm) - self.custom_object_client.patch_namespaced_custom_object = MagicMock(return_value=mock_vm) - - result = self.plugin.patch_vm_spec("test-vm", "default", False) - - self.assertTrue(result) - self.custom_object_client.patch_namespaced_custom_object.assert_called_once() - - def test_patch_vm_spec_api_exception(self): - """ - Test patch_vm_spec handles ApiException - """ - - api_error = ApiException(status=404, reason="Not Found") - self.custom_object_client.get_namespaced_custom_object = MagicMock(side_effect=api_error) - - result = self.plugin.patch_vm_spec("test-vm", "default", False) - - self.assertFalse(result) - - def test_patch_vm_spec_general_exception(self): - """ - Test patch_vm_spec handles general exceptions - """ - self.custom_object_client.get_namespaced_custom_object = MagicMock( - side_effect=Exception("Connection error") - ) - - result = self.plugin.patch_vm_spec("test-vm", "default", False) - - self.assertFalse(result) - def test_delete_vmi_api_exception_404(self): """ Test delete_vmi handles 404 ApiException during deletion """ - # Initialize required attributes self.plugin.original_vmi = self.mock_vmi.copy() self.plugin.original_vmi['metadata']['creationTimestamp'] = '2023-01-01T00:00:00Z' @@ -426,6 +518,103 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): self.assertEqual(result, 1) + def test_patch_vm_spec_api_exception(self): + """ + Test patch_vm_spec handles ApiException + """ + api_error = ApiException(status=404, reason="Not Found") + self.custom_object_client.get_namespaced_custom_object = MagicMock(side_effect=api_error) + + result = self.plugin.patch_vm_spec("test-vm", "default", False) + + self.assertFalse(result) + + def test_patch_vm_spec_general_exception(self): + """ + Test patch_vm_spec handles general exceptions + """ + self.custom_object_client.get_namespaced_custom_object = MagicMock( + side_effect=Exception("Connection error") + ) + + result = self.plugin.patch_vm_spec("test-vm", "default", False) + + self.assertFalse(result) + + # ==================== Helper Method Tests ==================== + + def test_get_vmis_with_regex_matching(self): + """ + Test get_vmis successfully filters VMIs by regex pattern + """ + # Mock namespace list + self.k8s_client.list_namespaces_by_regex = MagicMock(return_value=["default", "test-ns"]) + + # Mock VMI list with multiple VMIs + vmi_list = { + "items": [ + {"metadata": {"name": "test-vm-1"}, "status": {"phase": "Running"}}, + {"metadata": {"name": "test-vm-2"}, "status": {"phase": "Running"}}, + {"metadata": {"name": "other-vm"}, "status": {"phase": "Running"}}, + ] + } + self.custom_object_client.list_namespaced_custom_object = MagicMock(return_value=vmi_list) + + # Test with regex pattern that matches test-vm-* + self.plugin.get_vmis("test-vm-.*", "default") + + # Should have 4 VMs (2 per namespace * 2 namespaces) + self.assertEqual(len(self.plugin.vmis_list), 4) + # Verify only test-vm-* were added + for vmi in self.plugin.vmis_list: + self.assertTrue(vmi["metadata"]["name"].startswith("test-vm-")) + + def test_patch_vm_spec_success(self): + """ + Test patch_vm_spec successfully patches VM + """ + mock_vm = { + "metadata": {"name": "test-vm", "namespace": "default"}, + "spec": {"running": True} + } + + self.custom_object_client.get_namespaced_custom_object = MagicMock(return_value=mock_vm) + self.custom_object_client.patch_namespaced_custom_object = MagicMock(return_value=mock_vm) + + result = self.plugin.patch_vm_spec("test-vm", "default", False) + + self.assertTrue(result) + self.custom_object_client.patch_namespaced_custom_object.assert_called_once() + + def test_validate_environment_exception(self): + """ + Test validate_environment handles exceptions + """ + self.custom_object_client.list_namespaced_custom_object = MagicMock( + side_effect=Exception("Connection error") + ) + + result = self.plugin.validate_environment("test-vm", "default") + + self.assertFalse(result) + + def test_validate_environment_vmi_not_found(self): + """ + Test validate_environment when VMI doesn't exist + """ + # Mock CRDs exist + mock_crd_list = MagicMock() + mock_crd_list.items = MagicMock(return_value=["item1"]) + self.custom_object_client.list_namespaced_custom_object = MagicMock(return_value=mock_crd_list) + + # Mock VMI not found + with patch.object(self.plugin, 'get_vmi', return_value=None): + result = self.plugin.validate_environment("test-vm", "default") + + self.assertFalse(result) + + # ==================== Delete VMI Tests ==================== + def test_delete_vmi_successful_recreation(self): """ Test delete_vmi succeeds when VMI is recreated with new creationTimestamp @@ -479,22 +668,7 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): # When VMI stays deleted (None), delete_vmi waits for recreation and times out self.assertEqual(result, 1) - def test_wait_for_running_timeout(self): - """ - Test wait_for_running times out when VMI doesn't reach Running state - """ - self.plugin.affected_pod = AffectedPod(pod_name="test-vm", namespace="default") - - # Mock VMI in Pending state - pending_vmi = self.mock_vmi.copy() - pending_vmi['status']['phase'] = 'Pending' - - with patch.object(self.plugin, 'get_vmi', return_value=pending_vmi): - with patch('time.sleep'): - with patch('time.time', side_effect=[0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 121]): - result = self.plugin.wait_for_running("test-vm", "default", 120) - - self.assertEqual(result, 1) + # ==================== Wait for Running Tests ==================== def test_wait_for_running_vmi_not_exists(self): """ @@ -508,13 +682,15 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): with patch.object(self.plugin, 'get_vmi', side_effect=[None, None, running_vmi]): with patch('time.sleep'): - # time.time() called: start_time (0), while loop iteration 1 (1), iteration 2 (2), iteration 3 (3), end_time (3) + # time.time() called: start_time (0), iteration 1 (1), iteration 2 (2), iteration 3 (3), end_time (3) with patch('time.time', side_effect=[0, 1, 2, 3, 3]): result = self.plugin.wait_for_running("test-vm", "default", 120) self.assertEqual(result, 0) self.assertIsNotNone(self.plugin.affected_pod.pod_readiness_time) + # ==================== Recovery Tests ==================== + def test_recover_no_original_vmi(self): """ Test recover fails when no original VMI is captured @@ -542,6 +718,8 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): self.assertEqual(result, 1) + # ==================== Execute Scenario Tests ==================== + def test_execute_scenario_missing_vm_name(self): """ Test execute_scenario fails when vm_name is missing @@ -641,38 +819,12 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase): # Should have unrecovered pod self.assertEqual(len(result.unrecovered), 1) - def test_validate_environment_exception(self): - """ - Test validate_environment handles exceptions - """ - self.custom_object_client.list_namespaced_custom_object = MagicMock( - side_effect=Exception("Connection error") - ) - - result = self.plugin.validate_environment("test-vm", "default") - - self.assertFalse(result) - - def test_validate_environment_vmi_not_found(self): - """ - Test validate_environment when VMI doesn't exist - """ - # Mock CRDs exist - mock_crd_list = MagicMock() - mock_crd_list.items = MagicMock(return_value=["item1"]) - self.custom_object_client.list_namespaced_custom_object = MagicMock(return_value=mock_crd_list) - - # Mock VMI not found - with patch.object(self.plugin, 'get_vmi', return_value=None): - result = self.plugin.validate_environment("test-vm", "default") - - self.assertFalse(result) + # ==================== Initialization Tests ==================== def test_init_clients(self): """ Test init_clients initializes k8s client correctly """ - mock_k8s = MagicMock(spec=KrknKubernetes) mock_custom_client = MagicMock() mock_k8s.custom_object_client = mock_custom_client