#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from __future__ import absolute_import

from uuid import uuid4

from proton import *

from . import common

class Test(common.Test):

  def setUp(self):
    self.msg = Message()

  def tearDown(self):
    self.msg = None


class AccessorsTest(Test):

  def _test(self, name, default, values):
    d = getattr(self.msg, name)
    assert d == default, (d, default)
    for v in values:
      setattr(self.msg, name, v)
      gotten = getattr(self.msg, name)
      assert gotten == v, gotten

  def _test_symbol(self, name):
    self._test(name, symbol(None), (symbol(u"abc.123.#$%"), symbol(u"hello.world")))

  def _test_str(self, name):
    self._test(name, None, (u"asdf", u"fdsa", u""))

  def _test_time(self, name):
    self._test(name, 0, (0, 123456789, 987654321))

  def testId(self):
    self._test("id", None, ("bytes", None, 123, u"string", uuid4()))

  def testCorrelationId(self):
    self._test("correlation_id", None, ("bytes", None, 123, u"string", uuid4()))

  def testDurable(self):
    self._test("durable", False, (True, False))

  def testPriority(self):
    self._test("priority", Message.DEFAULT_PRIORITY, range(0, 255))

  def testTtl(self):
    self._test("ttl", 0, range(12345, 54321))

  def testFirstAcquirer(self):
    self._test("first_acquirer", False, (True, False))

  def testDeliveryCount(self):
    self._test("delivery_count", 0, range(0, 1024))

  def testUserId(self):
    self._test("user_id", b"", (b"asdf", b"fdsa", b"asd\x00fdsa", b""))

  def testAddress(self):
    self._test_str("address")

  def testSubject(self):
    self._test_str("subject")

  def testReplyTo(self):
    self._test_str("reply_to")

  def testContentType(self):
    self._test_symbol("content_type")

  def testContentEncoding(self):
    self._test_symbol("content_encoding")

  def testExpiryTime(self):
    self._test_time("expiry_time")

  def testCreationTime(self):
    self._test_time("creation_time")

  def testGroupId(self):
    self._test_str("group_id")

  def testGroupSequence(self):
    self._test("group_sequence", 0, (0, 1, 10, 20, 4294967294, 4294967295))

  def testReplyToGroupId(self):
    self._test_str("reply_to_group_id")

