#!/usr/bin/env python3 """ Comprehensive Test Suite for Enhanced Personal Internet Cell API Tests all new components and integrations """ import unittest import json import tempfile import os import shutil from datetime import datetime, timedelta from unittest.mock import Mock, patch, MagicMock import sys import threading import time # Add the api directory to the path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from base_service_manager import BaseServiceManager from config_manager import ConfigManager from service_bus import ServiceBus, EventType, Event from log_manager import LogManager, LogLevel from network_manager import NetworkManager from enhanced_cli import APIClient, ConfigManager as CLIConfigManager, EnhancedCLI class TestBaseServiceManager(unittest.TestCase): """Test the base service manager functionality""" def setUp(self): self.temp_dir = tempfile.mkdtemp() self.data_dir = os.path.join(self.temp_dir, 'data') self.config_dir = os.path.join(self.temp_dir, 'config') os.makedirs(self.data_dir, exist_ok=True) os.makedirs(self.config_dir, exist_ok=True) # Create a concrete implementation for testing class TestServiceManager(BaseServiceManager): def get_status(self): return {'running': True, 'status': 'online'} def test_connectivity(self): return {'success': True, 'message': 'Connected'} self.service_manager = TestServiceManager('test_service', self.data_dir, self.config_dir) def tearDown(self): shutil.rmtree(self.temp_dir) def test_initialization(self): """Test service manager initialization""" self.assertEqual(self.service_manager.service_name, 'test_service') self.assertEqual(self.service_manager.data_dir, self.data_dir) self.assertEqual(self.service_manager.config_dir, self.config_dir) self.assertTrue(os.path.exists(self.data_dir)) self.assertTrue(os.path.exists(self.config_dir)) def test_get_status(self): """Test get_status method""" status = self.service_manager.get_status() self.assertEqual(status['running'], True) self.assertEqual(status['status'], 'online') def test_test_connectivity(self): """Test test_connectivity method""" connectivity = self.service_manager.test_connectivity() self.assertEqual(connectivity['success'], True) self.assertEqual(connectivity['message'], 'Connected') def test_get_logs(self): """Test get_logs method""" # Create a test log file log_file = os.path.join(self.data_dir, 'test_service.log') with open(log_file, 'w') as f: f.write("Test log line 1\n") f.write("Test log line 2\n") logs = self.service_manager.get_logs(lines=2) self.assertEqual(len(logs), 2) self.assertIn("Test log line 1", logs[0]) self.assertIn("Test log line 2", logs[1]) def test_get_config(self): """Test get_config method""" # Create a test config file config_file = os.path.join(self.config_dir, 'test_service.json') test_config = {'key': 'value', 'number': 42} with open(config_file, 'w') as f: json.dump(test_config, f) config = self.service_manager.get_config() self.assertEqual(config['key'], 'value') self.assertEqual(config['number'], 42) def test_update_config(self): """Test update_config method""" test_config = {'new_key': 'new_value', 'number': 100} success = self.service_manager.update_config(test_config) self.assertTrue(success) # Verify config was saved config = self.service_manager.get_config() self.assertEqual(config['new_key'], 'new_value') self.assertEqual(config['number'], 100) def test_validate_config(self): """Test validate_config method""" test_config = {'key': 'value'} validation = self.service_manager.validate_config(test_config) self.assertTrue(validation['valid']) self.assertEqual(len(validation['errors']), 0) def test_get_metrics(self): """Test get_metrics method""" metrics = self.service_manager.get_metrics() self.assertEqual(metrics['service'], 'test_service') self.assertIn('timestamp', metrics) self.assertEqual(metrics['status'], 'unknown') def test_handle_error(self): """Test handle_error method""" test_error = ValueError("Test error") error_info = self.service_manager.handle_error(test_error, "test_context") self.assertEqual(error_info['error'], "Test error") self.assertEqual(error_info['type'], "ValueError") self.assertEqual(error_info['context'], "test_context") self.assertEqual(error_info['service'], 'test_service') self.assertIn('traceback', error_info) def test_health_check(self): """Test health_check method""" health = self.service_manager.health_check() self.assertEqual(health['service'], 'test_service') self.assertIn('timestamp', health) self.assertIn('status', health) self.assertIn('connectivity', health) self.assertIn('metrics', health) self.assertIn('healthy', health) self.assertTrue(health['healthy']) class TestConfigManager(unittest.TestCase): """Test the configuration manager functionality""" def setUp(self): self.temp_dir = tempfile.mkdtemp() self.config_dir = os.path.join(self.temp_dir, 'config') self.data_dir = os.path.join(self.temp_dir, 'data') os.makedirs(self.config_dir, exist_ok=True) os.makedirs(self.data_dir, exist_ok=True) self.config_file = os.path.join(self.config_dir, 'cell_config.json') assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] TestConfigManager.setUp: self.config_file = {self.config_file}") # Ensure the config file exists and is a valid JSON file if not os.path.exists(self.config_file): with open(self.config_file, 'w') as f: json.dump({}, f) self.config_manager = ConfigManager(self.config_file, self.data_dir) def tearDown(self): shutil.rmtree(self.temp_dir) if os.path.exists(self.config_file): os.remove(self.config_file) def test_initialization(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_initialization: self.config_file = {self.config_file}") """Test config manager initialization""" self.assertTrue(os.path.exists(self.config_dir)) self.assertTrue(os.path.exists(self.data_dir)) self.assertTrue(os.path.exists(self.config_manager.backup_dir)) self.assertIsNotNone(self.config_manager.service_schemas) def test_get_service_config(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_get_service_config: self.config_file = {self.config_file}") """Test getting service configuration""" # Test with non-existent service with self.assertRaises(ValueError): self.config_manager.get_service_config('nonexistent_service') # Test with valid service config = self.config_manager.get_service_config('network') self.assertEqual(config, {}) def test_update_service_config(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_update_service_config: self.config_file = {self.config_file}") """Test updating service configuration""" test_config = { 'dns_port': 53, 'dhcp_range': '10.0.0.100-10.0.0.200', 'ntp_servers': ['pool.ntp.org'] } success = self.config_manager.update_service_config('network', test_config) self.assertTrue(success) # Verify config was saved config = self.config_manager.get_service_config('network') self.assertEqual(config['dns_port'], 53) self.assertEqual(config['dhcp_range'], '10.0.0.100-10.0.0.200') self.assertEqual(config['ntp_servers'], ['pool.ntp.org']) def test_validate_config(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_validate_config: self.config_file = {self.config_file}") """Test configuration validation""" # Test valid config valid_config = { 'dns_port': 53, 'dhcp_range': '10.0.0.100-10.0.0.200', 'ntp_servers': ['pool.ntp.org'] } validation = self.config_manager.validate_config('network', valid_config) self.assertTrue(validation['valid']) self.assertEqual(len(validation['errors']), 0) # Test invalid config (missing required field) invalid_config = { 'dns_port': 53 # Missing dhcp_range and ntp_servers } validation = self.config_manager.validate_config('network', invalid_config) self.assertFalse(validation['valid']) self.assertGreater(len(validation['errors']), 0) # Test invalid config (wrong type) invalid_type_config = { 'dns_port': 'not_a_number', 'dhcp_range': '10.0.0.100-10.0.0.200', 'ntp_servers': ['pool.ntp.org'] } validation = self.config_manager.validate_config('network', invalid_type_config) self.assertFalse(validation['valid']) self.assertGreater(len(validation['errors']), 0) def test_backup_and_restore(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_backup_and_restore: self.config_file = {self.config_file}") """Test configuration backup and restore""" # Create some test configurations test_configs = { 'network': {'dns_port': 53, 'dhcp_range': '10.0.0.100-10.0.0.200', 'ntp_servers': ['pool.ntp.org']}, 'wireguard': {'port': 51820, 'private_key': 'test_key', 'address': '10.0.0.1/24'} } for service, config in test_configs.items(): self.config_manager.update_service_config(service, config) # Create backup backup_id = self.config_manager.backup_config() self.assertIsNotNone(backup_id) # List backups backups = self.config_manager.list_backups() self.assertEqual(len(backups), 1) self.assertEqual(backups[0]['backup_id'], backup_id) # Modify config self.config_manager.update_service_config('network', {'dns_port': 5353}) # Restore backup success = self.config_manager.restore_config(backup_id) self.assertTrue(success) # Verify restoration config = self.config_manager.get_service_config('network') self.assertEqual(config['dns_port'], 53) # Should be restored value def test_export_import_config(self): assert not os.path.isdir(self.config_file), f"self.config_file is a directory: {self.config_file}" print(f"[DEBUG] test_export_import_config: self.config_file = {self.config_file}") """Test configuration export and import""" # Create test configurations test_configs = { 'network': {'dns_port': 53, 'dhcp_range': '10.0.0.100-10.0.0.200', 'ntp_servers': ['pool.ntp.org']}, 'wireguard': {'port': 51820, 'private_key': 'test_key', 'address': '10.0.0.1/24'} } for service, config in test_configs.items(): self.config_manager.update_service_config(service, config) # Export configuration exported_json = self.config_manager.export_config('json') exported_yaml = self.config_manager.export_config('yaml') self.assertIsInstance(exported_json, str) self.assertIsInstance(exported_yaml, str) # Clear unified config file if os.path.exists(self.config_file): os.remove(self.config_file) # Import configuration success = self.config_manager.import_config(exported_json, 'json') self.assertTrue(success) # Verify import for service, expected_config in test_configs.items(): config = self.config_manager.get_service_config(service) for key, value in expected_config.items(): self.assertEqual(config[key], value) # Also verify that required fields are present (even if with default values) schema = self.config_manager.service_schemas[service] for field in schema['required']: self.assertIn(field, config) class TestServiceBus(unittest.TestCase): """Test the service bus functionality""" def setUp(self): self.service_bus = ServiceBus() def test_initialization(self): """Test service bus initialization""" self.assertFalse(self.service_bus.running) self.assertEqual(len(self.service_bus.service_registry), 0) self.assertEqual(len(self.service_bus.event_handlers), 0) def test_start_stop(self): """Test service bus start and stop""" self.service_bus.start() self.assertTrue(self.service_bus.running) self.assertIsNotNone(self.service_bus.event_loop_thread) self.service_bus.stop() self.assertFalse(self.service_bus.running) def test_register_unregister_service(self): """Test service registration and unregistration""" mock_service = Mock() mock_service.get_status.return_value = {'running': True} # Register service self.service_bus.register_service('test_service', mock_service) self.assertIn('test_service', self.service_bus.service_registry) self.assertEqual(self.service_bus.service_registry['test_service'], mock_service) # Unregister service self.service_bus.unregister_service('test_service') self.assertNotIn('test_service', self.service_bus.service_registry) def test_publish_subscribe_events(self): """Test event publishing and subscription""" events_received = [] def event_handler(event): events_received.append(event) # Subscribe to events self.service_bus.subscribe_to_event(EventType.SERVICE_STARTED, event_handler) # Start service bus self.service_bus.start() # Publish event test_data = {'service': 'test_service', 'timestamp': datetime.utcnow().isoformat()} self.service_bus.publish_event(EventType.SERVICE_STARTED, 'test_service', test_data) # Wait for event processing time.sleep(0.1) # Check if event was received self.assertEqual(len(events_received), 1) self.assertEqual(events_received[0].event_type, EventType.SERVICE_STARTED) self.assertEqual(events_received[0].source, 'test_service') self.assertEqual(events_received[0].data, test_data) self.service_bus.stop() def test_call_service(self): """Test service method calling""" # Create a real service class instead of Mock class TestService: def test_method(self, arg1=None): return 'test_result' test_service = TestService() self.service_bus.register_service('test_service', test_service) # Call service method result = self.service_bus.call_service('test_service', 'test_method', arg1='value1') self.assertEqual(result, 'test_result') # Test calling non-existent service with self.assertRaises(ValueError): self.service_bus.call_service('nonexistent_service', 'test_method') # Test calling non-existent method with self.assertRaises(ValueError): self.service_bus.call_service('test_service', 'nonexistent_method') def test_service_orchestration(self): """Test service orchestration""" mock_service = Mock() mock_service.start = Mock() mock_service.stop = Mock() self.service_bus.register_service('test_service', mock_service) # Test service start orchestration success = self.service_bus.orchestrate_service_start('test_service') self.assertTrue(success) mock_service.start.assert_called_once() # Test service stop orchestration success = self.service_bus.orchestrate_service_stop('test_service') self.assertTrue(success) mock_service.stop.assert_called_once() # Test service restart orchestration success = self.service_bus.orchestrate_service_restart('test_service') self.assertTrue(success) self.assertEqual(mock_service.start.call_count, 2) self.assertEqual(mock_service.stop.call_count, 2) def test_event_history(self): """Test event history functionality""" self.service_bus.start() # Publish some events for i in range(5): self.service_bus.publish_event(EventType.SERVICE_STARTED, f'service_{i}', {'index': i}) # Wait for event processing time.sleep(0.1) # Get event history events = self.service_bus.get_event_history(limit=3) self.assertEqual(len(events), 3) # Test filtering by event type started_events = self.service_bus.get_event_history(EventType.SERVICE_STARTED, limit=2) self.assertEqual(len(started_events), 2) for event in started_events: self.assertEqual(event.event_type, EventType.SERVICE_STARTED) # Test filtering by source service_0_events = self.service_bus.get_event_history(source='service_0') self.assertEqual(len(service_0_events), 1) self.assertEqual(service_0_events[0].source, 'service_0') self.service_bus.stop() class TestLogManager(unittest.TestCase): """Test the log manager functionality""" def setUp(self): self.temp_dir = tempfile.mkdtemp() self.log_dir = os.path.join(self.temp_dir, 'logs') os.makedirs(self.log_dir, exist_ok=True) self.log_manager = LogManager(self.log_dir) def tearDown(self): self.log_manager.stop() shutil.rmtree(self.temp_dir) def test_initialization(self): """Test log manager initialization""" self.assertTrue(os.path.exists(self.log_dir)) self.assertIsNotNone(self.log_manager.formatters) self.assertIsNotNone(self.log_manager.handlers) self.assertTrue(self.log_manager.running) def test_add_service_logger(self): """Test adding service loggers""" config = {'level': 'INFO', 'formatter': 'json', 'console': False} self.log_manager.add_service_logger('test_service', config) self.assertIn('test_service', self.log_manager.service_loggers) self.assertIn('test_service', self.log_manager.handlers) def test_get_service_logs(self): """Test getting service logs""" # Add service logger first config = {'level': 'INFO', 'formatter': 'json', 'console': False} self.log_manager.add_service_logger('test_service', config) # Create a test log file in the correct location log_file = self.log_manager.log_dir / 'test_service.log' with open(log_file, 'w') as f: f.write('{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "message": "Test log 1"}\n') f.write('{"timestamp": "2024-01-01T10:01:00Z", "level": "ERROR", "message": "Test log 2"}\n') f.write('{"timestamp": "2024-01-01T10:02:00Z", "level": "INFO", "message": "Test log 3"}\n') # Test getting all logs logs = self.log_manager.get_service_logs_parsed('test_service', level='ALL', lines=3) self.assertEqual(len(logs), 3) # Test filtering by level error_logs = self.log_manager.get_service_logs_parsed('test_service', level='ERROR', lines=10) self.assertEqual(len(error_logs), 1) self.assertEqual(error_logs[0]['level'], 'ERROR') def test_search_logs(self): """Test log search functionality""" # Add service loggers first config = {'level': 'INFO', 'formatter': 'json', 'console': False} services = ['service1', 'service2'] for service in services: self.log_manager.add_service_logger(service, config) # Create test log files in the correct location for service in services: log_file = self.log_manager.log_dir / f'{service}.log' with open(log_file, 'w') as f: f.write('{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "message": "Test message for ' + service + '"}\n') f.write('{"timestamp": "2024-01-01T10:01:00Z", "level": "ERROR", "message": "Error in ' + service + '"}\n') # Test search across all services results = self.log_manager.search_logs('Test message') self.assertEqual(len(results), 2) # Test search with service filter results = self.log_manager.search_logs('Error', services=['service1']) self.assertEqual(len(results), 1) self.assertIn('service1', results[0]['service']) # Test search with level filter results = self.log_manager.search_logs('', level='ERROR') self.assertEqual(len(results), 2) for result in results: self.assertEqual(result['level'], 'ERROR') def test_export_logs(self): """Test log export functionality""" # Add service logger first config = {'level': 'INFO', 'formatter': 'json', 'console': False} self.log_manager.add_service_logger('test_service', config) # Create test log file in the correct location log_file = self.log_manager.log_dir / 'test_service.log' with open(log_file, 'w') as f: f.write('{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "message": "Test log"}\n') # Test JSON export json_export = self.log_manager.export_logs('json') self.assertIsInstance(json_export, str) self.assertIn('Test log', json_export) # Test CSV export csv_export = self.log_manager.export_logs('csv') self.assertIsInstance(csv_export, str) self.assertIn('Test log', csv_export) # Test text export text_export = self.log_manager.export_logs('text') self.assertIsInstance(text_export, str) self.assertIn('Test log', text_export) def test_log_statistics(self): """Test log statistics functionality""" # Create test log file log_file = os.path.join(self.log_dir, 'test_service.log') with open(log_file, 'w') as f: f.write('{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "message": "Info log"}\n') f.write('{"timestamp": "2024-01-01T10:01:00Z", "level": "ERROR", "message": "Error log"}\n') f.write('{"timestamp": "2024-01-01T10:02:00Z", "level": "WARNING", "message": "Warning log"}\n') # Get statistics stats = self.log_manager.get_log_statistics('test_service') self.assertIn('test_service', stats) self.assertEqual(stats['test_service']['total_entries'], 3) self.assertIn('level_counts', stats['test_service']) self.assertEqual(stats['test_service']['level_counts']['INFO'], 1) self.assertEqual(stats['test_service']['level_counts']['ERROR'], 1) self.assertEqual(stats['test_service']['level_counts']['WARNING'], 1) class TestEnhancedCLI(unittest.TestCase): """Test the enhanced CLI functionality""" def setUp(self): self.cli = EnhancedCLI() def test_api_client(self): """Test API client functionality""" client = APIClient() self.assertEqual(client.base_url, "http://localhost:3000/api") self.assertIsNotNone(client.session) def test_cli_config_manager(self): """Test CLI configuration manager""" config_manager = CLIConfigManager() self.assertIsNotNone(config_manager.config) # Test get/set config_manager.set('test_key', 'test_value') self.assertEqual(config_manager.get('test_key'), 'test_value') # Test export/import exported = config_manager.export_config('json') self.assertIsInstance(exported, str) self.assertIn('test_key', exported) def test_cli_commands(self): """Test CLI commands""" # Test status command with patch.object(self.cli.api_client, 'request') as mock_request: mock_request.return_value = { 'cell_name': 'test-cell', 'domain': 'test.local', 'peers_count': 2, 'services': {'network': {'running': True}} } # Capture print output from io import StringIO import sys old_stdout = sys.stdout sys.stdout = StringIO() try: self.cli.do_status("") output = sys.stdout.getvalue() self.assertIn('test-cell', output) self.assertIn('test.local', output) finally: sys.stdout = old_stdout class TestNetworkManagerIntegration(unittest.TestCase): """Test NetworkManager integration with BaseServiceManager""" def setUp(self): self.temp_dir = tempfile.mkdtemp() self.data_dir = os.path.join(self.temp_dir, 'data') self.config_dir = os.path.join(self.temp_dir, 'config') os.makedirs(self.data_dir, exist_ok=True) os.makedirs(self.config_dir, exist_ok=True) self.network_manager = NetworkManager(self.data_dir, self.config_dir) def tearDown(self): shutil.rmtree(self.temp_dir) def test_inheritance(self): """Test that NetworkManager inherits from BaseServiceManager""" self.assertIsInstance(self.network_manager, BaseServiceManager) self.assertEqual(self.network_manager.service_name, 'network') def test_get_status(self): """Test NetworkManager get_status method""" status = self.network_manager.get_status() self.assertIn('timestamp', status) self.assertIn('network', status) def test_test_connectivity(self): """Test NetworkManager test_connectivity method""" connectivity = self.network_manager.test_connectivity() self.assertIn('timestamp', connectivity) self.assertIn('network', connectivity) def run_tests(): """Run all tests""" # Create test suite test_suite = unittest.TestSuite() # Add test classes test_classes = [ TestBaseServiceManager, TestConfigManager, TestServiceBus, TestLogManager, TestEnhancedCLI, TestNetworkManagerIntegration ] for test_class in test_classes: tests = unittest.TestLoader().loadTestsFromTestCase(test_class) test_suite.addTests(tests) # Run tests runner = unittest.TextTestRunner(verbosity=2) result = runner.run(test_suite) # Print summary print(f"\n{'='*50}") print(f"Test Summary:") print(f"Tests run: {result.testsRun}") print(f"Failures: {len(result.failures)}") print(f"Errors: {len(result.errors)}") print(f"Success rate: {((result.testsRun - len(result.failures) - len(result.errors)) / result.testsRun * 100):.1f}%") print(f"{'='*50}") return result.wasSuccessful() if __name__ == '__main__': success = run_tests() sys.exit(0 if success else 1)