#!/usr/bin/env python3
"""Local smoke test for the device automation and load-test scripts."""

from __future__ import annotations

import csv
import http.server
import os
import socketserver
import subprocess
import sys
import tempfile
import threading
import urllib.parse
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]


class ApiHandler(http.server.BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        self.rfile.read(length)
        parsed = urllib.parse.urlparse(self.path)
        route = urllib.parse.parse_qs(parsed.query).get("s", [""])[0]
        if parsed.path == "/p/login/login":
            payload = b'{"code":0,"data":{"_token":"MOCK-PC-TOKEN"}}'
        elif parsed.path == "/p/staff/tree":
            payload = (
                b'{"code":0,"data":{"department":'
                b'[{"uuid":"ROOT","children":[{"uuid":"MOCK-DEPARTMENT"}]}]}}'
            )
        elif parsed.path == "/p/staff/add":
            payload = (
                b'{"code":0,"data":{"uuid":"MOCK-STAFF-UUID",'
                b'"ai_user_id":"1000001"}}'
            )
        elif parsed.path == "/p/Equipment/pushList":
            payload = (
                b'{"code":0,"data":{"rows":'
                b'[{"id":120,"code":"MOCK460001"}],"total":1}}'
            )
        elif parsed.path == "/p/mealOrder/list":
            payload = b'{"code":0,"data":{"rows":[{"id":1001}],"total":1}}'
        elif route == "/api/passport/loginAccount":
            payload = b'{"status":200,"data":{"token":"MOCK-MINI-TOKEN"}}'
        elif route == "/api/mealOrder/list":
            payload = (
                b'{"status":200,"data":{"list":{"data":[{"id":1001}],'
                b'"total":1}}}'
            )
        elif route.startswith("/api/"):
            payload = b'{"status":200,"message":"mock success","data":[]}'
        else:
            payload = b'{"code":0,"message":"mock success","data":[]}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, _format: str, *_args: object) -> None:
        return


class MqttHandler(socketserver.BaseRequestHandler):
    def handle(self) -> None:
        self.request.recv(4096)
        self.request.sendall(b"\x20\x02\x00\x00")


class ReusableTcpServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


def run(command: list[str], env: dict[str, str] | None = None) -> None:
    subprocess.run(command, cwd=ROOT, env=env, check=True)


def main() -> int:
    http_server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), ApiHandler)
    mqtt_server = ReusableTcpServer(("127.0.0.1", 0), MqttHandler)
    for server in (http_server, mqtt_server):
        threading.Thread(target=server.serve_forever, daemon=True).start()

    try:
        with tempfile.TemporaryDirectory(prefix="csfzx-device-test-") as temp_name:
            temp = Path(temp_name)
            face_image = temp / "face.jpg"
            face_image.write_bytes(b"mock-face-image")
            environment = os.environ.copy()
            environment.update(
                {
                    "BASE_URL": f"http://127.0.0.1:{http_server.server_port}",
                    "MQTT_HOST": "127.0.0.1",
                    "MQTT_PORT": str(mqtt_server.server_address[1]),
                    "DEVICE_CODE": "MOCK460001",
                    "MATE_CODE": "MOCK-MATE-CODE",
                    "DISH_UUID": "MOCK-DISH-UUID",
                    "FACE_IMAGE": str(face_image),
                    "PC_PASSWORD": "mock-pc-password",
                    "MINIAPP_PASSWORD": "mock-mini-password",
                    "REPORT_ROOT": str(temp / "automation"),
                }
            )
            run(["bash", "csfzx-device-automation-test.sh"], environment)
            automation_reports = list((temp / "automation").glob("device-full-flow-*"))
            assert len(automation_reports) == 1
            report = automation_reports[0]
            assert "RESULT=PASS" in (report / "summary.txt").read_text(encoding="utf-8")
            assert "MOCK-PC-TOKEN" not in (report / "pc_login.json").read_text(
                encoding="utf-8"
            )
            assert "MOCK-MINI-TOKEN" not in (
                report / "miniapp_login.json"
            ).read_text(encoding="utf-8")

            common = [
                sys.executable,
                "csfzx-device-load-test.py",
                "--device-prefix",
                "MOCK460",
                "--device-count",
                "5",
                "--workers",
                "3",
                "--execute",
            ]
            run(
                common
                + [
                    "--mode",
                    "heartbeat",
                    "--base-url",
                    environment["BASE_URL"],
                    "--report-dir",
                    str(temp / "heartbeat"),
                ]
            )
            run(
                common
                + [
                    "--mode",
                    "mqtt",
                    "--mqtt-host",
                    "127.0.0.1",
                    "--mqtt-port",
                    environment["MQTT_PORT"],
                    "--report-dir",
                    str(temp / "mqtt"),
                ]
            )

            order_cases = temp / "order-cases.csv"
            with order_cases.open("w", encoding="utf-8", newline="") as handle:
                writer = csv.DictWriter(
                    handle,
                    fieldnames=(
                        "staff_uuid",
                        "equipment_code",
                        "dishes_uuid",
                        "weight",
                    ),
                )
                writer.writeheader()
                for index in range(3):
                    writer.writerow(
                        {
                            "staff_uuid": f"MOCK-STAFF-{index}",
                            "equipment_code": f"MOCK460{index:03d}",
                            "dishes_uuid": "MOCK-DISH-UUID",
                            "weight": "120",
                        }
                    )
            run(
                [
                    sys.executable,
                    "csfzx-device-load-test.py",
                    "--mode",
                    "order",
                    "--base-url",
                    environment["BASE_URL"],
                    "--order-cases",
                    str(order_cases),
                    "--workers",
                    "2",
                    "--execute",
                    "--confirm-order-load",
                    "--report-dir",
                    str(temp / "order"),
                ]
            )
    finally:
        http_server.shutdown()
        mqtt_server.shutdown()

    print("MOCK_SMOKE_TEST=PASS")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())