krkn/tests/test_node_network_chaos.py

#!/usr/bin/env python3

"""
Test suite for NodeNetworkChaosModule class

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

Assisted By: Claude Code
"""

import unittest
import queue
from unittest.mock import MagicMock, patch, call

from krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos import (
    NodeNetworkChaosModule,
)
from krkn.scenario_plugins.network_chaos_ng.models import (
    NetworkChaosConfig,
    NetworkChaosScenarioType,
)


class TestNodeNetworkChaosModule(unittest.TestCase):

    def setUp(self):
        """
        Set up test fixtures for NodeNetworkChaosModule
        """
        self.mock_kubecli = MagicMock()
        self.mock_kubernetes = MagicMock()
        self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes

        self.config = NetworkChaosConfig(
            id="test-node-network-chaos",
            image="test-image",
            wait_duration=1,
            test_duration=30,
            label_selector="",
            service_account="",
            taints=[],
            namespace="default",
            instance_count=1,
            target="worker-1",
            execution="parallel",
            interfaces=["eth0"],
            ingress=True,
            egress=True,
            latency="100ms",
            loss="10",
            bandwidth="100mbit",
            force=False,
        )

        self.module = NodeNetworkChaosModule(self.config, self.mock_kubecli)

    def test_initialization(self):
        """
        Test NodeNetworkChaosModule initialization
        """
        self.assertEqual(self.module.config, self.config)
        self.assertEqual(self.module.kubecli, self.mock_kubecli)
        self.assertEqual(self.module.base_network_config, self.config)

    def test_get_config(self):
        """
        Test get_config returns correct scenario type and config
        """
        scenario_type, config = self.module.get_config()

        self.assertEqual(scenario_type, NetworkChaosScenarioType.Node)
        self.assertEqual(config, self.config)

    def test_get_targets_with_target_name(self):
        """
        Test get_targets with specific node target name
        """
        self.config.label_selector = ""
        self.config.target = "worker-1"
        self.mock_kubernetes.list_nodes.return_value = ["worker-1", "worker-2"]

        targets = self.module.get_targets()

        self.assertEqual(targets, ["worker-1"])

    def test_get_targets_with_label_selector(self):
        """
        Test get_targets with label selector
        """
        self.config.label_selector = "node-role.kubernetes.io/worker="
        self.mock_kubernetes.list_nodes.return_value = ["worker-1", "worker-2"]

        targets = self.module.get_targets()

        self.assertEqual(targets, ["worker-1", "worker-2"])
        self.mock_kubernetes.list_nodes.assert_called_once_with(
            "node-role.kubernetes.io/worker="
        )

    def test_get_targets_node_not_found(self):
        """
        Test get_targets raises exception when node doesn't exist
        """
        self.config.label_selector = ""
        self.config.target = "non-existent-node"
        self.mock_kubernetes.list_nodes.return_value = ["worker-1", "worker-2"]

        with self.assertRaises(Exception) as context:
            self.module.get_targets()

        self.assertIn("not found", str(context.exception))

    def test_get_targets_no_target_or_selector(self):
        """
        Test get_targets raises exception when neither target nor selector specified
        """
        self.config.label_selector = ""
        self.config.target = ""

        with self.assertRaises(Exception) as context:
            self.module.get_targets()

        self.assertIn("neither", str(context.exception))

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.time.sleep")
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_delete_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_set_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_success(
        self,
        mock_log_info,
        mock_setup,
        mock_set_rules,
        mock_delete_rules,
        mock_sleep,
        mock_qdisc_is_simple,
    ):
        """
        Test successful run of node network chaos
        """
        # Mock setup returns container_ids and interfaces
        mock_setup.return_value = (["container-123"], ["eth0"])

        # Mock qdisc check - simple qdisc
        mock_qdisc_is_simple.return_value = True

        self.module.run("worker-1")

        # Verify setup was called with node name
        mock_setup.assert_called_once()
        setup_args = mock_setup.call_args[0]
        # Node name should be passed as target and is_node=True (8th arg, index 7)
        self.assertEqual(setup_args[7], True)  # is_node flag

        # Verify qdisc was checked
        mock_qdisc_is_simple.assert_called_once()

        # Verify tc rules were set (with pids=None for node scenario)
        mock_set_rules.assert_called_once()
        set_call_args = mock_set_rules.call_args
        # pids should be None (last argument)
        self.assertIsNone(set_call_args[0][-1])

        # Verify sleep for test duration
        mock_sleep.assert_called_once_with(30)

        # Verify tc rules were deleted
        mock_delete_rules.assert_called_once()
        delete_call_args = mock_delete_rules.call_args
        # pids should be None (7th argument, index 6)
        self.assertIsNone(delete_call_args[0][6])

        # Verify cleanup pod was deleted
        self.assertEqual(self.mock_kubernetes.delete_pod.call_count, 1)

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_error")
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_no_interfaces_detected(
        self, mock_log_info, mock_log_error, mock_setup, mock_qdisc_is_simple
    ):
        """
        Test run handles case when no network interfaces detected
        """
        # Mock setup returns empty interfaces
        mock_setup.return_value = (["container-123"], [])

        # Set config to auto-detect interfaces
        self.config.interfaces = []

        self.module.run("worker-1")

        # Verify error was logged
        mock_log_error.assert_called()
        self.assertIn("no network interface", str(mock_log_error.call_args))

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_warning"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_complex_qdisc_without_force(
        self, mock_log_info, mock_log_warning, mock_setup, mock_qdisc_is_simple
    ):
        """
        Test run skips chaos when complex qdisc exists and force=False
        """
        # Mock setup
        mock_setup.return_value = (["container-123"], ["eth0"])

        # Mock qdisc check - complex qdisc
        mock_qdisc_is_simple.return_value = False

        # force is False
        self.config.force = False

        self.module.run("worker-1")

        # Verify warning was logged
        mock_log_warning.assert_called()
        self.assertIn("already has tc rules", str(mock_log_warning.call_args))

        # Verify cleanup pod was still deleted
        self.assertEqual(self.mock_kubernetes.delete_pod.call_count, 1)

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.time.sleep")
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_delete_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_set_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_warning"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_complex_qdisc_with_force(
        self,
        mock_log_info,
        mock_log_warning,
        mock_setup,
        mock_set_rules,
        mock_delete_rules,
        mock_sleep,
        mock_qdisc_is_simple,
    ):
        """
        Test run proceeds with chaos when complex qdisc exists and force=True
        """
        # Mock setup
        mock_setup.return_value = (["container-123"], ["eth0"])

        # Mock qdisc check - complex qdisc
        mock_qdisc_is_simple.return_value = False

        # force is True
        self.config.force = True

        self.module.run("worker-1")

        # Verify warning was logged about forcing
        mock_log_warning.assert_called()
        self.assertIn("forcing", str(mock_log_warning.call_args))

        # Verify sleep for safety warning (10 seconds)
        sleep_calls = [call[0][0] for call in mock_sleep.call_args_list]
        self.assertIn(10, sleep_calls)

        # Verify tc rules were set
        mock_set_rules.assert_called_once()

        # Verify tc rules were deleted
        mock_delete_rules.assert_called_once()

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.time.sleep")
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_delete_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_set_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_uses_configured_interfaces(
        self,
        mock_log_info,
        mock_setup,
        mock_set_rules,
        mock_delete_rules,
        mock_sleep,
        mock_qdisc_is_simple,
    ):
        """
        Test run uses configured interfaces instead of detected ones
        """
        # Mock setup returns different interfaces
        mock_setup.return_value = (["container-123"], ["eth0", "eth1"])

        # Mock qdisc check - simple qdisc
        mock_qdisc_is_simple.return_value = True

        # Set specific interfaces in config
        self.config.interfaces = ["eth2"]

        self.module.run("worker-1")

        # Verify set_rules was called with configured interfaces
        call_args = mock_set_rules.call_args
        # interfaces is the 3rd positional argument (index 2)
        self.assertEqual(call_args[0][2], ["eth2"])

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_with_error_queue(
        self, mock_log_info, mock_setup, mock_qdisc_is_simple
    ):
        """
        Test run with error_queue for parallel execution
        """
        # Mock setup to raise exception
        mock_setup.side_effect = Exception("Test error")

        error_queue = queue.Queue()
        self.module.run("worker-1", error_queue)

        # Verify error was put in queue instead of raising
        self.assertFalse(error_queue.empty())
        error = error_queue.get()
        self.assertEqual(error, "Test error")

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.time.sleep")
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_delete_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_set_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_ingress_egress_flags(
        self,
        mock_log_info,
        mock_setup,
        mock_set_rules,
        mock_delete_rules,
        mock_sleep,
        mock_qdisc_is_simple,
    ):
        """
        Test run passes ingress and egress flags correctly
        """
        # Mock setup
        mock_setup.return_value = (["container-123"], ["eth0"])

        # Mock qdisc check
        mock_qdisc_is_simple.return_value = True

        # Set specific ingress/egress config
        self.config.ingress = False
        self.config.egress = True

        self.module.run("worker-1")

        # Verify set_rules was called with correct egress/ingress flags
        set_call_args = mock_set_rules.call_args
        # egress is 1st arg (index 0), ingress is 2nd arg (index 1)
        self.assertEqual(set_call_args[0][0], True)  # egress
        self.assertEqual(set_call_args[0][1], False)  # ingress

        # Verify delete_rules was called with correct flags
        delete_call_args = mock_delete_rules.call_args
        self.assertEqual(delete_call_args[0][0], True)  # egress
        self.assertEqual(delete_call_args[0][1], False)  # ingress

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_warning"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_mixed_simple_and_complex_qdisc(
        self, mock_log_info, mock_log_warning, mock_setup, mock_qdisc_is_simple
    ):
        """
        Test run with multiple interfaces where some have complex qdisc
        """
        # Mock setup with multiple interfaces
        mock_setup.return_value = (["container-123"], ["eth0", "eth1"])

        # Set config to use detected interfaces
        self.config.interfaces = []
        self.config.force = False

        # Mock qdisc check - eth0 simple, eth1 complex
        mock_qdisc_is_simple.side_effect = [True, False]

        self.module.run("worker-1")

        # Verify warning about complex qdisc on eth1
        mock_log_warning.assert_called()
        warning_message = str(mock_log_warning.call_args)
        self.assertIn("already has tc rules", warning_message)

    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.node_qdisc_is_simple"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.time.sleep")
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_delete_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.common_set_limit_rules"
    )
    @patch(
        "krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.setup_network_chaos_ng_scenario"
    )
    @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_network_chaos.log_info")
    def test_run_checks_qdisc_for_all_interfaces(
        self,
        mock_log_info,
        mock_setup,
        mock_set_rules,
        mock_delete_rules,
        mock_sleep,
        mock_qdisc_is_simple,
    ):
        """
        Test run checks qdisc for all interfaces
        """
        # Mock setup with multiple interfaces
        mock_setup.return_value = (["container-123"], ["eth0", "eth1", "eth2"])

        # Set config to use detected interfaces
        self.config.interfaces = []

        # All interfaces simple
        mock_qdisc_is_simple.return_value = True

        self.module.run("worker-1")

        # Verify qdisc was checked for all 3 interfaces
        self.assertEqual(mock_qdisc_is_simple.call_count, 3)


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