"""Synthetic collection/watch tests. No Bluetooth hardware or private captures."""

import asyncio
import importlib.util
import struct
from argparse import Namespace
from pathlib import Path
from unittest.mock import AsyncMock, Mock

import pytest

from bmspy.ecoflow_bt import EcoFlowBTDevice, EcoFlowPacket, _State
from bmspy.ecoflow_proto import parse_protobuf_fields

ADDRESS = "02:00:00:00:00:03"


@pytest.fixture
def script():
    path = Path(__file__).resolve().parents[1] / "tools" / "ecoflow-bt.py"
    spec = importlib.util.spec_from_file_location("ecoflow_live_diagnostics", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def varint(value):
    result = bytearray()
    while value > 127:
        result.append((value & 127) | 128)
        value >>= 7
    result.append(value)
    return bytes(result)


def wire_field(number, wire_type=0, value=0):
    body = struct.pack("<f", value) if wire_type == 5 else varint(value)
    return varint(number << 3 | wire_type) + body


def report(payload, src=2, cmd_set=0xFE, cmd_id=0x15):
    return EcoFlowPacket(src, 0x21, cmd_set, cmd_id, payload)


def required_payload(script, group=None, omit=()):
    return b"".join(
        wire_field(number, wire_type)
        for name, fields in script.RIVER3_REQUIRED_GROUPS.items()
        if group is None or group == name
        for number, wire_type in fields.items() if number not in omit
    )


def device(prefix="R651"):
    return EcoFlowBTDevice(ADDRESS, prefix + "TEST00000001", "test-user")


def attach(script, dev, watch=False):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES[dev.serial_number[:4]], watch)
    dev.on_packet(collector.receive)
    return collector


def river2_bypass_packets():
    """Synthetic first burst: three core heartbeats, MPPT, and an empty reply."""
    pd = bytearray(121)
    struct.pack_into("<BHHi", pd, 14, 88, 35, 35, 5218)
    ems = bytearray(46)
    struct.pack_into("<II", ems, 17, 5999, 5999)
    struct.pack_into("<f", ems, 26, 87.5)
    bms = bytearray(69)
    bms[11], bms[20] = 88, 28
    struct.pack_into("<I", bms, 12, 13)
    struct.pack_into("<fIII", bms, 53, 87.75, 0, 0, 0)
    return [
        EcoFlowPacket(2, 0x21, 0x20, 2, bytes(pd), version=2),
        EcoFlowPacket(0x35, 0x21, 1, 0x52, b"", version=2),
        EcoFlowPacket(3, 0x21, 0x20, 2, bytes(ems), version=2),
        EcoFlowPacket(3, 0x21, 0x20, 0x32, bytes(bms), version=2),
        EcoFlowPacket(5, 0x21, 0x20, 2, bytes(94), version=2),
    ]


@pytest.mark.parametrize("order", [("measurements", "configuration"), ("configuration", "measurements")])
def test_checklist_combines_reports_in_either_order_and_counts_zero(script, order, capsys):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R651"])
    collector.receive(report(required_payload(script, order[0])))
    assert collector.missing()
    first_output = capsys.readouterr().out
    assert "DisplayPropertyUpload" in first_output  # Printed before collection finishes.
    collector.receive(report(required_payload(script, order[1])))
    assert collector.missing() == set()
    assert collector.progress() == "measurements 31/31, configuration 22/22"
    assert "DisplayPropertyUpload" in capsys.readouterr().out
    lines = "\n".join(collector.decoded_packets()["ef_pkt_02_fe_15"])
    assert "pow_in_sum_w: 0.000 W" in lines
    assert "plug_in_info_ac_in_chg_pow_max: 0 W" in lines


def test_only_new_or_changed_fields_print_in_normal_mode(script, capsys):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R651"])
    packet = report(wire_field(262, 5, 89))
    collector.receive(packet)
    assert "cms_batt_soc: 89.000 %" in capsys.readouterr().out
    collector.receive(packet)
    assert capsys.readouterr().out == ""
    collector.receive(report(wire_field(262, 5, 88)))
    assert "cms_batt_soc: 88.000 %" in capsys.readouterr().out
    assert len(collector.fields) == 1


