from __future__ import annotations

import json
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path

import pytest

from cognitive_agent.application.self_repair import SelfRepairService
from cognitive_agent.auth import AuthConfig, AuthRuntime
from cognitive_agent.server import SeatDemoHandler


def _auth_disabled() -> AuthRuntime:
    return AuthRuntime(
        AuthConfig(
            enabled=False,
            username="",
            password_hash="",
            session_secret=b"",
            session_ttl_seconds=3600,
            login_attempts=5,
            login_window_seconds=300,
            requests_per_minute=120,
            trust_proxy_headers=False,
        )
    )


class TestSelfRepairApi:
    @classmethod
    def setup_class(cls) -> None:
        cls.repair = SelfRepairService()
        cls.original_repair = SeatDemoHandler.repair
        cls.original_auth = SeatDemoHandler.auth
        SeatDemoHandler.repair = cls.repair
        SeatDemoHandler.auth = _auth_disabled()
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SeatDemoHandler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def teardown_class(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=2)
        SeatDemoHandler.repair = cls.original_repair
        SeatDemoHandler.auth = cls.original_auth

    def request(
        self,
        path: str,
        *,
        method: str = "GET",
        payload: dict[str, object] | None = None,
        role: str | None = "admin",
    ) -> urllib.request.Request:
        headers = {"Accept": "application/json"}
        data = None
        if role is not None:
            headers["X-Cognitive-Agent-Role"] = role
        if payload is not None:
            headers["Content-Type"] = "application/json"
            data = json.dumps(payload).encode("utf-8")
        return urllib.request.Request(
            f"{self.base_url}{path}", method=method, headers=headers, data=data
        )

    def test_user_cannot_call_self_repair_api(self) -> None:
        for request in (
            self.request("/api/self-repair/state", role="user"),
            self.request(
                "/api/self-repair/healing/start",
                method="POST",
                payload={"case_id": "FI-FIELD-01", "execution_mode": "shadow"},
                role="user",
            ),
        ):
            with pytest.raises(urllib.error.HTTPError) as caught:
                urllib.request.urlopen(request, timeout=3)
            assert caught.value.code == 403
            body = json.loads(caught.value.read().decode("utf-8"))
            assert body["error"] == "admin_required"

    def test_admin_can_read_and_start_shadow_repair(self) -> None:
        with urllib.request.urlopen(
            self.request("/api/self-repair/state"), timeout=3
        ) as response:
            state = json.loads(response.read().decode("utf-8"))
        assert state["schema"] == "seat-self-healing-console-state-v1"
        assert len(state["scenarios"]) == 6
        assert state["hardware"]["connector_enabled_count"] == 21

        with urllib.request.urlopen(
            self.request(
                "/api/self-repair/runtime",
                method="POST",
                payload={"active": True, "execution_mode": "shadow"},
            ),
            timeout=3,
        ) as response:
            runtime = json.loads(response.read().decode("utf-8"))["runtime"]
        assert runtime["active"] is True

        with urllib.request.urlopen(
            self.request(
                "/api/self-repair/healing/start",
                method="POST",
                payload={"case_id": "FI-FIELD-01", "execution_mode": "shadow"},
            ),
            timeout=3,
        ) as response:
            started = json.loads(response.read().decode("utf-8"))["healing"]
        assert started["case_id"] == "FI-FIELD-01"
        assert started["execution_mode"] == "shadow"
        assert started["injection_scope"] == "SHADOW_TELEMETRY_ONLY"

        with urllib.request.urlopen(
            self.request(
                "/api/self-repair/healing/reset",
                method="POST",
                payload={"execution_mode": "shadow"},
            ),
            timeout=3,
        ) as response:
            reset = json.loads(response.read().decode("utf-8"))
        assert reset["ok"] is True
        assert reset["healing"]["phase"] == "IDLE"

    def test_rule_self_test_is_available_without_hardware(self) -> None:
        with urllib.request.urlopen(
            self.request(
                "/api/self-repair/hardware/self-test", method="POST", payload={}
            ),
            timeout=5,
        ) as response:
            result = json.loads(response.read().decode("utf-8"))
        assert result["ok"] is True
        assert result["self_test"]["result"] == "PASS"

    def test_frontend_assets_and_role_transition_are_served(self) -> None:
        with urllib.request.urlopen(f"{self.base_url}/", timeout=3) as response:
            html = response.read().decode("utf-8")
        with urllib.request.urlopen(
            f"{self.base_url}/self-repair.js?v=repair-20", timeout=3
        ) as response:
            script = response.read().decode("utf-8").replace("\r\n", "\n")
        with urllib.request.urlopen(
            f"{self.base_url}/self-repair.css?v=repair-21", timeout=3
        ) as response:
            styles = response.read().decode("utf-8")
        with urllib.request.urlopen(
            f"{self.base_url}/self-repair-seat3d.js?v=repair-19", timeout=3
        ) as response:
            seat3d = response.read().decode("utf-8")
        with urllib.request.urlopen(
            f"{self.base_url}/console-v2.js?v=desktop-refactor-31", timeout=3
        ) as response:
            console_script = response.read().decode("utf-8")

        assert 'href="/self-repair.css?v=repair-21"' in html
        assert 'src="/self-repair.js?v=repair-20"' in html
        assert 'class="v2-module-heading"' in script
        assert 'SELF-REPAIR' in script
        assert 'padding: var(--v2-module-title-block) 0 0' in styles
        assert 'margin-inline: var(--v2-module-title-inline)' in styles
        assert 'src="/self-repair-seat3d.js?v=repair-19"' in html
        assert "量产故障注入" in script
        assert "运行与自修复日志" in script
        assert "启动座椅运行" in script
        assert "开始正常任务" not in script
        assert 'id="repairSeatCanvas"' in script
        assert 'id="repairSeatViewer"' in script
        assert "self-repair-seat-reset" in script
        assert "seat-healing-state" in script
        assert 'id="repairConfirmDialog"' in script
        assert "confirmAction(" in script
        assert "window.confirm" not in script
        assert 'hardware.link?.adapter === "OK"' in script
        assert 'PASSED: ["21针通过", "online"]' in script
        assert "connectorRows" in script
        assert "Polling refreshes data only" in script
        assert "renderRecords();\n    showView(activeView);" not in script
        assert 'let latest = fallbackState();' in script
        assert 'renderAll();\n    window.lucide?.createIcons' in script
        assert '"WDT复位后失能"' in script
        assert "--repair-deep-blue" in styles
        assert '.console-v2[data-module="self-repair"] .section-kicker' in styles
        assert 'grid-template-areas: "twin cases logs"' in styles
        assert ".v2-repair-event-stream" in styles
        assert "background: rgba(255, 255, 255, 0.94)" in styles
        assert "border-color: color-mix(in srgb, var(--v2-accent) 18%, white)" in styles
        assert ".v2-repair-toolbar button.primary:hover:not(:disabled)" in styles
        assert "color: #fff;" in styles
        assert "seat-healing-state" in seat3d
        assert "function backrestShape" in seat3d
        assert "new THREE.CapsuleGeometry" in seat3d
        assert 'cssThemeColor("--v2-accent", "#315eaa")' in seat3d
        assert 'viewer.addEventListener("pointerdown"' in seat3d
        assert 'window.addEventListener("pointermove"' in seat3d
        assert "window.addEventListener(resetEvent" in seat3d
        assert "function viewerHasLayout(viewer)" in seat3d
        assert "function deferSeat3dUntilVisible(viewer, options)" in seat3d
        assert "if (!viewerHasLayout(viewer))" in seat3d
        assert 'window.addEventListener("cognitive-agent:module-change"' in seat3d
        assert 'canvas.addEventListener("webglcontextlost"' in seat3d
        assert '["self-calibration", "self-repair"].includes(module)' in console_script
        assert 'roleSelect.disabled = ui.activeModule === "self-repair"' in console_script


def test_self_repair_bridge_uses_the_backend_runtime_directory() -> None:
    source = (
        Path(__file__).resolve().parents[2]
        / "tools"
        / "zcan"
        / "seat_self_repair_bridge.py"
    ).read_text(encoding="utf-8")

    assert '"artifacts", "tmp", "self-repair", "runtime"' in source
    assert 'os.path.join(BASE_DIR, "seat_function_runtime")' not in source
    assert "A silent or disconnected CAN bus must not kill" in source
    assert 'self.message("GUI STOP_ALL发送失败' in source
    assert 'self.message("GUI thermal off发送失败' in source
    assert "if opcode == 0x01 and self.latest_system is not None" in source
    assert "A bounded\n            # timed command is deliberately allowed" in source