tests/test_vllm_manager.py

import asyncio
import unittest
import os
import sys
from unittest.mock import MagicMock, patch

# Ensure src is in python path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../src')))

from magentic_ui.backend.managers.vllm_manager import VLLMManager

class TestVLLMManager(unittest.TestCase):
    
    @patch("magentic_ui.backend.managers.vllm_manager.subprocess.Popen")
    def test_vllm_manager_start(self, mock_popen):
        manager = VLLMManager()
        
        # Mock process
        mock_process = MagicMock()
        mock_process.poll.return_value = None # Process running
        mock_popen.return_value = mock_process
        
        async def run_test():
            # Mock _wait_for_ready to avoid actual sleep loop in test
            with patch.object(manager, '_wait_for_ready', new_callable=AsyncMock):
                await manager.start()
                
        # Helper for async test
        class AsyncMock(MagicMock):
            async def __call__(self, *args, **kwargs):
                return super(AsyncMock, self).__call__(*args, **kwargs)

        loop = asyncio.new_event_loop()
        loop.run_until_complete(run_test())
        loop.close()
        
        self.assertTrue(manager.is_running())
        mock_popen.assert_called_once()

    def test_vllm_manager_stop(self):
        manager = VLLMManager()
        manager._process = MagicMock()
        manager._process.pid = 12345
        
        # Patch os.getpgid since that's called before killpg
        with patch("os.killpg") as mock_killpg, \
             patch("os.getpgid", return_value=12345):
            
            manager.stop()
            # The killpg call happens inside stop()
            mock_killpg.assert_called() 
            
        self.assertIsNone(manager._process)

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