416 lines
15 KiB
Python
416 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mt5Bridge end-to-end test.
|
|
|
|
Usage:
|
|
python test_all.py # default remote, read-only + check
|
|
python test_all.py --trades # enable trade test (open/close)
|
|
python test_all.py --base http://host:p # custom base URL
|
|
python test_all.py --symbol XAUUSD # custom symbol
|
|
"""
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
from typing import Callable
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
print("missing requests: pip install requests")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
import websocket
|
|
except ImportError:
|
|
print("missing websocket-client: pip install websocket-client")
|
|
sys.exit(1)
|
|
|
|
|
|
# ─────────── Colors ───────────
|
|
C_G = "\033[92m"
|
|
C_R = "\033[91m"
|
|
C_Y = "\033[93m"
|
|
C_B = "\033[96m"
|
|
C_X = "\033[0m"
|
|
C_D = "\033[2m"
|
|
|
|
|
|
# ─────────── Stats ───────────
|
|
PASS = 0
|
|
FAIL = 0
|
|
ERRORS: list[str] = []
|
|
|
|
|
|
def case(name: str, fn: Callable[[], bool]) -> None:
|
|
global PASS, FAIL
|
|
sys.stdout.write(f"{C_B}- {name}{C_X} ")
|
|
try:
|
|
ok = fn()
|
|
# lambda with walrus expressions returns tuple; take last element
|
|
if isinstance(ok, tuple):
|
|
ok = ok[-1]
|
|
if ok:
|
|
print(f"{C_G}PASS{C_X}")
|
|
PASS += 1
|
|
else:
|
|
print(f"{C_R}FAIL{C_X}")
|
|
FAIL += 1
|
|
ERRORS.append(name)
|
|
except AssertionError as e:
|
|
print(f"{C_R}FAIL{C_X} {C_D}{e}{C_X}")
|
|
FAIL += 1
|
|
ERRORS.append(f"{name}: {e}")
|
|
except Exception as e:
|
|
msg = f"{type(e).__name__}: {e}"
|
|
print(f"{C_R}ERROR{C_X} {C_D}{msg}{C_X}")
|
|
FAIL += 1
|
|
ERRORS.append(f"{name}: {msg}")
|
|
|
|
|
|
# ─────────── HTTP client ───────────
|
|
class Bridge:
|
|
def __init__(self, base: str, key: str, timeout: float = 15.0):
|
|
self.base = base.rstrip("/")
|
|
self.key = key
|
|
self.timeout = timeout
|
|
self.s = requests.Session()
|
|
self.s.headers.update({"X-API-Key": key})
|
|
|
|
def get(self, path: str, params: dict | None = None) -> dict:
|
|
r = self.s.get(f"{self.base}{path}", params=params, timeout=self.timeout)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def post(self, path: str, body: dict) -> dict:
|
|
r = self.s.post(f"{self.base}{path}", json=body, timeout=self.timeout)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def delete(self, path: str) -> dict:
|
|
r = self.s.delete(f"{self.base}{path}", timeout=self.timeout)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def expect_status(self, method: str, path: str, expected: int, **kw) -> int:
|
|
r = self.s.request(method, f"{self.base}{path}", timeout=self.timeout, **kw)
|
|
assert r.status_code == expected, f"HTTP {r.status_code} != {expected}: {r.text[:200]}"
|
|
return r.status_code
|
|
|
|
|
|
# ─────────── SSE ───────────
|
|
def sse_first(base: str, key: str, symbol: str, timeout_sec: float) -> tuple[bool, str]:
|
|
"""Open SSE, wait for first data: line or timeout."""
|
|
import urllib.request
|
|
url = f"{base}/stream/ticks-sse/{symbol}?key={key}"
|
|
req = urllib.request.Request(url, headers={"X-API-Key": key, "Accept": "text/event-stream"})
|
|
with urllib.request.urlopen(req, timeout=timeout_sec + 5) as resp:
|
|
deadline = time.time() + timeout_sec
|
|
while time.time() < deadline:
|
|
line = resp.readline()
|
|
if not line:
|
|
break
|
|
line = line.decode("utf-8", "ignore").rstrip("\r\n")
|
|
if line.startswith("data: "):
|
|
return True, line[6:80]
|
|
return False, ""
|
|
|
|
|
|
# ─────────── WebSocket ───────────
|
|
def ws_first(base: str, key: str, symbol: str, timeout_sec: float) -> tuple[bool, str]:
|
|
ws_url = "ws" + base[4:] if base.startswith("http") else base
|
|
ws_url = f"{ws_url}/stream/ticks/{symbol}?key={key}"
|
|
deadline = time.time() + timeout_sec
|
|
ws = websocket.create_connection(ws_url, timeout=timeout_sec + 5,
|
|
header=[f"X-API-Key: {key}"])
|
|
try:
|
|
ws.settimeout(1.0)
|
|
last_type = None
|
|
while time.time() < deadline:
|
|
try:
|
|
opcode, data = ws.recv_data()
|
|
except (websocket.WebSocketTimeoutException, TimeoutError):
|
|
continue
|
|
if opcode in (websocket.ABNF.OPCODE_TEXT,):
|
|
text = data.decode("utf-8", "ignore")
|
|
if '"type":"tick"' in text:
|
|
return True, text[:80]
|
|
last_type = "hb"
|
|
elif opcode == websocket.ABNF.OPCODE_CLOSE:
|
|
break
|
|
# fall back: got heartbeat means transport works, but MT5 may need time
|
|
return last_type == "hb", f"heartbeat (no tick yet)"
|
|
finally:
|
|
try:
|
|
ws.close()
|
|
except Exception:
|
|
pass
|
|
return False, ""
|
|
|
|
|
|
# ─────────── Warm-up ───────────
|
|
def warmup(b: Bridge, symbol: str, timeout_sec: float = 15.0) -> None:
|
|
try:
|
|
ok, detail = ws_first(b.base, b.key, symbol, timeout_sec)
|
|
if "tick" in detail:
|
|
print(f"{C_D} [WS warm-up ok, tick received]{C_X}")
|
|
elif "heartbeat" in detail:
|
|
print(f"{C_D} [WS warm-up: got heartbeat, waiting for MT5 to initialize quote]{C_X}")
|
|
time.sleep(3) # give MT5 time to populate symbol data
|
|
else:
|
|
print(f"{C_D} [WS warm-up failed (no data)]{C_X}")
|
|
except Exception as e:
|
|
print(f"{C_D} [WS warm-up exception: {e}]{C_X}")
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════
|
|
def run(args) -> int:
|
|
b = Bridge(args.base, args.key)
|
|
sym = args.symbol
|
|
|
|
print()
|
|
print(f"{C_Y}=== Mt5Bridge E2E test ==={C_X}")
|
|
print(f"Base : {args.base}")
|
|
print(f"Symbol : {sym}")
|
|
print(f"Trades : {'enabled (will place real orders)' if args.trades else 'disabled (--trades to enable)'}")
|
|
print()
|
|
|
|
# ── system / account ──
|
|
case("1. GET /health", lambda: (
|
|
r := b.get("/health"),
|
|
r["status"] == "healthy" and r["mt5_connected"] is True
|
|
))
|
|
|
|
case("2. GET /account", lambda: (
|
|
r := b.get("/account"),
|
|
a := r["data"][0],
|
|
print(f" login={a['login']} server={a['server']} balance={a['balance']} currency={a['currency']}"),
|
|
a["login"] > 0 and a["balance"] >= 0
|
|
))
|
|
|
|
# ── warm-up to add symbol to Market Watch ──
|
|
print(f"{C_D}-- warm-up (WS) --{C_X}")
|
|
warmup(b, sym, timeout_sec=6)
|
|
|
|
# ── quote ──
|
|
case("3. GET /symbols/" + sym, lambda: (
|
|
r := b.get(f"/symbols/{sym}"),
|
|
s := r["data"][0],
|
|
print(f" bid={s['bid']} ask={s['ask']} digits={s['digits']} point={s['point']}"),
|
|
s["bid"] > 0 and s["digits"] > 0
|
|
))
|
|
|
|
case("4. GET /symbols/" + sym + "/tick", lambda: (
|
|
r := b.get(f"/symbols/{sym}/tick"),
|
|
t := r["data"][0],
|
|
print(f" bid={t['bid']} ask={t['ask']} time={t['time']}"),
|
|
t["bid"] > 0 and t["ask"] >= t["bid"]
|
|
))
|
|
|
|
# ── rates ──
|
|
case("5. GET /rates/from-pos (H1, count=5)", lambda: (
|
|
r := b.get("/rates/from-pos", {"symbol": sym, "timeframe": "TIMEFRAME_H1",
|
|
"start_pos": 0, "count": 5}),
|
|
print(f" count={r['count']} first={r['data'][0]['time']} last={r['data'][-1]['time']}"),
|
|
r["count"] == 5
|
|
))
|
|
|
|
case("6. GET /rates/from-date (cross-day 7-7~7-8, H1)", lambda: (
|
|
r := b.get("/rates/from-date", {"symbol": sym, "timeframe": "TIMEFRAME_H1",
|
|
"date_from": "2026-07-07", "date_to": "2026-07-08"}),
|
|
print(f" count={r['count']} first={r['data'][0]['time']}"),
|
|
r["count"] > 0
|
|
))
|
|
|
|
case("6b. GET /rates/from-date (same day 7-7~7-7, M1)", lambda: (
|
|
r := b.get("/rates/from-date", {"symbol": sym, "timeframe": "TIMEFRAME_M1",
|
|
"date_from": "2026-07-07", "date_to": "2026-07-07"}),
|
|
print(f" count={r['count']} (bug fix: should be > 0)"),
|
|
r["count"] > 0
|
|
))
|
|
|
|
case("6c. GET /rates/from-date (date_from > date_to, expect 400)", lambda: (
|
|
b.expect_status("GET", "/rates/from-date", 400,
|
|
params={"symbol": sym, "timeframe": "TIMEFRAME_H1",
|
|
"date_from": "2026-07-08", "date_to": "2026-07-01"}),
|
|
print(" status=400 ok"),
|
|
True
|
|
))
|
|
|
|
# ── positions / orders / history ──
|
|
case("7. GET /positions", lambda: (
|
|
r := b.get("/positions"),
|
|
print(f" count={r['count']}"),
|
|
True
|
|
))
|
|
|
|
case("7b. GET /positions?symbol=__NOPE__", lambda: (
|
|
r := b.get("/positions", {"symbol": "__NOPE__"}),
|
|
print(f" count={r['count']}"),
|
|
r["count"] == 0
|
|
))
|
|
|
|
case("8. GET /orders", lambda: (
|
|
r := b.get("/orders"),
|
|
print(f" count={r['count']}"),
|
|
True
|
|
))
|
|
|
|
case("9. GET /history/deals (7-1~7-8)", lambda: (
|
|
r := b.get("/history/deals", {"date_from": "2026-07-01", "date_to": "2026-07-08",
|
|
"symbol": sym}),
|
|
print(f" count={r['count']}"),
|
|
True
|
|
))
|
|
|
|
# ── gvar ──
|
|
import uuid
|
|
gvar = f"TEST_{uuid.uuid4().hex[:8]}"
|
|
|
|
case(f"10a. POST /gvar/{gvar}", lambda: (
|
|
r := b.post(f"/gvar/{gvar}", {"value": 42.5}),
|
|
print(f" value={r['value']}"),
|
|
r["value"] == 42.5
|
|
))
|
|
|
|
case(f"10b. GET /gvar/{gvar}", lambda: (
|
|
r := b.get(f"/gvar/{gvar}"),
|
|
print(f" value={r['value']}"),
|
|
r["value"] == 42.5
|
|
))
|
|
|
|
case("10c. GET /gvar (expect at least 1 TEST_*)", lambda: (
|
|
r := b.get("/gvar"),
|
|
found := sum(1 for g in r["data"] if g["name"].startswith("TEST_")),
|
|
print(f" total={r['count']} TEST_*={found}"),
|
|
found >= 1
|
|
))
|
|
|
|
case(f"10d. DELETE /gvar/{gvar}", lambda: (
|
|
r := b.delete(f"/gvar/{gvar}"),
|
|
print(f" deleted={r['deleted']}"),
|
|
r["deleted"] == gvar
|
|
))
|
|
|
|
# ── order check ──
|
|
case("11. POST /order/check (buy 0.01 " + sym + ")", lambda: (
|
|
bid := b.get(f"/symbols/{sym}/tick")["data"][0]["bid"],
|
|
r := b.post("/order/check", {
|
|
"action": 1, "symbol": sym, "volume": 0.01, "order_type": 0,
|
|
"price": bid, "sl": 0, "tp": 0, "magic": 0, "comment": "test", "deviation": 10,
|
|
"type_filling": 0, # ORDER_FILLING_FOK - required by ICMarkets
|
|
}),
|
|
rc := r["data"]["retcode"],
|
|
print(f" retcode={rc} comment={r['data']['comment']}"),
|
|
rc == 0 # with type_filling set, should be 0
|
|
))
|
|
|
|
# ── SSE ──
|
|
case(f"12. SSE /stream/ticks-sse/{sym} (<=35s for tick or heartbeat)", lambda: (
|
|
result := sse_first(b.base, b.key, sym, timeout_sec=35),
|
|
ok := result[0],
|
|
snippet := result[1],
|
|
print(f" {snippet}"),
|
|
ok
|
|
))
|
|
|
|
# ── WebSocket ──
|
|
case(f"13. WS /stream/ticks/{sym} (<=35s for tick or heartbeat)", lambda: (
|
|
result := ws_first(b.base, b.key, sym, timeout_sec=35),
|
|
ok := result[0],
|
|
snippet := result[1],
|
|
print(f" {snippet}"),
|
|
ok
|
|
))
|
|
|
|
# ── trades (opt-in) ──
|
|
if args.trades:
|
|
print()
|
|
print(f"{C_Y}[!] TRADES enabled - will place real orders{C_X}")
|
|
print()
|
|
|
|
case("14. POST /order/send -> /position/close (open 0.01 then close)", lambda: (
|
|
bid := b.get(f"/symbols/{sym}/tick")["data"][0]["bid"],
|
|
r := b.post("/order/send", {"request": {
|
|
"action": 1, "symbol": sym, "volume": 0.01, "order_type": 0,
|
|
"price": bid, "sl": 0, "tp": 0, "magic": 99001,
|
|
"comment": "selftest", "deviation": 10, "type_filling": 0,
|
|
}}),
|
|
print(f" send retcode={r['data']['retcode']} order={r['data']['order']}"),
|
|
r["data"]["retcode"] == 10009,
|
|
time.sleep(0.5),
|
|
r2 := b.post("/position/close", {"ticket": r["data"]["order"]}),
|
|
print(f" close retcode={r2['data']['retcode']}"),
|
|
r2["data"]["retcode"] == 10009
|
|
))
|
|
|
|
case("15. POST /position/modify (change SL/TP, then restore)", lambda: (
|
|
r := b.get("/positions"),
|
|
print(f" positions count={r['count']}"),
|
|
(r["count"] == 0) and (print(" [no positions, skip]"), True)[-1]
|
|
or (
|
|
p := r["data"][0],
|
|
orig_sl := p["sl"], orig_tp := p["tp"],
|
|
new_sl := round(p["price_current"] - 50, 2),
|
|
new_tp := round(p["price_current"] + 50, 2),
|
|
r2 := b.post("/position/modify", {"ticket": p["ticket"], "sl": new_sl, "tp": new_tp}),
|
|
print(f" modify retcode={r2['data']['retcode']}"),
|
|
r2["data"]["retcode"] == 10009,
|
|
r3 := b.post("/position/modify", {"ticket": p["ticket"], "sl": orig_sl, "tp": orig_tp}),
|
|
print(f" restore retcode={r3['data']['retcode']}"),
|
|
r3["data"]["retcode"] == 10009,
|
|
)[-1]
|
|
))
|
|
|
|
case("16. POST /positions/close-batch (magic=99001)", lambda: (
|
|
r := b.post("/positions/close-batch", {"magic": 99001}),
|
|
print(f" closed={r['closed']} failed={r['failed']}"),
|
|
True
|
|
))
|
|
|
|
case("17. POST /order/cancel (place pending then cancel)", lambda: (
|
|
bid := b.get(f"/symbols/{sym}/tick")["data"][0]["bid"],
|
|
pend_price := int(bid * 0.9),
|
|
r := b.post("/order/send", {"request": {
|
|
"action": 5, "symbol": sym, "volume": 0.01, "order_type": 2,
|
|
"price": pend_price, "sl": 0, "tp": 0, "magic": 99002,
|
|
"comment": "selftest-pending", "deviation": 0, "type_filling": 0,
|
|
}}),
|
|
print(f" place retcode={r['data']['retcode']} order={r['data']['order']}"),
|
|
r["data"]["retcode"] == 10009,
|
|
time.sleep(0.5),
|
|
r2 := b.post("/order/cancel", {"ticket": r["data"]["order"]}),
|
|
print(f" cancel retcode={r2['data']['retcode']}"),
|
|
r2["data"]["retcode"] == 10009
|
|
))
|
|
else:
|
|
print()
|
|
print(f"{C_D}[skip trades -- use --trades to enable]{C_X}")
|
|
|
|
# ── summary ──
|
|
print()
|
|
print(f"{C_Y}=== Summary ==={C_X}")
|
|
print(f"Pass: {C_G}{PASS}{C_X}")
|
|
color = C_R if FAIL else C_G
|
|
print(f"Fail: {color}{FAIL}{C_X}")
|
|
if ERRORS:
|
|
print()
|
|
print(f"{C_R}Failed cases:{C_X}")
|
|
for e in ERRORS:
|
|
print(f" - {e}")
|
|
print()
|
|
return 0 if FAIL == 0 else 1
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description="Mt5Bridge end-to-end test")
|
|
p.add_argument("--base", default="http://61.164.252.86:13485", help="Bridge base URL")
|
|
p.add_argument("--key", default="UiHMqtaYLZzwBdcuS4RFmEGhgDO8N2eI", help="API Key")
|
|
p.add_argument("--symbol", default="XAUUSD", help="Test symbol")
|
|
p.add_argument("--trades", action="store_true", help="Enable trade test (places real orders)")
|
|
args = p.parse_args()
|
|
sys.exit(run(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |