"""Client-visible EcoFlow protocol fields, using synthetic reports only."""

import io
import json
import struct
from dataclasses import asdict
from unittest.mock import MagicMock, patch

import pytest

from bmspy.classes import BMSInfoField, BMSScalarField
from bmspy.client import read_data, socket_comms
from bmspy.ecoflow_bt import EcoFlowBTDevice, EcoFlowPacket
from bmspy.ecoflow_proto import decode_schema_values
from bmspy.ecoflow_river2 import MESSAGE_DEFINITIONS
from bmspy.ecoflow_river3_schema import ENUMS, MESSAGES
from bmspy.server import send_response


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


def integer(number, value):
    return varint(number << 3) + varint(value)


def floating(number, value):
    return varint(number << 3 | 5) + struct.pack("<f", value)


def blob(number, value):
    return varint(number << 3 | 2) + varint(len(value)) + value


def sample(number, kind):
    if kind in MESSAGES or kind == "string":
        return blob(number, b"")
    if kind == "float":
        return floating(number, 1.5)
    assert kind in ("bool", "uint32", "int32") or kind in ENUMS
    return integer(number, 0)


def device(prefix):
    return EcoFlowBTDevice("02:00:00:00:00:01", prefix + "TEST00000001", "test-user")


def record(dev, payload, route=(2, 0xFE, 0x15)):
    src, cmd_set, cmd_id = route
    dev._record(EcoFlowPacket(src, 0x21, cmd_set, cmd_id, payload))
    return dict(dev.snapshot.items())


def heartbeat(definitions):
    # Include skipped bytes, zeroes, and signed values, not just summary fields.
    values = [bytes([0xAB]) * struct.calcsize("<" + f.fmt) if f.fmt.endswith("s")
              else -3 if f.fmt in ("i", "h") else 1.5 if f.fmt == "f" else 0
              for f in definitions]
    return struct.pack("<" + "".join(f.fmt for f in definitions), *values), values


@pytest.mark.parametrize("route,namespace", [
    ((2, 0x20, 2), "pd"), ((3, 0x20, 2), "ems"),
    ((3, 0x20, 0x32), "bms"), ((4, 0x20, 2), "inv"), ((5, 0x20, 2), "mppt"),
])
def test_every_river2_layout_field_is_client_visible(route, namespace):
    definitions = MESSAGE_DEFINITIONS[route][1]
    payload, values = heartbeat(definitions)
    fields = record(device("R601"), payload, route)
    prefix = f"ef_river2_{namespace}_"
    assert {key for key in fields if key.startswith(prefix)} == {
        prefix + f.name for f in definitions
    }
    for definition, value in zip(definitions, values):
        field = fields[prefix + definition.name]
        if isinstance(value, bytes):
            assert isinstance(field, BMSInfoField)
            assert field.info == "0x" + value.hex()
        else:
            assert isinstance(field, BMSScalarField)
            assert field.raw_value == value * definition.scale
            assert field.units == (definition.unit or None)


def test_river2_short_firmware_reports_keep_previous_tail_and_explicit_zero():
    dev = device("R603")
    definitions = MESSAGE_DEFINITIONS[(2, 0x20, 2)][1]
    payload, _ = heartbeat(definitions)
    record(dev, payload + b"\x12\x34", (2, 0x20, 2))
    fields = record(dev, payload[:23], (2, 0x20, 2))
    assert fields["ef_river2_pd_watts_in_sum"].raw_value == 0
    assert fields["ef_river2_pd_relay_switch_count"].raw_value == 0
    assert fields["ef_river2_pd_trailing_bytes"].info == "0x1234"
    assert fields["ef_river2_pd_remain_time"].raw_value == -3
    assert "ef_remaining_time_minutes" not in fields
    fresh = record(device("R601"), payload[:23], (2, 0x20, 2))
    assert "ef_river2_pd_relay_switch_count" not in fresh


