import sys from pathlib import Path # Add api directory to path api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) import unittest import tempfile import shutil import os from unittest.mock import patch, MagicMock from routing_manager import RoutingManager import json class TestRoutingManager(unittest.TestCase): def setUp(self): self.test_dir = tempfile.mkdtemp() self.data_dir = os.path.join(self.test_dir, 'data') self.config_dir = os.path.join(self.test_dir, 'config') os.makedirs(self.data_dir, exist_ok=True) os.makedirs(self.config_dir, exist_ok=True) self.manager = RoutingManager(data_dir=self.data_dir, config_dir=self.config_dir) def tearDown(self): shutil.rmtree(self.test_dir) def test_initialization(self): # Test RoutingManager initialization and config creation self.assertTrue(os.path.exists(self.manager.routing_dir)) self.assertTrue(os.path.exists(self.manager.rules_file)) # Check that rules file contains default structure with open(self.manager.rules_file) as f: rules = json.load(f) self.assertIn('nat_rules', rules) self.assertIn('peer_routes', rules) self.assertIn('exit_nodes', rules) self.assertIn('bridge_routes', rules) self.assertIn('split_routes', rules) self.assertIn('firewall_rules', rules) self.assertIsInstance(rules['nat_rules'], list) self.assertIsInstance(rules['peer_routes'], dict) self.assertIsInstance(rules['exit_nodes'], list) self.assertIsInstance(rules['bridge_routes'], list) self.assertIsInstance(rules['split_routes'], list) self.assertIsInstance(rules['firewall_rules'], list) @patch.object(RoutingManager, '_apply_nat_rule', return_value=True) @patch.object(RoutingManager, '_remove_nat_rule', return_value=True) def test_add_and_remove_nat_rule(self, mock_remove_nat, mock_apply_nat): # Add a valid NAT rule result = self.manager.add_nat_rule('10.0.0.0/24', 'eth0') self.assertTrue(result) # Check that the rule is persisted with open(self.manager.rules_file) as f: rules = json.load(f) self.assertEqual(len(rules['nat_rules']), 1) rule = rules['nat_rules'][0] self.assertEqual(rule['source_network'], '10.0.0.0/24') self.assertEqual(rule['target_interface'], 'eth0') self.assertEqual(rule['nat_type'], 'MASQUERADE') self.assertTrue(rule['enabled']) # Remove the NAT rule rule_id = rule['id'] result = self.manager.remove_nat_rule(rule_id) self.assertTrue(result) with open(self.manager.rules_file) as f: rules = json.load(f) self.assertEqual(len(rules['nat_rules']), 0) # Test invalid NAT rule (bad CIDR) result = self.manager.add_nat_rule('bad-cidr', 'eth0') self.assertFalse(result) # Test invalid NAT rule (bad interface) result = self.manager.add_nat_rule('10.0.0.0/24', '') self.assertFalse(result) # Test invalid NAT rule (bad nat_type) result = self.manager.add_nat_rule('10.0.0.0/24', 'eth0', nat_type='INVALID') self.assertFalse(result) # Test invalid NAT rule (bad protocol) result = self.manager.add_nat_rule('10.0.0.0/24', 'eth0', protocol='INVALID') self.assertFalse(result) @patch.object(RoutingManager, '_apply_peer_route', return_value=True) @patch.object(RoutingManager, '_remove_peer_route', return_value=True) def test_add_and_remove_peer_route(self, mock_remove_peer, mock_apply_peer): # Add a valid peer route allowed_networks = ['10.0.0.0/24'] result = self.manager.add_peer_route('peer1', '10.0.0.2', allowed_networks) self.assertTrue(result) # Check that the route is persisted with open(self.manager.rules_file) as f: rules = json.load(f) self.assertIn('peer1', rules['peer_routes']) route = rules['peer_routes']['peer1'] self.assertEqual(route['peer_name'], 'peer1') self.assertEqual(route['peer_ip'], '10.0.0.2') self.assertEqual(route['allowed_networks'], allowed_networks) self.assertEqual(route['route_type'], 'lan') self.assertTrue(route['enabled']) # Remove the peer route result = self.manager.remove_peer_route('peer1') self.assertTrue(result) with open(self.manager.rules_file) as f: rules = json.load(f) self.assertNotIn('peer1', rules['peer_routes']) # Test invalid peer route (bad peer_name) result = self.manager.add_peer_route('', '10.0.0.2', allowed_networks) self.assertFalse(result) # Test invalid peer route (bad peer_ip) result = self.manager.add_peer_route('peer2', '', allowed_networks) self.assertFalse(result) # Test invalid peer route (bad allowed_networks) result = self.manager.add_peer_route('peer3', '10.0.0.3', ['bad-cidr']) self.assertFalse(result) # Test invalid peer route (bad route_type) result = self.manager.add_peer_route('peer4', '10.0.0.4', allowed_networks, route_type='invalid') self.assertFalse(result) def test_add_exit_node(self): pass # Test adding exit node configuration def test_add_bridge_route(self): pass # Test adding bridge route between peers def test_add_split_route(self): pass # Test adding split routing rule def test_add_firewall_rule(self): pass # Test adding firewall rule def test_get_routing_status(self): pass # Test routing status and monitoring def test_test_routing_connectivity(self): pass # Test routing connectivity def test_get_routing_logs(self): pass # Test log collection def test_error_handling(self): pass # Test error handling and edge cases def test_subprocess_command_execution(self): pass # Test subprocess command execution (mocked) def test_route_parsing_and_analysis(self): pass # Test route parsing and analysis if __name__ == '__main__': unittest.main()