import keyword
import dataclasses
from collections import defaultdict

import pytest
import zigpy.types as zigpy_t
import zigpy.zdo.types

import zigpy_znp.frames as frames
import zigpy_znp.commands as c
from zigpy_znp import types as t


def test_command_header():
    """Test CommandHeader class."""
    data = b"\x61\x02"
    extra = b"the rest of data\xaa\x55"
    r, rest = t.CommandHeader.deserialize(data + extra)
    assert rest == extra
    assert r.cmd0 == 0x61
    assert r.id == 0x02

    r = t.CommandHeader(0x0261)
    assert r.cmd0 == 0x61
    assert r.id == 0x02

    new1 = r.with_id(0xFF)
    assert new1.id == 0xFF
    assert new1.cmd0 == 0x61

    new2 = r.with_id(0x00)
    assert new2.id == 0x00
    assert new2.cmd0 == 0x61


def test_command_setters():
    """Test setters"""
    # the setter order should not matter
    command = t.CommandHeader(0xFFFF)
    for cmd_type in t.CommandType:
        for subsys in t.Subsystem:
            # There's probably no need to iterate over all 256 possible values
            for cmd_id in (0x00, 0xFF, 0x10, 0x01, 0xF0, 0x0F, 0x22, 0xEE):
                perms = [
                    command.with_id(cmd_id).with_type(cmd_type).with_subsystem(subsys),
                    command.with_type(cmd_type).with_id(cmd_id).with_subsystem(subsys),
                    command.with_type(cmd_type).with_subsystem(subsys).with_id(cmd_id),
                    command.with_subsystem(subsys).with_type(cmd_type).with_id(cmd_id),
                    command.with_subsystem(subsys).with_id(cmd_id).with_type(cmd_type),
                    command.with_id(cmd_id).with_subsystem(subsys).with_type(cmd_type),
                    t.CommandHeader(0xFFFF, id=cmd_id, subsystem=subsys, type=cmd_type),
                ]

                assert len(set(perms)) == 1
                assert perms[0].id == cmd_id
                assert perms[0].subsystem == subsys
                assert perms[0].type == cmd_type


def test_error_code():
    data = b"\x03"
    extra = b"the rest of the owl\x00\xff"

    r, rest = t.ErrorCode.deserialize(data + extra)
    assert rest == extra
    assert r == 0x03
    assert r.name == "INVALID_PARAMETER"

    r, rest = t.ErrorCode.deserialize(b"\xaa" + extra)
    assert rest == extra
    assert r.name == "undefined_0xaa"


def _validate_schema(schema):
    for index, param in enumerate(schema):
        assert isinstance(param.name, str)
        assert param.name.isidentifier()
        assert not keyword.iskeyword(param.name)
        assert isinstance(param.type, type)
        assert isinstance(param.description, str)

        # All optional params must be together at the end
        if param.optional:
            assert all(p.optional for p in schema[index:])

        # Trailing bytes must be at the very end
        if issubclass(param.type, t.TrailingBytes):
            assert not schema[index + 1 :]


def test_commands_schema():
    commands_by_id = defaultdict(list)

    for commands in c.ALL_COMMANDS:
        for cmd in commands:
            if cmd.type == t.CommandType.SREQ:
                assert cmd.type == cmd.Req.header.type
                assert cmd.Rsp.header.type == t.CommandType.SRSP
                assert (
                    cmd.subsystem
                    == cmd.Req.header.subsystem
                    == cmd.Rsp.header.subsystem
                )
                assert isinstance(cmd.Req.header, t.CommandHeader)
                assert isinstance(cmd.Rsp.header, t.CommandHeader)

                assert cmd.Req.Rsp is cmd.Rsp
                assert cmd.Rsp.Req is cmd.Req
                assert cmd.Callback is None

                _validate_schema(cmd.Req.schema)
                _validate_schema(cmd.Rsp.schema)

                commands_by_id[cmd.Req.header].append(cmd.Req)
                commands_by_id[cmd.Rsp.header].append(cmd.Rsp)
            elif cmd.type == t.CommandType.AREQ:
                # we call the AREQ Rsp a Callback
                assert cmd.Rsp is None

                # only one of them can be set
                assert (cmd.Callback is not None) ^ (cmd.Req is not None)

                if cmd.Callback is not None:
                    assert cmd.type == cmd.Callback.header.type
                    assert cmd.subsystem == cmd.Callback.header.subsystem
                    assert isinstance(cmd.Callback.header, t.CommandHeader)

                    _validate_schema(cmd.Callback.schema)

                    commands_by_id[cmd.Callback.header].append(cmd.Callback)
                elif cmd.Req is not None:
                    assert cmd.type == cmd.Req.header.type
                    assert cmd.subsystem == cmd.Req.header.subsystem
                    assert isinstance(cmd.Req.header, t.CommandHeader)

                    _validate_schema(cmd.Req.schema)

                    commands_by_id[cmd.Req.header].append(cmd.Req)
                else:
                    assert False, "Command is empty"  # noqa: B011
            elif cmd.type == t.CommandType.SRSP:
                # The one command like this is RPCError
                assert cmd is c.RPCError.CommandNotRecognized

                assert cmd.type == cmd.Rsp.header.type
                assert cmd.Req is None
                assert cmd.Callback is None
                assert cmd.Rsp.header.type == t.CommandType.SRSP
                assert cmd.subsystem == cmd.Rsp.header.subsystem
                assert isinstance(cmd.Rsp.header, t.CommandHeader)

                _validate_schema(cmd.Rsp.schema)

                commands_by_id[cmd.Rsp.header].append(cmd.Rsp)
            else:
                assert False, "Command has unknown type"  # noqa: B011

    duplicate_commands = {
        cmd: commands for cmd, commands in commands_by_id.items() if len(commands) > 1
    }
    assert not duplicate_commands

    assert len(commands_by_id.keys()) == len(c.COMMANDS_BY_ID.keys())


