134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import requests
|
|
import asyncio
|
|
import aiohttp
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(message)s')
|
|
logger = logging.getLogger('MT5-Manual-Test')
|
|
|
|
# 配置
|
|
CONFIG = {
|
|
'user': '76888962',
|
|
'password': 'LZ-trade666888',
|
|
'host': '18.163.85.196',
|
|
'port': 443,
|
|
'connectTimeoutSeconds': 30
|
|
}
|
|
|
|
BASE_URL = "http://43.167.188.220:5000"
|
|
|
|
|
|
def test_manual_connect():
|
|
"""手动测试连接"""
|
|
logger.info("🔧 手动测试连接...")
|
|
|
|
try:
|
|
url = f"{BASE_URL}/Connect"
|
|
|
|
logger.info(f"请求URL: {url}")
|
|
logger.info(f"请求参数: {CONFIG}")
|
|
|
|
response = requests.get(url, params=CONFIG, timeout=30)
|
|
|
|
logger.info(f"状态码: {response.status_code}")
|
|
logger.info(f"响应头: {dict(response.headers)}")
|
|
logger.info(f"响应内容: {response.text}")
|
|
|
|
if response.status_code == 200:
|
|
token = response.text
|
|
logger.info(f"✅ 连接成功! Token: {token}")
|
|
|
|
# 测试其他端点
|
|
test_other_endpoints(token)
|
|
return True
|
|
else:
|
|
logger.error(f"❌ 连接失败: {response.status_code}")
|
|
return False
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"❌ 请求异常: {e}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"❌ 未知错误: {e}")
|
|
return False
|
|
|
|
|
|
def test_other_endpoints(token):
|
|
"""测试其他端点"""
|
|
logger.info("\n🔧 测试其他端点...")
|
|
|
|
endpoints = [
|
|
'/CheckConnect',
|
|
'/AccountSummary',
|
|
'/Symbols',
|
|
'/GetQuote?symbol=EURUSD'
|
|
]
|
|
|
|
for endpoint in endpoints:
|
|
try:
|
|
url = f"{BASE_URL}{endpoint}"
|
|
params = {'id': token}
|
|
|
|
logger.info(f"测试端点: {endpoint}")
|
|
response = requests.get(url, params=params, timeout=10)
|
|
|
|
logger.info(f" 状态码: {response.status_code}")
|
|
if response.status_code == 200:
|
|
logger.info(f" ✅ 成功")
|
|
if len(response.text) < 100: # 避免输出过长
|
|
logger.info(f" 响应: {response.text}")
|
|
else:
|
|
logger.error(f" ❌ 失败")
|
|
|
|
except Exception as e:
|
|
logger.error(f" ❌ 错误: {e}")
|
|
|
|
|
|
async def test_async_manual():
|
|
"""异步手动测试"""
|
|
logger.info("\n🔧 异步手动测试...")
|
|
|
|
try:
|
|
url = f"{BASE_URL}/Connect"
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
logger.info(f"异步请求URL: {url}")
|
|
async with session.get(url, params=CONFIG, timeout=30) as response:
|
|
logger.info(f"异步状态码: {response.status}")
|
|
text = await response.text()
|
|
logger.info(f"异步响应: {text}")
|
|
|
|
if response.status == 200:
|
|
logger.info("✅ 异步连接成功!")
|
|
return True
|
|
else:
|
|
logger.error("❌ 异步连接失败!")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ 异步测试失败: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""运行手动测试"""
|
|
print("🚀 MT5 手动连接测试")
|
|
print("="*50)
|
|
|
|
# 同步测试
|
|
sync_success = test_manual_connect()
|
|
|
|
# 异步测试
|
|
async_success = asyncio.run(test_async_manual())
|
|
|
|
print("\n" + "="*50)
|
|
print("📊 手动测试结果")
|
|
print(f"同步测试: {'✅ 通过' if sync_success else '❌ 失败'}")
|
|
print(f"异步测试: {'✅ 通过' if async_success else '❌ 失败'}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |