pic/tests/test_cell_manager.py

#!/usr/bin/env python3
"""
Unit tests for CellManager class
"""

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 os
import json
import shutil
from unittest.mock import patch, MagicMock
from datetime import datetime

# Add parent directory to path for imports
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app import CellManager

class TestCellManager(unittest.TestCase):
    """Test cases for CellManager class"""
    
    def setUp(self):
        """Set up test environment"""
        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)
        # Use a unique config file for each test
        self.config_path = os.path.join(self.config_dir, 'cell_config_test.json')
        if os.path.exists(self.config_path):
            os.remove(self.config_path)
        self.env_patcher = patch.dict(os.environ, {
            'CELL_NAME': 'testcell',
            'DATA_DIR': self.data_dir,
            'CONFIG_DIR': self.config_dir
        })
        self.env_patcher.start()
        # Pass config_path to CellManager for isolation
        self.cell_manager = CellManager(config_path=self.config_path)
    
    def tearDown(self):
        """Clean up test environment"""
        self.env_patcher.stop()
        shutil.rmtree(self.test_dir)
    
    def test_initial_config_creation(self):
        """Test that initial configuration is created correctly"""
        config = self.cell_manager.config
        
        self.assertEqual(config['cell_name'], 'testcell')
        self.assertEqual(config['domain'], 'testcell.cell')
        self.assertEqual(config['ip_range'], '10.0.0.0/24')
        self.assertEqual(config['wireguard_port'], 51820)
        self.assertIn('created_at', config)
        # Only one config should exist
        self.assertIsInstance(config, dict)
    
    def test_config_persistence(self):
        """Test that configuration is saved and loaded correctly"""
        # Modify config
        self.cell_manager.config['cell_name'] = 'modified'
        self.cell_manager.save_config()
        
        # Create new instance to test loading (same config_path)
        new_manager = CellManager(config_path=self.config_path)
        self.assertEqual(new_manager.config['cell_name'], 'modified')
    
    def test_peer_management(self):
        """Test adding and removing peers"""
        # Test empty peers list
        peers = self.cell_manager.get_peers()
        self.assertEqual(len(peers), 0)
        
        # Test adding peer
        peer_data = {
            'name': 'testpeer',
            'ip': '192.168.1.100',
            'public_key': 'testkey123'
        }
        
        success, message = self.cell_manager.add_peer(peer_data)
        self.assertTrue(success)
        self.assertIn('successfully', message)
        
        # Test peer was added
        peers = self.cell_manager.get_peers()
        self.assertEqual(len(peers), 1)
        self.assertEqual(peers[0]['name'], 'testpeer')
        
        # Test adding duplicate peer (should fail)
        success, message = self.cell_manager.add_peer({'name': 'testpeer', 'ip': '192.168.1.100', 'public_key': 'testkey123'})
        self.assertFalse(success)
        self.assertIn('already exists', message)
        
        # Test removing peer
        success, message = self.cell_manager.remove_peer('testpeer')
        self.assertTrue(success)
        
        # Test peer was removed
        peers = self.cell_manager.get_peers()
        self.assertEqual(len(peers), 0)
    
    def test_peer_validation(self):
        """Test peer data validation"""
        # Test missing required fields
        invalid_peer = {'name': 'test'}
        success, message = self.cell_manager.add_peer(invalid_peer)
        self.assertFalse(success)
        self.assertIn('Missing required field', message)
        
        # Test valid peer
        valid_peer = {
            'name': 'testpeer',
            'ip': '192.168.1.100',
            'public_key': 'testkey123'
        }
        success, message = self.cell_manager.add_peer(valid_peer)
        self.assertTrue(success)
    
    def test_get_status(self):
        """Test status information retrieval"""
        status = self.cell_manager.get_status()
        
        self.assertIn('cell_name', status)
        self.assertIn('domain', status)
        self.assertIn('peers_count', status)
        self.assertIn('services', status)
        self.assertIn('uptime', status)
        
        self.assertEqual(status['cell_name'], 'testcell')
        self.assertEqual(status['domain'], 'testcell.cell')
        self.assertEqual(status['peers_count'], 0)
    
    @patch('subprocess.run')
    def test_service_status_check(self, mock_run):
        """Test service status checking"""
        mock_run.return_value.stdout = 'cell-dns\n'
        mock_run.return_value.returncode = 0
        services = self.cell_manager.get_services_status()
        # Accept either flat or nested dict structure
        if isinstance(services, dict) and 'dns' in services:
            self.assertIn('dns', services)
        elif 'network' in services and isinstance(services['network'], dict):
            self.assertIn('status', services['network'])
        # Mock failed service check
        mock_run.side_effect = Exception("Docker not available")
        services = self.cell_manager.get_services_status()
        # Accept either flat or nested dict structure
        if isinstance(services, dict) and 'dns' in services:
            self.assertFalse(services['dns'])
        elif 'network' in services and isinstance(services['network'], dict):
            self.assertIn('status', services['network'])
    
    def test_uptime_retrieval(self):
        """Test uptime retrieval"""
        uptime = self.cell_manager.get_uptime()
        self.assertIsInstance(uptime, int)
        self.assertGreaterEqual(uptime, 0)

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