from __future__ import annotations

import json
from collections import deque
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import paho.mqtt.client as mqtt
import pytest

from k1link.device_plugins.xgrids_k1.calibration_snapshot import (
    seal_factory_calibration_snapshot,
)
from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
    ApplicationControlAuthority,
    LiveDeviceControlBinding,
)
from k1link.device_plugins.xgrids_k1.protocol.calibration_file import (
    CALIBRATION_FILE_READ_COMMAND,
    CALIBRATION_FILE_REQUEST_TOPIC,
    CALIBRATION_FILE_RESPONSE_TOPIC,
    FACTORY_CALIBRATION_PATHS,
    FACTORY_CAMERA_CALIBRATION_PATH,
    FACTORY_CAMERA_LIDAR_EXTRINSIC_PATH,
    CalibrationFileProtocolError,
    CalibrationFileRejected,
    build_factory_calibration_file_read,
    decode_factory_calibration_file_response,
)
from k1link.device_plugins.xgrids_k1.protocol.calibration_mqtt import (
    CALIBRATION_READ_SUBSCRIPTIONS,
    FactoryCalibrationReadResult,
    ReviewedCalibrationMqttReader,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import OPENAPI_SUCCESS
from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import iter_fields

APPLICATION_KEY = "11111111-2222-3333-4444-555555555555"
VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
DEVICE_SERIAL = "K1SERIAL01"
PROFILE_ID = "xgrids.lixelkity-k1.fw-3.0.2.local-network.v2"
CALIBRATION_FIXTURE_ROOT = Path(__file__).parent / "fixtures" / "k1" / "calibration"


def _authority() -> ApplicationControlAuthority:
    return ApplicationControlAuthority(openapi_key=APPLICATION_KEY)


def _binding(**changes: object) -> LiveDeviceControlBinding:
    values = {
        "vendor_device_id": VENDOR_DEVICE_ID,
        "device_serial": DEVICE_SERIAL,
        "software_version": "V3.0.2",
        "system_version": "V3.0.2",
        "device_model": "LixelKity K1",
        "device_type": "A4",
        "is_activated": True,
    }
    values.update(changes)
    return LiveDeviceControlBinding(**values)  # type: ignore[arg-type]


def _varint(value: int) -> bytes:
    encoded = bytearray()
    while value > 0x7F:
        encoded.append((value & 0x7F) | 0x80)
        value >>= 7
    encoded.append(value)
    return bytes(encoded)


def _uint(number: int, value: int) -> bytes:
    return _varint(number << 3) + _varint(value)


def _bytes(number: int, value: bytes) -> bytes:
    return _varint((number << 3) | 2) + _varint(len(value)) + value


def _text(number: int, value: str) -> bytes:
    return _bytes(number, value.encode("utf-8"))


def _header(session_id: str) -> bytes:
    return b"".join(
        (
            _text(4, VENDOR_DEVICE_ID),
            _text(5, session_id),
            _text(6, APPLICATION_KEY),
        )
    )


def _device_info_response() -> bytes:
    base_info = b"".join(
        (
            _text(2, "V3.0.2"),
            _text(3, "V3.0.2"),
            _text(6, "LixelKity K1"),
            _text(7, DEVICE_SERIAL),
            _text(8, "A4"),
        )
    )
    working_status = _uint(1, 1)
    device_info = _bytes(2, base_info) + _bytes(7, working_status)
    return b"".join(
        (
            _bytes(1, _header(":DeviceInfoRequest")),
            _bytes(2, device_info),
            _bytes(15, _uint(1, OPENAPI_SUCCESS)),
        )
    )


def _calibration_response(
    path: str,
    content: bytes,
    *,
    command: int = CALIBRATION_FILE_READ_COMMAND,
    result_code: int = OPENAPI_SUCCESS,
    session_id: str | None = None,
) -> bytes:
    return b"".join(
        (
            _bytes(
                1,
                _header(
                    session_id or f"{VENDOR_DEVICE_ID}:CalibFileRequest"
                ),
            ),
            _uint(2, command),
            _text(3, path),
            _bytes(4, content),
            _bytes(15, _uint(1, result_code)),
        )
    )


def _content_for(path: str) -> bytes:
    if path == FACTORY_CAMERA_CALIBRATION_PATH:
        return (CALIBRATION_FIXTURE_ROOT / "camera.yaml").read_bytes()
    return (CALIBRATION_FIXTURE_ROOT / "extrinsic_camera_lidar.yaml").read_bytes()


def test_factory_file_request_is_command_five_and_has_no_content_field() -> None:
    request = build_factory_calibration_file_read(
        _authority(),
        _binding(),
        FACTORY_CAMERA_CALIBRATION_PATH,
    )
    fields = {field.number: field.value for field in iter_fields(request.payload)}
    header = fields[1]
    assert isinstance(header, bytes)
    header_fields = {field.number: field.value for field in iter_fields(header)}

    assert request.topic == CALIBRATION_FILE_REQUEST_TOPIC
    assert request.response_topic == CALIBRATION_FILE_RESPONSE_TOPIC
    assert request.mutates_device is False
    assert request.automatic_retry is False
    assert fields[2] == 5
    assert fields[3] == FACTORY_CAMERA_CALIBRATION_PATH.encode()
    assert 4 not in fields
    assert header_fields[4] == VENDOR_DEVICE_ID.encode()
    assert header_fields[5] == f"{VENDOR_DEVICE_ID}:CalibFileRequest".encode()
    assert header_fields[6] == APPLICATION_KEY.encode()


@pytest.mark.parametrize(
    "path",
    (
        "/mnt/system/factory-data/config/../Lixel.yaml",
        "/mnt/system/factory-data/config/extrinsic_camera_motor.yaml",
        "/tmp/camera.yaml",
        "",
    ),
)
def test_factory_file_request_rejects_every_path_outside_two_file_allowlist(
    path: str,
) -> None:
    with pytest.raises(CalibrationFileProtocolError, match="two-file allowlist"):
        build_factory_calibration_file_read(_authority(), _binding(), path)


def test_factory_file_request_rejects_a_nonreviewed_live_binding() -> None:
    with pytest.raises(CalibrationFileProtocolError, match="reviewed activated"):
        build_factory_calibration_file_read(
            _authority(),
            _binding(system_version="V3.0.3"),
            FACTORY_CAMERA_CALIBRATION_PATH,
        )


def test_correlated_factory_file_response_returns_exact_utf8_bytes() -> None:
    request = build_factory_calibration_file_read(
        _authority(),
        _binding(),
        FACTORY_CAMERA_CALIBRATION_PATH,
    )
    content = _content_for(request.path)

    result = decode_factory_calibration_file_response(
        _calibration_response(request.path, content),
        request,
        _authority(),
        _binding(),
    )

    assert result.path == request.path
    assert result.content == content
    assert result.content_bytes == len(content)
    assert len(result.content_sha256) == 64
    assert result.result_code == OPENAPI_SUCCESS


def test_factory_file_response_rejects_write_command_path_drift_and_duplicates() -> None:
    request = build_factory_calibration_file_read(
        _authority(),
        _binding(),
        FACTORY_CAMERA_CALIBRATION_PATH,
    )
    with pytest.raises(CalibrationFileProtocolError, match="not a file-read"):
        decode_factory_calibration_file_response(
            _calibration_response(request.path, b"yaml", command=6),
            request,
            _authority(),
            _binding(),
        )
    with pytest.raises(CalibrationFileProtocolError, match="path mismatch"):
        decode_factory_calibration_file_response(
            _calibration_response(FACTORY_CAMERA_LIDAR_EXTRINSIC_PATH, b"yaml"),
            request,
            _authority(),
            _binding(),
        )
    duplicate_path = _calibration_response(request.path, b"yaml") + _text(3, request.path)
    with pytest.raises(CalibrationFileProtocolError, match="duplicated"):
        decode_factory_calibration_file_response(
            duplicate_path,
            request,
            _authority(),
            _binding(),
        )


def test_factory_file_response_rejects_device_error_and_non_utf8_content() -> None:
    request = build_factory_calibration_file_read(
        _authority(),
        _binding(),
        FACTORY_CAMERA_CALIBRATION_PATH,
    )
    with pytest.raises(CalibrationFileRejected) as rejected:
        decode_factory_calibration_file_response(
            _calibration_response(request.path, b"yaml", result_code=123),
            request,
            _authority(),
            _binding(),
        )
    assert rejected.value.result_code == 123
    with pytest.raises(CalibrationFileProtocolError, match="valid UTF-8"):
        decode_factory_calibration_file_response(
            _calibration_response(request.path, b"\xff"),
            request,
            _authority(),
            _binding(),
        )


class FakeReasonCode:
    def __init__(self, *, failure: bool = False) -> None:
        self.is_failure = failure


class FakeCalibrationClient:
    def __init__(self) -> None:
        self.on_connect: Any = None
        self.on_subscribe: Any = None
        self.on_publish: Any = None
        self.on_message: Any = None
        self.on_disconnect: Any = None
        self.connect_timeout = 0.0
        self.events: deque[tuple[str, object]] = deque()
        self.connect_calls: list[tuple[str, int, int]] = []
        self.subscribe_calls: list[list[tuple[str, int]]] = []
        self.publish_calls: list[tuple[str, bytes, int, bool]] = []
        self.unsubscribe_calls: list[list[str]] = []
        self.next_mid = 20

    def connect(self, host: str, port: int, keepalive: int) -> mqtt.MQTTErrorCode:
        self.connect_calls.append((host, port, keepalive))
        self.events.append(("connect", FakeReasonCode()))
        return mqtt.MQTT_ERR_SUCCESS

    def subscribe(self, topics: list[tuple[str, int]]) -> tuple[mqtt.MQTTErrorCode, int]:
        self.subscribe_calls.append(topics)
        self.events.append(("subscribe", 7))
        return mqtt.MQTT_ERR_SUCCESS, 7

    def publish(
        self,
        topic: str,
        payload: bytes,
        qos: int,
        retain: bool,
    ) -> SimpleNamespace:
        self.publish_calls.append((topic, payload, qos, retain))
        mid = self.next_mid
        self.next_mid += 1
        self.events.append(("publish", mid))
        if topic == "lixel/application/request/device_info":
            response_topic = "lixel/application/response/device_info"
            response_payload = _device_info_response()
        else:
            fields = {field.number: field.value for field in iter_fields(payload)}
            raw_path = fields[3]
            assert isinstance(raw_path, bytes)
            path = raw_path.decode()
            response_topic = CALIBRATION_FILE_RESPONSE_TOPIC
            response_payload = _calibration_response(path, _content_for(path))
        self.events.append(("message", (response_topic, response_payload)))
        return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid)

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        del timeout
        if not self.events:
            return mqtt.MQTT_ERR_SUCCESS
        kind, value = self.events.popleft()
        if kind == "connect":
            self.on_connect(self, None, SimpleNamespace(), value, None)
        elif kind == "subscribe":
            self.on_subscribe(
                self,
                None,
                value,
                [FakeReasonCode() for _topic in CALIBRATION_READ_SUBSCRIPTIONS],
                None,
            )
        elif kind == "publish":
            self.on_publish(self, None, value, FakeReasonCode(), None)
        elif kind == "message":
            topic, payload = value
            self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload))
        return mqtt.MQTT_ERR_SUCCESS

    def unsubscribe(self, topics: list[str]) -> tuple[mqtt.MQTTErrorCode, int]:
        self.unsubscribe_calls.append(topics)
        return mqtt.MQTT_ERR_SUCCESS, 50

    def disconnect(self) -> mqtt.MQTTErrorCode:
        return mqtt.MQTT_ERR_SUCCESS


