Skip to content

GitLab

T
tangled
Commit 94c09453 authored by jamesr's avatar jamesr Committed by Commit bot
Browse files
Options

Remove mojo/python and gyp targets 

These are being developed and tested in the mojo repository. The GN
rules for the python public targets are still in place for developing
mojo applications in the chromium repo.

Review URL: https://codereview.chromium.org/675563002

Cr-Commit-Position: refs/heads/master@{#300740}
parent 44fecdd7
Hide whitespace changes
Inline Side-by-side
Showing with 0 additions and 1452 deletions
mojo/BUILD.gn
View file @ 94c09453
...@@ -29,9 +29,6 @@ group("mojo") { ...@@ -29,9 +29,6 @@ group("mojo") {
} }
if (is_linux) { if (is_linux) {
deps += [
"//mojo/python",
]
if (mojo_use_go) { if (mojo_use_go) {
deps += [ deps += [
"//mojo/go", "//mojo/go",
......
mojo/PRESUBMIT.py
View file @ 94c09453
...@@ -22,9 +22,6 @@ def CheckChangeOnUpload(input_api, output_api): ...@@ -22,9 +22,6 @@ def CheckChangeOnUpload(input_api, output_api):
# For the python bindings: # For the python bindings:
mojo_python_bindings_path = os.path.join( mojo_python_bindings_path = os.path.join(
input_api.PresubmitLocalPath(), "public", "python") input_api.PresubmitLocalPath(), "public", "python")
# For the python bindings tests:
mojo_python_bindings_tests_path = os.path.join(
input_api.PresubmitLocalPath(), "python", "tests")
# TODO(vtl): Don't lint these files until the (many) problems are fixed # TODO(vtl): Don't lint these files until the (many) problems are fixed
# (possibly by deleting/rewriting some files). # (possibly by deleting/rewriting some files).
temporary_black_list = input_api.DEFAULT_BLACK_LIST + \ temporary_black_list = input_api.DEFAULT_BLACK_LIST + \
...@@ -40,7 +37,6 @@ def CheckChangeOnUpload(input_api, output_api): ...@@ -40,7 +37,6 @@ def CheckChangeOnUpload(input_api, output_api):
third_party_path, third_party_path,
mojo_public_bindings_pylib_path, mojo_public_bindings_pylib_path,
mojo_python_bindings_path, mojo_python_bindings_path,
mojo_python_bindings_tests_path,
] ]
results += input_api.canned_checks.RunPylint( results += input_api.canned_checks.RunPylint(
input_api, output_api, extra_paths_list=pylint_extra_paths, input_api, output_api, extra_paths_list=pylint_extra_paths,
......
mojo/mojo.gyp
View file @ 94c09453
...@@ -66,14 +66,6 @@ ...@@ -66,14 +66,6 @@
'mojo_external_application_tests', 'mojo_external_application_tests',
], ],
}], }],
['component != "shared_library" and OS == "linux"', {
'dependencies': [
'mojo_python_bindings',
'mojo_python_embedder',
'mojo_python_system',
'mojo_python',
],
}],
] ]
}, },
{ {
...@@ -585,135 +577,6 @@ ...@@ -585,135 +577,6 @@
}, },
], ],
}], }],
['component!="shared_library" and OS=="linux"', {
'targets': [
{
# GN version: //mojo/public/python:system
'target_name': 'mojo_python_system',
'variables': {
'python_base_module': 'mojo',
'python_cython_module': 'system',
},
'sources': [
'public/python/mojo/c_core.pxd',
'public/python/mojo/c_environment.pxd',
'public/python/mojo/system.pyx',
'public/python/src/python_system_helper.cc',
'public/python/src/python_system_helper.h',
],
'dependencies': [
'public/mojo_public.gyp:mojo_environment_standalone',
'public/mojo_public.gyp:mojo_system',
'public/mojo_public.gyp:mojo_utility',
],
'includes': [ '../third_party/cython/cython_compiler.gypi' ],
},
{
# GN version: //mojo/python:embedder
'target_name': 'mojo_python_embedder',
'type': 'loadable_module',
'variables': {
'python_base_module': 'mojo',
'python_cython_module': 'embedder',
},
'sources': [
'python/system/mojo/embedder.pyx',
],
'dependencies': [
'edk/mojo_edk.gyp:mojo_system_impl',
],
'includes': [ '../third_party/cython/cython_compiler.gypi' ],
},
{
# GN version: //mojo/public/python:bindings
'target_name': 'mojo_python_bindings',
'type': 'none',
'variables': {
'python_base_module': 'mojo/bindings',
},
'sources': [
'public/python/mojo/bindings/__init__.py',
'public/python/mojo/bindings/descriptor.py',
'public/python/mojo/bindings/messaging.py',
'public/python/mojo/bindings/promise.py',
'public/python/mojo/bindings/reflection.py',
'public/python/mojo/bindings/serialization.py',
],
'dependencies': [
'mojo_python_system',
],
'includes': [ '../third_party/cython/python_module.gypi' ],
},
{
# GN version: //mojo/python
'target_name': 'mojo_python',
'type': 'none',
'variables': {
'python_base_module': 'mojo',
},
'sources': [
'public/python/mojo/__init__.py',
],
'dependencies': [
'mojo_python_bindings',
'mojo_python_embedder',
'mojo_python_system',
],
# The python module need to be copied to their destinations
'actions': [
{
'action_name': 'Copy system module.',
'inputs': [
'<(DEPTH)/build/cp.py',
'<(PRODUCT_DIR)/libmojo_python_system.so',
],
'outputs': [
'<(PRODUCT_DIR)/python/mojo/system.so',
],
'action': [
'python',
'<@(_inputs)',
'<@(_outputs)',
]
},
{
'action_name': 'Copy embedder module.',
'inputs': [
'<(DEPTH)/build/cp.py',
'<(PRODUCT_DIR)/libmojo_python_embedder.so',
],
'outputs': [
'<(PRODUCT_DIR)/python/mojo/embedder.so',
],
'action': [
'python',
'<@(_inputs)',
'<@(_outputs)',
]
},
],
'includes': [ '../third_party/cython/python_module.gypi' ],
},
],
}],
['component!="shared_library" and OS=="linux" and test_isolation_mode!="noop"', {
'targets': [
{
'target_name': 'mojo_python_unittests_run',
'type': 'none',
'dependencies': [
'mojo_python',
'public/mojo_public.gyp:mojo_public_test_interfaces',
],
'includes': [
'../build/isolate.gypi',
],
'sources': [
'mojo_python_unittests.isolate',
],
},
],
}],
['test_isolation_mode != "noop"', { ['test_isolation_mode != "noop"', {
'targets': [ 'targets': [
{ {
......
mojo/mojo_python_unittests.isolate deleted 100644 → 0
View file @ 44fecdd7
# 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.
{
'conditions': [
['OS=="linux"', {
'variables': {
'command': [
'tools/run_mojo_python_bindings_tests.py',
'--build-dir=<(PRODUCT_DIR)',
],
'files': [
'<(PRODUCT_DIR)/gen/mojo/public/interfaces/bindings/tests/',
'<(PRODUCT_DIR)/python/',
'python/tests/',
'tools/pylib/mojo_python_tests_runner.py',
'tools/run_mojo_python_bindings_tests.py',
],
},
}],
],
}
mojo/public/python/BUILD.gn
View file @ 94c09453
...@@ -12,7 +12,6 @@ group("python") { ...@@ -12,7 +12,6 @@ group("python") {
] ]
} }
# GYP version: mojo.gyp:mojo_python_system
python_binary_module("system") { python_binary_module("system") {
python_base_module = "mojo" python_base_module = "mojo"
sources = [ sources = [
...@@ -44,7 +43,6 @@ copy("base") { ...@@ -44,7 +43,6 @@ copy("base") {
] ]
} }
# GYP version: mojo.gyp:mojo_python_bindings
copy("bindings") { copy("bindings") {
sources = [ sources = [
"mojo/bindings/__init__.py", "mojo/bindings/__init__.py",
......
mojo/python/BUILD.gn deleted 100644 → 0
View file @ 44fecdd7
# 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("//third_party/cython/rules.gni")
# GYP version: mojo/mojo.gyp:mojo_python
group("python") {
deps = [
":embedder",
"//mojo/public/python",
]
}
# GYP version: mojo/mojo.gyp:mojo_python_embedder
python_binary_module("embedder") {
python_base_module = "mojo"
sources = [
"system/mojo/embedder.pyx",
]
deps = [
"//mojo/edk/system",
]
datadeps = [
"//mojo/public/python:system",
]
}
mojo/python/system/mojo/embedder.pyx deleted 100644 → 0
View file @ 44fecdd7
# 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.
# distutils: language = c++
from libc.stdint cimport uintptr_t
from mojo import system
cdef extern from "third_party/cython/python_export.h":
pass
cdef extern from "base/memory/scoped_ptr.h":
cdef cppclass scoped_ptr[T]:
scoped_ptr(T*)
cdef extern from "mojo/edk/embedder/platform_support.h" \
namespace "mojo::embedder" nogil:
cdef cppclass PlatformSupport:
pass
cdef extern from "mojo/edk/embedder/simple_platform_support.h" \
namespace "mojo::embedder" nogil:
cdef cppclass SimplePlatformSupport(PlatformSupport):
SimplePlatformSupport()
cdef extern from "mojo/edk/embedder/embedder.h" nogil:
cdef void InitCEmbedder "mojo::embedder::Init"(
scoped_ptr[PlatformSupport] platform_support)
cdef extern from "mojo/public/platform/native/system_thunks.h" nogil:
cdef struct MojoSystemThunks:
pass
cdef MojoSystemThunks MojoMakeSystemThunks()
def Init():
InitCEmbedder(scoped_ptr[PlatformSupport](
new SimplePlatformSupport()))
cdef MojoSystemThunks thunks = MojoMakeSystemThunks()
system.SetSystemThunks(<uintptr_t>(&thunks))
mojo/python/tests/async_wait_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 mojo_unittest
# pylint: disable=E0611
from mojo import system
class AsyncWaitTest(mojo_unittest.MojoTestCase):
def setUp(self):
super(AsyncWaitTest, self).setUp()
self.array = []
self.handles = system.MessagePipe()
self.cancel = self.handles.handle0.AsyncWait(system.HANDLE_SIGNAL_READABLE,
system.DEADLINE_INDEFINITE,
self._OnResult)
def tearDown(self):
self.cancel()
self.handles = None
self.array = None
super(AsyncWaitTest, self).tearDown()
def _OnResult(self, value):
self.array.append(value)
def _WriteToHandle(self):
self.handles.handle1.WriteMessage()
def _PostWriteAndRun(self):
self.loop.PostDelayedTask(self._WriteToHandle, 0)
self.loop.RunUntilIdle()
def testAsyncWait(self):
self._PostWriteAndRun()
self.assertEquals(len(self.array), 1)
self.assertEquals(system.RESULT_OK, self.array[0])
def testAsyncWaitCancel(self):
self.loop.PostDelayedTask(self.cancel, 0)
self._PostWriteAndRun()
self.assertEquals(len(self.array), 0)
def testAsyncWaitImmediateCancel(self):
self.cancel()
self._PostWriteAndRun()
self.assertEquals(len(self.array), 0)
mojo/python/tests/bindings_constants_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 math
import unittest
# Generated files
# pylint: disable=F0401
import sample_service_mojom
class ConstantBindingsTest(unittest.TestCase):
def testConstantGeneration(self):
self.assertEquals(sample_service_mojom.TWELVE, 12)
self.assertEquals(sample_service_mojom.TOO_BIG_FOR_SIGNED_INT64,
9999999999999999999)
self.assertEquals(sample_service_mojom.DOUBLE_INFINITY,
float('inf'))
self.assertEquals(sample_service_mojom.DOUBLE_NEGATIVE_INFINITY,
float('-inf'))
self.assertTrue(math.isnan(sample_service_mojom.DOUBLE_NA_N))
self.assertEquals(sample_service_mojom.FLOAT_INFINITY,
float('inf'))
self.assertEquals(sample_service_mojom.FLOAT_NEGATIVE_INFINITY,
float('-inf'))
self.assertTrue(math.isnan(sample_service_mojom.FLOAT_NA_N))
def testConstantOnStructGeneration(self):
self.assertEquals(sample_service_mojom.Foo.FOOBY, "Fooby")
def testStructImmutability(self):
with self.assertRaises(AttributeError):
sample_service_mojom.Foo.FOOBY = 0
with self.assertRaises(AttributeError):
del sample_service_mojom.Foo.FOOBY
with self.assertRaises(AttributeError):
sample_service_mojom.Foo.BAR = 1
mojo/python/tests/bindings_enums_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 unittest
# Generated files
# pylint: disable=F0401
import sample_import_mojom
import sample_service_mojom
class EnumBindingsTest(unittest.TestCase):
# Testing enum classes are in the right module.
def testModule(self):
self.assertEquals(sample_import_mojom.Shape.__module__,
'sample_import_mojom')
# Testing that enum class have expected constant values.
def testTopLevelEnumGeneration(self):
self.assertEquals(sample_import_mojom.Shape.RECTANGLE, 1)
self.assertEquals(sample_import_mojom.Shape.CIRCLE, 2)
self.assertEquals(sample_import_mojom.Shape.TRIANGLE, 3)
self.assertEquals(sample_import_mojom.Shape.LAST,
sample_import_mojom.Shape.TRIANGLE)
self.assertEquals(sample_import_mojom.AnotherShape.RECTANGLE, 10)
self.assertEquals(sample_import_mojom.AnotherShape.CIRCLE, 11)
self.assertEquals(sample_import_mojom.AnotherShape.TRIANGLE, 12)
self.assertEquals(sample_import_mojom.YetAnotherShape.RECTANGLE, 20)
self.assertEquals(sample_import_mojom.YetAnotherShape.CIRCLE, 21)
self.assertEquals(sample_import_mojom.YetAnotherShape.TRIANGLE, 22)
# Testing that internal enum class have expected constant values.
def testInternalEnumGeneration(self):
self.assertEquals(sample_service_mojom.Bar.Type.VERTICAL, 1)
self.assertEquals(sample_service_mojom.Bar.Type.HORIZONTAL, 2)
self.assertEquals(sample_service_mojom.Bar.Type.BOTH, 3)
self.assertEquals(sample_service_mojom.Bar.Type.INVALID, 4)
# Testing an enum class cannot be instantiated.
def testNonInstantiableEnum(self):
with self.assertRaises(TypeError):
sample_import_mojom.Shape()
# Testing an enum does not contain the VALUES constant.
def testNoVALUESConstant(self):
with self.assertRaises(AttributeError):
# pylint: disable=W0104
sample_import_mojom.Shape.VALUES
# Testing enum values are frozen.
def testEnumFrozen(self):
with self.assertRaises(AttributeError):
sample_import_mojom.Shape.RECTANGLE = 2
with self.assertRaises(AttributeError):
del sample_import_mojom.Shape.RECTANGLE
with self.assertRaises(AttributeError):
sample_import_mojom.Shape.NewShape = 4
mojo/python/tests/bindings_serialization_deserialization_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 math
import unittest
# pylint: disable=E0611,F0401
import mojo.system
# Generated files
# pylint: disable=F0401
import sample_import_mojom
import sample_import2_mojom
import sample_service_mojom
def _NewHandle():
return mojo.system.MessagePipe().handle0
def _NewBar():
bar_instance = sample_service_mojom.Bar()
bar_instance.alpha = 22
bar_instance.beta = 87
bar_instance.gamma = 122
bar_instance.type = sample_service_mojom.Bar.Type.BOTH
return bar_instance
def _NewFoo():
foo_instance = sample_service_mojom.Foo()
foo_instance.name = "Foo.name"
foo_instance.x = 23
foo_instance.y = -23
foo_instance.a = False
foo_instance.b = True
foo_instance.c = True
foo_instance.bar = _NewBar()
foo_instance.extra_bars = [
_NewBar(),
_NewBar(),
]
foo_instance.data = 'Hello world'
foo_instance.source = _NewHandle()
foo_instance.input_streams = [ _NewHandle() ]
foo_instance.output_streams = [ _NewHandle(), _NewHandle() ]
foo_instance.array_of_array_of_bools = [ [ True, False ], [] ]
foo_instance.multi_array_of_strings = [
[
[ "1", "2" ],
[],
[ "3", "4" ],
],
[],
]
foo_instance.array_of_bools = [ True, 0, 1, 2, 0, 0, 0, 0, 0, True ]
return foo_instance
class SerializationDeserializationTest(unittest.TestCase):
def testFooSerialization(self):
(data, _) = _NewFoo().Serialize()
self.assertTrue(len(data))
self.assertEquals(len(data) % 8, 0)
def testFooDeserialization(self):
(data, handles) = _NewFoo().Serialize()
self.assertTrue(
sample_service_mojom.Foo.Deserialize(data, handles))
def testFooSerializationDeserialization(self):
foo1 = _NewFoo()
(data, handles) = foo1.Serialize()
foo2 = sample_service_mojom.Foo.Deserialize(data, handles)
self.assertEquals(foo1, foo2)
def testDefaultsTestSerializationDeserialization(self):
v1 = sample_service_mojom.DefaultsTest()
v1.a18 = []
v1.a19 = ""
v1.a21 = sample_import_mojom.Point()
v1.a22.location = sample_import_mojom.Point()
v1.a22.size = sample_import2_mojom.Size()
(data, handles) = v1.Serialize()
v2 = sample_service_mojom.DefaultsTest.Deserialize(data, handles)
# NaN needs to be a special case.
self.assertNotEquals(v1, v2)
self.assertTrue(math.isnan(v2.a28))
self.assertTrue(math.isnan(v2.a31))
v1.a28 = v2.a28 = v1.a31 = v2.a31 = 0
self.assertEquals(v1, v2)
def testFooDeserializationError(self):
with self.assertRaises(Exception):
sample_service_mojom.Foo.Deserialize("", [])
mojo/python/tests/bindings_structs_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 math
import unittest
# pylint: disable=E0611,F0401
import mojo.system
# Generated files
# pylint: disable=F0401
import regression_tests_mojom
import sample_import_mojom
import sample_import2_mojom
import sample_service_mojom
class StructBindingsTest(unittest.TestCase):
def testModule(self):
self.assertEquals(sample_service_mojom.DefaultsTest.__module__,
'sample_service_mojom')
def testDefaultsTest(self):
defaults_test = sample_service_mojom.DefaultsTest()
self.assertEquals(defaults_test.a0, -12)
self.assertEquals(defaults_test.a1, 12)
self.assertEquals(defaults_test.a2, 1234)
self.assertEquals(defaults_test.a3, 34567)
self.assertEquals(defaults_test.a4, 123456)
self.assertEquals(defaults_test.a5, 3456789012)
self.assertEquals(defaults_test.a6, -111111111111)
self.assertEquals(defaults_test.a7, 9999999999999999999)
self.assertEquals(defaults_test.a8, 0x12345)
self.assertEquals(defaults_test.a9, -0x12345)
self.assertEquals(defaults_test.a10, 1234)
self.assertEquals(defaults_test.a11, True)
self.assertEquals(defaults_test.a12, False)
self.assertEquals(defaults_test.a13, 123.25)
self.assertEquals(defaults_test.a14, 1234567890.123)
self.assertEquals(defaults_test.a15, 1E10)
self.assertEquals(defaults_test.a16, -1.2E+20)
self.assertEquals(defaults_test.a17, 1.23E-20)
self.assertEquals(defaults_test.a18, None)
self.assertEquals(defaults_test.a19, None)
self.assertEquals(defaults_test.a20, sample_service_mojom.Bar.Type.BOTH)
self.assertEquals(defaults_test.a21, None)
self.assertTrue(isinstance(defaults_test.a22, sample_import2_mojom.Thing))
self.assertEquals(defaults_test.a23, 0xFFFFFFFFFFFFFFFF)
self.assertEquals(defaults_test.a24, 0x123456789)
self.assertEquals(defaults_test.a25, -0x123456789)
self.assertEquals(defaults_test.a26, float('inf'))
self.assertEquals(defaults_test.a27, float('-inf'))
self.assertTrue(math.isnan(defaults_test.a28))
self.assertEquals(defaults_test.a29, float('inf'))
self.assertEquals(defaults_test.a30, float('-inf'))
self.assertTrue(math.isnan(defaults_test.a31))
def testNoAliasing(self):
foo1 = sample_service_mojom.Foo()
foo2 = sample_service_mojom.Foo()
foo1.name = "foo1"
foo2.name = "foo2"
self.assertEquals(foo1.name, "foo1")
self.assertEquals(foo2.name, "foo2")
defaults_test1 = sample_service_mojom.DefaultsTest()
defaults_test2 = sample_service_mojom.DefaultsTest()
self.assertIsNot(defaults_test1.a22, defaults_test2.a22)
def testImmutableAttributeSet(self):
foo_instance = sample_service_mojom.Foo()
with self.assertRaises(AttributeError):
foo_instance.new_attribute = None
with self.assertRaises(AttributeError):
del foo_instance.name
def _TestIntegerField(self, entity, field_name, bits, signed):
if signed:
min_value = -(1 << (bits - 1))
max_value = (1 << (bits - 1)) - 1
else:
min_value = 0
max_value = (1 << bits) - 1
entity.__setattr__(field_name, min_value)
entity.__setattr__(field_name, max_value)
with self.assertRaises(TypeError):
entity.__setattr__(field_name, None)
with self.assertRaises(OverflowError):
entity.__setattr__(field_name, min_value - 1)
with self.assertRaises(OverflowError):
entity.__setattr__(field_name, max_value + 1)
with self.assertRaises(TypeError):
entity.__setattr__(field_name, 'hello world')
def testTypes(self):
defaults_test = sample_service_mojom.DefaultsTest()
# Integer types
self._TestIntegerField(defaults_test, 'a0', 8, True)
self._TestIntegerField(defaults_test, 'a1', 8, False)
self._TestIntegerField(defaults_test, 'a2', 16, True)
self._TestIntegerField(defaults_test, 'a3', 16, False)
self._TestIntegerField(defaults_test, 'a4', 32, True)
self._TestIntegerField(defaults_test, 'a5', 32, False)
self._TestIntegerField(defaults_test, 'a6', 64, True)
self._TestIntegerField(defaults_test, 'a7', 64, False)
# Boolean types
defaults_test.a11 = False
self.assertEquals(defaults_test.a11, False)
defaults_test.a11 = None
self.assertEquals(defaults_test.a11, False)
defaults_test.a11 = []
self.assertEquals(defaults_test.a11, False)
defaults_test.a12 = True
self.assertEquals(defaults_test.a12, True)
defaults_test.a12 = 1
self.assertEquals(defaults_test.a12, True)
defaults_test.a12 = [[]]
self.assertEquals(defaults_test.a12, True)
# Floating point types
with self.assertRaises(TypeError):
defaults_test.a13 = 'hello'
with self.assertRaises(TypeError):
defaults_test.a14 = 'hello'
# Array type
defaults_test.a18 = None
defaults_test.a18 = []
defaults_test.a18 = [ 0 ]
defaults_test.a18 = [ 255 ]
defaults_test.a18 = [ 0, 255 ]
with self.assertRaises(TypeError):
defaults_test.a18 = [[]]
with self.assertRaises(OverflowError):
defaults_test.a18 = [ -1 ]
with self.assertRaises(OverflowError):
defaults_test.a18 = [ 256 ]
# String type
defaults_test.a19 = None
defaults_test.a19 = ''
defaults_test.a19 = 'hello world'
with self.assertRaises(TypeError):
defaults_test.a19 = [[]]
with self.assertRaises(TypeError):
defaults_test.a19 = [ -1 ]
with self.assertRaises(TypeError):
defaults_test.a19 = [ 256 ]
# Structs
defaults_test.a21 = None
defaults_test.a21 = sample_import_mojom.Point()
with self.assertRaises(TypeError):
defaults_test.a21 = 1
with self.assertRaises(TypeError):
defaults_test.a21 = sample_import2_mojom.Thing()
# Handles
foo_instance = sample_service_mojom.Foo()
foo_instance.source = None
foo_instance.source = mojo.system.Handle()
with self.assertRaises(TypeError):
foo_instance.source = 1
with self.assertRaises(TypeError):
foo_instance.source = object()
def testConstructor(self):
bar_instance = sample_service_mojom.Bar()
foo_instance = sample_service_mojom.Foo(name="Foo",
x=-1,
y=5,
a=False,
bar=bar_instance)
self.assertEquals(foo_instance.name, "Foo")
self.assertEquals(foo_instance.x, -1)
self.assertEquals(foo_instance.y, 5)
self.assertEquals(foo_instance.a, False)
self.assertEquals(foo_instance.bar, bar_instance)
def testPositionalConstructor(self):
p = sample_import_mojom.Point()
self.assertEquals(p.x, 0)
self.assertEquals(p.y, 0)
p = sample_import_mojom.Point(34)
self.assertEquals(p.x, 34)
self.assertEquals(p.y, 0)
p = sample_import_mojom.Point(34, 12)
self.assertEquals(p.x, 34)
self.assertEquals(p.y, 12)
p = sample_import_mojom.Point(x=34, y=12)
self.assertEquals(p.x, 34)
self.assertEquals(p.y, 12)
p = sample_import_mojom.Point(34, y=12)
self.assertEquals(p.x, 34)
self.assertEquals(p.y, 12)
with self.assertRaises(TypeError):
p = sample_import_mojom.Point(0, 0, 0)
with self.assertRaises(TypeError):
p = sample_import_mojom.Point(0, x=0)
with self.assertRaises(TypeError):
p = sample_import_mojom.Point(c=0)
def testCyclicDefinition(self):
a = regression_tests_mojom.A()
b = regression_tests_mojom.B()
self.assertIsNone(a.b)
self.assertIsNone(b.a)
a.b = b
self.assertIs(a.b, b)
mojo/python/tests/messaging_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 unittest
import mojo_unittest
from mojo.bindings import messaging
# pylint: disable=E0611
from mojo import system
class _ForwardingConnectionErrorHandler(messaging.ConnectionErrorHandler):
def __init__(self, callback):
messaging.ConnectionErrorHandler.__init__(self)
self._callback = callback
def OnError(self, result):
self._callback(result)
class ConnectorTest(mojo_unittest.MojoTestCase):
def setUp(self):
super(ConnectorTest, self).setUp()
self.received_messages = []
self.received_errors = []
def _OnMessage(message):
self.received_messages.append(message)
return True
def _OnError(result):
self.received_errors.append(result)
handles = system.MessagePipe()
self.connector = messaging.Connector(handles.handle1)
self.connector.SetIncomingMessageReceiver(
messaging.ForwardingMessageReceiver(_OnMessage))
self.connector.SetErrorHandler(
_ForwardingConnectionErrorHandler(_OnError))
self.connector.Start()
self.handle = handles.handle0
def tearDown(self):
self.connector = None
self.handle = None
super(ConnectorTest, self).tearDown()
def testConnectorRead(self):
self.handle.WriteMessage()
self.loop.RunUntilIdle()
self.assertTrue(self.received_messages)
self.assertFalse(self.received_errors)
def testConnectorWrite(self):
self.connector.Accept(messaging.Message())
(result, _, _) = self.handle.ReadMessage()
self.assertEquals(result, system.RESULT_OK)
self.assertFalse(self.received_errors)
def testConnectorCloseRemoteHandle(self):
self.handle.Close()
self.loop.RunUntilIdle()
self.assertFalse(self.received_messages)
self.assertTrue(self.received_errors)
self.assertEquals(self.received_errors[0],
system.RESULT_FAILED_PRECONDITION)
def testConnectorDeleteConnector(self):
self.connector = None
(result, _, _) = self.handle.ReadMessage()
self.assertEquals(result, system.RESULT_FAILED_PRECONDITION)
class HeaderTest(unittest.TestCase):
def testSimpleMessageHeader(self):
header = messaging.MessageHeader(0xdeadbeaf, messaging.NO_FLAG)
self.assertEqual(header.message_type, 0xdeadbeaf)
self.assertFalse(header.has_request_id)
self.assertFalse(header.expects_response)
self.assertFalse(header.is_response)
data = header.Serialize()
other_header = messaging.MessageHeader.Deserialize(data)
self.assertEqual(other_header.message_type, 0xdeadbeaf)
self.assertFalse(other_header.has_request_id)
self.assertFalse(other_header.expects_response)
self.assertFalse(other_header.is_response)
def testMessageHeaderWithRequestID(self):
# Request message.
header = messaging.MessageHeader(0xdeadbeaf,
messaging.MESSAGE_EXPECTS_RESPONSE_FLAG)
self.assertEqual(header.message_type, 0xdeadbeaf)
self.assertTrue(header.has_request_id)
self.assertTrue(header.expects_response)
self.assertFalse(header.is_response)
self.assertEqual(header.request_id, 0)
data = header.Serialize()
other_header = messaging.MessageHeader.Deserialize(data)
self.assertEqual(other_header.message_type, 0xdeadbeaf)
self.assertTrue(other_header.has_request_id)
self.assertTrue(other_header.expects_response)
self.assertFalse(other_header.is_response)
self.assertEqual(other_header.request_id, 0)
header.request_id = 0xdeadbeafdeadbeaf
data = header.Serialize()
other_header = messaging.MessageHeader.Deserialize(data)
self.assertEqual(other_header.request_id, 0xdeadbeafdeadbeaf)
# Response message.
header = messaging.MessageHeader(0xdeadbeaf,
messaging.MESSAGE_IS_RESPONSE_FLAG,
0xdeadbeafdeadbeaf)
self.assertEqual(header.message_type, 0xdeadbeaf)
self.assertTrue(header.has_request_id)
self.assertFalse(header.expects_response)
self.assertTrue(header.is_response)
self.assertEqual(header.request_id, 0xdeadbeafdeadbeaf)
data = header.Serialize()
other_header = messaging.MessageHeader.Deserialize(data)
self.assertEqual(other_header.message_type, 0xdeadbeaf)
self.assertTrue(other_header.has_request_id)
self.assertFalse(other_header.expects_response)
self.assertTrue(other_header.is_response)
self.assertEqual(other_header.request_id, 0xdeadbeafdeadbeaf)
class RouterTest(mojo_unittest.MojoTestCase):
def setUp(self):
super(RouterTest, self).setUp()
self.received_messages = []
self.received_errors = []
def _OnMessage(message):
self.received_messages.append(message)
return True
def _OnError(result):
self.received_errors.append(result)
handles = system.MessagePipe()
self.router = messaging.Router(handles.handle1)
self.router.SetIncomingMessageReceiver(
messaging.ForwardingMessageReceiver(_OnMessage))
self.router.SetErrorHandler(
_ForwardingConnectionErrorHandler(_OnError))
self.router.Start()
self.handle = handles.handle0
def tearDown(self):
self.router = None
self.handle = None
super(RouterTest, self).tearDown()
def testSimpleMessage(self):
header_data = messaging.MessageHeader(0, messaging.NO_FLAG).Serialize()
message = messaging.Message(header_data)
self.router.Accept(message)
self.loop.RunUntilIdle()
self.assertFalse(self.received_errors)
self.assertFalse(self.received_messages)
(res, data, _) = self.handle.ReadMessage(bytearray(len(header_data)))
self.assertEquals(system.RESULT_OK, res)
self.assertEquals(data[0], header_data)
def testSimpleReception(self):
header_data = messaging.MessageHeader(0, messaging.NO_FLAG).Serialize()
self.handle.WriteMessage(header_data)
self.loop.RunUntilIdle()
self.assertFalse(self.received_errors)
self.assertEquals(len(self.received_messages), 1)
self.assertEquals(self.received_messages[0].data, header_data)
def testRequestResponse(self):
header_data = messaging.MessageHeader(
0, messaging.MESSAGE_EXPECTS_RESPONSE_FLAG).Serialize()
message = messaging.Message(header_data)
back_messages = []
def OnBackMessage(message):
back_messages.append(message)
self.router.AcceptWithResponder(message,
messaging.ForwardingMessageReceiver(
OnBackMessage))
self.loop.RunUntilIdle()
self.assertFalse(self.received_errors)
self.assertFalse(self.received_messages)
(res, data, _) = self.handle.ReadMessage(bytearray(len(header_data)))
self.assertEquals(system.RESULT_OK, res)
message_header = messaging.MessageHeader.Deserialize(data[0])
self.assertNotEquals(message_header.request_id, 0)
response_header_data = messaging.MessageHeader(
0,
messaging.MESSAGE_IS_RESPONSE_FLAG,
message_header.request_id).Serialize()
self.handle.WriteMessage(response_header_data)
self.loop.RunUntilIdle()
self.assertFalse(self.received_errors)
self.assertEquals(len(back_messages), 1)
self.assertEquals(back_messages[0].data, response_header_data)
mojo/python/tests/mojo_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 unittest
# pylint: disable=E0611,F0401
import mojo.embedder
import mojo.system as system
class MojoTestCase(unittest.TestCase):
def setUp(self):
mojo.embedder.Init()
self.loop = system.RunLoop()
def tearDown(self):
self.loop = None
mojo/python/tests/promise_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 unittest
# pylint: disable=F0401
from mojo.bindings import promise
class PromiseTest(unittest.TestCase):
def setUp(self):
self.accumulated = []
def _AddToAccumulated(self, res):
self.accumulated.append(res)
return res
def testResolve(self):
p = promise.Promise.Resolve(0)
self.assertEquals(p.state, promise.Promise.STATE_FULLFILLED)
p.Then(self._AddToAccumulated)
self.assertEquals(self.accumulated, [0])
def testResolveToPromise(self):
p = promise.Promise.Resolve(0)
self.assertEquals(p.state, promise.Promise.STATE_FULLFILLED)
q = promise.Promise.Resolve(p)
self.assertEquals(p.state, promise.Promise.STATE_FULLFILLED)
q.Then(self._AddToAccumulated)
self.assertEquals(self.accumulated, [0])
def testReject(self):
p = promise.Promise.Reject(0)
self.assertEquals(p.state, promise.Promise.STATE_REJECTED)
p.Then(onRejected=self._AddToAccumulated)
self.assertEquals(self.accumulated, [0])
def testGeneratorFunctionResolve(self):
(p, resolve, _) = _GetPromiseAndFunctions()
self.assertEquals(p.state, promise.Promise.STATE_PENDING)
p.Then(self._AddToAccumulated)
resolve(0)
self.assertEquals(p.state, promise.Promise.STATE_FULLFILLED)
self.assertEquals(self.accumulated, [0])
def testGeneratorFunctionReject(self):
(p, _, reject) = _GetPromiseAndFunctions()
self.assertEquals(p.state, promise.Promise.STATE_PENDING)
p.Then(None, self._AddToAccumulated)
reject(0)
self.assertEquals(p.state, promise.Promise.STATE_REJECTED)
self.assertEquals(self.accumulated, [0])
def testGeneratorFunctionResolveToPromise(self):
(p1, resolve, _) = _GetPromiseAndFunctions()
p2 = promise.Promise(lambda x, y: x(p1))
self.assertEquals(p2.state, promise.Promise.STATE_PENDING)
p2.Then(self._AddToAccumulated)
resolve(promise.Promise.Resolve(0))
self.assertEquals(self.accumulated, [0])
def testComputation(self):
(p, resolve, _) = _GetPromiseAndFunctions()
p.Then(lambda x: x+1).Then(lambda x: x+2).Then(self._AddToAccumulated)
self.assertEquals(self.accumulated, [])
resolve(0)
self.assertEquals(self.accumulated, [3])
def testRecoverAfterException(self):
(p, resolve, _) = _GetPromiseAndFunctions()
p.Then(_ThrowException).Catch(self._AddToAccumulated)
self.assertEquals(self.accumulated, [])
resolve(0)
self.assertEquals(len(self.accumulated), 1)
self.assertIsInstance(self.accumulated[0], RuntimeError)
self.assertEquals(self.accumulated[0].message, 0)
def testMultipleRejectResolve(self):
(p, resolve, reject) = _GetPromiseAndFunctions()
p.Then(self._AddToAccumulated, self._AddToAccumulated)
resolve(0)
self.assertEquals(self.accumulated, [0])
resolve(0)
self.assertEquals(self.accumulated, [0])
reject(0)
self.assertEquals(self.accumulated, [0])
self.accumulated = []
(p, resolve, reject) = _GetPromiseAndFunctions()
p.Then(self._AddToAccumulated, self._AddToAccumulated)
reject(0)
self.assertEquals(self.accumulated, [0])
resolve(0)
self.assertEquals(self.accumulated, [0])
reject(0)
self.assertEquals(self.accumulated, [0])
def testAll(self):
promises_and_functions = [_GetPromiseAndFunctions() for x in xrange(10)]
promises = [x[0] for x in promises_and_functions]
all_promise = promise.Promise.All(*promises)
res = []
def AddToRes(values):
res.append(values)
all_promise.Then(AddToRes, AddToRes)
for i, (_, resolve, _) in enumerate(promises_and_functions):
self.assertEquals(len(res), 0)
resolve(i)
self.assertEquals(len(res), 1)
self.assertEquals(res[0], [i for i in xrange(10)])
self.assertEquals(all_promise.state, promise.Promise.STATE_FULLFILLED)
def testAllFailure(self):
promises_and_functions = [_GetPromiseAndFunctions() for x in xrange(10)]
promises = [x[0] for x in promises_and_functions]
all_promise = promise.Promise.All(*promises)
res = []
def AddToRes(values):
res.append(values)
all_promise.Then(AddToRes, AddToRes)
for i in xrange(10):
if i <= 5:
self.assertEquals(len(res), 0)
else:
self.assertEquals(len(res), 1)
if i != 5:
promises_and_functions[i][1](i)
else:
promises_and_functions[i][2]('error')
self.assertEquals(len(res), 1)
self.assertEquals(res[0], 'error')
self.assertEquals(all_promise.state, promise.Promise.STATE_REJECTED)
def testRace(self):
promises_and_functions = [_GetPromiseAndFunctions() for x in xrange(10)]
promises = [x[0] for x in promises_and_functions]
race_promise = promise.Promise.Race(*promises)
res = []
def AddToRes(values):
res.append(values)
race_promise.Then(AddToRes, AddToRes)
self.assertEquals(len(res), 0)
promises_and_functions[7][1]('success')
self.assertEquals(len(res), 1)
for i, (f) in enumerate(promises_and_functions):
f[1 + (i % 2)](i)
self.assertEquals(len(res), 1)
self.assertEquals(res[0], 'success')
self.assertEquals(race_promise.state, promise.Promise.STATE_FULLFILLED)
def testRaceFailure(self):
promises_and_functions = [_GetPromiseAndFunctions() for x in xrange(10)]
promises = [x[0] for x in promises_and_functions]
race_promise = promise.Promise.Race(*promises)
res = []
def AddToRes(values):
res.append(values)
race_promise.Then(AddToRes, AddToRes)
self.assertEquals(len(res), 0)
promises_and_functions[7][2]('error')
self.assertEquals(len(res), 1)
for i, (f) in enumerate(promises_and_functions):
f[1 + (i % 2)](i)
self.assertEquals(len(res), 1)
self.assertEquals(res[0], 'error')
self.assertEquals(race_promise.state, promise.Promise.STATE_REJECTED)
def _GetPromiseAndFunctions():
functions = {}
def GeneratorFunction(resolve, reject):
functions['resolve'] = resolve
functions['reject'] = reject
p = promise.Promise(GeneratorFunction)
return (p, functions['resolve'], functions['reject'])
def _ThrowException(x):
raise RuntimeError(x)
mojo/python/tests/runloop_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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 mojo_unittest
# pylint: disable=E0611
from mojo import system
def _Increment(array):
def _Closure():
array.append(0)
return _Closure
class RunLoopTest(mojo_unittest.MojoTestCase):
def testRunLoop(self):
array = []
for _ in xrange(10):
self.loop.PostDelayedTask(_Increment(array))
self.loop.RunUntilIdle()
self.assertEquals(len(array), 10)
def testRunLoopWithException(self):
def Throw():
raise Exception("error")
array = []
self.loop.PostDelayedTask(Throw)
self.loop.PostDelayedTask(_Increment(array))
with self.assertRaisesRegexp(Exception, '^error$'):
self.loop.Run()
self.assertEquals(len(array), 0)
self.loop.RunUntilIdle()
self.assertEquals(len(array), 1)
def testCurrent(self):
self.assertEquals(system.RunLoop.Current(), self.loop)
self.loop = None
self.assertIsNone(system.RunLoop.Current())
mojo/python/tests/system_unittest.py deleted 100644 → 0
View file @ 44fecdd7
# 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
from mojo import 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):
def testResults(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 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.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 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
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.WriteMessage()
self.assertEquals(
system.RESULT_OK,
handle.Wait(
system.HANDLE_SIGNAL_READABLE,
system.DEADLINE_INDEFINITE))
def testWaitOverManyMessagePipe(self):
handles = system.MessagePipe()
handle0 = handles.handle0
handle1 = handles.handle1
self.assertEquals(
0,
system.WaitMany(
[(handle0, system.HANDLE_SIGNAL_WRITABLE),
(handle1, system.HANDLE_SIGNAL_WRITABLE)],
system.DEADLINE_INDEFINITE))
self.assertEquals(
system.RESULT_DEADLINE_EXCEEDED,
system.WaitMany(
[(handle0, system.HANDLE_SIGNAL_READABLE),
(handle1, system.HANDLE_SIGNAL_READABLE)], 0))
handle0.WriteMessage()
self.assertEquals(
1,
system.WaitMany(
[(handle0, system.HANDLE_SIGNAL_READABLE),
(handle1, system.HANDLE_SIGNAL_READABLE)],
system.DEADLINE_INDEFINITE))
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 testCreateDataPipeWithDiscardFlag(self):
options = system.CreateDataPipeOptions()
options.flags = system.CreateDataPipeOptions.FLAG_MAY_DISCARD
self._TestDataHandleCreation(system.DataPipe(options))
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)))
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)
mojo/tools/mojob.sh
View file @ 94c09453
...@@ -68,13 +68,6 @@ do_perftests() { ...@@ -68,13 +68,6 @@ do_perftests() {
"out/$1/mojo_public_system_perftests" || exit 1 "out/$1/mojo_public_system_perftests" || exit 1
} }
do_pytests() {
echo "Running python tests in out/$1 ..."
python mojo/tools/run_mojo_python_tests.py || exit 1
python mojo/tools/run_mojo_python_bindings_tests.py "--build-dir=out/$1" || \
exit 1
}
do_gn() { do_gn() {
local gn_args="$(make_gn_args $1)" local gn_args="$(make_gn_args $1)"
echo "Running gn with --args=\"${gn_args}\" ..." echo "Running gn with --args=\"${gn_args}\" ..."
...@@ -164,10 +157,6 @@ for arg in "$@"; do ...@@ -164,10 +157,6 @@ for arg in "$@"; do
should_do_Debug && do_perftests Debug should_do_Debug && do_perftests Debug
should_do_Release && do_perftests Release should_do_Release && do_perftests Release
;; ;;
pytest)
should_do_Debug && do_pytests Debug
should_do_Release && do_pytests Release
;;
gn) gn)
should_do_Debug && do_gn Debug should_do_Debug && do_gn Debug
should_do_Release && do_gn Release should_do_Release && do_gn Release
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment