Home | History | Annotate | Download | only in tests 
      1 # Copyright 2014 The Chromium Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 import random
      6 import sys
      7 import time
      8 import unittest
      9 
     10 # pylint: disable=F0401
     11 import mojo.embedder
     12 from mojo import system
     13 
     14 DATA_SIZE = 1024
     15 
     16 
     17 def _GetRandomBuffer(size):
     18   random.seed(size)
     19   return bytearray(''.join(chr(random.randint(0, 255)) for i in xrange(size)))
     20 
     21 
     22 class BaseMojoTest(unittest.TestCase):
     23 
     24   def setUp(self):
     25     mojo.embedder.Init()
     26 
     27 
     28 class CoreTest(BaseMojoTest):
     29 
     30   def testResults(self):
     31     self.assertEquals(system.RESULT_OK, 0)
     32     self.assertLess(system.RESULT_CANCELLED, 0)
     33     self.assertLess(system.RESULT_UNKNOWN, 0)
     34     self.assertLess(system.RESULT_INVALID_ARGUMENT, 0)
     35     self.assertLess(system.RESULT_DEADLINE_EXCEEDED, 0)
     36     self.assertLess(system.RESULT_NOT_FOUND, 0)
     37     self.assertLess(system.RESULT_ALREADY_EXISTS, 0)
     38     self.assertLess(system.RESULT_PERMISSION_DENIED, 0)
     39     self.assertLess(system.RESULT_RESOURCE_EXHAUSTED, 0)
     40     self.assertLess(system.RESULT_FAILED_PRECONDITION, 0)
     41     self.assertLess(system.RESULT_ABORTED, 0)
     42     self.assertLess(system.RESULT_OUT_OF_RANGE, 0)
     43     self.assertLess(system.RESULT_UNIMPLEMENTED, 0)
     44     self.assertLess(system.RESULT_INTERNAL, 0)
     45     self.assertLess(system.RESULT_UNAVAILABLE, 0)
     46     self.assertLess(system.RESULT_DATA_LOSS, 0)
     47     self.assertLess(system.RESULT_BUSY, 0)
     48     self.assertLess(system.RESULT_SHOULD_WAIT, 0)
     49 
     50   def testConstants(self):
     51     self.assertGreaterEqual(system.DEADLINE_INDEFINITE, 0)
     52     self.assertGreaterEqual(system.HANDLE_SIGNAL_NONE, 0)
     53     self.assertGreaterEqual(system.HANDLE_SIGNAL_READABLE, 0)
     54     self.assertGreaterEqual(system.HANDLE_SIGNAL_WRITABLE, 0)
     55     self.assertGreaterEqual(system.WRITE_MESSAGE_FLAG_NONE, 0)
     56     self.assertGreaterEqual(system.READ_MESSAGE_FLAG_NONE, 0)
     57     self.assertGreaterEqual(system.READ_MESSAGE_FLAG_MAY_DISCARD, 0)
     58     self.assertGreaterEqual(system.WRITE_DATA_FLAG_NONE, 0)
     59     self.assertGreaterEqual(system.WRITE_DATA_FLAG_ALL_OR_NONE, 0)
     60     self.assertGreaterEqual(system.READ_DATA_FLAG_NONE, 0)
     61     self.assertGreaterEqual(system.READ_DATA_FLAG_ALL_OR_NONE, 0)
     62     self.assertGreaterEqual(system.READ_DATA_FLAG_DISCARD, 0)
     63     self.assertGreaterEqual(system.READ_DATA_FLAG_QUERY, 0)
     64     self.assertGreaterEqual(system.MAP_BUFFER_FLAG_NONE, 0)
     65 
     66   def testGetTimeTicksNow(self):
     67     pt1 = time.time()
     68     v1 = system.GetTimeTicksNow()
     69     time.sleep(1e-3)
     70     v2 = system.GetTimeTicksNow()
     71     pt2 = time.time()
     72     self.assertGreater(v1, 0)
     73     self.assertGreater(v2, v1 + 1000)
     74     self.assertGreater(1e6 * (pt2 - pt1), v2 - v1)
     75 
     76   def _testHandlesCreation(self, *args):
     77     for handle in args:
     78       self.assertTrue(handle.IsValid())
     79       handle.Close()
     80       self.assertFalse(handle.IsValid())
     81 
     82   def _TestMessageHandleCreation(self, handles):
     83     self._testHandlesCreation(handles.handle0, handles.handle1)
     84 
     85   def testCreateMessagePipe(self):
     86     self._TestMessageHandleCreation(system.MessagePipe())
     87 
     88   def testCreateMessagePipeWithNoneOptions(self):
     89     self._TestMessageHandleCreation(system.MessagePipe(None))
     90 
     91   def testCreateMessagePipeWithOptions(self):
     92     self._TestMessageHandleCreation(
     93         system.MessagePipe(system.CreateMessagePipeOptions()))
     94 
     95   def testWaitOverMessagePipe(self):
     96     handles = system.MessagePipe()
     97     handle = handles.handle0
     98 
     99     self.assertEquals(system.RESULT_OK, handle.Wait(
    100         system.HANDLE_SIGNAL_WRITABLE, system.DEADLINE_INDEFINITE))
    101     self.assertEquals(system.RESULT_DEADLINE_EXCEEDED,
    102                       handle.Wait(system.HANDLE_SIGNAL_READABLE, 0))
    103 
    104     handles.handle1.WriteMessage()
    105 
    106     self.assertEquals(
    107         system.RESULT_OK,
    108         handle.Wait(
    109             system.HANDLE_SIGNAL_READABLE,
    110             system.DEADLINE_INDEFINITE))
    111 
    112   def testWaitOverManyMessagePipe(self):
    113     handles = system.MessagePipe()
    114     handle0 = handles.handle0
    115     handle1 = handles.handle1
    116 
    117     self.assertEquals(
    118         0,
    119         system.WaitMany(
    120             [(handle0, system.HANDLE_SIGNAL_WRITABLE),
    121              (handle1, system.HANDLE_SIGNAL_WRITABLE)],
    122             system.DEADLINE_INDEFINITE))
    123     self.assertEquals(
    124         system.RESULT_DEADLINE_EXCEEDED,
    125         system.WaitMany(
    126             [(handle0, system.HANDLE_SIGNAL_READABLE),
    127              (handle1, system.HANDLE_SIGNAL_READABLE)], 0))
    128 
    129     handle0.WriteMessage()
    130 
    131     self.assertEquals(
    132         1,
    133         system.WaitMany(
    134             [(handle0, system.HANDLE_SIGNAL_READABLE),
    135              (handle1, system.HANDLE_SIGNAL_READABLE)],
    136             system.DEADLINE_INDEFINITE))
    137 
    138   def testSendBytesOverMessagePipe(self):
    139     handles = system.MessagePipe()
    140     data = _GetRandomBuffer(DATA_SIZE)
    141     handles.handle0.WriteMessage(data)
    142     (res, buffers, next_message) = handles.handle1.ReadMessage()
    143     self.assertEquals(system.RESULT_RESOURCE_EXHAUSTED, res)
    144     self.assertEquals(None, buffers)
    145     self.assertEquals((DATA_SIZE, 0), next_message)
    146     result = bytearray(DATA_SIZE)
    147     (res, buffers, next_message) = handles.handle1.ReadMessage(result)
    148     self.assertEquals(system.RESULT_OK, res)
    149     self.assertEquals(None, next_message)
    150     self.assertEquals((data, []), buffers)
    151 
    152   def testSendEmptyDataOverMessagePipe(self):
    153     handles = system.MessagePipe()
    154     handles.handle0.WriteMessage(None)
    155     (res, buffers, next_message) = handles.handle1.ReadMessage()
    156 
    157     self.assertEquals(system.RESULT_OK, res)
    158     self.assertEquals(None, next_message)
    159     self.assertEquals((None, []), buffers)
    160 
    161   def testSendHandleOverMessagePipe(self):
    162     handles = system.MessagePipe()
    163     handles_to_send = system.MessagePipe()
    164     handles.handle0.WriteMessage(handles=[handles_to_send.handle0,
    165                                            handles_to_send.handle1])
    166     (res, buffers, next_message) = handles.handle1.ReadMessage(
    167         max_number_of_handles=2)
    168 
    169     self.assertFalse(handles_to_send.handle0.IsValid())
    170     self.assertFalse(handles_to_send.handle1.IsValid())
    171     self.assertEquals(system.RESULT_OK, res)
    172     self.assertEquals(None, next_message)
    173     self.assertEquals(None, buffers[0])
    174     self.assertEquals(2, len(buffers[1]))
    175 
    176     handles = buffers[1]
    177     for handle in handles:
    178       self.assertTrue(handle.IsValid())
    179       (res, buffers, next_message) = handle.ReadMessage()
    180       self.assertEquals(system.RESULT_SHOULD_WAIT, res)
    181 
    182     for handle in handles:
    183       handle.WriteMessage()
    184 
    185     for handle in handles:
    186       (res, buffers, next_message) = handle.ReadMessage()
    187       self.assertEquals(system.RESULT_OK, res)
    188 
    189   def _TestDataHandleCreation(self, handles):
    190     self._testHandlesCreation(
    191         handles.producer_handle, handles.consumer_handle)
    192 
    193   def testCreateDataPipe(self):
    194     self._TestDataHandleCreation(system.DataPipe())
    195 
    196   def testCreateDataPipeWithNoneOptions(self):
    197     self._TestDataHandleCreation(system.DataPipe(None))
    198 
    199   def testCreateDataPipeWithDefaultOptions(self):
    200     self._TestDataHandleCreation(
    201         system.DataPipe(system.CreateDataPipeOptions()))
    202 
    203   def testCreateDataPipeWithDiscardFlag(self):
    204     options = system.CreateDataPipeOptions()
    205     options.flags = system.CreateDataPipeOptions.FLAG_MAY_DISCARD
    206     self._TestDataHandleCreation(system.DataPipe(options))
    207 
    208   def testCreateDataPipeWithElementSize(self):
    209     options = system.CreateDataPipeOptions()
    210     options.element_num_bytes = 5
    211     self._TestDataHandleCreation(system.DataPipe(options))
    212 
    213   def testCreateDataPipeWithCapacity(self):
    214     options = system.CreateDataPipeOptions()
    215     options.element_capacity_num_bytes = DATA_SIZE
    216     self._TestDataHandleCreation(system.DataPipe(options))
    217 
    218   def testCreateDataPipeWithIncorrectParameters(self):
    219     options = system.CreateDataPipeOptions()
    220     options.element_num_bytes = 5
    221     options.capacity_num_bytes = DATA_SIZE
    222     with self.assertRaises(system.MojoException) as cm:
    223       self._TestDataHandleCreation(system.DataPipe(options))
    224     self.assertEquals(system.RESULT_INVALID_ARGUMENT, cm.exception.mojo_result)
    225 
    226   def testSendEmptyDataOverDataPipe(self):
    227     pipes = system.DataPipe()
    228     self.assertEquals((system.RESULT_OK, 0), pipes.producer_handle.WriteData())
    229     self.assertEquals(
    230         (system.RESULT_OK, None), pipes.consumer_handle.ReadData())
    231 
    232   def testSendDataOverDataPipe(self):
    233     pipes = system.DataPipe()
    234     data = _GetRandomBuffer(DATA_SIZE)
    235     self.assertEquals((system.RESULT_OK, DATA_SIZE),
    236                       pipes.producer_handle.WriteData(data))
    237     self.assertEquals((system.RESULT_OK, data),
    238                       pipes.consumer_handle.ReadData(bytearray(DATA_SIZE)))
    239 
    240   def testTwoPhaseWriteOnDataPipe(self):
    241     pipes = system.DataPipe()
    242     (res, buf) = pipes.producer_handle.BeginWriteData(DATA_SIZE)
    243     self.assertEquals(system.RESULT_OK, res)
    244     self.assertGreaterEqual(len(buf.buffer), DATA_SIZE)
    245     data = _GetRandomBuffer(DATA_SIZE)
    246     buf.buffer[0:DATA_SIZE] = data
    247     self.assertEquals(system.RESULT_OK, buf.End(DATA_SIZE))
    248     self.assertEquals((system.RESULT_OK, data),
    249                       pipes.consumer_handle.ReadData(bytearray(DATA_SIZE)))
    250 
    251   def testTwoPhaseReadOnDataPipe(self):
    252     pipes = system.DataPipe()
    253     data = _GetRandomBuffer(DATA_SIZE)
    254     self.assertEquals((system.RESULT_OK, DATA_SIZE),
    255                       pipes.producer_handle.WriteData(data))
    256     (res, buf) = pipes.consumer_handle.BeginReadData()
    257     self.assertEquals(system.RESULT_OK, res)
    258     self.assertEquals(DATA_SIZE, len(buf.buffer))
    259     self.assertEquals(data, buf.buffer)
    260     self.assertEquals(system.RESULT_OK, buf.End(DATA_SIZE))
    261 
    262   def testCreateSharedBuffer(self):
    263     self._testHandlesCreation(system.CreateSharedBuffer(DATA_SIZE))
    264 
    265   def testCreateSharedBufferWithNoneOptions(self):
    266     self._testHandlesCreation(system.CreateSharedBuffer(DATA_SIZE, None))
    267 
    268   def testCreateSharedBufferWithDefaultOptions(self):
    269     self._testHandlesCreation(
    270         system.CreateSharedBuffer(
    271             DATA_SIZE,
    272             system.CreateSharedBufferOptions()))
    273 
    274   def testDuplicateSharedBuffer(self):
    275     handle = system.CreateSharedBuffer(DATA_SIZE)
    276     self._testHandlesCreation(handle.Duplicate())
    277 
    278   def testDuplicateSharedBufferWithNoneOptions(self):
    279     handle = system.CreateSharedBuffer(DATA_SIZE)
    280     self._testHandlesCreation(handle.Duplicate(None))
    281 
    282   def testDuplicateSharedBufferWithDefaultOptions(self):
    283     handle = system.CreateSharedBuffer(DATA_SIZE)
    284     self._testHandlesCreation(
    285         handle.Duplicate(system.DuplicateSharedBufferOptions()))
    286 
    287   def testSendBytesOverSharedBuffer(self):
    288     handle = system.CreateSharedBuffer(DATA_SIZE)
    289     duplicated = handle.Duplicate()
    290     data = _GetRandomBuffer(DATA_SIZE)
    291     (res1, buf1) = handle.Map(0, DATA_SIZE)
    292     (res2, buf2) = duplicated.Map(0, DATA_SIZE)
    293     self.assertEquals(system.RESULT_OK, res1)
    294     self.assertEquals(system.RESULT_OK, res2)
    295     self.assertEquals(DATA_SIZE, len(buf1.buffer))
    296     self.assertEquals(DATA_SIZE, len(buf2.buffer))
    297     self.assertEquals(buf1.buffer, buf2.buffer)
    298 
    299     buf1.buffer[:] = data
    300     self.assertEquals(data, buf1.buffer)
    301     self.assertEquals(data, buf2.buffer)
    302     self.assertEquals(buf1.buffer, buf2.buffer)
    303 
    304 
    305 if __name__ == '__main__':
    306   suite = unittest.TestLoader().loadTestsFromTestCase(CoreTest)
    307   test_results = unittest.TextTestRunner(verbosity=0).run(suite)
    308   if not test_results.wasSuccessful():
    309     sys.exit(1)
    310   sys.exit(0)
    311