krkn/tests/test_managed_cluster_scenario_plugin.py

#!/usr/bin/env python3

"""
Test suite for ManagedClusterScenarioPlugin class

Usage:
    python -m coverage run -a -m unittest tests/test_managed_cluster_scenario_plugin.py -v

Assisted By: Claude Code
"""

import unittest
from unittest.mock import Mock, patch, call
from krkn_lib.k8s import KrknKubernetes
from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift

from krkn.scenario_plugins.managed_cluster.managed_cluster_scenario_plugin import ManagedClusterScenarioPlugin
from krkn.scenario_plugins.managed_cluster import common_functions


class TestManagedClusterScenarioPlugin(unittest.TestCase):

    def setUp(self):
        """
        Set up test fixtures for ManagedClusterScenarioPlugin
        """
        self.plugin = ManagedClusterScenarioPlugin()
        self.mock_kubecli = Mock(spec=KrknKubernetes)

    def test_get_scenario_types(self):
        """
        Test get_scenario_types returns correct scenario type
        """
        result = self.plugin.get_scenario_types()

        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):
    """
    Test suite for common_functions module
    """

    def setUp(self):
        """
        Set up test fixtures for common_functions tests
        """
        self.mock_kubecli = Mock(spec=KrknKubernetes)

    def test_get_managedcluster_with_specific_name_exists(self):
        """
        Test get_managedcluster returns the specified cluster when it exists
        """
        self.mock_kubecli.list_killable_managedclusters.return_value = ["cluster1", "cluster2", "cluster3"]

        result = common_functions.get_managedcluster(
            "cluster1", "", 1, self.mock_kubecli
        )

        self.assertEqual(result, ["cluster1"])
        self.mock_kubecli.list_killable_managedclusters.assert_called_once_with()

    def test_get_managedcluster_with_specific_name_not_exists(self):
        """
        Test get_managedcluster falls back to label selector when specified cluster doesn't exist
        """
        self.mock_kubecli.list_killable_managedclusters.side_effect = [
            ["cluster2", "cluster3"],
            ["cluster2", "cluster3"]
        ]

        result = common_functions.get_managedcluster(
            "cluster1", "env=test", 1, self.mock_kubecli
        )

        self.assertEqual(len(result), 1)
        self.assertIn(result[0], ["cluster2", "cluster3"])

    def test_get_managedcluster_with_label_selector(self):
        """
        Test get_managedcluster returns clusters matching label selector
        """
        self.mock_kubecli.list_killable_managedclusters.side_effect = [
            ["cluster1", "cluster2", "cluster3"],
            ["cluster1", "cluster2", "cluster3"],
        ]

        result = common_functions.get_managedcluster(
            "", "env=production", 2, self.mock_kubecli
        )

        self.assertEqual(len(result), 2)
        # Should be called once without and once with label_selector
        self.assertEqual(
            self.mock_kubecli.list_killable_managedclusters.call_count,
            2,
        )
        self.mock_kubecli.list_killable_managedclusters.assert_has_calls(
            [call(), call("env=production")]
        )

    def test_get_managedcluster_no_available_clusters(self):
        """
        Test get_managedcluster raises exception when no clusters are available
        """
        self.mock_kubecli.list_killable_managedclusters.return_value = []

        with self.assertRaises(Exception) as context:
            common_functions.get_managedcluster(
                "", "env=nonexistent", 1, self.mock_kubecli
            )

        self.assertIn("Available managedclusters with the provided label selector do not exist", str(context.exception))

    def test_get_managedcluster_kill_count_equals_available(self):
        """
        Test get_managedcluster returns all clusters when instance_kill_count equals available clusters
        """
        available_clusters = ["cluster1", "cluster2", "cluster3"]
        self.mock_kubecli.list_killable_managedclusters.return_value = available_clusters

        result = common_functions.get_managedcluster(
            "", "env=test", 3, self.mock_kubecli
        )

        self.assertEqual(result, available_clusters)
        self.assertEqual(len(result), 3)

    @patch('logging.info')
    def test_get_managedcluster_return_empty_when_count_is_zero(self, mock_logging):
        """
        Test get_managedcluster returns empty list when instance_kill_count is 0
        """
        available_clusters = ["cluster1", "cluster2", "cluster3"]
        self.mock_kubecli.list_killable_managedclusters.return_value = available_clusters

        result = common_functions.get_managedcluster(
            "", "env=test", 0, self.mock_kubecli
        )

        self.assertEqual(result, [])
        mock_logging.assert_called()

    @patch('random.randint')
    def test_get_managedcluster_random_selection(self, mock_randint):
        """
        Test get_managedcluster randomly selects the specified number of clusters
        """
        available_clusters = ["cluster1", "cluster2", "cluster3", "cluster4", "cluster5"]
        self.mock_kubecli.list_killable_managedclusters.return_value = available_clusters.copy()
        mock_randint.side_effect = [1, 0, 2]

        result = common_functions.get_managedcluster(
            "", "env=test", 3, self.mock_kubecli
        )

        self.assertEqual(len(result), 3)
        for cluster in result:
            self.assertIn(cluster, available_clusters)
        # Ensure no duplicates
        self.assertEqual(len(result), len(set(result)))

    @patch('logging.info')
    def test_get_managedcluster_logs_available_clusters(self, mock_logging):
        """
        Test get_managedcluster logs available clusters with label selector
        """
        available_clusters = ["cluster1", "cluster2"]
        self.mock_kubecli.list_killable_managedclusters.return_value = available_clusters

        common_functions.get_managedcluster(
            "", "env=test", 1, self.mock_kubecli
        )

        mock_logging.assert_called()
        call_args = str(mock_logging.call_args)
        self.assertIn("Available managedclusters with the label selector", call_args)

    @patch('logging.info')
    def test_get_managedcluster_logs_when_name_not_found(self, mock_logging):
        """
        Test get_managedcluster logs when specified cluster name doesn't exist
        """
        self.mock_kubecli.list_killable_managedclusters.side_effect = [
            ["cluster2"],
            ["cluster2"]
        ]

        common_functions.get_managedcluster(
            "nonexistent-cluster", "env=test", 1, self.mock_kubecli
        )
        # Check that logging was called multiple times (including the info message about unavailable cluster)
        self.assertGreaterEqual(mock_logging.call_count, 1)  
        # Check all calls for the expected message
        all_calls = [str(call) for call in mock_logging.call_args_list]
        found_message = any("managedcluster with provided managedcluster_name does not exist" in call
                           for call in all_calls)
        self.assertTrue(found_message)

    def test_wait_for_available_status(self):
        """
        Test wait_for_available_status calls watch_managedcluster_status with correct parameters
        """
        common_functions.wait_for_available_status(
            "test-cluster", 300, self.mock_kubecli
        )

        self.mock_kubecli.watch_managedcluster_status.assert_called_once_with(
            "test-cluster", "True", 300
        )

    def test_wait_for_unavailable_status(self):
        """
        Test wait_for_unavailable_status calls watch_managedcluster_status with correct parameters
        """
        common_functions.wait_for_unavailable_status(
            "test-cluster", 300, self.mock_kubecli
        )

        self.mock_kubecli.watch_managedcluster_status.assert_called_once_with(
            "test-cluster", "Unknown", 300
        )


if __name__ == "__main__":
    unittest.main()