Gan, Jimmy
56a8ff28ad
Fix 20 pre-existing test failures: - Update mock conversation DB schema with missing columns - Add in-memory OPC database mock to avoid PostgreSQL dependency in CI - Fix file download endpoint to check auth before revealing file existence - Reload routers.system and routers.security in test_app for fresh config - Reset cached Docker client in security router between tests - Fix websocket auth test to properly check for connection rejection Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
117 lines
4.2 KiB
Python
117 lines
4.2 KiB
Python
"""Integration tests for WebSocket security (terminal and OPC)"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
def test_terminal_ws_requires_authentication(test_app):
|
|
"""Test terminal WebSocket requires authentication"""
|
|
# Connection without auth should be rejected — either the upgrade fails
|
|
# (403) or the server immediately closes the connection after upgrade.
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect("/api/terminal/ws?host=nas") as websocket:
|
|
websocket.receive_text()
|
|
|
|
|
|
def test_terminal_ws_rejects_invalid_token(test_app):
|
|
"""Test terminal WebSocket rejects invalid tokens"""
|
|
# Try to connect with invalid token
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect(
|
|
"/api/terminal/ws?host=nas", cookies={"nas_access_token": "invalid-token"}
|
|
) as websocket:
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_rejects_expired_token(test_app, expired_access_token):
|
|
"""Test terminal WebSocket rejects expired tokens"""
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect(
|
|
"/api/terminal/ws?host=nas", cookies={"nas_access_token": expired_access_token}
|
|
) as websocket:
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_requires_page_access(test_app, valid_access_token):
|
|
"""Test terminal WebSocket requires terminal page access"""
|
|
# This test would need a token for a user without terminal access
|
|
# For now, we verify the endpoint exists and has auth
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect(
|
|
"/api/terminal/ws?host=nas", cookies={"nas_access_token": "invalid"}
|
|
) as websocket:
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_rejects_unknown_host(test_app, valid_access_token):
|
|
"""Test terminal WebSocket rejects unknown hosts"""
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect(
|
|
"/api/terminal/ws?host=unknown", cookies={"nas_access_token": valid_access_token}
|
|
) as websocket:
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_enforces_session_limit(test_app, valid_access_token):
|
|
"""Test terminal WebSocket enforces max sessions per user"""
|
|
# This would require actually establishing multiple connections
|
|
# For now, we verify the endpoint exists
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_validates_known_hosts(test_app, valid_access_token):
|
|
"""Test terminal WebSocket validates SSH known_hosts"""
|
|
# The endpoint should fail if known_hosts is not available
|
|
# This is tested by the _known_hosts_available check
|
|
pass
|
|
|
|
|
|
def test_opc_ws_requires_authentication(test_app):
|
|
"""Test OPC WebSocket requires authentication"""
|
|
# OPC WebSocket should also require authentication
|
|
with pytest.raises(Exception):
|
|
with test_app.websocket_connect("/api/opc/ws") as websocket:
|
|
pass
|
|
|
|
|
|
def test_opc_ws_accepts_ping(test_app, valid_access_token):
|
|
"""Test OPC WebSocket responds to ping"""
|
|
# This would require mocking the WebSocket connection
|
|
pass
|
|
|
|
|
|
def test_opc_ws_broadcasts_task_updates(test_app, admin_headers):
|
|
"""Test OPC WebSocket broadcasts task updates"""
|
|
# This would require establishing a WebSocket connection and creating a task
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_handles_resize_messages(test_app, valid_access_token):
|
|
"""Test terminal WebSocket handles terminal resize messages"""
|
|
# Terminal resize messages start with 0x01 byte followed by cols/rows
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_handles_ping_pong(test_app, valid_access_token):
|
|
"""Test terminal WebSocket handles ping/pong keepalive"""
|
|
# Terminal should respond to __ping__ with __pong__
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_closes_on_ssh_error(test_app, valid_access_token):
|
|
"""Test terminal WebSocket closes gracefully on SSH errors"""
|
|
# Should close connection if SSH connection fails
|
|
pass
|
|
|
|
|
|
def test_terminal_ws_releases_session_on_close(test_app, valid_access_token):
|
|
"""Test terminal WebSocket releases session slot on close"""
|
|
# Should decrement active session count when connection closes
|
|
pass
|
|
|
|
|
|
def test_opc_ws_filters_messages_by_authorization(test_app, admin_headers):
|
|
"""Test OPC WebSocket only sends messages for authorized tasks"""
|
|
# Users should only receive updates for tasks they have access to
|
|
pass
|