def test_river2_malformed_report_preserves_previous_protocol_fields():
    dev = device("R601")
    definitions = MESSAGE_DEFINITIONS[(3, 0x20, 2)][1]
    payload, _ = heartbeat(definitions)
    record(dev, payload, (3, 0x20, 2))
    for bad in (payload[:24], payload[:26] + struct.pack("<f", float("nan"))):
        fields = record(dev, bad, (3, 0x20, 2))
        assert "ef_river2_ems_decode_error" in fields
        assert fields["ef_river2_ems_f32_lcd_show_soc"].raw_value == 1.5
        assert fields["ef_pkt_03_20_02"].info == bad.hex()
    fields = record(dev, payload, (3, 0x20, 2))
    assert "ef_river2_ems_decode_error" not in fields


def test_every_river3_display_field_is_client_visible_without_defaults():
    for number, (name, kind, repeated) in MESSAGES["DisplayPropertyUpload"].items():
        fields = record(device("R651"), sample(number, kind))
        protocol = {k: v for k, v in fields.items() if k.startswith("ef_river3_display_")}
        key = "ef_river3_display_" + name
        assert set(protocol) == {key}
        field = protocol[key]
        if repeated or kind in MESSAGES:
            assert isinstance(field, BMSInfoField)
            assert json.loads(field.info) == ([{}] if repeated else {})
        elif kind == "string":
            assert field.info == ""
        else:
            assert isinstance(field, BMSScalarField)
            assert field.raw_value == (1.5 if kind == "float" else 0)
            if kind == "bool":
                assert field.raw_value is False


def test_river3_partial_reports_preserve_values_and_do_not_invent_units():
    dev = device("R651")
    record(dev, floating(11, -14.322) + floating(262, 89))
    fields = record(dev, integer(25, 0) + integer(258, -7))
    assert fields["ef_river3_display_pow_get_typec1"].raw_value == pytest.approx(-14.322)
    assert fields["ef_river3_display_pow_get_typec1"].units == "W"
    assert fields["ef_river3_display_cms_batt_soc"].raw_value == 89
    assert fields["ef_river3_display_cms_batt_soc"].units == "%"
    assert fields["bms_capacity_charge_ratio"].raw_value == .89
    assert fields["ef_river3_display_xboost_en"].raw_value is False
    assert fields["ef_river3_display_bms_min_cell_temp"].raw_value == -7
    assert fields["ef_river3_display_bms_min_cell_temp"].units == "°C"
    fields = record(dev, integer(269, 0xFFFFFFFF))
    assert fields["ef_river3_display_cms_chg_rem_time"].raw_value == 0xFFFFFFFF
    assert "ef_charge_remaining_time_minutes" not in fields


def test_nested_repeated_enum_strings_and_empty_values_keep_their_structure():
    dev = device("R651")
    stats = blob(1, integer(1, 2) + integer(2, 17)) + blob(1, integer(1, 999)) + blob(1, b"")
    fields = record(dev, blob(463, stats) + blob(433, b"") + blob(91, b""))
    value = json.loads(fields["ef_river3_display_display_statistics_sum"].info)
    assert value == {"list_info": [
        {"statistics_object": {"code": 2, "name": "STATISTICS_OBJECT_AC_OUT_ENERGY"}, "statistics_content": 17},
        {"statistics_object": {"code": 999, "name": None}}, {},
    ]}
    assert fields["ef_river3_display_plug_in_info_dcp_sn"].info == ""
    assert fields["ef_river3_display_plug_in_info_dcp2_resv"].info == "{}"
    # A later report replaces a reported group, rather than keeping stale children.
    fields = record(dev, blob(463, b""))
    assert fields["ef_river3_display_display_statistics_sum"].info == "{}"