def test_command_param_binding():
    # No params
    c.SYS.Ping.Req()

    # Invalid param name
    with pytest.raises(KeyError):
        c.SYS.Ping.Rsp(asd=123)

    # Valid param name
    c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)

    # Too many params, one valid
    with pytest.raises(KeyError):
        c.SYS.Ping.Rsp(foo="asd", Capabilities=t.MTCapabilities.SYS)

    # Not enough params
    with pytest.raises(KeyError):
        c.SYS.Ping.Rsp()

    # Invalid type
    with pytest.raises(ValueError):
        c.UTIL.TimeAlive.Rsp(Seconds=b"asd")

    # Valid type but invalid value
    with pytest.raises(ValueError):
        c.UTIL.SetPreConfigKey.Req(PreConfigKey=t.KeyData([1, 2, 3]))

    # Coerced numerical type
    a = c.UTIL.TimeAlive.Rsp(Seconds=12)
    b = c.UTIL.TimeAlive.Rsp(Seconds=t.uint32_t(12))

    assert a == b
    assert a.Seconds == b.Seconds
    assert type(a.Seconds) == type(b.Seconds) == t.uint32_t  # noqa: E721

    # Overflowing integer types
    with pytest.raises(ValueError):
        c.UTIL.TimeAlive.Rsp(Seconds=10**20)

    # Integers will not be coerced to enums
    assert t.MTCapabilities.SYS == 0x0001

    with pytest.raises(ValueError):
        c.SYS.Ping.Rsp(Capabilities=0x0001)

    # Parameters can be looked up by name
    ping_rsp = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
    assert ping_rsp.Capabilities == t.MTCapabilities.SYS

    # Invalid ones cannot
    with pytest.raises(AttributeError):
        ping_rsp.Oops

    # bytes are converted into t.ShortBytes
    cmd = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x0000, Value=b"asdfoo"
    )
    assert isinstance(cmd.Value, t.ShortBytes)

    # Lists are converted to typed LVLists
    c.UTIL.BindAddEntry.Req(
        DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
        DstEndpoint=0x56,
        ClusterIdList=[0x12, 0x45],
    )

    # Type errors within containers are also caught
    with pytest.raises(ValueError):
        c.UTIL.BindAddEntry.Req(
            DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
            DstEndpoint=0x56,
            ClusterIdList=[0x12, 0x457890],  # 0x457890 doesn't fit into a uint8_t
        )


def test_command_optional_params():
    # Optional params values don't need a value
    short_version_rsp = c.SYS.Version.Rsp(
        TransportRev=0,
        ProductId=1,
        MajorRel=2,
        MinorRel=3,
        MaintRel=4,
    )

    # Some can still be passed
    medium_version_rsp = c.SYS.Version.Rsp(
        TransportRev=0, ProductId=1, MajorRel=2, MinorRel=3, MaintRel=4, CodeRevision=5
    )

    # As can all
    long_version_rsp = c.SYS.Version.Rsp(
        TransportRev=0,
        ProductId=1,
        MajorRel=2,
        MinorRel=3,
        MaintRel=4,
        CodeRevision=5,
        BootloaderBuildType=c.sys.BootloaderBuildType.NON_BOOTLOADER_BUILD,
        BootloaderRevision=0xFFFFFFFF,
    )

    short_data = short_version_rsp.to_frame().data
    medium_data = medium_version_rsp.to_frame().data
    long_data = long_version_rsp.to_frame().data

    assert len(long_data) == len(medium_data) + 5 == len(short_data) + 9

    assert long_data.startswith(medium_data)
    assert medium_data.startswith(short_data)

    # Deserialization is greedy
    Version = c.SYS.Version.Rsp
    assert Version.from_frame(long_version_rsp.to_frame()) == long_version_rsp
    assert Version.from_frame(medium_version_rsp.to_frame()) == medium_version_rsp
    assert Version.from_frame(short_version_rsp.to_frame()) == short_version_rsp

    # Deserialization still fails if the frame is incomplete
    with pytest.raises(ValueError):
        Version.from_frame(
            frames.GeneralFrame(
                header=long_version_rsp.to_frame().header, data=long_data[:-1]
            )
        )

    # Deserialization will fail if the frame is incomplete but has no truncated fields
    with pytest.raises(ValueError):
        Version.from_frame(
            frames.GeneralFrame(
                header=long_version_rsp.to_frame().header, data=long_data[:4]
            )
        )

    with pytest.raises(ValueError):
        Version.from_frame(
            frames.GeneralFrame(
                header=long_version_rsp.to_frame().header, data=long_data + b"\x00"
            )
        )


