// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

namespace DotNetty.Codecs.Mqtt.Tests
{
    using System.Collections.Generic;
    using System.Linq;
    using DotNetty.Buffers;
    using DotNetty.Codecs.Mqtt.Packets;
    using DotNetty.Transport.Channels;
    using Moq;
    using Xunit;

    public class MqttCodecTests
    {
        static readonly IByteBufferAllocator Allocator = new UnpooledByteBufferAllocator();

        readonly MqttDecoder serverDecoder;
        readonly MqttDecoder clientDecoder;
        readonly Mock<IChannelHandlerContext> contextMock;

        public MqttCodecTests()
        {
            this.serverDecoder = new MqttDecoder(true, 256 * 1024);
            this.clientDecoder = new MqttDecoder(false, 256 * 1024);
            this.contextMock = new Mock<IChannelHandlerContext>(MockBehavior.Strict);
            this.contextMock.Setup(x => x.Removed).Returns(false);
            this.contextMock.Setup(x => x.Allocator).Returns(UnpooledByteBufferAllocator.Default);
        }

        [Theory]
        [InlineData("a", true, 0, null, null, "will/topic/name", new byte[] { 5, 3, 255, 6, 5 }, QualityOfService.ExactlyOnce, true)]
        [InlineData("11a_2", false, 1, "user1", null, "will", new byte[0], QualityOfService.AtLeastOnce, false)]
        [InlineData("abc/ж", false, 10, "", "pwd", null, null, null, false)]
        [InlineData("", true, 1000, "имя", "密碼", null, null, null, false)]
        public void TestConnectMessage(string clientId, bool cleanSession, int keepAlive, string userName,
            string password, string willTopicName, byte[] willMessage, QualityOfService? willQos, bool willRetain)
        {
            var packet = new ConnectPacket();
            packet.ClientId = clientId;
            packet.CleanSession = cleanSession;
            packet.KeepAliveInSeconds = keepAlive;
            if (userName != null)
            {
                packet.Username = userName;
                if (password != null)
                {
                    packet.Password = password;
                }
            }
            if (willTopicName != null)
            {
                packet.WillTopicName = willTopicName;
                packet.WillMessage = Unpooled.WrappedBuffer(willMessage);
                packet.WillQualityOfService = willQos ?? QualityOfService.AtMostOnce;
                packet.WillRetain = willRetain;
            }

            ConnectPacket recoded = this.RecodePacket(packet, true, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<ConnectPacket>()), Times.Once);
            Assert.Equal(packet.ClientId, recoded.ClientId);
            Assert.Equal(packet.CleanSession, recoded.CleanSession);
            Assert.Equal(packet.KeepAliveInSeconds, recoded.KeepAliveInSeconds);
            Assert.Equal(packet.HasUsername, recoded.HasUsername);
            if (packet.HasUsername)
            {
                Assert.Equal(packet.Username, recoded.Username);
            }
            Assert.Equal(packet.HasPassword, recoded.HasPassword);
            if (packet.HasPassword)
            {
                Assert.Equal(packet.Password, recoded.Password);
            }
            if (packet.HasWill)
            {
                Assert.Equal(packet.WillTopicName, recoded.WillTopicName);
                Assert.True(ByteBufferUtil.Equals(Unpooled.WrappedBuffer(willMessage), recoded.WillMessage));
                Assert.Equal(packet.WillQualityOfService, recoded.WillQualityOfService);
                Assert.Equal(packet.WillRetain, recoded.WillRetain);
            }
        }

