mirror of
https://github.com/johndoe6345789/docker-swarm-termina.git
synced 2026-04-24 13:45:01 +00:00
4e928db0a8
Added comprehensive edge case tests to reach 100% coverage: - test_complete_coverage.py: 16 tests for handler and utility edge cases - test_final_coverage.py: 9 tests for remaining uncovered code paths Coverage improvements: - handlers/terminal/disconnect.py: 38% → 100% - handlers/terminal/resize.py: 83% → 100% - utils/diagnostics/docker_env.py: 95% → 100% - utils/docker_client.py: 89% → 100% - utils/exec_helpers.py: 93% → 100% - utils/terminal_helpers.py: 92% → 100% Final metrics: - Pylint: 10.00/10 ✓ - Test Coverage: 100% ✓ - Tests Passing: 113/113 ✓ https://claude.ai/code/session_011PzvkCnVrsatoxbY3HbGXz
379 lines
13 KiB
Python
379 lines
13 KiB
Python
"""Tests to achieve 100% code coverage."""
|
|
import pytest
|
|
import os
|
|
import time
|
|
from unittest.mock import MagicMock, patch, Mock, PropertyMock
|
|
from flask_socketio import SocketIOTestClient
|
|
|
|
|
|
class TestHandlerEdgeCases:
|
|
"""Test edge cases in terminal handlers"""
|
|
|
|
@pytest.fixture
|
|
def socketio_client(self, app):
|
|
"""Create a SocketIO test client"""
|
|
from app import socketio
|
|
return socketio.test_client(app, namespace='/terminal')
|
|
|
|
def test_disconnect_handler_exception_during_cleanup(self):
|
|
"""Test disconnect handler when exec.kill() raises exception"""
|
|
from handlers.terminal.disconnect import handle_disconnect
|
|
from config import active_terminals
|
|
from flask import Flask
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context():
|
|
with patch('handlers.terminal.disconnect.request') as mock_request:
|
|
mock_request.sid = 'test_exception_sid'
|
|
|
|
# Create exec that raises exception on kill
|
|
mock_exec = MagicMock()
|
|
mock_exec.kill.side_effect = Exception("Kill failed")
|
|
active_terminals['test_exception_sid'] = {'exec': mock_exec}
|
|
|
|
# Should not raise, just clean up
|
|
handle_disconnect()
|
|
assert 'test_exception_sid' not in active_terminals
|
|
|
|
def test_input_handler_no_active_terminal(self):
|
|
"""Test input handler when no active terminal exists"""
|
|
from handlers.terminal.input import handle_input
|
|
from flask import Flask
|
|
from flask_socketio import emit
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context():
|
|
with patch('handlers.terminal.input.request') as mock_request:
|
|
with patch('handlers.terminal.input.emit') as mock_emit:
|
|
mock_request.sid = 'nonexistent_sid'
|
|
|
|
handle_input({'data': 'test'})
|
|
|
|
# Should emit error
|
|
mock_emit.assert_called_once()
|
|
args = mock_emit.call_args[0]
|
|
assert args[0] == 'error'
|
|
assert 'No active terminal session' in args[1]['error']
|
|
|
|
def test_input_handler_exception(self):
|
|
"""Test input handler when sendall raises exception"""
|
|
from handlers.terminal.input import handle_input
|
|
from config import active_terminals
|
|
from flask import Flask
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context():
|
|
with patch('handlers.terminal.input.request') as mock_request:
|
|
with patch('handlers.terminal.input.emit') as mock_emit:
|
|
mock_request.sid = 'error_sid'
|
|
|
|
# Mock the _sock attribute which is checked first
|
|
mock_inner_sock = MagicMock()
|
|
mock_inner_sock.sendall.side_effect = Exception("Send failed")
|
|
|
|
mock_sock = MagicMock()
|
|
mock_sock._sock = mock_inner_sock
|
|
|
|
mock_exec = MagicMock()
|
|
mock_exec.output = mock_sock
|
|
|
|
active_terminals['error_sid'] = {'exec': mock_exec}
|
|
|
|
handle_input({'data': 'test'})
|
|
|
|
# Should emit error
|
|
mock_emit.assert_called()
|
|
error_call = [c for c in mock_emit.call_args_list if c[0][0] == 'error']
|
|
assert len(error_call) > 0
|
|
|
|
def test_resize_handler_exception(self):
|
|
"""Test resize handler when it raises exception"""
|
|
from handlers.terminal.resize import handle_resize
|
|
from config import active_terminals
|
|
from flask import Flask
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context():
|
|
with patch('handlers.terminal.resize.request') as mock_request:
|
|
mock_request.sid = 'resize_error_sid'
|
|
active_terminals['resize_error_sid'] = {'exec': MagicMock()}
|
|
|
|
# Force an exception by passing invalid data
|
|
with patch('handlers.terminal.resize.logger') as mock_logger:
|
|
# This should trigger the exception handler
|
|
handle_resize(None) # None instead of dict
|
|
|
|
# Should have logged error
|
|
assert mock_logger.error.called
|
|
|
|
|
|
class TestDockerDiagnostics:
|
|
"""Test docker diagnostics edge cases"""
|
|
|
|
@patch('os.path.exists')
|
|
@patch('os.listdir')
|
|
def test_diagnose_var_run_not_exists(self, mock_listdir, mock_exists):
|
|
"""Test diagnostics when /var/run doesn't exist"""
|
|
from utils.diagnostics.docker_env import diagnose_docker_environment
|
|
|
|
mock_exists.return_value = False
|
|
|
|
# Should not raise exception
|
|
with patch('utils.diagnostics.docker_env.logger'):
|
|
diagnose_docker_environment()
|
|
|
|
@patch('os.path.exists')
|
|
@patch('os.listdir')
|
|
def test_diagnose_var_run_error(self, mock_listdir, mock_exists):
|
|
"""Test diagnostics when /var/run listing fails"""
|
|
from utils.diagnostics.docker_env import diagnose_docker_environment
|
|
|
|
def exists_side_effect(path):
|
|
if path == '/var/run':
|
|
return True
|
|
return False
|
|
|
|
mock_exists.side_effect = exists_side_effect
|
|
mock_listdir.side_effect = Exception("Permission denied")
|
|
|
|
# Should handle exception
|
|
with patch('utils.diagnostics.docker_env.logger'):
|
|
diagnose_docker_environment()
|
|
|
|
@patch('os.path.exists')
|
|
@patch('os.stat')
|
|
@patch('os.access')
|
|
@patch('os.getuid')
|
|
@patch('os.getgid')
|
|
@patch('os.getgroups')
|
|
def test_diagnose_docker_socket_permissions(
|
|
self, mock_getgroups, mock_getgid, mock_getuid,
|
|
mock_access, mock_stat, mock_exists
|
|
):
|
|
"""Test diagnostics for docker socket with permissions check"""
|
|
from utils.diagnostics.docker_env import diagnose_docker_environment
|
|
import pwd
|
|
import grp
|
|
|
|
def exists_side_effect(path):
|
|
if path == '/var/run':
|
|
return False
|
|
if path == '/var/run/docker.sock':
|
|
return True
|
|
return False
|
|
|
|
mock_exists.side_effect = exists_side_effect
|
|
|
|
# Mock stat for socket
|
|
mock_stat_result = MagicMock()
|
|
mock_stat_result.st_mode = 0o666
|
|
mock_stat_result.st_uid = 0
|
|
mock_stat_result.st_gid = 0
|
|
mock_stat.return_value = mock_stat_result
|
|
|
|
# Mock access - not readable/writable
|
|
mock_access.return_value = False
|
|
|
|
# Mock user info
|
|
mock_getuid.return_value = 0
|
|
mock_getgid.return_value = 0
|
|
mock_getgroups.return_value = [0, 1]
|
|
|
|
with patch('utils.diagnostics.docker_env.logger'):
|
|
with patch('pwd.getpwuid') as mock_getpwuid:
|
|
with patch('grp.getgrgid') as mock_getgrgid:
|
|
mock_user = MagicMock()
|
|
mock_user.pw_name = 'root'
|
|
mock_getpwuid.return_value = mock_user
|
|
|
|
mock_group = MagicMock()
|
|
mock_group.gr_name = 'root'
|
|
mock_getgrgid.return_value = mock_group
|
|
|
|
diagnose_docker_environment()
|
|
|
|
@patch('os.path.exists')
|
|
@patch('os.getuid')
|
|
def test_diagnose_user_info_error(self, mock_getuid, mock_exists):
|
|
"""Test diagnostics when user info lookup fails"""
|
|
from utils.diagnostics.docker_env import diagnose_docker_environment
|
|
|
|
mock_exists.return_value = False
|
|
mock_getuid.side_effect = Exception("No user info")
|
|
|
|
with patch('utils.diagnostics.docker_env.logger'):
|
|
diagnose_docker_environment()
|
|
|
|
@patch('os.path.exists')
|
|
@patch('os.getuid')
|
|
@patch('os.getgid')
|
|
@patch('os.getgroups')
|
|
def test_diagnose_group_lookup_error(self, mock_getgroups, mock_getgid, mock_getuid, mock_exists):
|
|
"""Test diagnostics when group lookup fails"""
|
|
from utils.diagnostics.docker_env import diagnose_docker_environment
|
|
import pwd
|
|
import grp
|
|
|
|
mock_exists.return_value = False
|
|
mock_getuid.return_value = 0
|
|
mock_getgid.return_value = 0
|
|
mock_getgroups.return_value = [999] # Non-existent group
|
|
|
|
with patch('utils.diagnostics.docker_env.logger'):
|
|
with patch('pwd.getpwuid') as mock_getpwuid:
|
|
with patch('grp.getgrgid') as mock_getgrgid:
|
|
mock_user = MagicMock()
|
|
mock_user.pw_name = 'test'
|
|
mock_getpwuid.return_value = mock_user
|
|
|
|
# Make group lookup fail
|
|
mock_getgrgid.side_effect = KeyError("Group not found")
|
|
|
|
diagnose_docker_environment()
|
|
|
|
|
|
class TestDockerClientEdgeCases:
|
|
"""Test docker client edge cases"""
|
|
|
|
@patch('docker.from_env')
|
|
@patch('docker.DockerClient')
|
|
def test_get_docker_client_unexpected_error(self, mock_docker_client, mock_from_env):
|
|
"""Test get_docker_client with unexpected error"""
|
|
from utils.docker_client import get_docker_client
|
|
|
|
# Make both methods raise unexpected errors
|
|
mock_from_env.side_effect = RuntimeError("Unexpected error")
|
|
mock_docker_client.side_effect = RuntimeError("Unexpected error")
|
|
|
|
with patch('utils.docker_client.diagnose_docker_environment'):
|
|
client = get_docker_client()
|
|
assert client is None
|
|
|
|
|
|
class TestExecHelpersEdgeCases:
|
|
"""Test exec helpers edge cases"""
|
|
|
|
def test_decode_output_empty(self):
|
|
"""Test decode_output with empty output"""
|
|
from utils.exec_helpers import decode_output
|
|
|
|
mock_exec = MagicMock()
|
|
mock_exec.output = None
|
|
|
|
result = decode_output(mock_exec)
|
|
assert result == ''
|
|
|
|
def test_decode_output_latin1_fallback(self):
|
|
"""Test decode_output falls back to latin-1"""
|
|
from utils.exec_helpers import decode_output
|
|
|
|
mock_exec = MagicMock()
|
|
# Create invalid UTF-8 that will force latin-1 fallback
|
|
mock_exec.output = bytes([0xff, 0xfe, 0xfd])
|
|
|
|
result = decode_output(mock_exec)
|
|
assert isinstance(result, str)
|
|
|
|
def test_extract_workdir_cd_command(self):
|
|
"""Test extract_workdir with cd command"""
|
|
from utils.exec_helpers import extract_workdir
|
|
|
|
output = "/home/user"
|
|
result_output, result_workdir = extract_workdir(output, "/app", True)
|
|
|
|
assert result_output == ''
|
|
assert result_workdir == "/home/user"
|
|
|
|
|
|
class TestTerminalHelpersEdgeCases:
|
|
"""Test terminal helpers edge cases"""
|
|
|
|
@patch('utils.terminal_helpers.threading.Thread')
|
|
def test_create_output_reader_unicode_decode_error(self, mock_thread):
|
|
"""Test output reader handles unicode decode errors"""
|
|
from utils.terminal_helpers import create_output_reader
|
|
from config import active_terminals
|
|
|
|
mock_socketio = MagicMock()
|
|
mock_sock = MagicMock()
|
|
|
|
# Return invalid UTF-8, then empty to end loop
|
|
mock_sock.recv.side_effect = [
|
|
bytes([0x80, 0x81]), # Invalid UTF-8
|
|
b'' # EOF
|
|
]
|
|
mock_sock.close = MagicMock()
|
|
|
|
mock_exec = MagicMock()
|
|
mock_exec.output = mock_sock
|
|
|
|
sid = 'unicode_test_sid'
|
|
active_terminals[sid] = {'exec': mock_exec}
|
|
|
|
# Get the actual thread function that would be called
|
|
def capture_thread_target(*args, **kwargs):
|
|
# Run the target function
|
|
kwargs['target']()
|
|
return MagicMock()
|
|
|
|
mock_thread.side_effect = capture_thread_target
|
|
|
|
create_output_reader(mock_socketio, sid, mock_exec)
|
|
|
|
# Should have emitted with latin-1 decoded data
|
|
assert mock_socketio.emit.called
|
|
|
|
@patch('utils.terminal_helpers.threading.Thread')
|
|
def test_create_output_reader_socket_recv_error(self, mock_thread):
|
|
"""Test output reader handles recv errors"""
|
|
from utils.terminal_helpers import create_output_reader
|
|
from config import active_terminals
|
|
|
|
mock_socketio = MagicMock()
|
|
mock_sock = MagicMock()
|
|
mock_sock.recv.side_effect = Exception("Socket error")
|
|
mock_sock.close = MagicMock()
|
|
|
|
mock_exec = MagicMock()
|
|
mock_exec.output = mock_sock
|
|
|
|
sid = 'socket_error_sid'
|
|
active_terminals[sid] = {'exec': mock_exec}
|
|
|
|
def capture_thread_target(*args, **kwargs):
|
|
kwargs['target']()
|
|
return MagicMock()
|
|
|
|
mock_thread.side_effect = capture_thread_target
|
|
|
|
create_output_reader(mock_socketio, sid, mock_exec)
|
|
|
|
# Should have cleaned up
|
|
assert sid not in active_terminals
|
|
|
|
@patch('utils.terminal_helpers.threading.Thread')
|
|
def test_create_output_reader_socket_close_error(self, mock_thread):
|
|
"""Test output reader handles close errors"""
|
|
from utils.terminal_helpers import create_output_reader
|
|
from config import active_terminals
|
|
|
|
mock_socketio = MagicMock()
|
|
mock_sock = MagicMock()
|
|
mock_sock.recv.return_value = b'' # EOF
|
|
mock_sock.close.side_effect = Exception("Close failed")
|
|
|
|
mock_exec = MagicMock()
|
|
mock_exec.output = mock_sock
|
|
|
|
sid = 'close_error_sid'
|
|
active_terminals[sid] = {'exec': mock_exec}
|
|
|
|
def capture_thread_target(*args, **kwargs):
|
|
kwargs['target']()
|
|
return MagicMock()
|
|
|
|
mock_thread.side_effect = capture_thread_target
|
|
|
|
# Should not raise exception
|
|
create_output_reader(mock_socketio, sid, mock_exec)
|