def test_command_optional_params_failures():
    with pytest.raises(KeyError):
        # Optional params cannot be skipped over
        c.SYS.Version.Rsp(
            TransportRev=0,
            ProductId=1,
            MajorRel=2,
            MinorRel=3,
            MaintRel=4,
            # CodeRevision=5,
            BootloaderBuildType=c.sys.BootloaderBuildType.BUILT_AS_HEX,
        )

    # Unless it's a partial command
    partial = c.SYS.Version.Rsp(
        TransportRev=0,
        ProductId=1,
        MajorRel=2,
        MinorRel=3,
        MaintRel=4,
        # CodeRevision=5,
        BootloaderBuildType=c.sys.BootloaderBuildType.BUILT_AS_HEX,
        partial=True,
    )

    # In which case, it cannot be serialized
    with pytest.raises(ValueError):
        partial.to_frame()


def test_simple_descriptor():
    lvlist16_type = zigpy_t.LVList[t.uint16_t]

    simple_descriptor = zigpy.zdo.types.SimpleDescriptor()
    simple_descriptor.endpoint = zigpy_t.uint8_t(1)
    simple_descriptor.profile = zigpy_t.uint16_t(260)
    simple_descriptor.device_type = zigpy_t.uint16_t(257)
    simple_descriptor.device_version = zigpy_t.uint8_t(0)
    simple_descriptor.input_clusters = lvlist16_type([0, 3, 4, 5, 6, 8, 2821, 1794])
    simple_descriptor.output_clusters = lvlist16_type([10, 25])

    c1 = c.ZDO.SimpleDescRsp.Callback(
        Src=t.NWK(0x1234),
        Status=t.ZDOStatus.SUCCESS,
        NWK=t.NWK(0x1234),
        SimpleDescriptor=simple_descriptor,
    )

    sp_simple_descriptor = zigpy.zdo.types.SizePrefixedSimpleDescriptor()
    sp_simple_descriptor.endpoint = zigpy_t.uint8_t(1)
    sp_simple_descriptor.profile = zigpy_t.uint16_t(260)
    sp_simple_descriptor.device_type = zigpy_t.uint16_t(257)
    sp_simple_descriptor.device_version = zigpy_t.uint8_t(0)
    sp_simple_descriptor.input_clusters = lvlist16_type([0, 3, 4, 5, 6, 8, 2821, 1794])
    sp_simple_descriptor.output_clusters = lvlist16_type([10, 25])

    c2 = c.ZDO.SimpleDescRsp.Callback(
        Src=t.NWK(0x1234),
        Status=t.ZDOStatus.SUCCESS,
        NWK=t.NWK(0x1234),
        SimpleDescriptor=sp_simple_descriptor,
    )

    assert c1.to_frame() == c2.to_frame()
    # assert c1 == c2


def test_command_str_repr():
    command = c.UTIL.BindAddEntry.Req(
        DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
        DstEndpoint=0x56,
        ClusterIdList=[0x12, 0x34],
    )

    assert repr(command) == str(command)
    assert str([command]) == f"[{str(command)}]"


def test_command_immutability():
    command1 = c.SYS.NVWrite.Req(
        partial=True, SysId=None, ItemId=0x1234, SubId=None, Offset=None, Value=None
    )

    command2 = c.SYS.NVWrite.Req(
        partial=True, SysId=None, ItemId=0x1234, SubId=None, Offset=None, Value=None
    )

    d = {command1: True}

    assert command1 == command2
    assert command2 in d
    assert {command1: True} == {command2: True}

    with pytest.raises(RuntimeError):
        command1.partial = False

    with pytest.raises(RuntimeError):
        command1.SysId = 0x10

    with pytest.raises(RuntimeError):
        command1.ItemId = 0x1234

    with pytest.raises(RuntimeError):
        del command1.ItemId

    assert command1 == command2


