mirror of
https://github.com/flutter/flutter.git
synced 2026-02-20 02:29:02 +08:00
328 lines
13 KiB
Python
328 lines
13 KiB
Python
# Copyright 2014 The Chromium Authors. All rights reserved.
|
|
# Use of this source code is governed by a BSD-style license that can be
|
|
# found in the LICENSE file.
|
|
|
|
import random
|
|
import time
|
|
|
|
import mojo_unittest
|
|
|
|
# pylint: disable=E0611
|
|
import mojo_system as system
|
|
|
|
DATA_SIZE = 1024
|
|
|
|
|
|
def _GetRandomBuffer(size):
|
|
random.seed(size)
|
|
return bytearray(''.join(chr(random.randint(0, 255)) for i in xrange(size)))
|
|
|
|
|
|
class CoreTest(mojo_unittest.MojoTestCase):
|
|
|
|
HANDLE_SIGNAL_READWRITABLE = (system.HANDLE_SIGNAL_READABLE |
|
|
system.HANDLE_SIGNAL_WRITABLE)
|
|
HANDLE_SIGNAL_ALL = (system.HANDLE_SIGNAL_READABLE |
|
|
system.HANDLE_SIGNAL_WRITABLE | system.HANDLE_SIGNAL_PEER_CLOSED)
|
|
|
|
def testResults(self):
|
|
self.assertEquals(system.RESULT_OK, 0)
|
|
self.assertGreater(system.RESULT_CANCELLED, 0)
|
|
self.assertGreater(system.RESULT_UNKNOWN, 0)
|
|
self.assertGreater(system.RESULT_INVALID_ARGUMENT, 0)
|
|
self.assertGreater(system.RESULT_DEADLINE_EXCEEDED, 0)
|
|
self.assertGreater(system.RESULT_NOT_FOUND, 0)
|
|
self.assertGreater(system.RESULT_ALREADY_EXISTS, 0)
|
|
self.assertGreater(system.RESULT_PERMISSION_DENIED, 0)
|
|
self.assertGreater(system.RESULT_RESOURCE_EXHAUSTED, 0)
|
|
self.assertGreater(system.RESULT_FAILED_PRECONDITION, 0)
|
|
self.assertGreater(system.RESULT_ABORTED, 0)
|
|
self.assertGreater(system.RESULT_OUT_OF_RANGE, 0)
|
|
self.assertGreater(system.RESULT_UNIMPLEMENTED, 0)
|
|
self.assertGreater(system.RESULT_INTERNAL, 0)
|
|
self.assertGreater(system.RESULT_UNAVAILABLE, 0)
|
|
self.assertGreater(system.RESULT_DATA_LOSS, 0)
|
|
self.assertGreater(system.RESULT_BUSY, 0)
|
|
self.assertGreater(system.RESULT_SHOULD_WAIT, 0)
|
|
|
|
def testConstants(self):
|
|
self.assertGreaterEqual(system.DEADLINE_INDEFINITE, 0)
|
|
self.assertGreaterEqual(system.HANDLE_SIGNAL_NONE, 0)
|
|
self.assertGreaterEqual(system.HANDLE_SIGNAL_READABLE, 0)
|
|
self.assertGreaterEqual(system.HANDLE_SIGNAL_WRITABLE, 0)
|
|
self.assertGreaterEqual(system.HANDLE_SIGNAL_PEER_CLOSED, 0)
|
|
self.assertGreaterEqual(system.WRITE_MESSAGE_FLAG_NONE, 0)
|
|
self.assertGreaterEqual(system.READ_MESSAGE_FLAG_NONE, 0)
|
|
self.assertGreaterEqual(system.READ_MESSAGE_FLAG_MAY_DISCARD, 0)
|
|
self.assertGreaterEqual(system.WRITE_DATA_FLAG_NONE, 0)
|
|
self.assertGreaterEqual(system.WRITE_DATA_FLAG_ALL_OR_NONE, 0)
|
|
self.assertGreaterEqual(system.READ_DATA_FLAG_NONE, 0)
|
|
self.assertGreaterEqual(system.READ_DATA_FLAG_ALL_OR_NONE, 0)
|
|
self.assertGreaterEqual(system.READ_DATA_FLAG_DISCARD, 0)
|
|
self.assertGreaterEqual(system.READ_DATA_FLAG_QUERY, 0)
|
|
self.assertGreaterEqual(system.READ_DATA_FLAG_PEEK, 0)
|
|
self.assertGreaterEqual(system.MAP_BUFFER_FLAG_NONE, 0)
|
|
|
|
def testGetTimeTicksNow(self):
|
|
v1 = system.GetTimeTicksNow()
|
|
time.sleep(1e-3)
|
|
v2 = system.GetTimeTicksNow()
|
|
self.assertGreater(v1, 0)
|
|
self.assertGreater(v2, v1 + 1e2)
|
|
self.assertLess(v2, v1 + 1e5)
|
|
|
|
def _testHandlesCreation(self, *args):
|
|
for handle in args:
|
|
self.assertTrue(handle.IsValid())
|
|
handle.Close()
|
|
self.assertFalse(handle.IsValid())
|
|
|
|
def _TestMessageHandleCreation(self, handles):
|
|
self._testHandlesCreation(handles.handle0, handles.handle1)
|
|
|
|
def testCreateMessagePipe(self):
|
|
self._TestMessageHandleCreation(system.MessagePipe())
|
|
|
|
def testCreateMessagePipeWithNoneOptions(self):
|
|
self._TestMessageHandleCreation(system.MessagePipe(None))
|
|
|
|
def testCreateMessagePipeWithOptions(self):
|
|
self._TestMessageHandleCreation(
|
|
system.MessagePipe(system.CreateMessagePipeOptions()))
|
|
|
|
def testWaitOverMessagePipe(self):
|
|
handles = system.MessagePipe()
|
|
handle = handles.handle0
|
|
|
|
(res, states) = handle.Wait(
|
|
system.HANDLE_SIGNAL_WRITABLE, system.DEADLINE_INDEFINITE)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1])
|
|
|
|
(res, states) = handle.Wait(system.HANDLE_SIGNAL_READABLE, 0)
|
|
self.assertEquals(system.RESULT_DEADLINE_EXCEEDED, res)
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1])
|
|
|
|
handles.handle1.WriteMessage()
|
|
|
|
(res, states) = handle.Wait(
|
|
system.HANDLE_SIGNAL_READABLE,
|
|
system.DEADLINE_INDEFINITE)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_READWRITABLE, states[0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1])
|
|
|
|
def testWaitInvalidArgument(self):
|
|
(res, index, states) = system.WaitMany(
|
|
[(system.Handle(0), system.HANDLE_SIGNAL_WRITABLE),
|
|
(system.Handle(0), system.HANDLE_SIGNAL_WRITABLE)],
|
|
system.DEADLINE_INDEFINITE)
|
|
self.assertEquals(system.RESULT_INVALID_ARGUMENT, res)
|
|
self.assertEquals(0, index)
|
|
self.assertEquals(states, None)
|
|
|
|
def testWaitOverManyMessagePipe(self):
|
|
handles = system.MessagePipe()
|
|
handle0 = handles.handle0
|
|
handle1 = handles.handle1
|
|
|
|
(res, index, states) = system.WaitMany(
|
|
[(handle0, system.HANDLE_SIGNAL_WRITABLE),
|
|
(handle1, system.HANDLE_SIGNAL_WRITABLE)],
|
|
system.DEADLINE_INDEFINITE)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(0, index)
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[0][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[0][1])
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[1][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1][1])
|
|
|
|
(res, index, states) = system.WaitMany(
|
|
[(handle0, system.HANDLE_SIGNAL_READABLE),
|
|
(handle1, system.HANDLE_SIGNAL_READABLE)], 0)
|
|
self.assertEquals(system.RESULT_DEADLINE_EXCEEDED, res)
|
|
self.assertEquals(None, index)
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[0][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[0][1])
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[1][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1][1])
|
|
|
|
handle0.WriteMessage()
|
|
|
|
(res, index, states) = system.WaitMany(
|
|
[(handle0, system.HANDLE_SIGNAL_READABLE),
|
|
(handle1, system.HANDLE_SIGNAL_READABLE)],
|
|
system.DEADLINE_INDEFINITE)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(1, index)
|
|
self.assertEquals(system.HANDLE_SIGNAL_WRITABLE, states[0][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[0][1])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_READWRITABLE, states[1][0])
|
|
self.assertEquals(CoreTest.HANDLE_SIGNAL_ALL, states[1][1])
|
|
|
|
def testSendBytesOverMessagePipe(self):
|
|
handles = system.MessagePipe()
|
|
data = _GetRandomBuffer(DATA_SIZE)
|
|
handles.handle0.WriteMessage(data)
|
|
(res, buffers, next_message) = handles.handle1.ReadMessage()
|
|
self.assertEquals(system.RESULT_RESOURCE_EXHAUSTED, res)
|
|
self.assertEquals(None, buffers)
|
|
self.assertEquals((DATA_SIZE, 0), next_message)
|
|
result = bytearray(DATA_SIZE)
|
|
(res, buffers, next_message) = handles.handle1.ReadMessage(result)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(None, next_message)
|
|
self.assertEquals((data, []), buffers)
|
|
|
|
def testSendEmptyDataOverMessagePipe(self):
|
|
handles = system.MessagePipe()
|
|
handles.handle0.WriteMessage(None)
|
|
(res, buffers, next_message) = handles.handle1.ReadMessage()
|
|
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(None, next_message)
|
|
self.assertEquals((None, []), buffers)
|
|
|
|
def testSendHandleOverMessagePipe(self):
|
|
handles = system.MessagePipe()
|
|
handles_to_send = system.MessagePipe()
|
|
handles.handle0.WriteMessage(handles=[handles_to_send.handle0,
|
|
handles_to_send.handle1])
|
|
(res, buffers, next_message) = handles.handle1.ReadMessage(
|
|
max_number_of_handles=2)
|
|
|
|
self.assertFalse(handles_to_send.handle0.IsValid())
|
|
self.assertFalse(handles_to_send.handle1.IsValid())
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(None, next_message)
|
|
self.assertEquals(None, buffers[0])
|
|
self.assertEquals(2, len(buffers[1]))
|
|
|
|
handles = buffers[1]
|
|
for handle in handles:
|
|
self.assertTrue(handle.IsValid())
|
|
(res, buffers, next_message) = handle.ReadMessage()
|
|
self.assertEquals(system.RESULT_SHOULD_WAIT, res)
|
|
|
|
for handle in handles:
|
|
handle.WriteMessage()
|
|
|
|
for handle in handles:
|
|
(res, buffers, next_message) = handle.ReadMessage()
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
|
|
def _TestDataHandleCreation(self, handles):
|
|
self._testHandlesCreation(
|
|
handles.producer_handle, handles.consumer_handle)
|
|
|
|
def testCreateDataPipe(self):
|
|
self._TestDataHandleCreation(system.DataPipe())
|
|
|
|
def testCreateDataPipeWithNoneOptions(self):
|
|
self._TestDataHandleCreation(system.DataPipe(None))
|
|
|
|
def testCreateDataPipeWithDefaultOptions(self):
|
|
self._TestDataHandleCreation(
|
|
system.DataPipe(system.CreateDataPipeOptions()))
|
|
|
|
def testCreateDataPipeWithElementSize(self):
|
|
options = system.CreateDataPipeOptions()
|
|
options.element_num_bytes = 5
|
|
self._TestDataHandleCreation(system.DataPipe(options))
|
|
|
|
def testCreateDataPipeWithCapacity(self):
|
|
options = system.CreateDataPipeOptions()
|
|
options.element_capacity_num_bytes = DATA_SIZE
|
|
self._TestDataHandleCreation(system.DataPipe(options))
|
|
|
|
def testCreateDataPipeWithIncorrectParameters(self):
|
|
options = system.CreateDataPipeOptions()
|
|
options.element_num_bytes = 5
|
|
options.capacity_num_bytes = DATA_SIZE
|
|
with self.assertRaises(system.MojoException) as cm:
|
|
self._TestDataHandleCreation(system.DataPipe(options))
|
|
self.assertEquals(system.RESULT_INVALID_ARGUMENT, cm.exception.mojo_result)
|
|
|
|
def testSendEmptyDataOverDataPipe(self):
|
|
pipes = system.DataPipe()
|
|
self.assertEquals((system.RESULT_OK, 0), pipes.producer_handle.WriteData())
|
|
self.assertEquals(
|
|
(system.RESULT_OK, None), pipes.consumer_handle.ReadData())
|
|
|
|
def testSendDataOverDataPipe(self):
|
|
pipes = system.DataPipe()
|
|
data = _GetRandomBuffer(DATA_SIZE)
|
|
self.assertEquals((system.RESULT_OK, DATA_SIZE),
|
|
pipes.producer_handle.WriteData(data))
|
|
self.assertEquals((system.RESULT_OK, data),
|
|
pipes.consumer_handle.ReadData(
|
|
bytearray(DATA_SIZE), system.READ_DATA_FLAG_PEEK))
|
|
self.assertEquals((system.RESULT_OK, data),
|
|
pipes.consumer_handle.ReadData(bytearray(DATA_SIZE)))
|
|
|
|
def testTwoPhaseWriteOnDataPipe(self):
|
|
pipes = system.DataPipe()
|
|
(res, buf) = pipes.producer_handle.BeginWriteData(DATA_SIZE)
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertGreaterEqual(len(buf.buffer), DATA_SIZE)
|
|
data = _GetRandomBuffer(DATA_SIZE)
|
|
buf.buffer[0:DATA_SIZE] = data
|
|
self.assertEquals(system.RESULT_OK, buf.End(DATA_SIZE))
|
|
self.assertEquals((system.RESULT_OK, data),
|
|
pipes.consumer_handle.ReadData(bytearray(DATA_SIZE)))
|
|
|
|
def testTwoPhaseReadOnDataPipe(self):
|
|
pipes = system.DataPipe()
|
|
data = _GetRandomBuffer(DATA_SIZE)
|
|
self.assertEquals((system.RESULT_OK, DATA_SIZE),
|
|
pipes.producer_handle.WriteData(data))
|
|
(res, buf) = pipes.consumer_handle.BeginReadData()
|
|
self.assertEquals(system.RESULT_OK, res)
|
|
self.assertEquals(DATA_SIZE, len(buf.buffer))
|
|
self.assertEquals(data, buf.buffer)
|
|
self.assertEquals(system.RESULT_OK, buf.End(DATA_SIZE))
|
|
|
|
def testCreateSharedBuffer(self):
|
|
self._testHandlesCreation(system.CreateSharedBuffer(DATA_SIZE))
|
|
|
|
def testCreateSharedBufferWithNoneOptions(self):
|
|
self._testHandlesCreation(system.CreateSharedBuffer(DATA_SIZE, None))
|
|
|
|
def testCreateSharedBufferWithDefaultOptions(self):
|
|
self._testHandlesCreation(
|
|
system.CreateSharedBuffer(
|
|
DATA_SIZE,
|
|
system.CreateSharedBufferOptions()))
|
|
|
|
def testDuplicateSharedBuffer(self):
|
|
handle = system.CreateSharedBuffer(DATA_SIZE)
|
|
self._testHandlesCreation(handle.Duplicate())
|
|
|
|
def testDuplicateSharedBufferWithNoneOptions(self):
|
|
handle = system.CreateSharedBuffer(DATA_SIZE)
|
|
self._testHandlesCreation(handle.Duplicate(None))
|
|
|
|
def testDuplicateSharedBufferWithDefaultOptions(self):
|
|
handle = system.CreateSharedBuffer(DATA_SIZE)
|
|
self._testHandlesCreation(
|
|
handle.Duplicate(system.DuplicateSharedBufferOptions()))
|
|
|
|
def testSendBytesOverSharedBuffer(self):
|
|
handle = system.CreateSharedBuffer(DATA_SIZE)
|
|
duplicated = handle.Duplicate()
|
|
data = _GetRandomBuffer(DATA_SIZE)
|
|
(res1, buf1) = handle.Map(0, DATA_SIZE)
|
|
(res2, buf2) = duplicated.Map(0, DATA_SIZE)
|
|
self.assertEquals(system.RESULT_OK, res1)
|
|
self.assertEquals(system.RESULT_OK, res2)
|
|
self.assertEquals(DATA_SIZE, len(buf1.buffer))
|
|
self.assertEquals(DATA_SIZE, len(buf2.buffer))
|
|
self.assertEquals(buf1.buffer, buf2.buffer)
|
|
|
|
buf1.buffer[:] = data
|
|
self.assertEquals(data, buf1.buffer)
|
|
self.assertEquals(data, buf2.buffer)
|
|
self.assertEquals(buf1.buffer, buf2.buffer)
|