# -*- coding: utf-8 -*-
#
# Copyright (c) The PyAMF Project.
# See LICENSE.txt for details.

"""
Flex Messaging compatibility tests.

@since: 0.3.2
"""

import unittest
import datetime
import uuid

import pyamf
from pyamf.flex import messaging


class AbstractMessageTestCase(unittest.TestCase):
    def test_repr(self):
        a = messaging.AbstractMessage()

        a.body = u'é,è'

        try:
            repr(a)
        except:
            raise


class EncodingTestCase(unittest.TestCase):
    """
    Encoding tests for L{messaging}
    """

    def test_AcknowledgeMessage(self):
        m = messaging.AcknowledgeMessage()
        m.correlationId = '1234'

        self.assertEqual(
            pyamf.encode(m).getvalue(),
            b'\n\x81\x0bUflex.messaging.messages.AcknowledgeMessage\tbody'
            b'\x11clientId\x1bcorrelationId\x17destination\x0fheaders\x13'
            b'messageId\x15timeToLive\x13timestamp\x01\x01\x06\t1234\x01\n\x0b'
            b'\x01\x01\x01\x01\x01\x01'
        )

    def test_CommandMessage(self):
        m = messaging.CommandMessage(operation='foo.bar')

        self.assertEqual(
            pyamf.encode(m).getvalue(),
            b'\n\x81\x1bMflex.messaging.messages.CommandMessage\tbody\x11client'
            b'Id\x1bcorrelationId\x17destination\x0fheaders\x13messageId\x13ope'
            b'ration\x15timeToLive\x13timestamp\x01\x01\x01\x01\n\x0b\x01\x01'
            b'\x01\x06\x0ffoo.bar\x01\x01\x01'
        )

    def test_ErrorMessage(self):
        m = messaging.ErrorMessage(faultString='ValueError')

        self.assertEqual(
            pyamf.encode(m).getvalue(),
            b'\n\x81[Iflex.messaging.messages.ErrorMessage\tbody\x11clientId'
            b'\x1bcorrelationId\x17destination\x19extendedData\x13faultCode'
            b'\x17faultDetail\x17faultString\x0fheaders\x13messageId\x13rootCau'
            b'se\x15timeToLive\x13timestamp\x01\x01\x01\x01\n\x0b\x01\x01\x01'
            b'\x01\x06\x15ValueError\n\x05\x01\x01\n\x05\x01\x01\x01\x01'
        )

    def test_RemotingMessage(self):
        m = messaging.RemotingMessage(source='foo.bar')

        self.assertEqual(
            pyamf.encode(m).getvalue(),
            b'\n\x81\x1bOflex.messaging.messages.RemotingMessage\tbody\x11clien'
            b'tId\x17destination\x0fheaders\x13messageId\x13operation\rsource'
            b'\x15timeToLive\x13timestamp\x01\x01\x01\n\x0b\x01\x01\x01\x01\x06'
            b'\x0ffoo.bar\x01\x01\x01'
        )