def test_packed_unpacked_and_duplicate_message_fields_merge_within_report():
    packed = blob(1, varint(0) + varint(128))
    payload = blob(432, packed) + blob(432, integer(1, 7))
    values = decode_schema_values(payload, "DisplayPropertyUpload")
    assert values["plug_in_info_dcp_resv"] == {"resv_info": [0, 128, 7]}
    empty = decode_schema_values(blob(432, blob(1, b"")), "DisplayPropertyUpload")
    assert empty["plug_in_info_dcp_resv"] == {"resv_info": []}
    assert decode_schema_values(floating(3, 1) + floating(3, 2), "DisplayPropertyUpload")["pow_in_sum_w"] == 2


def test_unknown_fields_keep_all_wire_encodings_and_repetitions():
    unknown = integer(999, 7) + integer(999, 0)
    unknown += blob(1000, b"opaque") + floating(1001, 1.5)
    unknown += varint(1002 << 3 | 1) + bytes(8)
    fields = record(device("R651"), unknown)
    for number, wire in ((999, 0), (1000, 2), (1001, 5), (1002, 1)):
        items = json.loads(fields[f"ef_river3_display_field_{number}"].info)
        assert all(item["wire_type"] == wire for item in items)
        assert all(bytes.fromhex(item["raw"]) in unknown for item in items)
    assert [item["value"] for item in json.loads(fields["ef_river3_display_field_999"].info)] == [7, 0]


@pytest.mark.parametrize("bad", [
    b"\x00", b"\x80", integer(4, 1), floating(4, float("inf")),
    integer(269, 1 << 32), blob(433, b"\xff"), blob(141, b"\x00"),
    blob(432, blob(1, b"\x80")),
])
def test_invalid_river3_protocol_report_is_atomic_and_recoverable(bad):
    dev = device("R651")
    record(dev, floating(3, 10) + floating(4, 20))
    payload = floating(3, 99) + bad
    fields = record(dev, payload)
    assert fields["ef_river3_display_pow_in_sum_w"].raw_value == 10
    assert fields["ef_river3_display_pow_out_sum_w"].raw_value == 20
    assert "ef_river3_display_decode_error" in fields
    assert fields["ef_pkt_02_fe_15"].info == payload.hex()
    fields = record(dev, floating(3, 0))
    assert fields["ef_river3_display_pow_in_sum_w"].raw_value == 0
    assert fields["ef_river3_display_pow_out_sum_w"].raw_value == 20
    assert "ef_river3_display_decode_error" not in fields


def test_schema_value_decoding_has_a_nesting_limit(monkeypatch):
    monkeypatch.setitem(MESSAGES, "Recursive", {1: ("child", "Recursive", False)})
    payload = b""
    for _ in range(20):
        payload = blob(1, payload)
    with pytest.raises(ValueError, match="maximum nesting depth"):
        decode_schema_values(payload, "Recursive")


def test_unconfirmed_runtime_and_unknown_routes_remain_raw_only():
    dev = device("R651")
    for route in ((2, 0xFE, 0x16), (3, 0xFE, 0x15), (0x35, 0x35, 0x20)):
        fields = record(dev, floating(244, 19.2), route)
        assert not any(key.startswith("ef_river3_") for key in fields)
        assert fields[f"ef_pkt_{route[0]:02x}_{route[1]:02x}_{route[2]:02x}"].info


def full_snapshots():
    river2, river3 = device("R601"), device("R651")
    for route, (_, definitions, _) in MESSAGE_DEFINITIONS.items():
        record(river2, heartbeat(definitions)[0], route)
    record(river3, b"".join(sample(number, kind) for number, (_, kind, _) in MESSAGES["DisplayPropertyUpload"].items()))
    return {"desktop": river2.snapshot, "nas": river3.snapshot}


def test_all_fields_survive_large_fragmented_server_response_to_client():
    snapshots = full_snapshots()
    server_socket = MagicMock()
    send_response(server_socket, snapshots, "test")
    frame = server_socket.sendall.call_args.args[0]
    assert len(frame) > 65536
    stream = io.BytesIO(frame)
    sock = MagicMock()
    sock.recv.side_effect = lambda n: stream.read(min(n, 2 if stream.tell() < 4 else 127))
    with patch("bmspy.client.socket.socket", return_value=sock):
        result = read_data("/fake/socket", "test")
    expected = json.loads(frame[4:])
    actual = {device: {key: asdict(value) for key, value in snap.items()} for device, snap in result.items()}
    assert actual == expected
    assert stream.read() == b""
    sock.close.assert_called_once()


