import asyncio

import pytest

import zigpy.device
import zigpy.types as t
import zigpy.zdo as zdo
import zigpy.zdo.types as zdo_types

from .async_mock import AsyncMock, MagicMock, patch, sentinel


def test_commands():
    for cmdid, cmdspec in zdo.types.CLUSTERS.items():
        assert 0 <= cmdid <= 0xFFFF
        assert isinstance(cmdspec, tuple)
        for paramname, paramtype in zip(cmdspec[0], cmdspec[1]):
            assert isinstance(paramname, str)
            assert hasattr(paramtype, "serialize")
            assert hasattr(paramtype, "deserialize")


@pytest.fixture
def METHOD_NAME(app):
    ieee = t.EUI64(map(t.uint8_t, [0, 1, 2, 3, 4, 5, 6, 7]))
    dev = zigpy.device.Device(app, ieee, 65535)
    dev.request = AsyncMock()

    app.devices[dev.ieee] = dev

    return zdo.ZDO(dev)


def test_deserialize(METHOD_NAME):
    hdr, args = METHOD_NAME.deserialize(2, b"\x01\x02\x03xx")
    assert hdr.tsn == 1
    assert hdr.is_reply is False
    assert args == [0x0302]


def test_deserialize_unknown(METHOD_NAME):
    with pytest.raises(ValueError):
        hdr, args = METHOD_NAME.deserialize(0x0100, b"\x01")


async def test_request(METHOD_NAME):
    await METHOD_NAME.request(2, 65535)
    app = METHOD_NAME._device._application
    assert METHOD_NAME.device.request.call_count == 1
    assert app.get_sequence.call_count == 1


async def test_bind(METHOD_NAME):
    cluster = MagicMock()
    cluster.endpoint.endpoint_id = 1
    cluster.cluster_id = 1026
    await METHOD_NAME.bind(cluster)
    assert METHOD_NAME.device.request.call_count == 1
    assert METHOD_NAME.device.request.call_args[0][1] == 0x0021


async def test_unbind(METHOD_NAME):
    cluster = MagicMock()
    cluster.endpoint.endpoint_id = 1
    cluster.cluster_id = 1026
    await METHOD_NAME.unbind(cluster)
    assert METHOD_NAME.device.request.call_count == 1
    assert METHOD_NAME.device.request.call_args[0][1] == 0x0022


@pytest.mark.parametrize(
    "remove_children, rejoin, flags",
    (
        (False, False, 0),
        (False, True, 0x80),
        (True, False, 0x40),
        (True, True, 0xC0),
    ),
)
async def test_leave(METHOD_NAME, remove_children, rejoin, flags):
    """Test ZDO leave request options."""
    with patch.object(METHOD_NAME, "request", AsyncMock()) as req_mock:
        await METHOD_NAME.leave(remove_children, rejoin)
    assert req_mock.await_count == 1
    assert req_mock.await_args[0][0] == 0x0034
    assert req_mock.await_args[0][1] == t.EUI64.convert("07:06:05:04:03:02:01:00")
    assert req_mock.await_args[0][2] == flags


async def test_permit(METHOD_NAME):
    await METHOD_NAME.permit()
    assert METHOD_NAME.device.request.call_count == 1
    assert METHOD_NAME.device.request.call_args[0][1] == 0x0036


async def test_broadcast(app):
    await zigpy.zdo.broadcast(app, 0x0036, 0, 0, 60, 0)

    assert app.send_packet.call_count == 1

    packet = app.send_packet.mock_calls[0].args[0]
    assert packet.dst.addr_mode == t.AddrMode.Broadcast
    assert packet.cluster_id == 0x0036


def _handle_match_desc(METHOD_NAME, profile):
    METHOD_NAME.reply = AsyncMock()
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd.Match_Desc_req
    METHOD_NAME.handle_message(5, 0x0006, hdr, [None, profile, [], []])
    assert METHOD_NAME.reply.call_count == 1


async def test_handle_match_desc_zha(METHOD_NAME):
    _handle_match_desc(METHOD_NAME, 260)
    await asyncio.wait(asyncio.all_tasks(), return_when=asyncio.FIRST_COMPLETED)
    assert METHOD_NAME.reply.await_count == 1
    assert METHOD_NAME.reply.call_args[0][3]


