# Python imports
import unittest
import time
import os
import numbers
import sys

# Project imports
import sysv_ipc
from .base import Base, make_key, sleep_past_granularity

# Not tested --
# - mode seems to be settable and readable, but ignored by the OS


class MessageQueueTestBase(Base):
    """base class for MessageQueue test classes"""
    def setUp(self):
        self.mq = sysv_ipc.MessageQueue(None, sysv_ipc.IPC_CREX)

    def tearDown(self):
        if self.mq:
            self.mq.remove()

    def assertWriteToReadOnlyPropertyFails(self, property_name, value):
        """test that writing to a readonly property raises TypeError"""
        Base.assertWriteToReadOnlyPropertyFails(self, self.mq, property_name, value)


class TestMessageQueueCreation(MessageQueueTestBase):
    """Exercise stuff related to creating MessageQueue"""

    def test_no_flags(self):
        """tests that opening a MessageQueue with no flags opens the existing
        MessageQueue and doesn't create a new MessageQueue"""
        mem_copy = sysv_ipc.MessageQueue(self.mq.key)
        self.assertEqual(self.mq.key, mem_copy.key)

    def test_IPC_CREAT_existing(self):
        """tests sysv_ipc.IPC_CREAT to open an existing MessageQueue without IPC_EXCL"""
        mem_copy = sysv_ipc.MessageQueue(self.mq.key, sysv_ipc.IPC_CREAT)

        self.assertEqual(self.mq.key, mem_copy.key)

    def test_IPC_CREAT_new(self):
        """tests sysv_ipc.IPC_CREAT to create a new MessageQueue without IPC_EXCL"""
        # I can't pass None for the name unless I also pass IPC_EXCL.
        key = make_key()

        # Note: this method of finding an unused key is vulnerable to a race
        # condition. It's good enough for test, but don't copy it for use in
        # production code!
        key_is_available = False
        while not key_is_available:
            try:
                mem = sysv_ipc.MessageQueue(key)
                mem.detach()
                mem.remove()
            except sysv_ipc.ExistentialError:
                key_is_available = True
            else:
                key = make_key()

        mq = sysv_ipc.MessageQueue(key, sysv_ipc.IPC_CREAT)

        self.assertIsNotNone(mq)

        mq.remove()

    def test_IPC_EXCL(self):
        """tests IPC_CREAT | IPC_EXCL prevents opening an existing MessageQueue"""
        with self.assertRaises(sysv_ipc.ExistentialError):
            sysv_ipc.MessageQueue(self.mq.key, sysv_ipc.IPC_CREX)

    def test_randomly_generated_key(self):
        """tests that the randomly-generated key works"""
        # This is tested implicitly elsewhere but I want to test it explicitly
        mq = sysv_ipc.MessageQueue(None, sysv_ipc.IPC_CREX)
        self.assertIsNotNone(mq.key)
        self.assertGreaterEqual(mq.key, sysv_ipc.KEY_MIN)
        self.assertLessEqual(mq.key, sysv_ipc.KEY_MAX)
        mq.remove()

    # don't bother testing mode, it's ignored by the OS?

    def test_default_flags(self):
        """tests that the flag is 0 by default (==> open existing)"""
        mq = sysv_ipc.MessageQueue(self.mq.key)
        self.assertEqual(self.mq.id, mq.id)

    def test_kwargs(self):
        """ensure init accepts keyword args as advertised"""
        # mode 0x180 = 0600. Octal is difficult to express in Python 2/3 compatible code.
        mq = sysv_ipc.MessageQueue(None, flags=sysv_ipc.IPC_CREX, mode=0x180,
                                   max_message_size=256)
        mq.remove()


