Fix test failures and thread context warnings

Fixed two issues:
1. test_terminal_sendall_with_container: Changed sock.recv() to sock._sock.recv() to use the correct SocketIO API
2. Thread context warnings: Captured request.sid before starting read_output thread to avoid "Working outside of request context" errors
3. test_input_with_direct_socket_fallback: Updated mock socket to block instead of returning empty immediately, which was causing premature thread cleanup

All 79 tests now pass with no warnings.

https://claude.ai/code/session_01DLxxKWp6dmtGD4ZUQrReTb
This commit is contained in:
Claude
2026-01-31 02:00:32 +00:00
parent 649c4dd2e7
commit aac0d5a509
3 changed files with 23 additions and 7 deletions

View File

@@ -514,13 +514,16 @@ def handle_start_terminal(data):
'container_id': container_id
}
# Capture request.sid before starting thread to avoid context issues
sid = request.sid
# Start a thread to read from the container and send to client
def read_output():
sock = exec_instance.output
try:
while True:
# Check if socket is still connected
if request.sid not in active_terminals:
if sid not in active_terminals:
break
try:
@@ -536,20 +539,20 @@ def handle_start_terminal(data):
decoded_data = data.decode('latin-1', errors='replace')
socketio.emit('output', {'data': decoded_data},
namespace='/terminal', room=request.sid)
namespace='/terminal', room=sid)
except Exception as e:
logger.error(f"Error reading from container: {e}")
break
finally:
# Clean up
if request.sid in active_terminals:
del active_terminals[request.sid]
if sid in active_terminals:
del active_terminals[sid]
try:
sock.close()
except:
pass
socketio.emit('exit', {'code': 0},
namespace='/terminal', room=request.sid)
namespace='/terminal', room=sid)
# Start the output reader thread
output_thread = threading.Thread(target=read_output, daemon=True)

View File

@@ -377,15 +377,24 @@ class TestWebSocketCoverage:
def test_input_with_direct_socket_fallback(self, mock_get_client, socketio_client, auth_token):
"""Test that input works with direct socket (no _sock attribute)"""
import app
import threading
mock_client = MagicMock()
mock_container = MagicMock()
mock_exec_instance = MagicMock()
# Create an event to control when the socket returns empty
stop_event = threading.Event()
def mock_recv(size):
# Block until stop_event is set, then return empty to exit thread
stop_event.wait(timeout=1.0)
return b''
# Create socket WITHOUT _sock attribute (direct socket)
mock_socket = MagicMock(spec=['sendall', 'recv', 'close'])
mock_socket.sendall = MagicMock()
mock_socket.recv = MagicMock(return_value=b'')
mock_socket.recv = MagicMock(side_effect=mock_recv)
mock_socket.close = MagicMock()
# Ensure it has NO _sock attribute
@@ -415,3 +424,7 @@ class TestWebSocketCoverage:
# Verify sendall was called on the socket itself (not _sock)
mock_socket.sendall.assert_called_with(b'echo test\n')
# Signal the thread to exit and clean up
stop_event.set()
time.sleep(0.1)

View File

@@ -57,7 +57,7 @@ class TestContainerSocketBehavior:
if not is_simulated:
# Only test actual output with real Docker
time.sleep(0.2)
output = sock.recv(4096)
output = sock._sock.recv(4096)
# Verify we got output without errors
assert output is not None