def test_unavailable_time_counts_as_received_not_missing(script):
    dev = device()
    collector = attach(script, dev)
    dev._record(report(required_payload(script, omit=(268,))))
    assert "measurements:cms_dsg_rem_time" in collector.missing()
    dev._record(report(wire_field(268, value=0xFFFFFFFF)))
    assert collector.missing() == set()
    assert dev.snapshot.ef_discharge_remaining_time_minutes is None


@pytest.mark.parametrize("route", [(3, 0xFE, 0x15), (2, 0xFE, 0x16), (2, 0x20, 2)])
def test_unrelated_routes_cannot_satisfy_river3_checklist(script, route):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R651"])
    original = collector.missing()
    collector.receive(report(required_payload(script), *route))
    assert collector.missing() == original
    assert collector.fields == {}


def test_malformed_packet_is_atomic_and_wrong_wire_type_does_not_count(script):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R651"])
    collector.receive(report(required_payload(script) + b"\x80"))
    assert not collector.fields
    assert not collector.seen
    assert collector.error is None
    collector.receive(report(required_payload(script, omit=(3,)) + wire_field(3)))
    assert collector.missing() == {"measurements:pow_in_sum_w"}
    assert 3 not in collector.fields
    collector.receive(report(wire_field(3, 5)))
    assert not collector.missing()


def test_nested_repeated_unknown_fields_remain_grouped_and_optional(script):
    collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R651"])
    child = wire_field(1, value=7)
    item = varint(463 << 3 | 2) + varint(len(child)) + child
    collector.receive(report(item + item))
    assert collector.fields[463] == item + item
    assert len(parse_protobuf_fields(collector.fields[463])) == 2
    assert len(collector.missing()) == 53
    collector.receive(report(required_payload(script)))
    assert not collector.missing()
    assert collector.fields[463] == item + item
    collector.receive(report(item))
    assert collector.fields[463] == item  # Replace the repeated group, do not append forever.
    assert not collector.missing()


def test_each_packet_callback_runs_after_snapshot_updates_even_for_duplicates():
    dev = device()
    packets, updates = [], []
    dev.on_packet(Mock(side_effect=ValueError("bad listener")))
    dev.on_packet(lambda pkt: packets.append((pkt, dev.snapshot.ef_input_power_watts.raw_value)))
    dev.on_update(lambda: updates.append(True))
    pkt = report(wire_field(3, 5, 12))
    dev._record(pkt)
    dev._record(pkt)
    assert packets == [(pkt, 12), (pkt, 12)]
    assert updates == [True, True]


def test_completed_reports_finish_without_waiting_for_a_deadline(script, monkeypatch, capsys):
    dev = device()
    collector = attach(script, dev)
    dev._record(report(required_payload(script)))
    sleep = AsyncMock(side_effect=AssertionError("must not wait"))
    monkeypatch.setattr(script.asyncio, "sleep", sleep)
    asyncio.run(script.collect([(ADDRESS, dev, set())], max_wait=0, collectors={ADDRESS: collector}))
    sleep.assert_not_awaited()
    assert "all required information received" in capsys.readouterr().out


def test_timeout_reports_named_missing_fields_without_claiming_completion(script, capsys):
    dev = device()
    collector = attach(script, dev)
    dev._record(report(required_payload(script, omit=(209,))))
    asyncio.run(script.collect([(ADDRESS, dev, set())], max_wait=0, collectors={ADDRESS: collector}))
    out = capsys.readouterr().out
    assert "timed out, incomplete" in out
    assert "configuration:plug_in_info_ac_in_chg_pow_max" in out
    assert "all required information received" not in out