class CodecTest(Test):

  def testProperties(self):
    self.msg.properties = {}
    self.msg.properties['key'] = 'value'
    data = self.msg.encode()

    msg2 = Message()
    msg2.decode(data)

    assert msg2.properties['key'] == 'value', msg2.properties['key']

  def testAnnotationsSymbolicAndUlongKey(self, a={symbol('one'): 1, 'two': 2, ulong(3): 'three'}):
    self.msg.annotations = a
    data = self.msg.encode()

    msg2 = Message()
    msg2.decode(data)
    # both keys must be symbols
    assert msg2.annotations == a

  def testRoundTrip(self):
    self.msg.id = "asdf"
    self.msg.correlation_id = uuid4()
    self.msg.ttl = 3
    self.msg.priority = 100
    self.msg.address = "address"
    self.msg.subject = "subject"
    self.msg.body = 'Hello World!'

    data = self.msg.encode()

    msg2 = Message()
    msg2.decode(data)

    assert self.msg.id == msg2.id, (self.msg.id, msg2.id)
    assert self.msg.correlation_id == msg2.correlation_id, (self.msg.correlation_id, msg2.correlation_id)
    assert self.msg.ttl == msg2.ttl, (self.msg.ttl, msg2.ttl)
    assert self.msg.priority == msg2.priority, (self.msg.priority, msg2.priority)
    assert self.msg.address == msg2.address, (self.msg.address, msg2.address)
    assert self.msg.subject == msg2.subject, (self.msg.subject, msg2.subject)
    assert self.msg.body == msg2.body, (self.msg.body, msg2.body)

  def testExpiryEncodeAsNull(self):
    self.msg.group_id = "A" # Force creation and expiry fields to be present
    data = self.msg.encode()

    decoder = Data()

    # Skip past the headers
    consumed = decoder.decode(data)
    decoder.clear()
    data = data[consumed:]

    decoder.decode(data)
    dproperties = decoder.get_py_described()
    # Check we've got the correct described list
    assert dproperties.descriptor == 0x73, (dproperties.descriptor)

    properties = dproperties.value
    assert properties[8] == None, properties[8]

  def testCreationEncodeAsNull(self):
    self.msg.group_id = "A" # Force creation and expiry fields to be present
    data = self.msg.encode()

    decoder = Data()

    # Skip past the headers
    consumed = decoder.decode(data)
    decoder.clear()
    data = data[consumed:]

    decoder.decode(data)
    dproperties = decoder.get_py_described()
    # Check we've got the correct described list
    assert dproperties.descriptor == 0x73, (dproperties.descriptor)

    properties = dproperties.value
    assert properties[9] == None, properties[9]

  def testGroupSequenceEncodeAsNull(self):
    self.msg.reply_to_group_id = "R" # Force group_id and group_sequence fields to be present
    data = self.msg.encode()

    decoder = Data()

    # Skip past the headers
    consumed = decoder.decode(data)
    decoder.clear()
    data = data[consumed:]

    decoder.decode(data)
    dproperties = decoder.get_py_described()
    # Check we've got the correct described list
    assert dproperties.descriptor == 0x73, (dproperties.descriptor)

    properties = dproperties.value
    assert properties[10] == None, properties[10]
    assert properties[11] == None, properties[11]

  def testGroupSequenceEncodeAsNonNull(self):
    self.msg.group_id = "G"
    self.msg.reply_to_group_id = "R" # Force group_id and group_sequence fields to be present
    data = self.msg.encode()

    decoder = Data()

    # Skip past the headers
    consumed = decoder.decode(data)
    decoder.clear()
    data = data[consumed:]

    decoder.decode(data)
    dproperties = decoder.get_py_described()
    # Check we've got the correct described list
    assert dproperties.descriptor == 0x73, (dproperties.descriptor)

    properties = dproperties.value
    assert properties[10] == 'G', properties[10]
    assert properties[11] == 0, properties[11]

  def testDefaultCreationExpiryDecode(self):
    # This is a message with everything filled explicitly as null or zero in LIST32 HEADER and PROPERTIES lists
    data = b'\x00\x53\x70\xd0\x00\x00\x00\x0a\x00\x00\x00\x05\x42\x40\x40\x42\x52\x00\x00\x53\x73\xd0\x00\x00\x00\x12\x00\x00\x00\x0d\x40\x40\x40\x40\x40\x40\x40\x40\x40\x40\x40\x52\x00\x40'
    msg2 = Message()
    msg2.decode(data)
    assert msg2.expiry_time == 0, (msg2.expiry_time)
    assert msg2.creation_time == 0, (msg2.creation_time)

    # The same message with LIST8s instead
    data = b'\x00\x53\x70\xc0\x07\x05\x42\x40\x40\x42\x52\x00\x00\x53\x73\xc0\x0f\x0d\x40\x40\x40\x40\x40\x40\x40\x40\x40\x40\x40\x52\x00\x40'
    msg3 = Message()
    msg3.decode(data)
    assert msg2.expiry_time == 0, (msg2.expiry_time)
    assert msg2.creation_time == 0, (msg2.creation_time)

    # Minified message with zero length HEADER and PROPERTIES lists
    data = b'\x00\x53\x70\x45' b'\x00\x53\x73\x45'
    msg4 = Message()
    msg4.decode(data)
    assert msg2.expiry_time == 0, (msg2.expiry_time)
    assert msg2.creation_time == 0, (msg2.creation_time)

  def testDefaultPriorityEncode(self):
    assert self.msg.priority == 4, (self.msg.priority)
    self.msg.ttl = 0.003 # field after priority, so forces priority to be present
    data = self.msg.encode()

    decoder = Data()
    decoder.decode(data)

    dheaders = decoder.get_py_described()
    # Check we've got the correct described list
    assert dheaders.descriptor == 0x70, (dheaders.descriptor)

    # Check that the priority field (second field) is encoded as null
    headers = dheaders.value
    assert headers[1] == None, (headers[1])

  def testDefaultPriorityDecode(self):
    # This is a message with everything filled explicitly as null or zero in LIST32 HEADER and PROPERTIES lists
    data = b'\x00\x53\x70\xd0\x00\x00\x00\x0a\x00\x00\x00\x05\x42\x40\x40\x42\x52\x00\x00\x53\x73\xd0\x00\x00\x00\x22\x00\x00\x00\x0d\x40\x40\x40\x40\x40\x40\x40\x40\x83\x00\x00\x00\x00\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00\x40\x52\x00\x40'
    msg2 = Message()
    msg2.decode(data)
    assert msg2.priority == 4, (msg2.priority)

    # The same message with LIST8s instead
    data = b'\x00\x53\x70\xc0\x07\x05\x42\x40\x40\x42\x52\x00\x00\x53\x73\xc0\x1f\x0d\x40\x40\x40\x40\x40\x40\x40\x40\x83\x00\x00\x00\x00\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00\x40\x52\x00\x40'
    msg3 = Message()
    msg3.decode(data)
    assert msg3.priority == 4, (msg3.priority)

    # Minified message with zero length HEADER and PROPERTIES lists
    data = b'\x00\x53\x70\x45' b'\x00\x53\x73\x45'
    msg4 = Message()
    msg4.decode(data)
    assert msg4.priority == 4, (msg4.priority)