async def test_handle_match_desc_generic(METHOD_NAME):
    _handle_match_desc(METHOD_NAME, 0)
    await asyncio.wait(asyncio.all_tasks(), return_when=asyncio.FIRST_COMPLETED)
    assert METHOD_NAME.reply.await_count == 1
    assert not METHOD_NAME.reply.call_args[0][3]


async def test_handle_nwk_addr(METHOD_NAME):
    ieee = METHOD_NAME._device.application.state.node_info.ieee
    METHOD_NAME.reply = MagicMock()
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd.NWK_addr_req
    METHOD_NAME.handle_message(5, 0x0000, hdr, [ieee, 0x00])
    assert METHOD_NAME.reply.call_count == 1


async def test_handle_ieee_addr(METHOD_NAME):
    nwk = METHOD_NAME._device.application.state.node_info.nwk
    METHOD_NAME.reply = MagicMock()
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd.IEEE_addr_req
    METHOD_NAME.handle_message(5, 0x0001, hdr, [nwk, 0x00])
    assert METHOD_NAME.reply.call_count == 1


def test_handle_announce(METHOD_NAME):
    dev = METHOD_NAME._device
    listener = MagicMock()
    METHOD_NAME.add_listener(listener)

    dev._application.devices.pop(dev.ieee)
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd.Device_annce
    METHOD_NAME.handle_message(
        5, 0x0013, hdr, [dev.nwk, dev.ieee, 0], dst_addressing=sentinel.dst_addr
    )

    assert listener.device_announce.call_count == 1
    assert listener.device_announce.call_args[0][0] is dev

    assert listener.zdo_device_annce.call_count == 1
    assert listener.zdo_device_annce.call_args[0][0] is dev
    assert listener.zdo_device_annce.call_args[0][1] is sentinel.dst_addr
    assert listener.zdo_device_annce.call_args[0][2] is hdr
    assert listener.zdo_device_annce.call_args[0][3] == [dev.nwk, dev.ieee, 0]


def test_handle_permit_join(METHOD_NAME):
    listener = MagicMock()
    METHOD_NAME.add_listener(listener)
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd.Mgmt_Permit_Joining_req
    METHOD_NAME.handle_message(5, 0x0036, hdr, [100, 1])
    assert listener.permit_duration.call_count == 1


def test_handle_unsupported(METHOD_NAME):
    listener = MagicMock()
    METHOD_NAME.add_listener(listener)
    hdr = MagicMock()
    hdr.command_id = zdo_types.ZDOCmd(0xFFFF)
    assert hdr.command_id not in list(zdo_types.ZDOCmd)
    METHOD_NAME.request = MagicMock()
    METHOD_NAME.reply = MagicMock()
    METHOD_NAME.handle_message(5, 0xFFFF, hdr, [])

    assert listener.zdo_undefined_0xffff.call_count == 1
    assert METHOD_NAME.request.call_count == 0
    assert METHOD_NAME.reply.call_count == 0


def test_device_accessor(METHOD_NAME):
    assert METHOD_NAME.device.nwk == 65535


async def test_reply(METHOD_NAME):
    METHOD_NAME.device.request = AsyncMock()
    await METHOD_NAME.reply(0x0005)
    assert METHOD_NAME.device.request.call_count == 1


def test_get_attr_error(METHOD_NAME):
    with pytest.raises(AttributeError):
        METHOD_NAME.no_such_attribute()


async def test_reply_tsn_override(METHOD_NAME, monkeypatch):
    clusters = MagicMock()
    clusters.__getitem__.return_value = (
        sentinel.param_names,
        sentinel.scheam,
    )
    monkeypatch.setattr(zdo_types, "CLUSTERS", clusters)
    mock_ser = MagicMock()
    mock_ser.return_value = b"\xaa\x55"
    monkeypatch.setattr(t, "serialize", mock_ser)
    await METHOD_NAME.reply(sentinel.cmd, sentinel.arg1, sentinel.arg2)
    seq = METHOD_NAME.device.request.call_args[0][4]
    data = METHOD_NAME.device.request.call_args[0][5]
    assert seq == 123
    assert data[0] == 123
    assert data[1:3] == b"\xaa\x55"

    # override tsn
    tsn = 0x23
    await METHOD_NAME.reply(sentinel.cmd, sentinel.arg1, sentinel.arg2, tsn=tsn)
    seq = METHOD_NAME.device.request.call_args[0][4]
    data = METHOD_NAME.device.request.call_args[0][5]
    assert seq == tsn
    assert data[0] == tsn
    assert data[1:3] == b"\xaa\x55"