def test_command_serialization():
    command = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x0000, Value=b"asdfoo"
    )
    frame = command.to_frame()

    assert frame.data == bytes.fromhex("12 5634 9078 0000 06") + b"asdfoo"

    # Partial frames cannot be serialized
    with pytest.raises(ValueError):
        partial1 = c.SYS.NVWrite.Req(partial=True, SysId=0x12)
        partial1.to_frame()

    # Partial frames cannot be serialized, even if all params are filled out
    with pytest.raises(ValueError):
        partial2 = c.SYS.NVWrite.Req(
            partial=True, SysId=None, ItemId=0x1234, SubId=None, Offset=None, Value=None
        )
        partial2.to_frame()


def test_command_equality():
    command1 = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
    )

    command2 = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
    )

    command3 = c.SYS.NVWrite.Req(
        SysId=0xFF, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
    )

    assert command1 == command1
    assert command1.matches(command1)
    assert command2 == command1
    assert command1 == command2

    assert command1 != command3
    assert command3 != command1

    assert command1.matches(command2)  # Matching is a superset of equality
    assert command2.matches(command1)
    assert not command1.matches(command3)
    assert not command3.matches(command1)

    assert not command1.matches(c.SYS.NVWrite.Req(partial=True))
    assert c.SYS.NVWrite.Req(partial=True).matches(command1)

    # parameters can be specified explicitly as None
    assert c.SYS.NVWrite.Req(partial=True, SubId=None).matches(command1)
    assert c.SYS.NVWrite.Req(partial=True, SubId=0x7890).matches(command1)
    assert not c.SYS.NVWrite.Req(partial=True, SubId=123).matches(command1)

    # Different frame types do not match, even if they have the same structure
    assert not c.SYS.NVWrite.Rsp(Status=t.Status.SUCCESS).matches(
        c.SYS.NVDelete.Rsp(partial=True)
    )


def test_command_deserialization():
    command = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
    )

    assert type(command).from_frame(command.to_frame()) == command
    assert command.to_frame() == type(command).from_frame(command.to_frame()).to_frame()

    # Deserialization fails if there is unparsed data at the end of the frame
    frame = command.to_frame()
    bad_frame = dataclasses.replace(frame, data=frame.data + b"\x00")

    with pytest.raises(ValueError):
        type(command).from_frame(bad_frame)

    # Deserialization fails if you attempt to deserialize the wrong frame
    with pytest.raises(ValueError):
        c.SYS.NVWrite.Req.from_frame(c.SYS.Ping.Req().to_frame())


def test_command_not_recognized():
    command = c.RPCError.CommandNotRecognized.Rsp(
        ErrorCode=c.rpc_error.ErrorCode.InvalidSubsystem,
        RequestHeader=t.CommandHeader(0xABCD),
    )

    transport_frame = frames.TransportFrame(command.to_frame())

    assert transport_frame.serialize()[:-1] == bytes.fromhex("FE  03  60 00  01  CD AB")


def test_command_replace_normal():
    command1 = c.SYS.NVWrite.Req(
        SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
    )

    command2 = c.SYS.NVWrite.Req(
        SysId=0x13, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoos"
    )

    assert command1.replace() == command1
    assert command1.replace(SysId=0x13, Value=b"asdfoos") == command2


def test_command_replace_partial():
    command1 = c.SYS.NVWrite.Req(partial=True, SysId=0x12)

    command2 = c.SYS.NVWrite.Req(partial=True, SysId=0x13)

    assert command1.replace() == command1
    assert command1.replace(SysId=0x13) == command2


def test_command_possibly_empty_payload():
    class TestSubsystem(t.CommandsBase, subsystem=t.Subsystem.SYS):
        Test = t.CommandDef(
            t.CommandType.AREQ,
            0x00,
            rsp_schema=(t.Param("Data", t.Bytes, "Can be any length"),),
        )

    Test = TestSubsystem.Test.Callback

    assert Test.from_frame(
        frames.GeneralFrame(header=Test.header, data=b"test")
    ) == Test(Data=t.Bytes(b"test"))

    assert Test.from_frame(frames.GeneralFrame(header=Test.header, data=b"")) == Test(
        Data=t.Bytes(b"")
    )


def test_neighbors_missing_payload():
    frame = frames.GeneralFrame(
        header=t.CommandHeader(
            id=0xB1,
            subsystem=t.Subsystem.ZDO,
            type=t.CommandType.AREQ,
        ),
        data=b"\x1F\x82\x84",
    )

    assert c.ZDO.MgmtLqiRsp.Callback.from_frame(frame) == c.ZDO.MgmtLqiRsp.Callback(
        Src=0x821F,
        Status=t.ZDOStatus.NOT_SUPPORTED,
    )