274 lines
8.6 KiB
Python
274 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
最小功能测试框架 - Flask + SocketIO + threading
|
||
用于验证打包exe后HTTP和WebSocket服务的可用性
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from flask import Flask, render_template_string, jsonify
|
||
from flask_socketio import SocketIO, emit
|
||
import logging
|
||
from datetime import datetime
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 创建Flask应用
|
||
app = Flask(__name__)
|
||
app.config['SECRET_KEY'] = 'minimal-test-secret-key-2024'
|
||
|
||
# 初始化SocketIO，强制使用threading模式
|
||
try:
|
||
logger.info("初始化SocketIO（threading模式）...")
|
||
socketio = SocketIO(
|
||
app,
|
||
cors_allowed_origins='*',
|
||
async_mode='threading',
|
||
logger=False,
|
||
engineio_logger=False
|
||
)
|
||
logger.info(f"SocketIO初始化成功，异步模式: {socketio.async_mode}")
|
||
except Exception as e:
|
||
logger.error(f"SocketIO初始化失败: {e}")
|
||
sys.exit(1)
|
||
|
||
# HTML测试页面模板
|
||
HTML_TEMPLATE = """
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head>
|
||
<title>最小功能测试 - Flask + SocketIO</title>
|
||
<meta charset="utf-8">
|
||
<style>
|
||
body { font-family: Arial, sans-serif; margin: 40px; }
|
||
.container { max-width: 800px; margin: 0 auto; }
|
||
.section { margin: 20px 0; padding: 20px; border: 1px solid #ddd; border-radius: 5px; }
|
||
.success { color: green; }
|
||
.error { color: red; }
|
||
.info { color: blue; }
|
||
button { padding: 10px 20px; margin: 5px; cursor: pointer; }
|
||
#messages { height: 200px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; background: #f9f9f9; }
|
||
.message { margin: 5px 0; }
|
||
</style>
|
||
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<h1>最小功能测试框架</h1>
|
||
<p class="info">测试Flask HTTP服务和SocketIO WebSocket服务</p>
|
||
|
||
<!-- HTTP API测试 -->
|
||
<div class="section">
|
||
<h2>HTTP API 测试</h2>
|
||
<button onclick="testHttpApi()">测试 HTTP API</button>
|
||
<div id="http-result"></div>
|
||
</div>
|
||
|
||
<!-- WebSocket测试 -->
|
||
<div class="section">
|
||
<h2>WebSocket 测试</h2>
|
||
<button onclick="connectSocket()">连接 WebSocket</button>
|
||
<button onclick="disconnectSocket()">断开连接</button>
|
||
<button onclick="sendTestMessage()">发送测试消息</button>
|
||
<div>连接状态: <span id="connection-status" class="error">未连接</span></div>
|
||
<div id="messages"></div>
|
||
</div>
|
||
|
||
<!-- 系统信息 -->
|
||
<div class="section">
|
||
<h2>系统信息</h2>
|
||
<div id="system-info">
|
||
<p>服务器时间: {{ server_time }}</p>
|
||
<p>SocketIO模式: {{ socketio_mode }}</p>
|
||
<p>Flask版本: {{ flask_version }}</p>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<script>
|
||
let socket = null;
|
||
|
||
function addMessage(message, type = 'info') {
|
||
const messages = document.getElementById('messages');
|
||
const div = document.createElement('div');
|
||
div.className = `message ${type}`;
|
||
div.innerHTML = `[${new Date().toLocaleTimeString()}] ${message}`;
|
||
messages.appendChild(div);
|
||
messages.scrollTop = messages.scrollHeight;
|
||
}
|
||
|
||
function testHttpApi() {
|
||
const resultDiv = document.getElementById('http-result');
|
||
resultDiv.innerHTML = '测试中...';
|
||
|
||
fetch('/api/test')
|
||
.then(response => response.json())
|
||
.then(data => {
|
||
resultDiv.innerHTML = `<span class="success">✓ HTTP API 正常</span><br>响应: ${JSON.stringify(data, null, 2)}`;
|
||
})
|
||
.catch(error => {
|
||
resultDiv.innerHTML = `<span class="error">✗ HTTP API 错误: ${error}</span>`;
|
||
});
|
||
}
|
||
|
||
function connectSocket() {
|
||
if (socket && socket.connected) {
|
||
addMessage('WebSocket已连接', 'info');
|
||
return;
|
||
}
|
||
|
||
socket = io();
|
||
|
||
socket.on('connect', function() {
|
||
document.getElementById('connection-status').innerHTML = '<span class="success">已连接</span>';
|
||
addMessage('WebSocket连接成功', 'success');
|
||
});
|
||
|
||
socket.on('disconnect', function() {
|
||
document.getElementById('connection-status').innerHTML = '<span class="error">未连接</span>';
|
||
addMessage('WebSocket连接断开', 'error');
|
||
});
|
||
|
||
socket.on('test_response', function(data) {
|
||
addMessage(`收到服务器响应: ${JSON.stringify(data)}`, 'success');
|
||
});
|
||
|
||
socket.on('server_message', function(data) {
|
||
addMessage(`服务器消息: ${data.message}`, 'info');
|
||
});
|
||
}
|
||
|
||
function disconnectSocket() {
|
||
if (socket) {
|
||
socket.disconnect();
|
||
socket = null;
|
||
}
|
||
}
|
||
|
||
function sendTestMessage() {
|
||
if (!socket || !socket.connected) {
|
||
addMessage('请先连接WebSocket', 'error');
|
||
return;
|
||
}
|
||
|
||
const testData = {
|
||
message: 'Hello from client',
|
||
timestamp: new Date().toISOString()
|
||
};
|
||
|
||
socket.emit('test_message', testData);
|
||
addMessage(`发送消息: ${JSON.stringify(testData)}`, 'info');
|
||
}
|
||
|
||
// 页面加载时自动连接
|
||
window.onload = function() {
|
||
connectSocket();
|
||
};
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
# HTTP路由
|
||
@app.route('/')
|
||
def index():
|
||
"""主页面"""
|
||
import flask
|
||
return render_template_string(HTML_TEMPLATE,
|
||
server_time=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
socketio_mode=socketio.async_mode,
|
||
flask_version=flask.__version__
|
||
)
|
||
|
||
@app.route('/api/test')
|
||
def api_test():
|
||
"""HTTP API测试接口"""
|
||
return jsonify({
|
||
'status': 'success',
|
||
'message': 'HTTP API工作正常',
|
||
'timestamp': datetime.now().isoformat(),
|
||
'server_info': {
|
||
'socketio_mode': socketio.async_mode,
|
||
'working_directory': os.getcwd()
|
||
}
|
||
})
|
||
|
||
@app.route('/health')
|
||
def health_check():
|
||
"""健康检查接口"""
|
||
return jsonify({
|
||
'status': 'healthy',
|
||
'services': {
|
||
'http': 'running',
|
||
'websocket': 'running',
|
||
'socketio_mode': socketio.async_mode
|
||
},
|
||
'timestamp': datetime.now().isoformat()
|
||
})
|
||
|
||
# SocketIO事件处理
|
||
@socketio.on('connect')
|
||
def handle_connect():
|
||
"""客户端连接事件"""
|
||
logger.info(f"客户端连接: {request.sid if 'request' in globals() else 'unknown'}")
|
||
emit('server_message', {
|
||
'message': 'WebSocket连接成功',
|
||
'timestamp': datetime.now().isoformat(),
|
||
'socketio_mode': socketio.async_mode
|
||
})
|
||
|
||
@socketio.on('disconnect')
|
||
def handle_disconnect():
|
||
"""客户端断开连接事件"""
|
||
logger.info(f"客户端断开连接: {request.sid if 'request' in globals() else 'unknown'}")
|
||
|
||
@socketio.on('test_message')
|
||
def handle_test_message(data):
|
||
"""处理测试消息"""
|
||
logger.info(f"收到测试消息: {data}")
|
||
|
||
# 回复客户端
|
||
emit('test_response', {
|
||
'status': 'received',
|
||
'original_message': data,
|
||
'server_timestamp': datetime.now().isoformat(),
|
||
'socketio_mode': socketio.async_mode
|
||
})
|
||
|
||
def main():
|
||
"""主函数"""
|
||
logger.info("="*50)
|
||
logger.info("最小功能测试框架启动")
|
||
logger.info("="*50)
|
||
logger.info(f"工作目录: {os.getcwd()}")
|
||
logger.info(f"Python版本: {sys.version}")
|
||
logger.info(f"SocketIO异步模式: {socketio.async_mode}")
|
||
|
||
try:
|
||
# 启动服务器
|
||
logger.info("启动服务器 http://localhost:5000")
|
||
logger.info("按 Ctrl+C 停止服务器")
|
||
|
||
socketio.run(
|
||
app,
|
||
host='0.0.0.0',
|
||
port=5000,
|
||
debug=False,
|
||
allow_unsafe_werkzeug=True
|
||
)
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("用户中断，正在关闭服务器...")
|
||
except Exception as e:
|
||
logger.error(f"服务器启动失败: {e}")
|
||
input("按回车键退出...")
|
||
sys.exit(1)
|
||
|
||
if __name__ == '__main__':
|
||
main() |