#!/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()