class SmallMessageTestCase(unittest.TestCase):
    """
    Tests for L{messaging.SmallMessageMixIn}
    """

    def setUp(self):
        self.decoder = pyamf.get_decoder(pyamf.AMF3)
        self.buffer = self.decoder.stream

    def test_acknowledge(self):
        bytes = (
            b'\n\x07\x07DSK\xa8\x03\n\x0b\x01%DSMessagingVersion\x05?\xf0'
            b'\x00\x00\x00\x00\x00\x00\tDSId\x06IEE0D161D-C11D-25CB-8DBE-3B77B'
            b'54B55D9\x01\x05Br3&m\x85\x10\x00\x0c!\xee\r\x16\x1d\xc1(&[\xc9'
            b'\x80RK\x9bE\xc6\xc4\x0c!\xee\r\x16\x1d\xc1=\x8e\xa3\xe0\x10\xef'
            b'\xad;\xe5\xc5j\x02\x0c!S\x84\x83\xdb\xa9\xc8\xcaM`\x952f\xdbQ'
            b'\xc9<\x00'
        )
        self.buffer.write(bytes)
        self.buffer.seek(0)

        msg = self.decoder.readElement()

        self.assertTrue(isinstance(msg, messaging.AcknowledgeMessageExt))
        self.assertEqual(msg.body, None)
        self.assertEqual(msg.destination, None)
        self.assertEqual(msg.timeToLive, None)

        self.assertEqual(
            msg.timestamp,
            datetime.datetime(2009, 8, 19, 11, 24, 43, 985000)
        )
        self.assertEqual(msg.headers, {
            'DSMessagingVersion': 1.0,
            'DSId': u'EE0D161D-C11D-25CB-8DBE-3B77B54B55D9'
        })
        self.assertEqual(
            msg.clientId,
            uuid.UUID('ee0d161d-c128-265b-c980-524b9b45c6c4')
        )
        self.assertEqual(
            msg.messageId,
            uuid.UUID('ee0d161d-c13d-8ea3-e010-efad3be5c56a')
        )
        self.assertEqual(
            msg.correlationId,
            uuid.UUID('538483db-a9c8-ca4d-6095-3266db51c93c')
        )
        self.assertEqual(self.buffer.remaining(), 0)

        # now encode the msg to check that encoding is byte for byte the same
        buffer = pyamf.encode(msg, encoding=pyamf.AMF3).getvalue()

        try:
            self.assertEqual(buffer, bytes)
        except:
            self.assertEqual(buffer,
                             b'\n\x07\x07DSK\xa8\x03\n\x0b\x01\t'
                             b'DSId\x06IEE0D161D-C11D-25CB-8DBE-3B77B54B55D9%DSMessagingVersion'
                             b'\x05?\xf0\x00\x00\x00\x00\x00\x00\x01\x05Br3&m\x85\x10\x00\x0c!'
                             b'\xee\r\x16\x1d\xc1(&[\xc9\x80RK\x9bE\xc6\xc4\x0c!\xee\r\x16\x1d'
                             b'\xc1=\x8e\xa3\xe0\x10\xef\xad;\xe5\xc5j\x02\x0c!S\x84\x83\xdb\xa9'
                             b'\xc8\xcaM`\x952f\xdbQ\xc9<\x00')

    def test_command(self):
        bytes = (
            b'\n\x07\x07DSC\x88\x02\n\x0b\x01\tDSId\x06IEE0D161D-C11D-'
            b'25CB-8DBE-3B77B54B55D9\x01\x0c!\xc0\xdf\xb7|\xd6\xee$1s\x152f'
            b'\xe11\xa8f\x01\x06\x01\x01\x04\x02'
        )

        self.buffer.write(bytes)
        self.buffer.seek(0)

        msg = self.decoder.readElement()

        self.assertTrue(isinstance(msg, messaging.CommandMessageExt))
        self.assertEqual(msg.body, None)
        self.assertEqual(msg.destination, None)
        self.assertEqual(msg.timeToLive, None)

        self.assertEqual(msg.timestamp, None)
        self.assertEqual(msg.headers, {
            'DSId': u'EE0D161D-C11D-25CB-8DBE-3B77B54B55D9'
        })
        self.assertEqual(msg.clientId, None)
        self.assertEqual(
            msg.messageId,
            uuid.UUID('c0dfb77c-d6ee-2431-7315-3266e131a866')
        )
        self.assertEqual(msg.correlationId, u'')
        self.assertEqual(self.buffer.remaining(), 0)

        # now encode the msg to check that encoding is byte for byte the same
        buffer = pyamf.encode(msg, encoding=pyamf.AMF3).getvalue()

        self.assertEqual(buffer, bytes)

    def test_async(self):
        pass

    def test_getmessage(self):
        """
        Tests for `getSmallMessage`
        """
        for cls in ['AbstractMessage', 'ErrorMessage', 'RemotingMessage']:
            cls = getattr(messaging, cls)
            self.assertRaises(NotImplementedError, cls().getSmallMessage)

        kwargs = {
            'body': {'foo': 'bar'},
            'clientId': 'spam',
            'destination': 'eggs',
            'headers': {'blarg': 'whoop'},
            'messageId': 'baz',
            'timestamp': 1234,
            'timeToLive': 99
        }

        # test async
        a = messaging.AsyncMessage(correlationId='yay', **kwargs)
        m = a.getSmallMessage()

        k = kwargs.copy()
        k.update({'correlationId': 'yay'})

        self.assertTrue(isinstance(m, messaging.AsyncMessageExt))
        self.assertEqual(m.__dict__, k)

        # test command
        a = messaging.CommandMessage(operation='yay', **kwargs)
        m = a.getSmallMessage()

        k = kwargs.copy()
        k.update({'operation': 'yay', 'correlationId': None})

        self.assertTrue(isinstance(m, messaging.CommandMessageExt))
        self.assertEqual(m.__dict__, k)

        # test ack
        a = messaging.AcknowledgeMessage(**kwargs)
        m = a.getSmallMessage()

        k = kwargs.copy()
        k.update({'correlationId': None})

        self.assertTrue(isinstance(m, messaging.AcknowledgeMessageExt))
        self.assertEqual(m.__dict__, k)