@pytest.mark.parametrize("frame", [b"", b"\x00\x00", struct.pack("!I", 100) + b"{}"])
def test_incomplete_client_response_fails_without_looping(frame):
    sock = MagicMock()
    sock.recv.side_effect = io.BytesIO(frame).read
    with patch("bmspy.client.socket.socket", return_value=sock), pytest.raises(SystemExit):
        socket_comms("/fake/socket", {"command": "GET"})
    sock.close.assert_called_once()


def test_connection_failure_does_not_read_from_unconnected_socket():
    sock = MagicMock()
    sock.connect.side_effect = ConnectionRefusedError("offline")
    with patch("bmspy.client.socket.socket", return_value=sock), pytest.raises(ConnectionRefusedError):
        socket_comms("/fake/socket", {"command": "GET"})
    sock.recv.assert_not_called()
    sock.sendall.assert_not_called()
    sock.close.assert_called_once()


def test_all_fields_reach_influxdb():
    pytest.importorskip("influxdb_client_3")
    from bmspy.influxdb import influxdb_create_snapshot

    snapshots = full_snapshots()
    points = influxdb_create_snapshot(snapshots)
    expected = {(ups, key) for ups, snap in snapshots.items() for key, _ in snap.items()}
    assert {(point._tags["ups"], point._name) for point in points} == expected
    for point in points:
        field = dict(snapshots[point._tags["ups"]].items())[point._name]
        value = (field.info if isinstance(field, BMSInfoField) else field.raw_value
                 if isinstance(field, BMSScalarField) else field.raw_values[1])
        assert point._fields["value"] == value
        assert point.to_line_protocol()


def test_all_fields_reach_prometheus():
    prometheus = pytest.importorskip("prometheus_client")
    from bmspy.prometheus import prometheus_create_metric, prometheus_populate_metric

    snapshots = full_snapshots()
    registry = prometheus.CollectorRegistry()
    metrics = prometheus_create_metric(registry, snapshots)
    prometheus_populate_metric(metrics, snapshots)
    assert set(metrics) == {key for snap in snapshots.values() for key, _ in snap.items()}
    for ups, snap in snapshots.items():
        for key, field in snap.items():
            labels = {"ups": ups}
            if isinstance(field, BMSInfoField):
                labels[key] = field.info
                assert registry.get_sample_value(key + "_info", labels) == 1
            elif isinstance(field, BMSScalarField):
                assert registry.get_sample_value(key, labels) == field.raw_value
            else:
                for index, value in field.raw_values.items():
                    assert registry.get_sample_value(key, {**labels, field.label: str(index)}) == value


def test_prometheus_daemon_adds_fields_from_later_reports():
    prometheus = pytest.importorskip("prometheus_client")
    from bmspy.prometheus import prometheus_export

    first, later = device("R651"), device("R651")
    record(first, floating(3, 10))
    record(later, floating(3, 20) + floating(11, -14))
    snapshots = [{"nas": first.snapshot}, {"nas": later.snapshot}]
    captured = []

    def capture(port, registry):
        captured.append(registry)

    class Finished(Exception):
        pass

    with patch("bmspy.prometheus.client.read_data", side_effect=snapshots), \
            patch.object(prometheus, "start_http_server", side_effect=capture), \
            patch("bmspy.prometheus.time.sleep", side_effect=[None, Finished]):
        with pytest.raises(Finished):
            prometheus_export()
    registry = captured[0]
    assert registry.get_sample_value("ef_river3_display_pow_in_sum_w", {"ups": "nas"}) == 20
    assert registry.get_sample_value("ef_river3_display_pow_get_typec1", {"ups": "nas"}) == -14