# -*- 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(),
            '\n\x81\x0bUflex.messaging.messages.AcknowledgeMessage\tbody'
            '\x11clientId\x17destination\x0fheaders\x13messageId\x13timestamp'
            '\x15timeToLive\x1bcorrelationId\x01\x01\x01\n\x0b\x01\x01\x01\x01'
            '\x01\x06\t1234\x01')

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

        self.assertEqual(pyamf.encode(m).getvalue(),
            '\n\x81\x1bMflex.messaging.messages.CommandMessage\x1bcorrelationId'
            '\tbody\x11clientId\x17destination\x0fheaders\x13messageId\x13'
            'timestamp\x15timeToLive\x13operation\x01\x01\x01\x01\n\x0b\x01\x01'
            '\x01\x01\x01\x06\x0ffoo.bar\x01')

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

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

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

        self.assertEqual(pyamf.encode(m).getvalue(),
            '\n\x81\x1bOflex.messaging.messages.RemotingMessage\x15timeToLive'
            '\x13timestamp\x13messageId\x0fheaders\x17destination\x11clientId'
            '\tbody\x13operation\rsource\x01\x01\x01\n\x0b\x01\x01\x01\x01\x01'
            '\x01\x06\x0ffoo.bar\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 = ('\n\x07\x07DSK\xa8\x03\n\x0b\x01%DSMessagingVersion\x05?\xf0'
            '\x00\x00\x00\x00\x00\x00\tDSId\x06IEE0D161D-C11D-25CB-8DBE-3B77B'
            '54B55D9\x01\x05Br3&m\x85\x10\x00\x0c!\xee\r\x16\x1d\xc1(&[\xc9'
            '\x80RK\x9bE\xc6\xc4\x0c!\xee\r\x16\x1d\xc1=\x8e\xa3\xe0\x10\xef'
            '\xad;\xe5\xc5j\x02\x0c!S\x84\x83\xdb\xa9\xc8\xcaM`\x952f\xdbQ'
            '\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()

        self.assertEqual(buffer, bytes)

    def test_command(self):
        bytes = ('\n\x07\x07DSC\x88\x02\n\x0b\x01\tDSId\x06IEE0D161D-C11D-'
            '25CB-8DBE-3B77B54B55D9\x01\x0c!\xc0\xdf\xb7|\xd6\xee$1s\x152f'
            '\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)