class TestMessageQueueSendReceive(MessageQueueTestBase):
    """Exercise send() and receive()"""
    def test_simple_send_receive(self):
        test_string = b'abcdefg'
        self.mq.send(test_string)
        self.assertEqual(self.mq.receive(), (test_string, 1))

    def test_message_type_send(self):
        """test the msg type param of send()"""
        test_string = b'abcdefg'
        self.mq.send(test_string, type=2)
        self.assertEqual(self.mq.receive(), (test_string, 2))

        with self.assertRaises(ValueError):
            self.mq.send(test_string, type=-1)

    def test_message_type_receive_default_order(self):
        """test that receive() doesn't filter by type by default"""
        for i in range(1, 4):
            self.mq.send('type' + str(i), type=i)

        # default order is FIFO.
        self.assertEqual(self.mq.receive(), (b'type1', 1))
        self.assertEqual(self.mq.receive(), (b'type2', 2))
        self.assertEqual(self.mq.receive(), (b'type3', 3))

        self.assertEqual(self.mq.current_messages, 0)

    # The bug referenced below affects use of a negative type. Supposedly it's only on 32 binaries
    # running on 64 bit systems, but I see it using 64-bit Python under 64-bit Linux.
    # A less demanding version of this test follows so Linux doesn't go entirely untested.
    @unittest.skipIf(sys.platform.startswith('linux'),
                     'msgrcv() buggy on Linux: https://bugzilla.kernel.org/show_bug.cgi?id=94181')
    def test_message_type_receive_specific_order(self):
        # Place messsages in Q w/highest type first
        for i in range(4, 0, -1):
            self.mq.send('type' + str(i), type=i)

        # receive(type=-2) should get "the first message of the lowest type that is <= the absolute
        # value of type."
        self.assertEqual(self.mq.receive(type=-2), (b'type2', 2))

        # receive(type=3) should get "the first message of that type."
        self.assertEqual(self.mq.receive(type=3), (b'type3', 3))

        # Ensure the others are still there.
        self.assertEqual(self.mq.receive(), (b'type4', 4))
        self.assertEqual(self.mq.receive(), (b'type1', 1))

        self.assertEqual(self.mq.current_messages, 0)

    def test_message_type_receive_specific_order_no_negative_type(self):
        """test that receive() filters appropriately on positive msg type (softer test for Linux)"""
        # Place messsages in Q w/highest type first
        for i in range(4, 0, -1):
            self.mq.send('type' + str(i), type=i)

        # receive(type=3) should get "the first message of that type."
        self.assertEqual(self.mq.receive(type=3), (b'type3', 3))

        # Ensure the others are still there.
        self.assertEqual(self.mq.receive(), (b'type4', 4))
        self.assertEqual(self.mq.receive(), (b'type2', 2))
        self.assertEqual(self.mq.receive(), (b'type1', 1))

        self.assertEqual(self.mq.current_messages, 0)

    def test_send_non_blocking(self):
        """Test that send(block=False) raises BusyError as appropriate"""
        # This is a bit tricky since the OS has its own ideas about when the queue is full.
        # I would like to fill it precisely to the brim and then make one last call to send(),
        # but I don't know exactly when the OS will decide the Q is full. Instead, I just keep
        # stuffing messages in until I get some kind of an error. If it's a BusyError, all is well.
        done = False

        while not done:
            try:
                self.mq.send('x', block=False)
            except sysv_ipc.BusyError:
                done = True

    def test_receive_non_blocking(self):
        """Test that receive(block=False) raises BusyError as appropriate"""
        with self.assertRaises(sysv_ipc.BusyError):
            self.mq.receive(block=False)

        self.mq.send('x', type=3)
        with self.assertRaises(sysv_ipc.BusyError):
            self.mq.receive(type=2, block=False)

    def test_ascii_null(self):
        """ensure I can send & receive 0x00"""
        test_string = b'abc' + bytes(0) + b'def'
        self.mq.send(test_string)
        self.assertEqual(self.mq.receive(), (test_string, 1))

    def test_utf8(self):
        """Test writing encoded Unicode"""
        test_string = 'G' + '\u00F6' + 'teborg'
        test_string = test_string.encode('utf-8')
        self.mq.send(test_string)
        self.assertEqual(self.mq.receive(), (test_string, 1))

    def test_send_kwargs(self):
        """ensure send() accepts keyword args as advertised"""
        self.mq.send(b'x', block=True, type=1)

    def test_receive_kwargs(self):
        """ensure receive() accepts keyword args as advertised"""
        self.mq.send(b'x', block=True, type=1)
        self.mq.receive(block=False, type=0)

    def test_max_message_size_respected(self):
        '''ensure the max_message_size param is respected'''
        mq = sysv_ipc.MessageQueue(None, sysv_ipc.IPC_CREX, max_message_size=10)

        with self.assertRaises(ValueError):
            mq.send(b' ' * 11, block=False)

        mq.remove()


class TestMessageQueueRemove(MessageQueueTestBase):
    """Exercise mq.remove()"""
    def test_remove(self):
        """tests that mq.remove() works"""
        self.mq.remove()
        with self.assertRaises(sysv_ipc.ExistentialError):
            sysv_ipc.MessageQueue(self.mq.key)
        # Wipe this out so that self.tearDown() doesn't crash.
        self.mq = None


