git/docker-swarm-termina
1
0
Fork 0
mirror of https://github.com/johndoe6345789/docker-swarm-termina.git synced 2026-04-24 13:45:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #22 from johndoe6345789/claude/screenshot-top-bar-buttons-bmFT7

Browse Source 
Refactor: Modularize Flask app with blueprints and utility modules
This commit is contained in:
johndoe6345789
2026-02-01 05:46:56 +00:00
committed by GitHub
parent 71ee74ed5f 4e928db0a8
commit c2c08fe157
50 changed files with 2043 additions and 722 deletions
Download Patch File Download Diff File Expand all files Collapse all files

633
backend/app.py
View File

@@ -1,26 +1,27 @@
from flask import Flask, jsonify, request
"""Main application entry point - refactored modular architecture."""
from flask import Flask
from flask_cors import CORS
from flask_socketio import SocketIO, emit, disconnect
import docker
import os
import sys
import logging
import threading
import select
from datetime import datetime, timedelta
from flask_socketio import SocketIO
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
from config import logger
from routes.login import login_bp
from routes.logout import logout_bp
from routes.health import health_bp
from routes.containers.list import list_bp
from routes.containers.exec import exec_bp
from routes.containers.start import start_bp
from routes.containers.stop import stop_bp
from routes.containers.restart import restart_bp
from routes.containers.remove import remove_bp
from handlers.terminal.register import register_terminal_handlers
from utils.diagnostics.docker_env import diagnose_docker_environment
from utils.docker_client import get_docker_client
# Initialize Flask app
app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": "*"}})
# Initialize SocketIO
socketio = SocketIO(
app,
cors_allowed_origins="*",
@@ -31,590 +32,20 @@ socketio = SocketIO(
engineio_logger=True
)
# Simple in-memory session storage (in production, use proper session management)
sessions = {}
# Track working directory per session
session_workdirs = {}
# Register blueprints
app.register_blueprint(login_bp)
app.register_blueprint(logout_bp)
app.register_blueprint(health_bp)
app.register_blueprint(list_bp)
app.register_blueprint(exec_bp)
app.register_blueprint(start_bp)
app.register_blueprint(stop_bp)
app.register_blueprint(restart_bp)
app.register_blueprint(remove_bp)
# Register WebSocket handlers
register_terminal_handlers(socketio)
# Default credentials (should be environment variables in production)
ADMIN_USERNAME = os.getenv('ADMIN_USERNAME', 'admin')
ADMIN_PASSWORD = os.getenv('ADMIN_PASSWORD', 'admin123')
def diagnose_docker_environment():
"""Diagnose Docker environment and configuration"""
logger.info("=== Docker Environment Diagnosis ===")
# Check environment variables
docker_host = os.getenv('DOCKER_HOST', 'Not set')
docker_cert_path = os.getenv('DOCKER_CERT_PATH', 'Not set')
docker_tls_verify = os.getenv('DOCKER_TLS_VERIFY', 'Not set')
logger.info(f"DOCKER_HOST: {docker_host}")
logger.info(f"DOCKER_CERT_PATH: {docker_cert_path}")
logger.info(f"DOCKER_TLS_VERIFY: {docker_tls_verify}")
# Check what's in /var/run
logger.info("Checking /var/run directory contents:")
try:
if os.path.exists('/var/run'):
var_run_contents = os.listdir('/var/run')
logger.info(f" /var/run contains: {var_run_contents}")
# Check for any Docker-related files
docker_related = [f for f in var_run_contents if 'docker' in f.lower()]
if docker_related:
logger.info(f" Docker-related files/dirs found: {docker_related}")
else:
logger.warning(" /var/run directory doesn't exist")
except Exception as e:
logger.error(f" Error reading /var/run: {e}")
# Check Docker socket
socket_path = '/var/run/docker.sock'
logger.info(f"Checking Docker socket at {socket_path}")
if os.path.exists(socket_path):
logger.info(f"✓ Docker socket exists at {socket_path}")
# Check permissions
import stat
st = os.stat(socket_path)
logger.info(f" Socket permissions: {oct(st.st_mode)}")
logger.info(f" Socket owner UID: {st.st_uid}")
logger.info(f" Socket owner GID: {st.st_gid}")
# Check if readable/writable
readable = os.access(socket_path, os.R_OK)
writable = os.access(socket_path, os.W_OK)
logger.info(f" Readable: {readable}")
logger.info(f" Writable: {writable}")
if not (readable and writable):
logger.warning(f"⚠ Socket exists but lacks proper permissions!")
else:
logger.error(f"✗ Docker socket NOT found at {socket_path}")
logger.error(f" This means the Docker socket mount is NOT configured in CapRover")
logger.error(f" The serviceUpdateOverride in captain-definition may not be applied")
# Check current user
import pwd
try:
current_uid = os.getuid()
current_gid = os.getgid()
user_info = pwd.getpwuid(current_uid)
logger.info(f"Current user: {user_info.pw_name} (UID: {current_uid}, GID: {current_gid})")
# Check groups
import grp
groups = os.getgroups()
logger.info(f"User groups (GIDs): {groups}")
for gid in groups:
try:
group_info = grp.getgrgid(gid)
logger.info(f" - {group_info.gr_name} (GID: {gid})")
except:
logger.info(f" - Unknown group (GID: {gid})")
except Exception as e:
logger.error(f"Error checking user info: {e}")
logger.info("=== End Diagnosis ===")
def get_docker_client():
"""Get Docker client with enhanced error reporting"""
try:
logger.info("Attempting to connect to Docker...")
# Try default connection first
try:
client = docker.from_env()
# Test the connection
client.ping()
logger.info("✓ Successfully connected to Docker using docker.from_env()")
return client
except Exception as e:
logger.warning(f"docker.from_env() failed: {e}")
# Try explicit Unix socket connection
try:
logger.info("Trying explicit Unix socket connection...")
client = docker.DockerClient(base_url='unix:///var/run/docker.sock')
client.ping()
logger.info("✓ Successfully connected to Docker using Unix socket")
return client
except Exception as e:
logger.warning(f"Unix socket connection failed: {e}")
# If all fails, run diagnostics and return None
logger.error("All Docker connection attempts failed!")
diagnose_docker_environment()
return None
except Exception as e:
logger.error(f"Unexpected error in get_docker_client: {e}", exc_info=True)
return None
def format_uptime(created_at):
"""Format container uptime"""
created = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
now = datetime.now(created.tzinfo)
delta = now - created
days = delta.days
hours = delta.seconds // 3600
minutes = (delta.seconds % 3600) // 60
if days > 0:
return f"{days}d {hours}h"
elif hours > 0:
return f"{hours}h {minutes}m"
else:
return f"{minutes}m"
@app.route('/api/auth/login', methods=['POST'])
def login():
"""Authenticate user"""
data = request.get_json()
username = data.get('username')
password = data.get('password')
if username == ADMIN_USERNAME and password == ADMIN_PASSWORD:
# Create a simple session token (in production, use JWT or proper session management)
session_token = f"session_{username}_{datetime.now().timestamp()}"
sessions[session_token] = {
'username': username,
'created_at': datetime.now()
}
return jsonify({
'success': True,
'token': session_token,
'username': username
})
return jsonify({
'success': False,
'message': 'Invalid credentials'
}), 401
@app.route('/api/auth/logout', methods=['POST'])
def logout():
"""Logout user"""
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
if token in sessions:
del sessions[token]
return jsonify({'success': True})
@app.route('/api/containers', methods=['GET'])
def get_containers():
"""Get list of all containers"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
containers = client.containers.list(all=True)
container_list = []
for container in containers:
container_list.append({
'id': container.short_id,
'name': container.name,
'image': container.image.tags[0] if container.image.tags else 'unknown',
'status': container.status,
'uptime': format_uptime(container.attrs['Created']) if container.status == 'running' else 'N/A'
})
return jsonify({'containers': container_list})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/containers/<container_id>/exec', methods=['POST'])
def exec_container(container_id):
"""Execute command in container"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
data = request.get_json()
user_command = data.get('command', 'echo "No command provided"')
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
container = client.containers.get(container_id)
# Get or initialize session working directory
session_key = f"{token}_{container_id}"
if session_key not in session_workdirs:
# Get container's default working directory or use root
session_workdirs[session_key] = '/'
current_workdir = session_workdirs[session_key]
# Check if this is a cd command
cd_match = user_command.strip()
is_cd_command = cd_match.startswith('cd ')
# If it's a cd command, handle it specially
if is_cd_command:
target_dir = cd_match[3:].strip() or '~'
# Resolve the new directory and update session
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
bash_command = [
'/bin/bash',
'-c',
f'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin; {resolve_command}'
]
else:
# Regular command - execute in current working directory
bash_command = [
'/bin/bash',
'-c',
f'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"'
]
# Try bash first, fallback to sh if bash doesn't exist
try:
exec_instance = container.exec_run(
bash_command,
stdout=True,
stderr=True,
stdin=False,
tty=True,
environment={'TERM': 'xterm-256color', 'LANG': 'C.UTF-8'}
)
except Exception as bash_error:
logger.warning(f"Bash execution failed, trying sh: {bash_error}")
# Fallback to sh
if is_cd_command:
target_dir = cd_match[3:].strip() or '~'
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
sh_command = ['/bin/sh', '-c', f'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin; {resolve_command}']
else:
sh_command = ['/bin/sh', '-c', f'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"']
exec_instance = container.exec_run(
sh_command,
stdout=True,
stderr=True,
stdin=False,
tty=True,
environment={'TERM': 'xterm-256color', 'LANG': 'C.UTF-8'}
)
# Decode output with error handling
output = ''
if exec_instance.output:
try:
output = exec_instance.output.decode('utf-8')
except UnicodeDecodeError:
# Try latin-1 as fallback
output = exec_instance.output.decode('latin-1', errors='replace')
# Extract and update working directory from output
new_workdir = current_workdir
if is_cd_command:
# For cd commands, the output is the new pwd
new_workdir = output.strip()
session_workdirs[session_key] = new_workdir
output = '' # Don't show the pwd output for cd
else:
# Extract workdir marker from output
if '::WORKDIR::' in output:
parts = output.rsplit('::WORKDIR::', 1)
output = parts[0]
new_workdir = parts[1].strip()
session_workdirs[session_key] = new_workdir
return jsonify({
'output': output,
'exit_code': exec_instance.exit_code,
'workdir': new_workdir
})
except Exception as e:
logger.error(f"Error executing command: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/api/containers/<container_id>/start', methods=['POST'])
def start_container(container_id):
"""Start a stopped container"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
container = client.containers.get(container_id)
container.start()
logger.info(f"Started container {container_id}")
return jsonify({'success': True, 'message': f'Container {container_id} started'})
except Exception as e:
logger.error(f"Error starting container: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/api/containers/<container_id>/stop', methods=['POST'])
def stop_container(container_id):
"""Stop a running container"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
container = client.containers.get(container_id)
container.stop()
logger.info(f"Stopped container {container_id}")
return jsonify({'success': True, 'message': f'Container {container_id} stopped'})
except Exception as e:
logger.error(f"Error stopping container: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/api/containers/<container_id>/restart', methods=['POST'])
def restart_container(container_id):
"""Restart a container"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
container = client.containers.get(container_id)
container.restart()
logger.info(f"Restarted container {container_id}")
return jsonify({'success': True, 'message': f'Container {container_id} restarted'})
except Exception as e:
logger.error(f"Error restarting container: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/api/containers/<container_id>', methods=['DELETE'])
def remove_container(container_id):
"""Remove a container"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Unauthorized'}), 401
token = auth_header.split(' ')[1]
if token not in sessions:
return jsonify({'error': 'Invalid session'}), 401
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
container = client.containers.get(container_id)
# Force remove (including if running)
container.remove(force=True)
logger.info(f"Removed container {container_id}")
return jsonify({'success': True, 'message': f'Container {container_id} removed'})
except Exception as e:
logger.error(f"Error removing container: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
@app.route('/api/health', methods=['GET'])
def health():
"""Health check endpoint"""
return jsonify({'status': 'healthy'})
# WebSocket handlers for interactive terminal
active_terminals = {}
@socketio.on('connect', namespace='/terminal')
def handle_connect():
"""Handle WebSocket connection"""
logger.info(f"Client connected to terminal WebSocket: {request.sid}")
@socketio.on('disconnect', namespace='/terminal')
def handle_disconnect():
"""Handle WebSocket disconnection"""
logger.info(f"Client disconnected from terminal WebSocket: {request.sid}")
# Clean up any active terminal sessions
if request.sid in active_terminals:
try:
exec_instance = active_terminals[request.sid]['exec']
# Try to stop the exec instance
if hasattr(exec_instance, 'kill'):
exec_instance.kill()
except:
pass
del active_terminals[request.sid]
@socketio.on('start_terminal', namespace='/terminal')
def handle_start_terminal(data):
"""Start an interactive terminal session"""
try:
container_id = data.get('container_id')
token = data.get('token')
cols = data.get('cols', 80)
rows = data.get('rows', 24)
# Validate token
if not token or token not in sessions:
emit('error', {'error': 'Unauthorized'})
disconnect()
return
# Get Docker client and container
client = get_docker_client()
if not client:
emit('error', {'error': 'Cannot connect to Docker'})
return
container = client.containers.get(container_id)
# Create an interactive bash session with PTY
exec_instance = container.exec_run(
['/bin/bash'],
stdin=True,
stdout=True,
stderr=True,
tty=True,
socket=True,
environment={
'TERM': 'xterm-256color',
'COLUMNS': str(cols),
'LINES': str(rows),
'LANG': 'C.UTF-8'
}
)
# Store the exec instance
active_terminals[request.sid] = {
'exec': exec_instance,
'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 sid not in active_terminals:
break
try:
# Read data from container
data = sock.recv(4096)
if not data:
break
# Send to client
try:
decoded_data = data.decode('utf-8')
except UnicodeDecodeError:
decoded_data = data.decode('latin-1', errors='replace')
socketio.emit('output', {'data': decoded_data},
namespace='/terminal', room=sid)
except Exception as e:
logger.error(f"Error reading from container: {e}")
break
finally:
# Clean up
if sid in active_terminals:
del active_terminals[sid]
try:
sock.close()
except:
pass
socketio.emit('exit', {'code': 0},
namespace='/terminal', room=sid)
# Start the output reader thread
output_thread = threading.Thread(target=read_output, daemon=True)
output_thread.start()
emit('started', {'message': 'Terminal started'})
except Exception as e:
logger.error(f"Error starting terminal: {e}", exc_info=True)
emit('error', {'error': str(e)})
@socketio.on('input', namespace='/terminal')
def handle_input(data):
"""Handle input from the client"""
try:
if request.sid not in active_terminals:
emit('error', {'error': 'No active terminal session'})
return
terminal_data = active_terminals[request.sid]
exec_instance = terminal_data['exec']
input_data = data.get('data', '')
# Send input to the container
sock = exec_instance.output
# Access the underlying socket for sendall method
if hasattr(sock, '_sock'):
sock._sock.sendall(input_data.encode('utf-8'))
else:
# Fallback for direct socket objects
sock.sendall(input_data.encode('utf-8'))
except Exception as e:
logger.error(f"Error sending input: {e}", exc_info=True)
emit('error', {'error': str(e)})
@socketio.on('resize', namespace='/terminal')
def handle_resize(data):
"""Handle terminal resize"""
try:
cols = data.get('cols', 80)
rows = data.get('rows', 24)
if request.sid in active_terminals:
terminal_data = active_terminals[request.sid]
exec_instance = terminal_data['exec']
# Note: Docker exec_run doesn't support resizing after creation
# This is a limitation of the Docker API
# We acknowledge the resize but can't actually resize the PTY
logger.info(f"Terminal resize requested: {cols}x{rows}")
except Exception as e:
logger.error(f"Error resizing terminal: {e}", exc_info=True)
if __name__ == '__main__':
# Run diagnostics on startup

28
backend/config.py Normal file
View File

@@ -0,0 +1,28 @@
"""Application configuration and constants."""
import os
import sys
import logging
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# Default credentials (should be environment variables in production)
ADMIN_USERNAME = os.getenv('ADMIN_USERNAME', 'admin')
ADMIN_PASSWORD = os.getenv('ADMIN_PASSWORD', 'admin123')
# Simple in-memory session storage (in production, use proper session management)
sessions = {}
# Track working directory per session
session_workdirs = {}
# Active terminal sessions
active_terminals = {}

1
backend/handlers/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Socket.io handlers - one file per event."""

1
backend/handlers/terminal/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Terminal WebSocket handlers."""

8
backend/handlers/terminal/connect.py Normal file
View File

@@ -0,0 +1,8 @@
"""Terminal WebSocket connect handler."""
from flask import request
from config import logger
def handle_connect():
"""Handle WebSocket connection."""
logger.info("Client connected to terminal WebSocket: %s", request.sid)

17
backend/handlers/terminal/disconnect.py Normal file
View File

@@ -0,0 +1,17 @@
"""Terminal WebSocket disconnect handler."""
from flask import request
from config import logger, active_terminals
def handle_disconnect():
"""Handle WebSocket disconnection."""
logger.info("Client disconnected from terminal WebSocket: %s", request.sid)
# Clean up any active terminal sessions
if request.sid in active_terminals:
try:
exec_instance = active_terminals[request.sid]['exec']
if hasattr(exec_instance, 'kill'):
exec_instance.kill()
except Exception: # pylint: disable=broad-exception-caught
pass
del active_terminals[request.sid]

32
backend/handlers/terminal/input.py Normal file
View File

@@ -0,0 +1,32 @@
"""Terminal WebSocket input handler."""
from flask import request
from flask_socketio import emit
from config import logger, active_terminals
def handle_input(data):
"""Handle input from the client.
Args:
data: Input data containing the user's input string
"""
try:
if request.sid not in active_terminals:
emit('error', {'error': 'No active terminal session'})
return
terminal_data = active_terminals[request.sid]
exec_instance = terminal_data['exec']
input_data = data.get('data', '')
# Send input to the container
sock = exec_instance.output
# Access the underlying socket for sendall method
if hasattr(sock, '_sock'):
sock._sock.sendall(input_data.encode('utf-8')) # pylint: disable=protected-access
else:
sock.sendall(input_data.encode('utf-8'))
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error sending input: %s", e, exc_info=True)
emit('error', {'error': str(e)})

33
backend/handlers/terminal/register.py Normal file
View File

@@ -0,0 +1,33 @@
"""Register all terminal WebSocket handlers."""
from handlers.terminal.connect import handle_connect
from handlers.terminal.disconnect import handle_disconnect
from handlers.terminal.start import handle_start_terminal
from handlers.terminal.input import handle_input
from handlers.terminal.resize import handle_resize
def register_terminal_handlers(socketio):
"""Register all terminal WebSocket event handlers.
Args:
socketio: SocketIO instance to register handlers with
"""
@socketio.on('connect', namespace='/terminal')
def on_connect():
return handle_connect()
@socketio.on('disconnect', namespace='/terminal')
def on_disconnect():
return handle_disconnect()
@socketio.on('start_terminal', namespace='/terminal')
def on_start_terminal(data):
return handle_start_terminal(socketio, data)
@socketio.on('input', namespace='/terminal')
def on_input(data):
return handle_input(data)
@socketio.on('resize', namespace='/terminal')
def on_resize(data):
return handle_resize(data)

24
backend/handlers/terminal/resize.py Normal file
View File

@@ -0,0 +1,24 @@
"""Terminal WebSocket resize handler."""
from flask import request
from config import logger, active_terminals
def handle_resize(data):
"""Handle terminal resize.
Args:
data: Resize data containing cols and rows
Note:
Docker exec_run doesn't support resizing after creation.
This is a limitation of the Docker API.
"""
try:
cols = data.get('cols', 80)
rows = data.get('rows', 24)
if request.sid in active_terminals:
logger.info("Terminal resize requested: %sx%s", cols, rows)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error resizing terminal: %s", e, exc_info=True)

66
backend/handlers/terminal/start.py Normal file
View File

@@ -0,0 +1,66 @@
"""Terminal WebSocket start handler."""
# pylint: disable=duplicate-code # Auth/client setup pattern is intentional
from flask import request
from flask_socketio import emit, disconnect
from config import logger, sessions, active_terminals
from utils.docker_client import get_docker_client
from utils.terminal_helpers import create_output_reader
def handle_start_terminal(socketio, data):
"""Start an interactive terminal session.
Args:
socketio: SocketIO instance
data: Request data containing container_id, token, cols, rows
"""
try:
container_id = data.get('container_id')
token = data.get('token')
cols = data.get('cols', 80)
rows = data.get('rows', 24)
# Validate token
if not token or token not in sessions:
emit('error', {'error': 'Unauthorized'})
disconnect()
return
# Get Docker client and container
client = get_docker_client()
if not client:
emit('error', {'error': 'Cannot connect to Docker'})
return
container = client.containers.get(container_id)
# Create an interactive bash session with PTY
exec_instance = container.exec_run(
['/bin/bash'],
stdin=True,
stdout=True,
stderr=True,
tty=True,
socket=True,
environment={
'TERM': 'xterm-256color',
'COLUMNS': str(cols),
'LINES': str(rows),
'LANG': 'C.UTF-8'
}
)
# Store the exec instance
active_terminals[request.sid] = {
'exec': exec_instance,
'container_id': container_id
}
# Start output reader thread
create_output_reader(socketio, request.sid, exec_instance)
emit('started', {'message': 'Terminal started'})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error starting terminal: %s", e, exc_info=True)
emit('error', {'error': str(e)})

1
backend/routes/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""API routes - one file per endpoint for clarity."""

1
backend/routes/containers/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Container management routes - one file per endpoint."""

59
backend/routes/containers/exec.py Normal file
View File

@@ -0,0 +1,59 @@
"""Execute command in container route."""
from flask import Blueprint, request, jsonify
from config import logger, session_workdirs
from utils.auth import check_auth
from utils.docker_client import get_docker_client
from utils.exec_helpers import (
get_session_workdir,
execute_command_with_fallback,
decode_output,
extract_workdir
)
exec_bp = Blueprint('exec_container', __name__)
@exec_bp.route('/api/containers/<container_id>/exec', methods=['POST'])
def exec_container(container_id):
"""Execute command in container."""
is_valid, token, error_response = check_auth()
if not is_valid:
return error_response
data = request.get_json()
user_command = data.get('command', 'echo "No command provided"')
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
# Get session working directory
session_key, current_workdir = get_session_workdir(token, container_id, session_workdirs)
# Execute command with bash/sh fallback
exec_instance = execute_command_with_fallback(
client.containers.get(container_id),
current_workdir,
user_command,
user_command.strip().startswith('cd ')
)
# Decode and extract workdir from output
output, new_workdir = extract_workdir(
decode_output(exec_instance),
current_workdir,
user_command.strip().startswith('cd ')
)
# Update session workdir
session_workdirs[session_key] = new_workdir
return jsonify({
'output': output,
'exit_code': exec_instance.exit_code,
'workdir': new_workdir
})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error executing command: %s", e, exc_info=True)
return jsonify({'error': str(e)}), 500

37
backend/routes/containers/list.py Normal file
View File

@@ -0,0 +1,37 @@
"""List containers route."""
from flask import Blueprint, jsonify
from utils.auth import check_auth
from utils.docker_client import get_docker_client
from utils.formatters import format_uptime
list_bp = Blueprint('list_containers', __name__)
@list_bp.route('/api/containers', methods=['GET'])
def get_containers():
"""Get list of all containers."""
is_valid, _, error_response = check_auth()
if not is_valid:
return error_response
client = get_docker_client()
if not client:
return jsonify({'error': 'Cannot connect to Docker'}), 500
try:
containers = client.containers.list(all=True)
container_list = []
for container in containers:
container_list.append({
'id': container.short_id,
'name': container.name,
'image': container.image.tags[0] if container.image.tags else 'unknown',
'status': container.status,
'uptime': format_uptime(container.attrs['Created'])
if container.status == 'running' else 'N/A'
})
return jsonify({'containers': container_list})
except Exception as e: # pylint: disable=broad-exception-caught
return jsonify({'error': str(e)}), 500

22
backend/routes/containers/remove.py Normal file
View File

@@ -0,0 +1,22 @@
"""Remove container route."""
from flask import Blueprint, jsonify
from config import logger
from utils.container_helpers import get_auth_and_container
remove_bp = Blueprint('remove_container', __name__)
@remove_bp.route('/api/containers/<container_id>', methods=['DELETE'])
def remove_container(container_id):
"""Remove a container."""
container, error_response = get_auth_and_container(container_id)
if error_response:
return error_response
try:
container.remove(force=True)
logger.info("Removed container %s", container_id)
return jsonify({'success': True, 'message': f'Container {container_id} removed'})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error removing container: %s", e, exc_info=True)
return jsonify({'error': str(e)}), 500

22
backend/routes/containers/restart.py Normal file
View File

@@ -0,0 +1,22 @@
"""Restart container route."""
from flask import Blueprint, jsonify
from config import logger
from utils.container_helpers import get_auth_and_container
restart_bp = Blueprint('restart_container', __name__)
@restart_bp.route('/api/containers/<container_id>/restart', methods=['POST'])
def restart_container(container_id):
"""Restart a container."""
container, error_response = get_auth_and_container(container_id)
if error_response:
return error_response
try:
container.restart()
logger.info("Restarted container %s", container_id)
return jsonify({'success': True, 'message': f'Container {container_id} restarted'})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error restarting container: %s", e, exc_info=True)
return jsonify({'error': str(e)}), 500

22
backend/routes/containers/start.py Normal file
View File

@@ -0,0 +1,22 @@
"""Start container route."""
from flask import Blueprint, jsonify
from config import logger
from utils.container_helpers import get_auth_and_container
start_bp = Blueprint('start_container', __name__)
@start_bp.route('/api/containers/<container_id>/start', methods=['POST'])
def start_container(container_id):
"""Start a stopped container."""
container, error_response = get_auth_and_container(container_id)
if error_response:
return error_response
try:
container.start()
logger.info("Started container %s", container_id)
return jsonify({'success': True, 'message': f'Container {container_id} started'})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error starting container: %s", e, exc_info=True)
return jsonify({'error': str(e)}), 500

22
backend/routes/containers/stop.py Normal file
View File

@@ -0,0 +1,22 @@
"""Stop container route."""
from flask import Blueprint, jsonify
from config import logger
from utils.container_helpers import get_auth_and_container
stop_bp = Blueprint('stop_container', __name__)
@stop_bp.route('/api/containers/<container_id>/stop', methods=['POST'])
def stop_container(container_id):
"""Stop a running container."""
container, error_response = get_auth_and_container(container_id)
if error_response:
return error_response
try:
container.stop()
logger.info("Stopped container %s", container_id)
return jsonify({'success': True, 'message': f'Container {container_id} stopped'})
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error stopping container: %s", e, exc_info=True)
return jsonify({'error': str(e)}), 500

10
backend/routes/health.py Normal file
View File

@@ -0,0 +1,10 @@
"""Health check route."""
from flask import Blueprint, jsonify
health_bp = Blueprint('health', __name__)
@health_bp.route('/api/health', methods=['GET'])
def health():
"""Health check endpoint."""
return jsonify({'status': 'healthy'})

31
backend/routes/login.py Normal file
View File

@@ -0,0 +1,31 @@
"""Login route."""
from datetime import datetime
from flask import Blueprint, request, jsonify
from config import ADMIN_USERNAME, ADMIN_PASSWORD, sessions
login_bp = Blueprint('login', __name__)
@login_bp.route('/api/auth/login', methods=['POST'])
def login():
"""Authenticate user."""
data = request.get_json()
username = data.get('username')
password = data.get('password')
if username == ADMIN_USERNAME and password == ADMIN_PASSWORD:
session_token = f"session_{username}_{datetime.now().timestamp()}"
sessions[session_token] = {
'username': username,
'created_at': datetime.now()
}
return jsonify({
'success': True,
'token': session_token,
'username': username
})
return jsonify({
'success': False,
'message': 'Invalid credentials'
}), 401

17
backend/routes/logout.py Normal file
View File

@@ -0,0 +1,17 @@
"""Logout route."""
from flask import Blueprint, request, jsonify
from config import sessions
logout_bp = Blueprint('logout', __name__)
@logout_bp.route('/api/auth/logout', methods=['POST'])
def logout():
"""Logout user."""
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
if token in sessions:
del sessions[token]
return jsonify({'success': True})

378
backend/tests/test_complete_coverage.py Normal file
View File

@@ -0,0 +1,378 @@
"""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)

12
backend/tests/test_containers.py
View File

@@ -21,7 +21,7 @@ class TestContainerEndpoints:
data = response.get_json()
assert 'error' in data
@patch('app.get_docker_client')
@patch('routes.containers.list.get_docker_client')
def test_get_containers_success(self, mock_get_client, client, auth_headers):
"""Test getting containers successfully"""
# Mock Docker client
@@ -44,7 +44,7 @@ class TestContainerEndpoints:
assert data['containers'][0]['id'] == 'abc123'
assert data['containers'][0]['name'] == 'test-container'
@patch('app.get_docker_client')
@patch('routes.containers.list.get_docker_client')
def test_get_containers_docker_unavailable(self, mock_get_client, client, auth_headers):
"""Test getting containers when Docker is unavailable"""
mock_get_client.return_value = None
@@ -54,7 +54,7 @@ class TestContainerEndpoints:
data = response.get_json()
assert 'error' in data
@patch('app.get_docker_client')
@patch('utils.container_helpers.get_docker_client')
def test_start_container_success(self, mock_get_client, client, auth_headers):
"""Test starting a container"""
mock_container = MagicMock()
@@ -68,7 +68,7 @@ class TestContainerEndpoints:
assert data['success'] is True
mock_container.start.assert_called_once()
@patch('app.get_docker_client')
@patch('utils.container_helpers.get_docker_client')
def test_stop_container_success(self, mock_get_client, client, auth_headers):
"""Test stopping a container"""
mock_container = MagicMock()
@@ -82,7 +82,7 @@ class TestContainerEndpoints:
assert data['success'] is True
mock_container.stop.assert_called_once()
@patch('app.get_docker_client')
@patch('utils.container_helpers.get_docker_client')
def test_restart_container_success(self, mock_get_client, client, auth_headers):
"""Test restarting a container"""
mock_container = MagicMock()
@@ -96,7 +96,7 @@ class TestContainerEndpoints:
assert data['success'] is True
mock_container.restart.assert_called_once()
@patch('app.get_docker_client')
@patch('utils.container_helpers.get_docker_client')
def test_remove_container_success(self, mock_get_client, client, auth_headers):
"""Test removing a container"""
mock_container = MagicMock()

156
backend/tests/test_coverage_boost.py Normal file
View File

@@ -0,0 +1,156 @@
"""Tests to boost coverage to 100%."""
import pytest
from unittest.mock import MagicMock, patch, Mock
from flask import jsonify
class TestContainerExceptionHandling:
"""Test exception handling in container routes"""
@patch('utils.container_helpers.get_docker_client')
def test_start_container_exception(self, mock_get_client, client, auth_headers):
"""Test start container with exception"""
mock_container = MagicMock()
mock_container.start.side_effect = Exception("Container failed to start")
mock_client = MagicMock()
mock_client.containers.get.return_value = mock_container
mock_get_client.return_value = mock_client
response = client.post('/api/containers/test123/start', headers=auth_headers)
assert response.status_code == 500
data = response.get_json()
assert 'error' in data
@patch('utils.container_helpers.get_docker_client')
def test_stop_container_exception(self, mock_get_client, client, auth_headers):
"""Test stop container with exception"""
mock_container = MagicMock()
mock_container.stop.side_effect = Exception("Container failed to stop")
mock_client = MagicMock()
mock_client.containers.get.return_value = mock_container
mock_get_client.return_value = mock_client
response = client.post('/api/containers/test123/stop', headers=auth_headers)
assert response.status_code == 500
data = response.get_json()
assert 'error' in data
@patch('utils.container_helpers.get_docker_client')
def test_restart_container_exception(self, mock_get_client, client, auth_headers):
"""Test restart container with exception"""
mock_container = MagicMock()
mock_container.restart.side_effect = Exception("Container failed to restart")
mock_client = MagicMock()
mock_client.containers.get.return_value = mock_container
mock_get_client.return_value = mock_client
response = client.post('/api/containers/test123/restart', headers=auth_headers)
assert response.status_code == 500
data = response.get_json()
assert 'error' in data
@patch('utils.container_helpers.get_docker_client')
def test_remove_container_exception(self, mock_get_client, client, auth_headers):
"""Test remove container with exception"""
mock_container = MagicMock()
mock_container.remove.side_effect = Exception("Container failed to remove")
mock_client = MagicMock()
mock_client.containers.get.return_value = mock_container
mock_get_client.return_value = mock_client
response = client.delete('/api/containers/test123', headers=auth_headers)
assert response.status_code == 500
data = response.get_json()
assert 'error' in data
@patch('routes.containers.list.get_docker_client')
def test_list_containers_exception(self, mock_get_client, client, auth_headers):
"""Test list containers with exception"""
mock_client = MagicMock()
mock_client.containers.list.side_effect = Exception("Failed to list containers")
mock_get_client.return_value = mock_client
response = client.get('/api/containers', headers=auth_headers)
assert response.status_code == 500
data = response.get_json()
assert 'error' in data
class TestContainerHelpers:
"""Test container_helpers exception handling"""
@patch('utils.container_helpers.get_docker_client')
def test_get_auth_and_container_exception(self, mock_get_client):
"""Test get_auth_and_container when container.get raises exception"""
from utils.container_helpers import get_auth_and_container
from config import sessions
# Create a valid session
token = 'test_token_123'
sessions[token] = {'username': 'test'}
# Mock client that raises exception
mock_client = MagicMock()
mock_client.containers.get.side_effect = Exception("Container not found")
mock_get_client.return_value = mock_client
# This test needs to be called in request context
from flask import Flask
app = Flask(__name__)
with app.test_request_context(headers={'Authorization': f'Bearer {token}'}):
container, error = get_auth_and_container('test123')
assert container is None
assert error is not None
assert error[1] == 500
class TestExecHelpers:
"""Test exec_helpers edge cases"""
def test_decode_output_unicode_error(self):
"""Test decode_output with invalid UTF-8"""
from utils.exec_helpers import decode_output
mock_exec = MagicMock()
# Invalid UTF-8 sequence
mock_exec.output = b'\x80\x81\x82\x83'
result = decode_output(mock_exec)
# Should fallback to latin-1
assert result is not None
assert isinstance(result, str)
def test_extract_workdir_no_marker(self):
"""Test extract_workdir when no marker present"""
from utils.exec_helpers import extract_workdir
output = "some command output"
current_workdir = "/test"
result_output, result_workdir = extract_workdir(output, current_workdir, False)
assert result_output == output
assert result_workdir == current_workdir
def test_execute_command_bash_fallback(self):
"""Test execute_command_with_fallback when bash fails"""
from utils.exec_helpers import execute_command_with_fallback
mock_container = MagicMock()
# Make bash fail, sh succeed
mock_container.exec_run.side_effect = [
Exception("bash not found"),
MagicMock(output=b'success', exit_code=0)
]
result = execute_command_with_fallback(
mock_container, '/app', 'ls', False
)
assert result.exit_code == 0
assert mock_container.exec_run.call_count == 2

12
backend/tests/test_docker_client.py
View File

@@ -9,7 +9,7 @@ class TestDockerClient:
@patch('docker.from_env')
def test_get_docker_client_success(self, mock_from_env):
"""Test successful Docker client connection"""
from app import get_docker_client
from utils.docker_client import get_docker_client
mock_client = MagicMock()
mock_client.ping.return_value = True
@@ -23,7 +23,7 @@ class TestDockerClient:
@patch('docker.from_env')
def test_get_docker_client_fallback_to_socket(self, mock_from_env, mock_docker_client):
"""Test fallback to Unix socket when from_env fails"""
from app import get_docker_client
from utils.docker_client import get_docker_client
# Make from_env fail
mock_from_env.side_effect = Exception("Connection failed")
@@ -41,7 +41,7 @@ class TestDockerClient:
@patch('docker.from_env')
def test_get_docker_client_all_methods_fail(self, mock_from_env, mock_docker_client):
"""Test when all Docker connection methods fail"""
from app import get_docker_client
from utils.docker_client import get_docker_client
# Make both methods fail
mock_from_env.side_effect = Exception("from_env failed")
@@ -56,7 +56,7 @@ class TestFormatUptime:
def test_format_uptime_zero_minutes(self):
"""Test formatting for containers just started"""
from app import format_uptime
from utils.formatters import format_uptime
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)
@@ -69,7 +69,7 @@ class TestFormatUptime:
def test_format_uptime_exactly_one_day(self):
"""Test formatting for exactly 1 day"""
from app import format_uptime
from utils.formatters import format_uptime
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)
@@ -81,7 +81,7 @@ class TestFormatUptime:
def test_format_uptime_many_days(self):
"""Test formatting for many days"""
from app import format_uptime
from utils.formatters import format_uptime
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)

14
backend/tests/test_edge_cases.py
View File

@@ -26,7 +26,7 @@ class TestEdgeCases:
})
assert response.status_code in [200, 401]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_containers_with_docker_error(self, mock_get_client, client, auth_headers):
"""Test containers endpoint when Docker returns unexpected error"""
mock_client = MagicMock()
@@ -38,7 +38,7 @@ class TestEdgeCases:
# Should return 500 or handle error
assert response.status_code in [500, 200]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_exec_with_missing_fields(self, mock_get_client, client, auth_headers):
"""Test exec with missing command field"""
mock_get_client.return_value = MagicMock()
@@ -50,7 +50,7 @@ class TestEdgeCases:
# Should return 400 or handle error
assert response.status_code in [400, 500]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_start_container_not_found(self, mock_get_client, client, auth_headers):
"""Test starting non-existent container"""
from docker.errors import NotFound
@@ -64,7 +64,7 @@ class TestEdgeCases:
assert response.status_code in [404, 500]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_stop_container_error(self, mock_get_client, client, auth_headers):
"""Test stopping container with error"""
mock_client = MagicMock()
@@ -78,7 +78,7 @@ class TestEdgeCases:
assert response.status_code in [500, 200]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_restart_container_error(self, mock_get_client, client, auth_headers):
"""Test restarting container with error"""
mock_client = MagicMock()
@@ -92,7 +92,7 @@ class TestEdgeCases:
assert response.status_code in [500, 200]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_remove_container_error(self, mock_get_client, client, auth_headers):
"""Test removing container with error"""
mock_client = MagicMock()
@@ -121,7 +121,7 @@ class TestEdgeCases:
assert response.status_code in [400, 401]
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_exec_with_empty_command(self, mock_get_client, client, auth_headers):
"""Test exec with empty command string"""
mock_get_client.return_value = MagicMock()

10
backend/tests/test_exec.py
View File

@@ -12,7 +12,7 @@ class TestContainerExec:
})
assert response.status_code == 401
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_simple_command(self, mock_get_client, client, auth_headers, auth_token):
"""Test executing a simple command"""
# Mock exec result
@@ -37,7 +37,7 @@ class TestContainerExec:
assert 'file1.txt' in data['output']
assert data['workdir'] == '/app'
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_cd_command(self, mock_get_client, client, auth_headers, auth_token):
"""Test executing cd command"""
# Mock exec result for cd command
@@ -62,7 +62,7 @@ class TestContainerExec:
assert data['workdir'] == '/home/user'
assert data['output'] == ''
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_command_with_error(self, mock_get_client, client, auth_headers, auth_token):
"""Test executing a command that fails"""
# Mock exec result with error
@@ -86,7 +86,7 @@ class TestContainerExec:
assert data['exit_code'] == 127
assert 'command not found' in data['output']
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_docker_unavailable(self, mock_get_client, client, auth_headers):
"""Test exec when Docker is unavailable"""
mock_get_client.return_value = None
@@ -99,7 +99,7 @@ class TestContainerExec:
data = response.get_json()
assert 'error' in data
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_unicode_handling(self, mock_get_client, client, auth_headers, auth_token):
"""Test exec with unicode output"""
# Mock exec result with unicode

14
backend/tests/test_exec_advanced.py
View File

@@ -5,7 +5,7 @@ from unittest.mock import MagicMock, patch
class TestExecAdvanced:
"""Advanced tests for command execution"""
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_bash_fallback_to_sh(self, mock_get_client, client, auth_headers, auth_token):
"""Test fallback from bash to sh when bash doesn't exist"""
# Mock exec that fails for bash but succeeds for sh
@@ -33,7 +33,7 @@ class TestExecAdvanced:
data = response.get_json()
assert data['exit_code'] == 0
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_container_not_found(self, mock_get_client, client, auth_headers):
"""Test exec on non-existent container"""
mock_client = MagicMock()
@@ -48,7 +48,7 @@ class TestExecAdvanced:
data = response.get_json()
assert 'error' in data
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_preserves_working_directory(self, mock_get_client, client, auth_headers, auth_token):
"""Test that working directory is preserved across commands"""
mock_exec_result = MagicMock()
@@ -76,7 +76,7 @@ class TestExecAdvanced:
json={'command': 'ls'})
assert response2.status_code == 200
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_cd_with_tilde(self, mock_get_client, client, auth_headers, auth_token):
"""Test cd command with tilde expansion"""
mock_exec_result = MagicMock()
@@ -98,7 +98,7 @@ class TestExecAdvanced:
data = response.get_json()
assert data['workdir'] == '/home/user'
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_cd_no_args(self, mock_get_client, client, auth_headers, auth_token):
"""Test cd command without arguments (should go to home)"""
mock_exec_result = MagicMock()
@@ -122,7 +122,7 @@ class TestExecAdvanced:
# workdir should be extracted from ::WORKDIR:: marker
assert data['workdir'] == '/'
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_latin1_encoding_fallback(self, mock_get_client, client, auth_headers, auth_token):
"""Test fallback to latin-1 encoding for non-UTF-8 output"""
# Create binary data that's not valid UTF-8
@@ -149,7 +149,7 @@ class TestExecAdvanced:
assert data['exit_code'] == 0
assert 'output' in data
@patch('app.get_docker_client')
@patch('routes.containers.exec.get_docker_client')
def test_exec_empty_command(self, mock_get_client, client, auth_headers, auth_token):
"""Test exec with empty/no command"""
mock_exec_result = MagicMock()

262
backend/tests/test_final_coverage.py Normal file
View File

@@ -0,0 +1,262 @@
"""Tests for final 100% coverage."""
import pytest
from unittest.mock import MagicMock, patch, Mock, PropertyMock
class TestRemainingHandlerCoverage:
"""Test remaining handler edge cases"""
def test_resize_with_active_terminal(self):
"""Test resize handler with active terminal"""
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:
with patch('handlers.terminal.resize.logger') as mock_logger:
mock_request.sid = 'resize_sid'
active_terminals['resize_sid'] = {'exec': MagicMock()}
handle_resize({'cols': 120, 'rows': 40})
# Should log the resize request
mock_logger.info.assert_called()
# Clean up
del active_terminals['resize_sid']
class TestDockerClientOuterException:
"""Test docker client outer exception handler"""
@patch('utils.docker_client.docker.from_env')
@patch('utils.docker_client.docker.DockerClient')
@patch('utils.docker_client.diagnose_docker_environment')
def test_get_docker_client_outer_exception(self, mock_diagnose, mock_docker_client, mock_from_env):
"""Test get_docker_client when outer try block catches exception"""
from utils.docker_client import get_docker_client
# Make the initial logger.info call raise an exception
with patch('utils.docker_client.logger') as mock_logger:
# Raise exception on the first logger.info call
mock_logger.info.side_effect = Exception("Unexpected logger error")
client = get_docker_client()
assert client is None
mock_logger.error.assert_called()
class TestExecHelpersCdFallback:
"""Test exec helpers cd command fallback to sh"""
def test_cd_command_sh_fallback(self):
"""Test build_sh_command for cd commands"""
from utils.exec_helpers import build_sh_command
result = build_sh_command('/home/user', 'cd /tmp', True)
assert result[0] == '/bin/sh'
assert result[1] == '-c'
assert 'cd "/home/user"' in result[2]
assert 'cd /tmp' in result[2]
assert 'pwd' in result[2]
class TestDiagnosticsDockerRelated:
"""Test diagnostics docker-related files logging"""
@patch('os.path.exists')
@patch('os.listdir')
def test_diagnose_with_docker_related_files(self, mock_listdir, mock_exists):
"""Test diagnostics when docker-related files are found"""
from utils.diagnostics.docker_env import diagnose_docker_environment
def exists_side_effect(path):
if path == '/var/run':
return True
if path == '/var/run/docker.sock':
return False
return False
mock_exists.side_effect = exists_side_effect
mock_listdir.return_value = ['docker.pid', 'docker.sock.tmp', 'other.file']
with patch('utils.diagnostics.docker_env.logger') as mock_logger:
diagnose_docker_environment()
# Should log docker-related files
info_calls = [str(call) for call in mock_logger.info.call_args_list]
assert any('docker' in str(call).lower() for call in info_calls)
@patch('os.path.exists')
@patch('os.stat')
@patch('os.access')
def test_diagnose_socket_not_readable_writable(self, mock_access, mock_stat, mock_exists):
"""Test diagnostics when socket exists but not readable/writable"""
from utils.diagnostics.docker_env import diagnose_docker_environment
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
mock_stat_result = MagicMock()
mock_stat_result.st_mode = 0o600
mock_stat_result.st_uid = 0
mock_stat_result.st_gid = 0
mock_stat.return_value = mock_stat_result
# Make access return False for both R_OK and W_OK
mock_access.return_value = False
with patch('utils.diagnostics.docker_env.logger') as mock_logger:
diagnose_docker_environment()
# Should log warning about permissions
warning_calls = [str(call) for call in mock_logger.warning.call_args_list]
assert any('permission' in str(call).lower() for call in warning_calls)
class TestTerminalHelpersSidRemoval:
"""Test terminal helpers when sid is removed during execution"""
@patch('utils.terminal_helpers.threading.Thread')
def test_output_reader_sid_removed_during_loop(self, mock_thread):
"""Test output reader when sid is removed from active_terminals during loop"""
from utils.terminal_helpers import create_output_reader
from config import active_terminals
mock_socketio = MagicMock()
mock_sock = MagicMock()
# Setup to remove sid after first iteration
call_count = [0]
def recv_side_effect(size):
call_count[0] += 1
if call_count[0] == 1:
# First call: return data and remove sid
if 'removal_test_sid' in active_terminals:
del active_terminals['removal_test_sid']
return b'test data'
# Second call won't happen because sid was removed
return b''
mock_sock.recv.side_effect = recv_side_effect
mock_sock.close = MagicMock()
mock_exec = MagicMock()
mock_exec.output = mock_sock
sid = 'removal_test_sid'
active_terminals[sid] = {'exec': mock_exec}
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 the data and broken out of loop
assert mock_socketio.emit.called
@patch('utils.terminal_helpers.threading.Thread')
def test_output_reader_finally_with_sid_present(self, mock_thread):
"""Test output reader finally block when sid is still in active_terminals"""
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 immediately
mock_sock.close = MagicMock()
mock_exec = MagicMock()
mock_exec.output = mock_sock
sid = 'finally_test_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)
# sid should be removed in finally block
assert sid not in active_terminals
class TestDisconnectNoKillMethod:
"""Test disconnect handler when exec has no kill method"""
def test_disconnect_exec_without_kill(self):
"""Test disconnect when exec instance has no kill method"""
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 = 'no_kill_sid'
# Create exec without kill method
mock_exec = MagicMock(spec=['output', 'exit_code']) # Explicitly exclude 'kill'
del mock_exec.kill # Ensure kill is not available
active_terminals['no_kill_sid'] = {'exec': mock_exec}
handle_disconnect()
# Should still clean up
assert 'no_kill_sid' not in active_terminals
class TestDiagnosticsReadableWritableSocket:
"""Test diagnostics when socket is readable and writable"""
@patch('os.path.exists')
@patch('os.stat')
@patch('os.access')
def test_diagnose_socket_readable_and_writable(self, mock_access, mock_stat, mock_exists):
"""Test diagnostics when socket exists and is readable/writable"""
from utils.diagnostics.docker_env import diagnose_docker_environment
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
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
# Make access return True (readable and writable)
mock_access.return_value = True
with patch('utils.diagnostics.docker_env.logger') as mock_logger:
diagnose_docker_environment()
# Should log success messages, not warnings
info_calls = [str(call) for call in mock_logger.info.call_args_list]
assert any('Readable' in str(call) or 'Writable' in str(call) for call in info_calls)
# Should NOT log permission warning
warning_calls = [str(call) for call in mock_logger.warning.call_args_list]
assert not any('socket' in str(call).lower() and 'permission' in str(call).lower() for call in warning_calls)

2
backend/tests/test_utils.py
View File

@@ -1,6 +1,6 @@
import pytest
from datetime import datetime, timezone, timedelta
from app import format_uptime
from utils.formatters import format_uptime
class TestUtilityFunctions:

4
backend/tests/test_websocket.py
View File

@@ -24,7 +24,7 @@ class TestWebSocketHandlers:
socketio_client.disconnect(namespace='/terminal')
assert not socketio_client.is_connected('/terminal')
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_start_terminal_unauthorized(self, mock_get_client, socketio_client):
"""Test starting terminal without valid token"""
socketio_client.emit('start_terminal', {
@@ -42,7 +42,7 @@ class TestWebSocketHandlers:
# For testing purposes, we just verify the test didn't crash
assert True
@patch('app.get_docker_client')
@patch('utils.docker_client.get_docker_client')
def test_start_terminal_docker_unavailable(self, mock_get_client, socketio_client, auth_token):
"""Test starting terminal when Docker is unavailable"""
mock_get_client.return_value = None

20
backend/tests/test_websocket_coverage.py
View File

@@ -21,7 +21,7 @@ class TestWebSocketCoverage:
from app import socketio
return socketio.test_client(app, namespace='/terminal')
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_start_terminal_success_flow(self, mock_get_client, socketio_client, auth_token):
"""Test successful terminal start with mocked Docker"""
# Create mock Docker client and container
@@ -77,7 +77,7 @@ class TestWebSocketCoverage:
assert call_args[1]['environment']['COLUMNS'] == '100'
assert call_args[1]['environment']['LINES'] == '30'
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_start_terminal_creates_thread(self, mock_get_client, socketio_client, auth_token):
"""Test that starting terminal creates output thread"""
mock_client = MagicMock()
@@ -130,7 +130,7 @@ class TestWebSocketCoverage:
decoded = invalid_utf8.decode('latin-1', errors='replace')
assert decoded is not None # Should not crash
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_start_terminal_latin1_fallback(self, mock_get_client, socketio_client, auth_token):
"""Test latin-1 fallback for invalid UTF-8"""
mock_client = MagicMock()
@@ -166,7 +166,7 @@ class TestWebSocketCoverage:
decoding_errors = [msg for msg in error_msgs if 'decode' in str(msg).lower()]
assert len(decoding_errors) == 0
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_start_terminal_container_not_found(self, mock_get_client, socketio_client, auth_token):
"""Test error when container doesn't exist"""
mock_client = MagicMock()
@@ -186,7 +186,7 @@ class TestWebSocketCoverage:
assert len(error_msgs) > 0, "Should receive error message"
assert 'not found' in error_msgs[0]['args'][0]['error'].lower()
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_start_terminal_exec_error(self, mock_get_client, socketio_client, auth_token):
"""Test error during exec_run"""
mock_client = MagicMock()
@@ -207,7 +207,7 @@ class TestWebSocketCoverage:
assert len(error_msgs) > 0, "Should receive error message"
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_handle_input_error_handling(self, mock_get_client, socketio_client, auth_token):
"""Test error handling in handle_input when sendall fails"""
import app
@@ -250,7 +250,7 @@ class TestWebSocketCoverage:
# Should receive error about socket problem
assert len(error_msgs) > 0, "Should receive error from failed sendall"
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_disconnect_cleanup(self, mock_get_client, socketio_client, auth_token):
"""Test that disconnect properly cleans up active terminals"""
import app
@@ -313,7 +313,7 @@ class TestWebSocketCoverage:
error_msgs = [msg for msg in received if msg['name'] == 'error']
assert len(error_msgs) == 0, "Resize should not error"
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_socket_close_on_exit(self, mock_get_client, socketio_client, auth_token):
"""Test that socket is closed when thread exits"""
mock_client = MagicMock()
@@ -343,7 +343,7 @@ class TestWebSocketCoverage:
# but the code path is exercised
assert True
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
def test_default_terminal_size(self, mock_get_client, socketio_client, auth_token):
"""Test default terminal size when not specified"""
mock_client = MagicMock()
@@ -373,7 +373,7 @@ class TestWebSocketCoverage:
assert call_args[1]['environment']['COLUMNS'] == '80'
assert call_args[1]['environment']['LINES'] == '24'
@patch('app.get_docker_client')
@patch('handlers.terminal.start.get_docker_client')
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

1
backend/utils/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Utility modules."""

20
backend/utils/auth.py Normal file
View File

@@ -0,0 +1,20 @@
"""Authentication utilities."""
from flask import request, jsonify
from config import sessions
def check_auth():
"""Check if request has valid authentication.
Returns:
tuple: (is_valid, token, error_response)
"""
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return False, None, (jsonify({'error': 'Unauthorized'}), 401)
token = auth_header.split(' ')[1]
if token not in sessions:
return False, None, (jsonify({'error': 'Invalid session'}), 401)
return True, token, None

31
backend/utils/container_helpers.py Normal file
View File

@@ -0,0 +1,31 @@
"""Common helpers for container routes."""
from flask import jsonify
from utils.auth import check_auth
from utils.docker_client import get_docker_client
def get_auth_and_container(container_id):
"""Common auth check and container retrieval pattern.
Args:
container_id: Container ID to retrieve
Returns:
tuple: (container, error_response) where error_response is None on success
"""
# Check authentication
is_valid, _, error_response = check_auth()
if not is_valid:
return None, error_response
# Get Docker client
client = get_docker_client()
if not client:
return None, (jsonify({'error': 'Cannot connect to Docker'}), 500)
# Get container
try:
container = client.containers.get(container_id)
return container, None
except Exception as e: # pylint: disable=broad-exception-caught
return None, (jsonify({'error': str(e)}), 500)

1
backend/utils/diagnostics/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Docker diagnostics utilities."""

88
backend/utils/diagnostics/docker_env.py Normal file
View File

@@ -0,0 +1,88 @@
"""Docker environment diagnostics."""
import os
from config import logger
def diagnose_docker_environment(): # pylint: disable=too-many-locals,too-many-statements
"""Diagnose Docker environment and configuration.
This function intentionally performs many checks and has many local variables
as it needs to comprehensively diagnose the Docker environment.
"""
logger.info("=== Docker Environment Diagnosis ===")
# Check environment variables
docker_host = os.getenv('DOCKER_HOST', 'Not set')
docker_cert_path = os.getenv('DOCKER_CERT_PATH', 'Not set')
docker_tls_verify = os.getenv('DOCKER_TLS_VERIFY', 'Not set')
logger.info("DOCKER_HOST: %s", docker_host)
logger.info("DOCKER_CERT_PATH: %s", docker_cert_path)
logger.info("DOCKER_TLS_VERIFY: %s", docker_tls_verify)
# Check what's in /var/run
logger.info("Checking /var/run directory contents:")
try:
if os.path.exists('/var/run'):
var_run_contents = os.listdir('/var/run')
logger.info(" /var/run contains: %s", var_run_contents)
# Check for any Docker-related files
docker_related = [f for f in var_run_contents if 'docker' in f.lower()]
if docker_related:
logger.info(" Docker-related files/dirs found: %s", docker_related)
else:
logger.warning(" /var/run directory doesn't exist")
except Exception as e: # pylint: disable=broad-exception-caught
logger.error(" Error reading /var/run: %s", e)
# Check Docker socket
socket_path = '/var/run/docker.sock'
logger.info("Checking Docker socket at %s", socket_path)
if os.path.exists(socket_path):
logger.info("✓ Docker socket exists at %s", socket_path)
# Check permissions
st = os.stat(socket_path)
logger.info(" Socket permissions: %s", oct(st.st_mode))
logger.info(" Socket owner UID: %s", st.st_uid)
logger.info(" Socket owner GID: %s", st.st_gid)
# Check if readable/writable
readable = os.access(socket_path, os.R_OK)
writable = os.access(socket_path, os.W_OK)
logger.info(" Readable: %s", readable)
logger.info(" Writable: %s", writable)
if not (readable and writable):
logger.warning("⚠ Socket exists but lacks proper permissions!")
else:
logger.error("✗ Docker socket NOT found at %s", socket_path)
logger.error(" This means the Docker socket mount is NOT configured in CapRover")
logger.error(" The serviceUpdateOverride in captain-definition may not be applied")
# Check current user
import pwd # pylint: disable=import-outside-toplevel
try:
current_uid = os.getuid()
current_gid = os.getgid()
user_info = pwd.getpwuid(current_uid)
logger.info("Current user: %s (UID: %s, GID: %s)",
user_info.pw_name, current_uid, current_gid)
# Check groups
import grp # pylint: disable=import-outside-toplevel
groups = os.getgroups()
logger.info("User groups (GIDs): %s", groups)
for gid in groups:
try:
group_info = grp.getgrgid(gid)
logger.info(" - %s (GID: %s)", group_info.gr_name, gid)
except KeyError:
logger.info(" - Unknown group (GID: %s)", gid)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error checking user info: %s", e)
logger.info("=== End Diagnosis ===")

38
backend/utils/docker_client.py Normal file
View File

@@ -0,0 +1,38 @@
"""Docker client getter."""
import docker
from config import logger
from utils.diagnostics.docker_env import diagnose_docker_environment
def get_docker_client():
"""Get Docker client with enhanced error reporting."""
try:
logger.info("Attempting to connect to Docker...")
# Try default connection first
try:
client = docker.from_env()
client.ping()
logger.info("✓ Successfully connected to Docker using docker.from_env()")
return client
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning("docker.from_env() failed: %s", e)
# Try explicit Unix socket connection
try:
logger.info("Trying explicit Unix socket connection...")
client = docker.DockerClient(base_url='unix:///var/run/docker.sock')
client.ping()
logger.info("✓ Successfully connected to Docker using Unix socket")
return client
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning("Unix socket connection failed: %s", e)
# If all fails, run diagnostics and return None
logger.error("All Docker connection attempts failed!")
diagnose_docker_environment()
return None
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Unexpected error in get_docker_client: %s", e, exc_info=True)
return None

148
backend/utils/exec_helpers.py Normal file
View File

@@ -0,0 +1,148 @@
"""Helper functions for container exec operations."""
from config import logger
def get_session_workdir(token, container_id, session_workdirs):
"""Get or initialize session working directory.
Args:
token: Session token
container_id: Container ID
session_workdirs: Session workdir dictionary
Returns:
tuple: (session_key, current_workdir)
"""
session_key = f"{token}_{container_id}"
if session_key not in session_workdirs:
session_workdirs[session_key] = '/'
return session_key, session_workdirs[session_key]
def execute_command_with_fallback(container, current_workdir, user_command, is_cd_command):
"""Execute command in container with bash/sh fallback.
Args:
container: Docker container object
current_workdir: Current working directory
user_command: User's command
is_cd_command: Whether this is a cd command
Returns:
Docker exec instance
"""
# Try bash first
try:
bash_command = build_bash_command(current_workdir, user_command, is_cd_command)
return execute_in_container(container, bash_command)
except Exception as bash_error: # pylint: disable=broad-exception-caught
logger.warning("Bash execution failed, trying sh: %s", bash_error)
sh_command = build_sh_command(current_workdir, user_command, is_cd_command)
return execute_in_container(container, sh_command)
def build_bash_command(current_workdir, user_command, is_cd_command):
"""Build bash command for execution.
Args:
current_workdir: Current working directory
user_command: User's command
is_cd_command: Whether this is a cd command
Returns:
list: Command array for Docker exec
"""
path_export = 'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
if is_cd_command:
target_dir = user_command.strip()[3:].strip() or '~'
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
return ['/bin/bash', '-c', f'{path_export}; {resolve_command}']
return [
'/bin/bash', '-c',
f'{path_export}; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"'
]
def build_sh_command(current_workdir, user_command, is_cd_command):
"""Build sh command for execution (fallback).
Args:
current_workdir: Current working directory
user_command: User's command
is_cd_command: Whether this is a cd command
Returns:
list: Command array for Docker exec
"""
path_export = 'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
if is_cd_command:
target_dir = user_command.strip()[3:].strip() or '~'
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
return ['/bin/sh', '-c', f'{path_export}; {resolve_command}']
return [
'/bin/sh', '-c',
f'{path_export}; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"'
]
def execute_in_container(container, command):
"""Execute command in container.
Args:
container: Docker container object
command: Command to execute
Returns:
Docker exec instance
"""
return container.exec_run(
command,
stdout=True,
stderr=True,
stdin=False,
tty=True,
environment={'TERM': 'xterm-256color', 'LANG': 'C.UTF-8'}
)
def decode_output(exec_instance):
"""Decode exec output with fallback encoding.
Args:
exec_instance: Docker exec instance
Returns:
str: Decoded output
"""
if not exec_instance.output:
return ''
try:
return exec_instance.output.decode('utf-8')
except UnicodeDecodeError:
return exec_instance.output.decode('latin-1', errors='replace')
def extract_workdir(output, current_workdir, is_cd_command):
"""Extract working directory from command output.
Args:
output: Command output
current_workdir: Current working directory
is_cd_command: Whether this was a cd command
Returns:
tuple: (cleaned_output, new_workdir)
"""
if is_cd_command:
return '', output.strip()
if '::WORKDIR::' in output:
parts = output.rsplit('::WORKDIR::', 1)
return parts[0], parts[1].strip()
return output, current_workdir

26
backend/utils/formatters.py Normal file
View File

@@ -0,0 +1,26 @@
"""Formatting utility functions."""
from datetime import datetime
def format_uptime(created_at):
"""Format container uptime.
Args:
created_at: ISO format datetime string
Returns:
Formatted uptime string (e.g., "2d 3h", "5h 30m", "15m")
"""
created = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
now = datetime.now(created.tzinfo)
delta = now - created
days = delta.days
hours = delta.seconds // 3600
minutes = (delta.seconds % 3600) // 60
if days > 0:
return f"{days}d {hours}h"
if hours > 0:
return f"{hours}h {minutes}m"
return f"{minutes}m"

50
backend/utils/terminal_helpers.py Normal file
View File

@@ -0,0 +1,50 @@
"""Helper functions for terminal operations."""
import threading
from config import logger, active_terminals
def create_output_reader(socketio, sid, exec_instance):
"""Create and start output reader thread.
Args:
socketio: SocketIO instance
sid: Session ID
exec_instance: Docker exec instance
Returns:
Thread: Started output reader thread
"""
def read_output():
sock = exec_instance.output
try:
while True:
if sid not in active_terminals:
break
try:
data = sock.recv(4096)
if not data:
break
try:
decoded_data = data.decode('utf-8')
except UnicodeDecodeError:
decoded_data = data.decode('latin-1', errors='replace')
socketio.emit('output', {'data': decoded_data},
namespace='/terminal', room=sid)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error reading from container: %s", e)
break
finally:
if sid in active_terminals:
del active_terminals[sid]
try:
sock.close()
except Exception: # pylint: disable=broad-exception-caught
pass
socketio.emit('exit', {'code': 0}, namespace='/terminal', room=sid)
thread = threading.Thread(target=read_output, daemon=True)
thread.start()
return thread

3
frontend/app/layout.tsx
View File

@@ -1,4 +1,5 @@
import type { Metadata } from "next";
import Script from "next/script";
import "./globals.css";
import { ThemeProvider } from "@/lib/theme";
import { Providers } from "./providers";
@@ -22,9 +23,9 @@ export default function RootLayout({
href="https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;500;600;700&display=swap"
rel="stylesheet"
/>
<script src="/env.js" />
</head>
<body>
<Script src="/env.js" strategy="beforeInteractive" />
<ThemeProvider>
<Providers>
{children}

54
frontend/components/ContainerCard/ContainerHeader.tsx
View File

@@ -1,5 +1,5 @@
import React from 'react';
import { Box, Typography, Chip } from '@mui/material';
import { Box, Typography, Chip, Tooltip } from '@mui/material';
import { PlayArrow, Inventory2 } from '@mui/icons-material';
import { ContainerHeaderProps } from '@/lib/interfaces/container';
@@ -30,30 +30,34 @@ export default function ContainerHeader({ name, image, status }: ContainerHeader
<Inventory2 sx={{ color: 'secondary.main', fontSize: 20 }} />
</Box>
<Box sx={{ minWidth: 0, flex: 1 }}>
<Typography
variant="h3"
component="h3"
sx={{
fontFamily: '"JetBrains Mono", monospace',
fontWeight: 500,
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
}}
>
{name}
</Typography>
<Typography
variant="body2"
color="text.secondary"
sx={{
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
}}
>
{image}
</Typography>
<Tooltip title={name} placement="top" arrow>
<Typography
variant="h3"
component="h3"
sx={{
fontFamily: '"JetBrains Mono", monospace',
fontWeight: 500,
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
}}
>
{name}
</Typography>
</Tooltip>
<Tooltip title={image} placement="bottom" arrow>
<Typography
variant="body2"
color="text.secondary"
sx={{
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
}}
>
{image}
</Typography>
</Tooltip>
</Box>
</Box>

21
frontend/components/ContainerCard/ContainerInfo.tsx
View File

@@ -1,5 +1,5 @@
import React from 'react';
import { Box, Typography } from '@mui/material';
import { Box, Typography, Tooltip } from '@mui/material';
import { ContainerInfoProps } from '@/lib/interfaces/container';
export default function ContainerInfo({ id, uptime }: ContainerInfoProps) {
@@ -18,12 +18,19 @@ export default function ContainerInfo({ id, uptime }: ContainerInfoProps) {
>
Container ID
</Typography>
<Typography
variant="body2"
sx={{ fontFamily: '"JetBrains Mono", monospace' }}
>
{id}
</Typography>
<Tooltip title={id} placement="top" arrow>
<Typography
variant="body2"
sx={{
fontFamily: '"JetBrains Mono", monospace',
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
}}
>
{id}
</Typography>
</Tooltip>
</Box>
<Box>
<Typography

10
frontend/components/Dashboard/DashboardHeader.tsx
View File

@@ -65,15 +65,15 @@ export default function DashboardHeader({
{isMobile ? (
<>
<IconButton
color="inherit"
color="secondary"
onClick={onRefresh}
disabled={isRefreshing}
size="small"
>
{isRefreshing ? <CircularProgress size={20} /> : <Refresh />}
{isRefreshing ? <CircularProgress size={20} color="secondary" /> : <Refresh />}
</IconButton>
<IconButton
color="inherit"
color="secondary"
onClick={onLogout}
size="small"
>
@@ -84,15 +84,17 @@ export default function DashboardHeader({
<>
<Button
variant="outlined"
color="secondary"
size="small"
onClick={onRefresh}
disabled={isRefreshing}
startIcon={isRefreshing ? <CircularProgress size={16} /> : <Refresh />}
startIcon={isRefreshing ? <CircularProgress size={16} color="secondary" /> : <Refresh />}
>
Refresh
</Button>
<Button
variant="outlined"
color="secondary"
size="small"
onClick={onLogout}
startIcon={<Logout />}

11
frontend/components/TerminalModal/InteractiveTerminal.tsx
View File

@@ -9,15 +9,18 @@ export default function InteractiveTerminal({ terminalRef }: InteractiveTerminal
ref={terminalRef}
sx={{
height: { xs: '400px', sm: '500px' },
backgroundColor: '#300A24',
backgroundColor: '#2E3436',
borderRadius: '4px',
border: '1px solid #5E2750',
border: '1px solid #1C1F20',
overflow: 'hidden',
'& .xterm': {
padding: '8px',
padding: '10px',
},
'& .xterm-viewport': {
backgroundColor: '#300A24 !important',
backgroundColor: '#2E3436 !important',
},
'& .xterm-screen': {
backgroundColor: '#2E3436',
},
}}
/>

20
frontend/components/TerminalModal/TerminalOutput.tsx
View File

@@ -9,28 +9,28 @@ export default function TerminalOutput({ output, containerName, outputRef }: Ter
ref={outputRef}
elevation={0}
sx={{
backgroundColor: '#300A24',
color: '#F8F8F2',
fontFamily: '"Ubuntu Mono", "Courier New", monospace',
backgroundColor: '#2E3436',
color: '#D3D7CF',
fontFamily: '"Ubuntu Mono", "DejaVu Sans Mono", "Courier New", monospace',
fontSize: { xs: '12px', sm: '14px' },
padding: { xs: 1.5, sm: 2 },
minHeight: { xs: '300px', sm: '400px' },
maxHeight: { xs: '400px', sm: '500px' },
overflowY: 'auto',
mb: 2,
border: '1px solid #5E2750',
border: '1px solid #1C1F20',
borderRadius: '4px',
'&::-webkit-scrollbar': {
width: { xs: '6px', sm: '10px' },
},
'&::-webkit-scrollbar-track': {
background: '#2C0922',
background: '#1C1F20',
},
'&::-webkit-scrollbar-thumb': {
background: '#5E2750',
background: '#555753',
borderRadius: '5px',
'&:hover': {
background: '#772953',
background: '#729FCF',
}
},
}}
@@ -38,15 +38,15 @@ export default function TerminalOutput({ output, containerName, outputRef }: Ter
{output.length === 0 ? (
<Box>
<Typography sx={{
color: '#8BE9FD',
color: '#729FCF',
fontFamily: 'inherit',
fontSize: '13px',
mb: 1
}}>
Ubuntu-style Terminal - Connected to <span style={{ color: '#50FA7B', fontWeight: 'bold' }}>{containerName}</span>
Ubuntu-style Terminal - Connected to <span style={{ color: '#8AE234', fontWeight: 'bold' }}>{containerName}</span>
</Typography>
<Typography sx={{
color: '#6272A4',
color: '#555753',
fontFamily: 'inherit',
fontSize: '12px'
}}>

BIN
frontend/gnome-interactive-demo.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 110 KiB

96
frontend/lib/hooks/useInteractiveTerminal.ts
View File

@@ -26,51 +26,77 @@ export function useInteractiveTerminal({
const connectionAttempts = useRef(0);
useEffect(() => {
if (!open || !terminalRef.current) return;
if (!open) return;
let term: Terminal | null = null;
let fitAddon: FitAddon | null = null;
let socket: Socket | null = null;
let mounted = true;
const initTerminal = async () => {
try {
// Wait for ref to be available
let attempts = 0;
while (!terminalRef.current && attempts < 10) {
await new Promise(resolve => setTimeout(resolve, 100));
attempts++;
}
if (!terminalRef.current || !mounted) {
console.warn('Terminal ref not available after waiting');
return;
}
console.log('Initializing interactive terminal...');
const [{ Terminal }, { FitAddon }] = await Promise.all([
import('@xterm/xterm'),
import('@xterm/addon-fit'),
]);
if (!terminalRef.current) return;
if (!terminalRef.current || !mounted) return;
console.log('Creating terminal instance...');
term = new Terminal({
cursorBlink: true,
fontSize: isMobile ? 12 : 14,
fontFamily: '"Ubuntu Mono", "Courier New", monospace',
fontFamily: '"Ubuntu Mono", "DejaVu Sans Mono", "Courier New", monospace',
theme: {
background: '#300A24',
foreground: '#F8F8F2',
cursor: '#F8F8F2',
black: '#2C0922',
red: '#FF5555',
green: '#50FA7B',
yellow: '#F1FA8C',
blue: '#8BE9FD',
magenta: '#FF79C6',
cyan: '#8BE9FD',
white: '#F8F8F2',
brightBlack: '#6272A4',
brightRed: '#FF6E6E',
brightGreen: '#69FF94',
brightYellow: '#FFFFA5',
brightBlue: '#D6ACFF',
brightMagenta: '#FF92DF',
brightCyan: '#A4FFFF',
brightWhite: '#FFFFFF',
// GNOME Terminal color scheme
background: '#2E3436',
foreground: '#D3D7CF',
cursor: '#D3D7CF',
cursorAccent: '#2E3436',
selectionBackground: '#4A90D9',
selectionForeground: '#FFFFFF',
// Standard colors
black: '#2E3436',
red: '#CC0000',
green: '#4E9A06',
yellow: '#C4A000',
blue: '#3465A4',
magenta: '#75507B',
cyan: '#06989A',
white: '#D3D7CF',
// Bright colors
brightBlack: '#555753',
brightRed: '#EF2929',
brightGreen: '#8AE234',
brightYellow: '#FCE94F',
brightBlue: '#729FCF',
brightMagenta: '#AD7FA8',
brightCyan: '#34E2E2',
brightWhite: '#EEEEEC',
},
});
console.log('Loading fit addon...');
fitAddon = new FitAddon();
term.loadAddon(fitAddon);
console.log('Opening terminal in DOM...');
term.open(terminalRef.current);
console.log('Terminal opened successfully');
setTimeout(() => {
try {
@@ -83,6 +109,11 @@ export function useInteractiveTerminal({
xtermRef.current = term;
fitAddonRef.current = fitAddon;
// Expose terminal for debugging
if (typeof window !== 'undefined') {
(window as any)._debugTerminal = term;
}
const wsUrl = API_BASE_URL.replace(/^http/, 'ws');
socket = io(`${wsUrl}/terminal`, {
transports: ['polling', 'websocket'],
@@ -122,7 +153,10 @@ export function useInteractiveTerminal({
});
socket.on('output', (data: { data: string }) => {
term?.write(data.data);
console.log('Received output event:', data);
if (term && data && data.data) {
term.write(data.data);
}
});
socket.on('error', (data: { error: string }) => {
@@ -173,19 +207,29 @@ export function useInteractiveTerminal({
window.addEventListener('resize', handleResize);
return () => {
mounted = false;
window.removeEventListener('resize', handleResize);
if (term) term.dispose();
if (socket) socket.disconnect();
if (term) {
console.log('Disposing terminal...');
term.dispose();
}
if (socket) {
console.log('Disconnecting socket...');
socket.disconnect();
}
};
} catch (error) {
console.error('Failed to initialize terminal:', error);
onFallback('Failed to load terminal. Switching to simple mode.');
if (mounted) {
onFallback('Failed to load terminal. Switching to simple mode.');
}
}
};
const cleanup = initTerminal();
return () => {
mounted = false;
cleanup.then((cleanupFn) => {
if (cleanupFn) cleanupFn();
});

176
frontend/show-interactive-direct.js Normal file
View File

@@ -0,0 +1,176 @@
const { chromium } = require('playwright');
(async () => {
let browser;
try {
console.log('=== GNOME Terminal Interactive Mode Demo ===\n');
browser = await chromium.launch({
headless: true,
executablePath: '/root/.cache/ms-playwright/chromium-1194/chrome-linux/chrome',
args: ['--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage']
});
const page = await (await browser.newContext({ viewport: { width: 1400, height: 900 }})).newPage();
// Capture all console messages
const errors = [];
const warnings = [];
page.on('console', msg => {
const type = msg.type();
const text = msg.text();
if (type === 'error') {
errors.push(text);
console.log(`[BROWSER ERROR]`, text);
} else if (type === 'warning') {
warnings.push(text);
console.log(`[BROWSER WARNING]`, text);
}
});
// Mock containers API
await page.route('**/api/containers', async (route) => {
await route.fulfill({
status: 200,
body: JSON.stringify({
containers: [{
id: 'gnome-demo-123',
name: 'ubuntu-container',
image: 'ubuntu:22.04',
status: 'running',
uptime: '2 days'
}]
})
});
});
// Block Socket.io requests to prevent errors
await page.route('**/socket.io/**', async (route) => {
// Just hang the request - don't respond
await new Promise(() => {});
});
console.log('1. Loading application...');
await page.goto('http://localhost:3000');
await page.evaluate(() => {
localStorage.setItem('auth_token', 'demo-token');
localStorage.setItem('auth_username', 'admin');
});
await page.goto('http://localhost:3000/dashboard', { waitUntil: 'networkidle' });
await page.waitForTimeout(1000);
console.log('✓ Dashboard loaded\n');
console.log('2. Opening terminal...');
await page.click('button:has-text("Open Shell")');
await page.waitForSelector('[role="dialog"]');
console.log('✓ Terminal modal opened\n');
console.log('3. Waiting for xterm to initialize...');
// Wait for xterm to be initialized
await page.waitForFunction(() => {
return window._debugTerminal != null;
}, { timeout: 10000 });
console.log('4. Injecting terminal startup...');
await page.evaluate(() => {
const term = window._debugTerminal;
if (term) {
console.log('Writing startup messages...');
term.write('\x1b[0m\r\n');
term.write('\x1b[1;32m*** Interactive Terminal Started ***\x1b[0m\r\n');
term.write('You can now use sudo, nano, vim, and other interactive commands.\r\n\r\n');
term.write('\x1b[1;32mroot@ubuntu-container\x1b[0m:\x1b[1;34m/app\x1b[0m# ');
}
});
await page.waitForTimeout(500);
console.log('5. Simulating "ls -la" command...');
await page.evaluate(() => {
const term = window._debugTerminal;
if (term) {
// Simulate typing
term.write('ls -la\r\n');
// Simulate ls -la output with colors
term.write('total 48\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 8 root root 4096 Feb 1 03:17 \x1b[1;34m.\x1b[0m\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 12 root root 4096 Jan 28 10:42 \x1b[1;34m..\x1b[0m\r\n');
term.write('-rw-r--r-- 1 root root 220 Jan 1 2024 .bashrc\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 2 root root 4096 Jan 15 14:23 \x1b[1;34mbin\x1b[0m\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 3 root root 4096 Jan 20 09:10 \x1b[1;34mconfig\x1b[0m\r\n');
term.write('-rw-r--r-- 1 root root 1245 Jan 28 16:34 docker-compose.yml\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 5 root root 4096 Feb 1 03:17 \x1b[1;34mnode_modules\x1b[0m\r\n');
term.write('-rw-r--r-- 1 root root 2891 Jan 30 11:22 package.json\r\n');
term.write('-rw-r--r-- 1 root root 8234 Jan 30 11:22 package-lock.json\r\n');
term.write('-rw-r--r-- 1 root root 523 Jan 15 08:45 README.md\r\n');
term.write('\x1b[1;34mdrwxr-xr-x\x1b[0m 4 root root 4096 Jan 25 13:56 \x1b[1;34msrc\x1b[0m\r\n');
// Show prompt again
term.write('\x1b[1;32mroot@ubuntu-container\x1b[0m:\x1b[1;34m/app\x1b[0m# ');
}
});
console.log('6. Waiting for render...');
await page.waitForTimeout(1000);
// Check status
const status = await page.evaluate(() => {
const activeButton = document.querySelector('button[aria-pressed="true"]');
const mode = activeButton?.textContent || 'unknown';
let bufferText = '';
if (window._debugTerminal) {
try {
const buffer = window._debugTerminal.buffer.active;
for (let i = 0; i < Math.min(buffer.length, 10); i++) {
const line = buffer.getLine(i);
if (line) {
bufferText += line.translateToString(true) + '\n';
}
}
} catch (e) {
bufferText = 'Error: ' + e.message;
}
}
return {
mode,
bufferText: bufferText.trim()
};
});
console.log('Terminal Mode:', status.mode);
console.log('\nBuffer content:');
console.log('---');
console.log(status.bufferText);
console.log('---\n');
console.log('Errors captured:', errors.length);
console.log('Warnings captured:', warnings.length);
if (errors.length > 0) {
console.log('\nUnique errors:');
[...new Set(errors)].forEach((err, i) => console.log(`${i + 1}. ${err.substring(0, 150)}`));
}
await page.screenshot({
path: '/home/user/docker-swarm-termina/frontend/gnome-interactive-demo.png',
fullPage: false
});
console.log('✓ Screenshot saved\n');
if (status.mode.includes('Interactive') && status.bufferText.includes('Interactive Terminal Started')) {
console.log('✅ SUCCESS! Interactive GNOME terminal with visible output!');
} else {
console.log('⚠ Mode:', status.mode);
}
} catch (error) {
console.error('Error:', error.message);
console.error(error.stack);
process.exit(1);
} finally {
if (browser) await browser.close();
}
})();