#!/usr/bin/env python3 """ Test suite for ingress network shaping plugin This test suite covers the ingress_shaping module functions using mocks to avoid needing actual Kubernetes infrastructure. Test Coverage: - Serialization of config and output objects - Interface discovery: get_default_interface, verify_interface, get_node_interfaces - Virtual interface management: create_ifb, delete_ifb, create_virtual_interfaces, delete_virtual_interfaces - Job management: get_job_pods, wait_for_job, delete_jobs - Traffic shaping command generation: get_ingress_cmd - Filter application: apply_ingress_filter - End-to-end network_chaos flows: parallel, serial, error cases IMPORTANT: These tests use mocking and do NOT require any Kubernetes cluster. All API calls are mocked via unittest.mock. Usage: python -m unittest tests/test_ingress_network_plugin.py -v # Run with coverage python -m coverage run -a -m unittest tests/test_ingress_network_plugin.py -v Assisted By: Claude Code """ import unittest from unittest.mock import Mock, patch from arcaflow_plugin_sdk import plugin from krkn.scenario_plugins.native.network import ingress_shaping class NetworkScenariosTest(unittest.TestCase): def test_serialization(self): """Test serialization of configuration and output objects""" plugin.test_object_serialization( ingress_shaping.NetworkScenarioConfig( node_interface_name={"foo": ["bar"]}, network_params={ "latency": "50ms", "loss": "0.02", "bandwidth": "100mbit", }, ), self.fail, ) plugin.test_object_serialization( ingress_shaping.NetworkScenarioSuccessOutput( filter_direction="ingress", test_interfaces={"foo": ["bar"]}, network_parameters={ "latency": "50ms", "loss": "0.02", "bandwidth": "100mbit", }, execution_type="parallel", ), self.fail, ) plugin.test_object_serialization( ingress_shaping.NetworkScenarioErrorOutput( error="Hello World", ), self.fail, ) def test_get_default_interface(self): """Test getting default interface from a node""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml_content" mock_kubecli.create_pod.return_value = None mock_kubecli.exec_cmd_in_pod.return_value = ( "default via 192.168.1.1 dev eth0 proto dhcp metric 100\n" "172.17.0.0/16 dev docker0 proto kernel scope link src 172.17.0.1" ) mock_kubecli.delete_pod.return_value = None result = ingress_shaping.get_default_interface( node="test-node", pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(result, ["eth0"]) mock_kubecli.create_pod.assert_called_once() mock_kubecli.exec_cmd_in_pod.assert_called_once() mock_kubecli.delete_pod.assert_called_once() def test_verify_interface_with_empty_list(self): """Test verifying interface when input list is empty""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml_content" mock_kubecli.create_pod.return_value = None mock_kubecli.exec_cmd_in_pod.return_value = ( "default via 192.168.1.1 dev eth0 proto dhcp metric 100\n" ) mock_kubecli.delete_pod.return_value = None result = ingress_shaping.verify_interface( input_interface_list=[], node="test-node", pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(result, ["eth0"]) def test_verify_interface_with_valid_interfaces(self): """Test verifying interface with valid interface list""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml_content" mock_kubecli.create_pod.return_value = None mock_kubecli.exec_cmd_in_pod.return_value = ( "eth0 UP 192.168.1.10/24\n" "eth1 UP 10.0.0.5/24\n" "lo UNKNOWN 127.0.0.1/8\n" ) mock_kubecli.delete_pod.return_value = None result = ingress_shaping.verify_interface( input_interface_list=["eth0", "eth1"], node="test-node", pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(result, ["eth0", "eth1"]) def test_verify_interface_with_invalid_interface(self): """Test verifying interface with an interface that doesn't exist""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml_content" mock_kubecli.create_pod.return_value = None mock_kubecli.exec_cmd_in_pod.return_value = ( "eth0 UP 192.168.1.10/24\n" "lo UNKNOWN 127.0.0.1/8\n" ) mock_kubecli.delete_pod.return_value = None with self.assertRaises(Exception) as context: ingress_shaping.verify_interface( input_interface_list=["eth0", "eth99"], node="test-node", pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertIn("Interface eth99 not found", str(context.exception)) @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_default_interface') def test_get_node_interfaces_with_label_selector(self, mock_get_default_interface): """Test getting node interfaces using label selector""" mock_kubecli = Mock() mock_pod_template = Mock() mock_kubecli.get_node.return_value = ["node1", "node2"] mock_get_default_interface.return_value = ["eth0"] result = ingress_shaping.get_node_interfaces( node_interface_dict=None, label_selector="node-role.kubernetes.io/worker", instance_count=2, pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(result, {"node1": ["eth0"], "node2": ["eth0"]}) self.assertEqual(mock_get_default_interface.call_count, 2) @patch('krkn.scenario_plugins.native.network.ingress_shaping.verify_interface') def test_get_node_interfaces_with_node_dict(self, mock_verify_interface): """Test getting node interfaces with provided node interface dictionary""" mock_kubecli = Mock() mock_pod_template = Mock() mock_kubecli.get_node.return_value = ["node1"] mock_verify_interface.return_value = ["eth0", "eth1"] result = ingress_shaping.get_node_interfaces( node_interface_dict={"node1": ["eth0", "eth1"]}, label_selector=None, instance_count=1, pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(result, {"node1": ["eth0", "eth1"]}) mock_verify_interface.assert_called_once() def test_get_node_interfaces_no_selector_no_dict(self): """Test that exception is raised when both node dict and label selector are missing""" mock_kubecli = Mock() mock_pod_template = Mock() with self.assertRaises(Exception) as context: ingress_shaping.get_node_interfaces( node_interface_dict=None, label_selector=None, instance_count=1, pod_template=mock_pod_template, kubecli=mock_kubecli, image="quay.io/krkn-chaos/krkn:tools" ) self.assertIn("label selector must be provided", str(context.exception)) def test_create_ifb(self): """Test creating virtual interfaces""" mock_kubecli = Mock() mock_kubecli.exec_cmd_in_pod.return_value = None ingress_shaping.create_ifb(kubecli=mock_kubecli, number=2, pod_name="test-pod") # Should call modprobe once and ip link set for each interface self.assertEqual(mock_kubecli.exec_cmd_in_pod.call_count, 3) # Verify modprobe call — cmd is now the first positional arg first_call = mock_kubecli.exec_cmd_in_pod.call_args_list[0] self.assertIn("modprobe", first_call[0][0]) self.assertIn("numifbs=2", first_call[0][0]) def test_delete_ifb(self): """Test deleting virtual interfaces""" mock_kubecli = Mock() mock_kubecli.exec_cmd_in_pod.return_value = None ingress_shaping.delete_ifb(kubecli=mock_kubecli, pod_name="test-pod") mock_kubecli.exec_cmd_in_pod.assert_called_once() # cmd is now the first positional arg call_args = mock_kubecli.exec_cmd_in_pod.call_args[0][0] self.assertIn("modprobe", call_args) self.assertIn("-r", call_args) self.assertIn("ifb", call_args) def test_get_job_pods(self): """Test getting pods associated with a job""" mock_kubecli = Mock() mock_api_response = Mock() mock_api_response.metadata.labels = {"controller-uid": "test-uid-123"} mock_kubecli.list_pods.return_value = ["pod1", "pod2"] result = ingress_shaping.get_job_pods(kubecli=mock_kubecli, api_response=mock_api_response) self.assertEqual(result, "pod1") mock_kubecli.list_pods.assert_called_once_with( label_selector="controller-uid=test-uid-123", namespace="default" ) @patch('time.sleep', return_value=None) @patch('time.time') def test_wait_for_job_success(self, mock_time, mock_sleep): """Test waiting for jobs to complete successfully""" mock_kubecli = Mock() mock_time.side_effect = [0, 10, 20] mock_response1 = Mock() mock_response1.status.succeeded = 1 mock_response1.status.failed = None mock_response2 = Mock() mock_response2.status.succeeded = 1 mock_response2.status.failed = None mock_kubecli.get_job_status.side_effect = [mock_response1, mock_response2] ingress_shaping.wait_for_job( kubecli=mock_kubecli, job_list=["job1", "job2"], timeout=300 ) self.assertEqual(mock_kubecli.get_job_status.call_count, 2) @patch('time.sleep', return_value=None) @patch('time.time') def test_wait_for_job_timeout(self, mock_time, mock_sleep): """Test waiting for jobs times out""" mock_kubecli = Mock() mock_time.side_effect = [0, 350] mock_response = Mock() mock_response.status.succeeded = None mock_response.status.failed = None mock_kubecli.get_job_status.return_value = mock_response with self.assertRaises(Exception) as context: ingress_shaping.wait_for_job( kubecli=mock_kubecli, job_list=["job1"], timeout=300 ) self.assertIn("timeout", str(context.exception)) def test_delete_jobs(self): """Test deleting jobs""" mock_kubecli = Mock() mock_response = Mock() mock_response.status.failed = None mock_kubecli.get_job_status.return_value = mock_response mock_kubecli.delete_job.return_value = None ingress_shaping.delete_jobs( kubecli=mock_kubecli, job_list=["job1", "job2"] ) self.assertEqual(mock_kubecli.get_job_status.call_count, 2) self.assertEqual(mock_kubecli.delete_job.call_count, 2) @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_job_pods') def test_delete_jobs_with_failed_job(self, mock_get_job_pods): """Test deleting jobs when one has failed""" mock_kubecli = Mock() mock_response = Mock() mock_response.status.failed = 1 mock_pod_status = Mock() mock_pod_status.status.container_statuses = [] mock_log_response = Mock() mock_log_response.data.decode.return_value = "Error log content" mock_kubecli.get_job_status.return_value = mock_response mock_get_job_pods.return_value = "failed-pod" mock_kubecli.read_pod.return_value = mock_pod_status mock_kubecli.get_pod_log.return_value = mock_log_response mock_kubecli.delete_job.return_value = None ingress_shaping.delete_jobs( kubecli=mock_kubecli, job_list=["failed-job"] ) mock_kubecli.read_pod.assert_called_once() mock_kubecli.get_pod_log.assert_called_once() def test_get_ingress_cmd_basic(self): """Test generating ingress traffic shaping commands""" result = ingress_shaping.get_ingress_cmd( interface_list=["eth0"], network_parameters={"latency": "50ms"}, duration=120 ) self.assertIn("tc qdisc add dev eth0 handle ffff: ingress", result) self.assertIn("tc filter add dev eth0", result) self.assertIn("ifb0", result) self.assertIn("delay 50ms", result) self.assertIn("sleep 120", result) self.assertIn("tc qdisc del", result) def test_get_ingress_cmd_multiple_interfaces(self): """Test generating commands for multiple interfaces""" result = ingress_shaping.get_ingress_cmd( interface_list=["eth0", "eth1"], network_parameters={"latency": "50ms", "bandwidth": "100mbit"}, duration=120 ) self.assertIn("eth0", result) self.assertIn("eth1", result) self.assertIn("ifb0", result) self.assertIn("ifb1", result) self.assertIn("delay 50ms", result) self.assertIn("rate 100mbit", result) def test_get_ingress_cmd_all_parameters(self): """Test generating commands with all network parameters""" result = ingress_shaping.get_ingress_cmd( interface_list=["eth0"], network_parameters={ "latency": "50ms", "loss": "0.02", "bandwidth": "100mbit" }, duration=120 ) self.assertIn("delay 50ms", result) self.assertIn("loss 0.02", result) self.assertIn("rate 100mbit", result) def test_get_ingress_cmd_invalid_interface(self): """Test that invalid interface names raise an exception""" with self.assertRaises(Exception) as context: ingress_shaping.get_ingress_cmd( interface_list=["eth0; rm -rf /"], network_parameters={"latency": "50ms"}, duration=120 ) self.assertIn("does not match the required regex pattern", str(context.exception)) @patch('yaml.safe_load') @patch('krkn.scenario_plugins.native.network.ingress_shaping.create_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_ingress_cmd') def test_apply_ingress_filter(self, mock_get_cmd, mock_create_virtual, mock_yaml): """Test applying ingress filters to a node""" mock_kubecli = Mock() mock_pod_template = Mock() mock_job_template = Mock() mock_job_template.render.return_value = "job_yaml" mock_cfg = ingress_shaping.NetworkScenarioConfig( node_interface_name={"node1": ["eth0"]}, network_params={"latency": "50ms"}, test_duration=120 ) mock_yaml.return_value = {"metadata": {"name": "test-job"}} mock_get_cmd.return_value = "tc commands" mock_kubecli.create_job.return_value = Mock() result = ingress_shaping.apply_ingress_filter( cfg=mock_cfg, interface_list=["eth0"], node="node1", pod_template=mock_pod_template, job_template=mock_job_template, kubecli=mock_kubecli, create_interfaces=True, param_selector="all", image="quay.io/krkn-chaos/krkn:tools" ) mock_create_virtual.assert_called_once() mock_get_cmd.assert_called_once() mock_kubecli.create_job.assert_called_once() self.assertEqual(result, "test-job") def test_create_virtual_interfaces(self): """Test creating virtual interfaces on a node""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml" mock_kubecli.create_pod.return_value = None mock_kubecli.exec_cmd_in_pod.return_value = None mock_kubecli.delete_pod.return_value = None ingress_shaping.create_virtual_interfaces( kubecli=mock_kubecli, interface_list=["eth0", "eth1"], node="test-node", pod_template=mock_pod_template, image="quay.io/krkn-chaos/krkn:tools" ) mock_kubecli.create_pod.assert_called_once() mock_kubecli.delete_pod.assert_called_once() @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_ifb') def test_delete_virtual_interfaces(self, mock_delete_ifb): """Test deleting virtual interfaces from nodes""" mock_kubecli = Mock() mock_pod_template = Mock() mock_pod_template.render.return_value = "pod_yaml" mock_kubecli.create_pod.return_value = None mock_kubecli.delete_pod.return_value = None ingress_shaping.delete_virtual_interfaces( kubecli=mock_kubecli, node_list=["node1", "node2"], pod_template=mock_pod_template, image="quay.io/krkn-chaos/krkn:tools" ) self.assertEqual(mock_kubecli.create_pod.call_count, 2) self.assertEqual(mock_delete_ifb.call_count, 2) self.assertEqual(mock_kubecli.delete_pod.call_count, 2) @patch('krkn.scenario_plugins.native.network.ingress_shaping.Environment') @patch('krkn.scenario_plugins.native.network.ingress_shaping.FileSystemLoader') @patch('yaml.safe_load') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_jobs') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.wait_for_job') @patch('krkn.scenario_plugins.native.network.ingress_shaping.apply_ingress_filter') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_node_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.KrknKubernetes') def test_network_chaos_parallel_execution( self, mock_kubecli_class, mock_get_nodes, mock_apply_filter, mock_wait_job, mock_delete_virtual, mock_delete_jobs, mock_yaml, mock_file_loader, mock_env ): """Test network chaos with parallel execution""" mock_kubecli_class.return_value = Mock() mock_yaml.return_value = {"metadata": {"name": "test-pod"}} mock_get_nodes.return_value = {"node1": ["eth0"], "node2": ["eth1"]} mock_apply_filter.side_effect = ["job1", "job2"] cfg = ingress_shaping.NetworkScenarioConfig( label_selector="node-role.kubernetes.io/worker", instance_count=2, network_params={"latency": "50ms"}, execution_type="parallel", test_duration=120, wait_duration=30 ) output_id, output_data = ingress_shaping.network_chaos(params=cfg, run_id="test-run") self.assertEqual(output_id, "success") self.assertEqual(output_data.filter_direction, "ingress") self.assertEqual(output_data.execution_type, "parallel") self.assertEqual(mock_apply_filter.call_count, 2) mock_wait_job.assert_called_once() mock_delete_virtual.assert_called_once() @patch('krkn.scenario_plugins.native.network.ingress_shaping.Environment') @patch('krkn.scenario_plugins.native.network.ingress_shaping.FileSystemLoader') @patch('yaml.safe_load') @patch('time.sleep', return_value=None) @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_jobs') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.wait_for_job') @patch('krkn.scenario_plugins.native.network.ingress_shaping.apply_ingress_filter') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_node_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.KrknKubernetes') def test_network_chaos_serial_execution( self, mock_kubecli_class, mock_get_nodes, mock_apply_filter, mock_wait_job, mock_delete_virtual, mock_delete_jobs, mock_sleep, mock_yaml, mock_file_loader, mock_env ): """Test network chaos with serial execution""" mock_kubecli_class.return_value = Mock() mock_yaml.return_value = {"metadata": {"name": "test-pod"}} mock_get_nodes.return_value = {"node1": ["eth0"]} mock_apply_filter.return_value = "job1" cfg = ingress_shaping.NetworkScenarioConfig( label_selector="node-role.kubernetes.io/worker", instance_count=1, network_params={"latency": "50ms", "bandwidth": "100mbit"}, execution_type="serial", test_duration=120, wait_duration=30 ) output_id, output_data = ingress_shaping.network_chaos(params=cfg, run_id="test-run") self.assertEqual(output_id, "success") self.assertEqual(output_data.execution_type, "serial") self.assertEqual(mock_apply_filter.call_count, 2) self.assertEqual(mock_wait_job.call_count, 2) @patch('krkn.scenario_plugins.native.network.ingress_shaping.Environment') @patch('krkn.scenario_plugins.native.network.ingress_shaping.FileSystemLoader') @patch('yaml.safe_load') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_jobs') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_node_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.KrknKubernetes') def test_network_chaos_invalid_execution_type( self, mock_kubecli_class, mock_get_nodes, mock_delete_virtual, mock_delete_jobs, mock_yaml, mock_file_loader, mock_env ): """Test network chaos with invalid execution type""" mock_kubecli_class.return_value = Mock() mock_yaml.return_value = {"metadata": {"name": "test-pod"}} mock_get_nodes.return_value = {"node1": ["eth0"]} cfg = ingress_shaping.NetworkScenarioConfig( label_selector="node-role.kubernetes.io/worker", instance_count=1, network_params={"latency": "50ms"}, execution_type="invalid_type", test_duration=120 ) output_id, output_data = ingress_shaping.network_chaos(params=cfg, run_id="test-run") self.assertEqual(output_id, "error") self.assertIn("Invalid execution type", output_data.error) @patch('krkn.scenario_plugins.native.network.ingress_shaping.Environment') @patch('krkn.scenario_plugins.native.network.ingress_shaping.FileSystemLoader') @patch('yaml.safe_load') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_jobs') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_node_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.KrknKubernetes') def test_network_chaos_get_nodes_error( self, mock_kubecli_class, mock_get_nodes, mock_delete_virtual, mock_delete_jobs, mock_yaml, mock_file_loader, mock_env ): """Test network chaos when getting nodes fails""" mock_kubecli_class.return_value = Mock() mock_yaml.return_value = {"metadata": {"name": "test-pod"}} mock_get_nodes.side_effect = Exception("Failed to get nodes") cfg = ingress_shaping.NetworkScenarioConfig( label_selector="node-role.kubernetes.io/worker", instance_count=1, network_params={"latency": "50ms"} ) output_id, output_data = ingress_shaping.network_chaos(params=cfg, run_id="test-run") self.assertEqual(output_id, "error") self.assertIn("Failed to get nodes", output_data.error) @patch('krkn.scenario_plugins.native.network.ingress_shaping.Environment') @patch('krkn.scenario_plugins.native.network.ingress_shaping.FileSystemLoader') @patch('yaml.safe_load') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_jobs') @patch('krkn.scenario_plugins.native.network.ingress_shaping.delete_virtual_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.apply_ingress_filter') @patch('krkn.scenario_plugins.native.network.ingress_shaping.get_node_interfaces') @patch('krkn.scenario_plugins.native.network.ingress_shaping.KrknKubernetes') def test_network_chaos_apply_filter_error( self, mock_kubecli_class, mock_get_nodes, mock_apply_filter, mock_delete_virtual, mock_delete_jobs, mock_yaml, mock_file_loader, mock_env ): """Test network chaos when applying filter fails""" mock_kubecli_class.return_value = Mock() mock_yaml.return_value = {"metadata": {"name": "test-pod"}} mock_get_nodes.return_value = {"node1": ["eth0"]} mock_apply_filter.side_effect = Exception("Failed to apply filter") cfg = ingress_shaping.NetworkScenarioConfig( label_selector="node-role.kubernetes.io/worker", instance_count=1, network_params={"latency": "50ms"}, execution_type="parallel" ) output_id, output_data = ingress_shaping.network_chaos(params=cfg, run_id="test-run") self.assertEqual(output_id, "error") self.assertIn("Failed to apply filter", output_data.error) mock_delete_virtual.assert_called_once() if __name__ == "__main__": unittest.main()