class TestMessageQueuePropertiesAndAttributes(MessageQueueTestBase):
    """Exercise props and attrs"""
    def test_property_key(self):
        """exercise MessageQueue.key"""
        self.assertGreaterEqual(self.mq.key, sysv_ipc.KEY_MIN)
        self.assertLessEqual(self.mq.key, sysv_ipc.KEY_MAX)
        self.assertWriteToReadOnlyPropertyFails('key', 42)

    # The POSIX spec says "msgget() shall return a non-negative integer", but OS X sometimes
    # returns a negative number like -1765146624. My guess is that they're using a UINT somewhere
    # which exceeds INT_MAX and hence looks negative, or they just don't care about the spec.
    # msgget() ref: http://pubs.opengroup.org/onlinepubs/009695399/functions/msgget.html
    @unittest.skipIf(sys.platform.startswith('darwin'),
                     'OS X message queues sometimes return negative ids')
    def test_property_id(self):
        """exercise MessageQueue.id"""
        self.assertGreaterEqual(self.mq.id, 0)
        self.assertWriteToReadOnlyPropertyFails('id', 42)

    def test_property_id_weak_for_darwin(self):
        """exercise MessageQueue.id with the Darwin-failing test removed"""
        self.assertIsInstance(self.mq.id, numbers.Integral)
        self.assertWriteToReadOnlyPropertyFails('id', 42)

    def test_attribute_max_size(self):
        """exercise MessageQueue.max_size"""
        self.assertGreaterEqual(self.mq.max_size, 0)
        # writing to max_size should not fail, and I test that here. However, as documented,
        # setting it is no guarantee that the OS will respect it. Caveat emptor.
        self.mq.max_size = 2048

    def test_property_last_send_time(self):
        """exercise MessageQueue.last_send_time"""
        self.assertEqual(self.mq.last_send_time, 0)
        # I can't record exactly when this send() happens, but as long as it is within 5 seconds
        # of the assertion happening, this test will pass.
        self.mq.send('x')
        self.assertLess(self.mq.last_send_time - time.time(), 5)
        self.assertWriteToReadOnlyPropertyFails('last_send_time', 42)

    def test_property_last_receive_time(self):
        """exercise MessageQueue.last_receive_time"""
        self.assertEqual(self.mq.last_receive_time, 0)
        self.mq.send('x')
        self.mq.receive()
        # I can't record exactly when this send() happens, but as long as it is within 5 seconds
        # of the assertion happening, this test will pass.
        self.assertLess(self.mq.last_receive_time - time.time(), 5)
        self.assertWriteToReadOnlyPropertyFails('last_receive_time', 42)

    def test_property_last_change_time(self):
        """exercise MessageQueue.last_change_time"""
        # Note that last_change_time doesn't start out as 0 (unlike e.g. last_receive_time), so
        # I don't test that here.
        original_last_change_time = self.mq.last_change_time
        sleep_past_granularity()
        # This might seem like a no-op, but setting the UID to any value triggers a call that
        # should set last_change_time.
        self.mq.uid = self.mq.uid
        self.assertLess(self.mq.last_change_time - time.time(), 5)
        # Ensure the time actually changed.
        self.assertNotEqual(self.mq.last_change_time, original_last_change_time)
        self.assertWriteToReadOnlyPropertyFails('last_change_time', 42)

    def test_property_last_send_pid(self):
        """exercise MessageQueue.last_send_pid"""
        self.assertEqual(self.mq.last_send_pid, 0)
        self.mq.send('x')
        self.assertEqual(self.mq.last_send_pid, os.getpid())
        self.assertWriteToReadOnlyPropertyFails('last_send_pid', 42)

    def test_property_last_receive_pid(self):
        """exercise MessageQueue.last_receive_pid"""
        self.assertEqual(self.mq.last_receive_pid, 0)
        self.mq.send('x')
        self.mq.receive()
        self.assertEqual(self.mq.last_receive_pid, os.getpid())
        self.assertWriteToReadOnlyPropertyFails('last_receive_pid', 42)

    def test_property_current_messages(self):
        """exercise MessageQueue.current_messages"""
        self.assertEqual(self.mq.current_messages, 0)
        self.mq.send('x')
        self.mq.send('x')
        self.mq.send('x')
        self.assertEqual(self.mq.current_messages, 3)
        self.mq.receive()
        self.assertEqual(self.mq.current_messages, 2)
        self.mq.receive()
        self.assertEqual(self.mq.current_messages, 1)
        self.mq.receive()
        self.assertEqual(self.mq.current_messages, 0)
        self.assertWriteToReadOnlyPropertyFails('current_messages', 42)

    def test_attribute_uid(self):
        """exercise MessageQueue.uid"""
        self.assertEqual(self.mq.uid, os.geteuid())

    def test_attribute_gid(self):
        """exercise MessageQueue.gid"""
        self.assertEqual(self.mq.gid, os.getgid())

    def test_attribute_cuid(self):
        """exercise MessageQueue.cuid"""
        self.assertEqual(self.mq.cuid, os.geteuid())
        self.assertWriteToReadOnlyPropertyFails('cuid', 42)

    def test_attribute_cgid(self):
        """exercise MessageQueue.cgid"""
        self.assertEqual(self.mq.cgid, os.getgid())
        self.assertWriteToReadOnlyPropertyFails('cgid', 42)


if __name__ == '__main__':
    unittest.main()