Index: mojo/python/tests/test_core.py |
diff --git a/mojo/python/tests/test_core.py b/mojo/python/tests/test_core.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..be0253b01a3295af11b740126f8cf4bd8a373e64 |
--- /dev/null |
+++ b/mojo/python/tests/test_core.py |
@@ -0,0 +1,314 @@ |
+# 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 sys |
+import time |
+import unittest |
+ |
+# Setup sys path |
+for path in sys.argv[1:]: |
+ sys.path.append(path) |
+ |
+# pylint: disable=F0401 |
+from mojo.embedder import init as init_embedder |
+from mojo import system |
+ |
+DATA_SIZE = 1024 |
+ |
+ |
+def get_random_buffer(size): |
+ random.seed(size) |
+ return bytearray(''.join(chr(random.randint(0, 255)) for i in xrange(size))) |
+ |
+ |
+class BaseMojoTest(unittest.TestCase): |
+ |
+ def setUp(self): |
+ init_embedder() |
+ |
+ |
+class CoreTest(BaseMojoTest): |
+ |
+ def test_results(self): |
+ self.assertEquals(system.RESULT_OK, 0) |
+ self.assertLess(system.RESULT_CANCELLED, 0) |
+ self.assertLess(system.RESULT_UNKNOWN, 0) |
+ self.assertLess(system.RESULT_INVALID_ARGUMENT, 0) |
+ self.assertLess(system.RESULT_DEADLINE_EXCEEDED, 0) |
+ self.assertLess(system.RESULT_NOT_FOUND, 0) |
+ self.assertLess(system.RESULT_ALREADY_EXISTS, 0) |
+ self.assertLess(system.RESULT_PERMISSION_DENIED, 0) |
+ self.assertLess(system.RESULT_RESOURCE_EXHAUSTED, 0) |
+ self.assertLess(system.RESULT_FAILED_PRECONDITION, 0) |
+ self.assertLess(system.RESULT_ABORTED, 0) |
+ self.assertLess(system.RESULT_OUT_OF_RANGE, 0) |
+ self.assertLess(system.RESULT_UNIMPLEMENTED, 0) |
+ self.assertLess(system.RESULT_INTERNAL, 0) |
+ self.assertLess(system.RESULT_UNAVAILABLE, 0) |
+ self.assertLess(system.RESULT_DATA_LOSS, 0) |
+ self.assertLess(system.RESULT_BUSY, 0) |
+ self.assertLess(system.RESULT_SHOULD_WAIT, 0) |
+ |
+ def test_constants(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.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.MAP_BUFFER_FLAG_NONE, 0) |
+ |
+ def test_get_time_ticks_now(self): |
+ pt1 = time.time() |
+ v1 = system.get_time_ticks_now() |
+ time.sleep(1e-3) |
+ v2 = system.get_time_ticks_now() |
+ pt2 = time.time() |
+ self.assertGreater(v1, 0) |
+ self.assertGreater(v2, v1 + 1000) |
+ self.assertGreater(1e6 * (pt2 - pt1), v2 - v1) |
+ |
+ def _test_handles_creation(self, *args): |
+ for handle in args: |
+ self.assertTrue(handle.is_valid()) |
+ handle.close() |
+ self.assertFalse(handle.is_valid()) |
+ |
+ def _test_message_handle_creation(self, handles): |
+ self._test_handles_creation(handles.handle0, handles.handle1) |
+ |
+ def test_create_message_pipe(self): |
+ self._test_message_handle_creation(system.MessagePipe()) |
+ |
+ def test_create_message_pipe_with_none_options(self): |
+ self._test_message_handle_creation(system.MessagePipe(None)) |
+ |
+ def test_create_message_pipe_with_options(self): |
+ self._test_message_handle_creation( |
+ system.MessagePipe(system.CreateMessagePipeOptions())) |
+ |
+ def test_wait_over_message_pipe(self): |
+ handles = system.MessagePipe() |
+ handle = handles.handle0 |
+ |
+ self.assertEquals(system.RESULT_OK, handle.wait( |
+ system.HANDLE_SIGNAL_WRITABLE, system.DEADLINE_INDEFINITE)) |
+ self.assertEquals(system.RESULT_DEADLINE_EXCEEDED, |
+ handle.wait(system.HANDLE_SIGNAL_READABLE, 0)) |
+ |
+ handles.handle1.write_message() |
+ |
+ self.assertEquals( |
+ system.RESULT_OK, |
+ handle.wait( |
+ system.HANDLE_SIGNAL_READABLE, |
+ system.DEADLINE_INDEFINITE)) |
+ |
+ def test_wait_over_many_message_pipe(self): |
+ handles = system.MessagePipe() |
+ handle0 = handles.handle0 |
+ handle1 = handles.handle1 |
+ |
+ self.assertEquals( |
+ 0, |
+ system.wait_many( |
+ [(handle0, system.HANDLE_SIGNAL_WRITABLE), |
+ (handle1, system.HANDLE_SIGNAL_WRITABLE)], |
+ system.DEADLINE_INDEFINITE)) |
+ self.assertEquals( |
+ system.RESULT_DEADLINE_EXCEEDED, |
+ system.wait_many( |
+ [(handle0, system.HANDLE_SIGNAL_READABLE), |
+ (handle1, system.HANDLE_SIGNAL_READABLE)], 0)) |
+ |
+ handle0.write_message() |
+ |
+ self.assertEquals( |
+ 1, |
+ system.wait_many( |
+ [(handle0, system.HANDLE_SIGNAL_READABLE), |
+ (handle1, system.HANDLE_SIGNAL_READABLE)], |
+ system.DEADLINE_INDEFINITE)) |
+ |
+ def test_send_bytes_over_message_pipe(self): |
+ handles = system.MessagePipe() |
+ data = get_random_buffer(DATA_SIZE) |
+ handles.handle0.write_message(data) |
+ (res, buffers, next_message) = handles.handle1.read_message() |
+ 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.read_message(result) |
+ self.assertEquals(system.RESULT_OK, res) |
+ self.assertEquals(None, next_message) |
+ self.assertEquals((data, []), buffers) |
+ |
+ def test_send_empty_data_over_message_pipe(self): |
+ handles = system.MessagePipe() |
+ handles.handle0.write_message(None) |
+ (res, buffers, next_message) = handles.handle1.read_message() |
+ |
+ self.assertEquals(system.RESULT_OK, res) |
+ self.assertEquals(None, next_message) |
+ self.assertEquals((None, []), buffers) |
+ |
+ def test_send_handle_over_message_pipe(self): |
+ handles = system.MessagePipe() |
+ handles_to_send = system.MessagePipe() |
+ handles.handle0.write_message(handles=[handles_to_send.handle0, |
+ handles_to_send.handle1]) |
+ (res, buffers, next_message) = handles.handle1.read_message( |
+ max_number_of_handles=2) |
+ |
+ self.assertFalse(handles_to_send.handle0.is_valid()) |
+ self.assertFalse(handles_to_send.handle1.is_valid()) |
+ 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.is_valid()) |
+ (res, buffers, next_message) = handle.read_message() |
+ self.assertEquals(system.RESULT_SHOULD_WAIT, res) |
+ |
+ for handle in handles: |
+ handle.write_message() |
+ |
+ for handle in handles: |
+ (res, buffers, next_message) = handle.read_message() |
+ self.assertEquals(system.RESULT_OK, res) |
+ |
+ def _test_data_handle_creation(self, handles): |
+ self._test_handles_creation( |
+ handles.producer_handle, handles.consumer_handle) |
+ |
+ def test_create_data_pipe(self): |
+ self._test_data_handle_creation(system.DataPipe()) |
+ |
+ def test_create_data_pipe_with_none_options(self): |
+ self._test_data_handle_creation(system.DataPipe(None)) |
+ |
+ def test_create_data_pipe_with_default_options(self): |
+ self._test_data_handle_creation( |
+ system.DataPipe(system.CreateDataPipeOptions())) |
+ |
+ def test_create_data_pipe_with_discard_flag(self): |
+ options = system.CreateDataPipeOptions() |
+ options.flags = system.CreateDataPipeOptions.FLAG_MAY_DISCARD |
+ self._test_data_handle_creation(system.DataPipe(options)) |
+ |
+ def test_create_data_pipe_with_element_size(self): |
+ options = system.CreateDataPipeOptions() |
+ options.element_num_bytes = 5 |
+ self._test_data_handle_creation(system.DataPipe(options)) |
+ |
+ def test_create_data_pipe_with_capacity(self): |
+ options = system.CreateDataPipeOptions() |
+ options.element_capacity_num_bytes = DATA_SIZE |
+ self._test_data_handle_creation(system.DataPipe(options)) |
+ |
+ def test_create_data_pipe_with_incorrect_parameters(self): |
+ options = system.CreateDataPipeOptions() |
+ options.element_num_bytes = 5 |
+ options.capacity_num_bytes = DATA_SIZE |
+ with self.assertRaises(system.MojoException) as cm: |
+ self._test_data_handle_creation(system.DataPipe(options)) |
+ self.assertEquals(system.RESULT_INVALID_ARGUMENT, cm.exception.mojo_result) |
+ |
+ def test_send_empty_data_over_data_pipe(self): |
+ pipes = system.DataPipe() |
+ self.assertEquals((system.RESULT_OK, 0), pipes.producer_handle.write_data()) |
+ self.assertEquals( |
+ (system.RESULT_OK, None), pipes.consumer_handle.read_data()) |
+ |
+ def test_send_data_over_data_pipe(self): |
+ pipes = system.DataPipe() |
+ data = get_random_buffer(DATA_SIZE) |
+ self.assertEquals((system.RESULT_OK, DATA_SIZE), |
+ pipes.producer_handle.write_data(data)) |
+ self.assertEquals((system.RESULT_OK, data), |
+ pipes.consumer_handle.read_data(bytearray(DATA_SIZE))) |
+ |
+ def test_two_phase_write_on_data_pipe(self): |
+ pipes = system.DataPipe() |
+ (res, buf) = pipes.producer_handle.begin_write_data(DATA_SIZE) |
+ self.assertEquals(system.RESULT_OK, res) |
+ self.assertGreaterEqual(len(buf.buffer), DATA_SIZE) |
+ data = get_random_buffer(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.read_data(bytearray(DATA_SIZE))) |
+ |
+ def test_two_phase_read_on_data_pipe(self): |
+ pipes = system.DataPipe() |
+ data = get_random_buffer(DATA_SIZE) |
+ self.assertEquals((system.RESULT_OK, DATA_SIZE), |
+ pipes.producer_handle.write_data(data)) |
+ (res, buf) = pipes.consumer_handle.begin_read_data() |
+ 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 test_create_shared_buffer(self): |
+ self._test_handles_creation(system.create_shared_buffer(DATA_SIZE)) |
+ |
+ def test_create_shared_buffer_with_none_options(self): |
+ self._test_handles_creation(system.create_shared_buffer(DATA_SIZE, None)) |
+ |
+ def test_create_shared_buffer_with_default_options(self): |
+ self._test_handles_creation( |
+ system.create_shared_buffer( |
+ DATA_SIZE, |
+ system.CreateSharedBufferOptions())) |
+ |
+ def test_duplicate_shared_buffer(self): |
+ handle = system.create_shared_buffer(DATA_SIZE) |
+ self._test_handles_creation(handle.duplicate()) |
+ |
+ def test_duplicate_shared_buffer_with_none_options(self): |
+ handle = system.create_shared_buffer(DATA_SIZE) |
+ self._test_handles_creation(handle.duplicate(None)) |
+ |
+ def test_duplicate_shared_buffer_with_default_options(self): |
+ handle = system.create_shared_buffer(DATA_SIZE) |
+ self._test_handles_creation( |
+ handle.duplicate(system.DuplicateSharedBufferOptions())) |
+ |
+ def test_send_bytes_over_shared_buffer(self): |
+ handle = system.create_shared_buffer(DATA_SIZE) |
+ duplicated = handle.duplicate() |
+ data = get_random_buffer(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) |
+ |
+ |
+if __name__ == '__main__': |
+ suite = unittest.TestLoader().loadTestsFromTestCase(CoreTest) |
+ test_results = unittest.TextTestRunner(verbosity=0).run(suite) |
+ if not test_results.wasSuccessful(): |
+ sys.exit(1) |
+ sys.exit(0) |