        [Theory]
        [InlineData(false, ConnectReturnCode.Accepted)]
        [InlineData(true, ConnectReturnCode.Accepted)]
        [InlineData(false, ConnectReturnCode.RefusedUnacceptableProtocolVersion)]
        [InlineData(false, ConnectReturnCode.RefusedIdentifierRejected)]
        [InlineData(false, ConnectReturnCode.RefusedServerUnavailable)]
        [InlineData(false, ConnectReturnCode.RefusedBadUsernameOrPassword)]
        [InlineData(false, ConnectReturnCode.RefusedNotAuthorized)]
        public void TestConnAckMessage(bool sessionPresent, ConnectReturnCode returnCode)
        {
            var packet = new ConnAckPacket();
            packet.SessionPresent = sessionPresent;
            packet.ReturnCode = returnCode;

            ConnAckPacket recoded = this.RecodePacket(packet, false, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<ConnAckPacket>()), Times.Once);
            Assert.Equal(packet.SessionPresent, recoded.SessionPresent);
            Assert.Equal(packet.ReturnCode, recoded.ReturnCode);
        }

        [Theory]
        [InlineData(1, new[] { "+", "+/+", "//", "/#", "+//+" }, new[] { QualityOfService.ExactlyOnce, QualityOfService.AtLeastOnce, QualityOfService.AtMostOnce, QualityOfService.ExactlyOnce, QualityOfService.AtMostOnce })]
        [InlineData(ushort.MaxValue, new[] { "a" }, new[] { QualityOfService.AtLeastOnce })]
        public void TestSubscribeMessage(int packetId, string[] topicFilters, QualityOfService[] requestedQosValues)
        {
            var packet = new SubscribePacket(packetId, topicFilters.Zip(requestedQosValues, (topic, qos) => new SubscriptionRequest(topic, qos)).ToArray());

            SubscribePacket recoded = this.RecodePacket(packet, true, true);
            clientDecoder
            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<SubscribePacket>()), Times.Once);
            Assert.Equal(packet.Requests, recoded.Requests, EqualityComparer<SubscriptionRequest>.Default);
            Assert.Equal(packet.PacketId, recoded.PacketId);
        }

        [Theory]
        [InlineData(1, new[] { QualityOfService.ExactlyOnce, QualityOfService.AtLeastOnce, QualityOfService.AtMostOnce, QualityOfService.Failure, QualityOfService.AtMostOnce })]
        [InlineData(ushort.MaxValue, new[] { QualityOfService.AtLeastOnce })]
        public void TestSubAckMessage(int packetId, QualityOfService[] qosValues)
        {
            var packet = new SubAckPacket();
            packet.PacketId = packetId;
            packet.ReturnCodes = qosValues;

            SubAckPacket recoded = this.RecodePacket(packet, false, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<SubAckPacket>()), Times.Once);
            Assert.Equal(packet.ReturnCodes, recoded.ReturnCodes);
            Assert.Equal(packet.PacketId, recoded.PacketId);
        }

        [Theory]
        [InlineData(1, new[] { "+", "+/+", "//", "/#", "+//+" })]
        [InlineData(ushort.MaxValue, new[] { "a" })]
        public void TestUnsubscribeMessage(int packetId, string[] topicFilters)
        {
            var packet = new UnsubscribePacket(packetId, topicFilters);

            UnsubscribePacket recoded = this.RecodePacket(packet, true, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<UnsubscribePacket>()), Times.Once);
            Assert.Equal(packet.TopicFilters, recoded.TopicFilters);
            Assert.Equal(packet.PacketId, recoded.PacketId);
        }

        [Theory]
        [InlineData(QualityOfService.AtMostOnce, false, false, 1, "a", null)]
        [InlineData(QualityOfService.ExactlyOnce, true, false, ushort.MaxValue, "/", new byte[0])]
        [InlineData(QualityOfService.AtLeastOnce, false, true, 129, "a/b", new byte[] { 1, 2, 3 })]
        [InlineData(QualityOfService.ExactlyOnce, true, true, ushort.MaxValue - 1, "topic/name/that/is/longer/than/256/characters/topic/name/that/is/longer/than/256/characters/topic/name/that/is/longer/than/256/characters/topic/name/that/is/longer/than/256/characters/topic/name/that/is/longer/than/256/characters/topic/name/that/is/longer/than/256/characters/", new byte[] { 1 })]
        public void TestPublishMessage(QualityOfService qos, bool dup, bool retain, int packetId, string topicName, byte[] payload)
        {
            var packet = new PublishPacket(qos, dup, retain);
            packet.TopicName = topicName;
            if (qos > QualityOfService.AtMostOnce)
            {
                packet.PacketId = packetId;
            }
            packet.Payload = payload == null ? null : Unpooled.WrappedBuffer(payload);

            PublishPacket recoded = this.RecodePacket(packet, false, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<PublishPacket>()), Times.Once);
            Assert.Equal(packet.TopicName, recoded.TopicName);
            if (packet.QualityOfService > QualityOfService.AtMostOnce)
            {
                Assert.Equal(packet.PacketId, recoded.PacketId);
            }
            Assert.True(ByteBufferUtil.Equals(payload == null ? Unpooled.Empty : Unpooled.WrappedBuffer(payload), recoded.Payload));
        }

        [Theory]
        [InlineData(1)]
        [InlineData(127)]
        [InlineData(128)]
        [InlineData(256)]
        [InlineData(257)]
        [InlineData(ushort.MaxValue)]
        public void TestPacketIdOnlyResponseMessages(int packetId)
        {
            this.TestPublishResponseMessage<PubAckPacket>(packetId, true);
            this.TestPublishResponseMessage<PubAckPacket>(packetId, false);
            this.TestPublishResponseMessage<PubRecPacket>(packetId, true);
            this.TestPublishResponseMessage<PubRecPacket>(packetId, false);
            this.TestPublishResponseMessage<PubRelPacket>(packetId, true);
            this.TestPublishResponseMessage<PubRelPacket>(packetId, false);
            this.TestPublishResponseMessage<PubCompPacket>(packetId, true);
            this.TestPublishResponseMessage<PubCompPacket>(packetId, false);
            this.TestPublishResponseMessage<UnsubAckPacket>(packetId, false);
        }

        void TestPublishResponseMessage<T>(int packetId, bool useServer)
            where T : PacketWithId, new()
        {
            var packet = new T
            {
                PacketId = packetId
            };

            T recoded = this.RecodePacket(packet, useServer, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<T>()), Times.Once);
            this.contextMock.ResetCalls();
            Assert.Equal(packet.PacketId, recoded.PacketId);
        }

        [Fact]
        public void TestEmptyPacketMessages()
        {
            this.TestEmptyPacketMessage(PingReqPacket.Instance, true);
            this.TestEmptyPacketMessage(PingRespPacket.Instance, false);
            this.TestEmptyPacketMessage(DisconnectPacket.Instance, true);
        }

        void TestEmptyPacketMessage<T>(T packet, bool useServer)
            where T : Packet
        {
            T recoded = this.RecodePacket(packet, useServer, true);

            this.contextMock.Verify(x => x.FireChannelRead(It.IsAny<T>()), Times.Once);
        }

        T RecodePacket<T>(T packet, bool useServer, bool explodeForDecode)
            where T : Packet
        {
            var output = new List<object>();
            MqttEncoder.DoEncode(Allocator, packet, output);

            T observedPacket = null;
            this.contextMock.Setup(x => x.FireChannelRead(It.IsAny<T>()))
                .Callback((object message) => observedPacket = Assert.IsAssignableFrom<T>(message))
                .Returns(this.contextMock.Object);

            foreach (IByteBuffer message in output)
            {
                MqttDecoder mqttDecoder = useServer ? this.serverDecoder : this.clientDecoder;
                if (explodeForDecode)
                {
                    while (message.IsReadable())
                    {
                        IByteBuffer finalBuffer = message.ReadBytes(1);
                        mqttDecoder.ChannelRead(this.contextMock.Object, finalBuffer);
                    }
                }
                else
                {  
                    mqttDecoder.ChannelRead(this.contextMock.Object, message);
                }
            }
            return observedPacket;
        }
    }
}