1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
# Copyright 2021 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tests for zmake multiproc."""
import threading
import zmake.multiproc
def test_single_function_executor_success():
"""Test single function success."""
executor = zmake.multiproc.Executor()
executor.append(lambda: 0)
assert executor.wait() == 0
def test_single_function_executor_fail():
"""Test single function fail."""
executor = zmake.multiproc.Executor()
executor.append(lambda: -2)
assert executor.wait() == -2
def test_single_function_executor_raise():
"""Test single function raising an exception."""
executor = zmake.multiproc.Executor()
executor.append(lambda: 1 / 0)
assert executor.wait() != 0
def _lock_step(cond, predicate, step, return_value=0):
with cond:
cond.wait_for(predicate=lambda: step[0] == predicate)
step[0] += 1
cond.notify_all()
return return_value
def test_two_function_executor_wait_for_both():
"""Test two functions in executor."""
cond = threading.Condition()
step = [0]
executor = zmake.multiproc.Executor()
executor.append(lambda: _lock_step(cond=cond, predicate=0, step=step))
executor.append(lambda: _lock_step(cond=cond, predicate=1, step=step))
assert executor.wait() == 0
assert step[0] == 2
def test_two_function_executor_one_fails():
"""Test two functions in executor, when one fails."""
cond = threading.Condition()
step = [0]
executor = zmake.multiproc.Executor()
executor.append(
lambda: _lock_step(cond=cond, predicate=0, step=step, return_value=-1)
)
executor.append(lambda: _lock_step(cond=cond, predicate=1, step=step))
assert executor.wait() == -1
assert step[0] == 2
|