def test_reviewed_mqtt_reader_performs_one_identity_read_then_two_exact_file_reads() -> None:
    client = FakeCalibrationClient()
    reader = ReviewedCalibrationMqttReader(
        "10.255.254.54",
        client_factory=lambda: client,  # type: ignore[arg-type]
    )

    result = reader.read_factory_calibration(_authority())

    assert result.binding == _binding()
    assert [item.path for item in result.files] == list(FACTORY_CALIBRATION_PATHS)
    assert client.subscribe_calls == [list(CALIBRATION_READ_SUBSCRIPTIONS)]
    assert [topic for topic, _payload, _qos, _retain in client.publish_calls] == [
        "lixel/application/request/device_info",
        CALIBRATION_FILE_REQUEST_TOPIC,
        CALIBRATION_FILE_REQUEST_TOPIC,
    ]
    assert all(qos == 2 and not retain for _topic, _payload, qos, retain in client.publish_calls)
    assert result.transport["publish_attempts"] == 3
    assert result.transport["correlated_responses"] == 3
    assert result.transport["write_command_available"] is False


def test_private_snapshot_is_new_only_and_hashes_both_exact_artifacts(tmp_path: Path) -> None:
    files = tuple(
        decode_factory_calibration_file_response(
            _calibration_response(path, _content_for(path)),
            build_factory_calibration_file_read(_authority(), _binding(), path),
            _authority(),
            _binding(),
        )
        for path in FACTORY_CALIBRATION_PATHS
    )
    result = FactoryCalibrationReadResult(
        binding=_binding(),
        files=(files[0], files[1]),
        transport={"automatic_retry": False, "publish_attempts": 3},
    )

    snapshot = seal_factory_calibration_snapshot(
        result,
        evidence_root=tmp_path,
        compatibility_profile_id=PROFILE_ID,
        captured_at=datetime(2026, 7, 20, 10, 30, tzinfo=UTC),
    )

    assert snapshot["status"] == "available"
    internal = snapshot["device_internal_calibration"]
    assert isinstance(internal, dict)
    snapshot_path = Path(str(internal["private_snapshot_path"]))
    assert snapshot_path.is_dir()
    assert (snapshot_path / "camera.yaml").read_bytes() == _content_for(
        FACTORY_CAMERA_CALIBRATION_PATH
    )
    assert (snapshot_path / "extrinsic_camera_lidar.yaml").read_bytes() == _content_for(
        FACTORY_CAMERA_LIDAR_EXTRINSIC_PATH
    )
    manifest = json.loads((snapshot_path / "manifest.json").read_text(encoding="utf-8"))
    assert manifest["source"]["request_command"] == 5
    assert manifest["source"]["write_command_available"] is False
    assert manifest["normalized_calibration"]["transform_notation"] == (
        "T_destination_from_source"
    )
    assert [item["source_path"] for item in manifest["artifacts"]] == list(
        FACTORY_CALIBRATION_PATHS
    )
    assert snapshot_path.stat().st_mode & 0o777 == 0o700
    assert all(path.stat().st_mode & 0o777 == 0o600 for path in snapshot_path.iterdir())
    assert internal["camera_stream_mapping"]["status"] == "firmware-profile-verified"