@pytest.mark.parametrize("prefix", ["R601", "R651"])
def test_watch_continues_after_completion_and_timeout_and_includes_late_unknown_packets(
    script, monkeypatch, capsys, prefix,
):
    dev = device(prefix)
    dev._state = _State.AUTHENTICATED
    collector = attach(script, dev, watch=True)
    if prefix == "R651":
        dev._record(report(required_payload(script)))
    else:
        for src, cmd_id in [(2, 2), (3, 2), (3, 0x32)]:
            dev._record(EcoFlowPacket(src, 0x21, 0x20, cmd_id, bytes(128), version=2))
    calls = 0

    async def receive_later(_):
        nonlocal calls
        calls += 1
        if calls == 1:
            dev._record(report(b"test-only-unknown", src=0x35, cmd_set=0x35, cmd_id=0x20))
            dev._record(report(wire_field(999, value=7)))
        else:
            raise asyncio.CancelledError()

    monkeypatch.setattr(script.asyncio, "sleep", receive_later)
    expected = script.MODEL_PROFILES[prefix]["expected_fields"]
    with pytest.raises(asyncio.CancelledError):
        asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=0,
                                   collectors={ADDRESS: collector}, watch=True))
    out = capsys.readouterr().out
    assert calls == 2
    assert "continuing watch for optional or unknown information" in out
    assert "ef_pkt_35_35_20" in out
    assert "raw_payload: " + b"test-only-unknown".hex() in out
    assert "No decoded fields; retain raw payload for review." in out
    assert "field 999: 7" in out
    assert "timed out" not in out


def test_watch_detects_disconnection_even_after_completion(script):
    dev = device()
    collector = attach(script, dev, watch=True)
    dev._record(report(required_payload(script)))
    with pytest.raises(ConnectionError, match="not authenticated"):
        asyncio.run(script.collect([(ADDRESS, dev, set())], max_wait=0,
                                   collectors={ADDRESS: collector}, watch=True))


@pytest.mark.parametrize("prefix", ["R601", "R651"])
@pytest.mark.parametrize("watch,failure", [
    (False, None), (False, "output"), (True, "cancel"), (True, "output"), (True, "disconnect"),
])
def test_run_registers_before_first_burst_and_always_disconnects(script, monkeypatch, capsys, watch, failure, prefix):
    dev = device(prefix)

    async def connect(**_):
        dev._state = _State.AUTHENTICATED
        if prefix == "R601":
            for packet in river2_bypass_packets():
                dev._record(packet)
        else:
            dev._record(report(required_payload(script, "measurements")))
            dev._record(report(required_payload(script, "configuration")))
        if failure == "disconnect":
            dev._state = _State.DISCONNECTED

    manager = Mock(add_device=Mock(return_value=dev), connect_all=AsyncMock(side_effect=connect),
                   disconnect_all=AsyncMock())
    monkeypatch.setattr(script, "EcoFlowBTManager", Mock(return_value=manager))
    if failure == "cancel":
        monkeypatch.setattr(script.asyncio, "sleep", AsyncMock(side_effect=asyncio.CancelledError()))
    if failure == "output":
        monkeypatch.setattr(script.PacketCollector, "_receive", Mock(side_effect=OSError("output failed")))
    args = Namespace(scan=False, device=ADDRESS, serial_number=dev.serial_number,
                     debug=0, timeout=0, watch=watch)
    if failure:
        error = {"cancel": asyncio.CancelledError, "output": OSError, "disconnect": ConnectionError}[failure]
        with pytest.raises(error):
            asyncio.run(script.run("test-user", args))
    else:
        asyncio.run(script.run("test-user", args))
        out = capsys.readouterr().out
        final = out.split("Snapshot:")[1]
        if prefix == "R601":
            assert "3/3 required heartbeats received (PD, EMS, BMS)" in out
            assert "Other packet types seen: ef_pkt_35_01_52 (empty payload)" in out
            assert "[PD heartbeat]" in final
            assert "latest fields collected this run" not in final
        else:
            assert "all required information received" in out
            assert "latest fields collected this run" in final
            assert "cms_batt_soc: 0.000 %" in final
            assert "plug_in_info_ac_in_chg_pow_max: 0 W" in final
        assert "timed out" not in out
    manager.disconnect_all.assert_awaited_once()