Add comprehensive SSH terminal drawer feature with embedded and popout modes, integrate playbook lint results API with local cache fallback, and enhance host management UI with terminal access buttons

tests/backend/test_terminal.py

@@ -0,0 +1,181 @@
+"""
+Tests for terminal SSH web feature.
+"""
+import hashlib
+import pytest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from app.services.terminal_service import (
+    TerminalService,
+    terminal_service,
+    TERMINAL_SESSION_TTL_MINUTES,
+    TERMINAL_MAX_SESSIONS_PER_USER,
+)
+
+
+class TestTerminalService:
+    """Tests for TerminalService class."""
+    
+    def test_generate_session_id(self):
+        """Test session ID generation produces unique IDs."""
+        service = TerminalService()
+        id1 = service.generate_session_id()
+        id2 = service.generate_session_id()
+        
+        assert len(id1) == 64  # 32 bytes hex = 64 chars
+        assert len(id2) == 64
+        assert id1 != id2
+    
+    def test_generate_session_token(self):
+        """Test token generation produces unique tokens and valid hashes."""
+        service = TerminalService()
+        token1, hash1 = service.generate_session_token()
+        token2, hash2 = service.generate_session_token()
+        
+        # Tokens should be unique
+        assert token1 != token2
+        
+        # Hashes should be valid SHA256
+        assert len(hash1) == 64
+        assert len(hash2) == 64
+        
+        # Hash should match token
+        computed_hash = hashlib.sha256(token1.encode()).hexdigest()
+        assert computed_hash == hash1
+    
+    def test_verify_token_valid(self):
+        """Test token verification with valid token."""
+        service = TerminalService()
+        token, token_hash = service.generate_session_token()
+        
+        assert service.verify_token(token, token_hash) is True
+    
+    def test_verify_token_invalid(self):
+        """Test token verification with invalid token."""
+        service = TerminalService()
+        token, token_hash = service.generate_session_token()
+        
+        # Wrong token
+        assert service.verify_token("wrong_token", token_hash) is False
+        
+        # Wrong hash
+        assert service.verify_token(token, "wrong_hash") is False
+    
+    @pytest.mark.asyncio
+    async def test_allocate_port(self):
+        """Test port allocation."""
+        service = TerminalService()
+        session_id = service.generate_session_id()
+        
+        port = await service.allocate_port(session_id)
+        
+        assert port >= 7680
+        assert port <= 7700
+        assert port in service._allocated_ports
+        assert service._allocated_ports[port] == session_id
+    
+    @pytest.mark.asyncio
+    async def test_allocate_port_unique(self):
+        """Test that allocated ports are unique."""
+        service = TerminalService()
+        
+        ports = []
+        for i in range(5):
+            session_id = service.generate_session_id()
+            port = await service.allocate_port(session_id)
+            ports.append(port)
+        
+        # All ports should be unique
+        assert len(ports) == len(set(ports))
+    
+    @pytest.mark.asyncio
+    async def test_release_port(self):
+        """Test port release."""
+        service = TerminalService()
+        session_id = service.generate_session_id()
+        
+        port = await service.allocate_port(session_id)
+        assert port in service._allocated_ports
+        
+        await service.release_port(port)
+        assert port not in service._allocated_ports
+    
+    def test_check_ttyd_available_not_found(self):
+        """Test ttyd availability check when not installed."""
+        service = TerminalService()
+        service._ttyd_available = None  # Reset cache
+        
+        with patch('shutil.which', return_value=None):
+            result = service.check_ttyd_available()
+        
+        assert result is False
+        assert service._ttyd_available is False
+    
+    def test_check_ttyd_available_found(self):
+        """Test ttyd availability check when installed."""
+        service = TerminalService()
+        service._ttyd_available = None  # Reset cache
+        
+        with patch('shutil.which', return_value='/usr/bin/ttyd'):
+            result = service.check_ttyd_available()
+        
+        assert result is True
+        assert service._ttyd_available is True
+    
+    def test_get_session_url_direct(self):
+        """Test session URL generation for direct access."""
+        service = TerminalService()
+        url = service.get_session_url(7680, "test_token")
+        
+        assert url == "http://localhost:7680/"
+    
+    def test_get_session_url_with_base(self):
+        """Test session URL generation with base URL."""
+        service = TerminalService()
+        url = service.get_session_url(7680, "test_token", base_url="https://example.com")
+        
+        assert url == "https://example.com/terminal/proxy/7680?token=test_token"
+    
+    def test_get_websocket_url(self):
+        """Test WebSocket URL generation."""
+        service = TerminalService()
+        url = service.get_websocket_url(7680)
+        
+        assert url == "ws://localhost:7680/ws"
+    
+    def test_get_active_session_count_empty(self):
+        """Test active session count when no sessions."""
+        service = TerminalService()
+        assert service.get_active_session_count() == 0
+    
+    @pytest.mark.asyncio
+    async def test_terminate_session_not_found(self):
+        """Test terminating non-existent session."""
+        service = TerminalService()
+        result = await service.terminate_session("nonexistent_id")
+        assert result is False
+
+
+class TestTerminalSessionTTL:
+    """Tests for session TTL configuration."""
+    
+    def test_session_ttl_default(self):
+        """Test default TTL value."""
+        assert TERMINAL_SESSION_TTL_MINUTES == 30
+    
+    def test_max_sessions_default(self):
+        """Test default max sessions per user."""
+        assert TERMINAL_MAX_SESSIONS_PER_USER == 3
+
+
+class TestTerminalServiceSingleton:
+    """Tests for terminal service singleton."""
+    
+    def test_global_instance(self):
+        """Test global terminal service instance exists."""
+        from app.services.terminal_service import get_terminal_service
+        
+        service = get_terminal_service()
+        assert service is terminal_service
+        assert isinstance(service, TerminalService)