diff options
Diffstat (limited to 'zephyr/zmake/tests/test_multiproc_executor.py')
-rw-r--r-- | zephyr/zmake/tests/test_multiproc_executor.py | 30 |
1 files changed, 20 insertions, 10 deletions
diff --git a/zephyr/zmake/tests/test_multiproc_executor.py b/zephyr/zmake/tests/test_multiproc_executor.py index ebc2be5e4f..c905ef03ec 100644 --- a/zephyr/zmake/tests/test_multiproc_executor.py +++ b/zephyr/zmake/tests/test_multiproc_executor.py @@ -1,52 +1,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(cv, predicate, step, return_value=0): - with cv: - cv.wait_for(predicate=lambda: step[0] == predicate) +def _lock_step(cond, predicate, step, return_value=0): + with cond: + cond.wait_for(predicate=lambda: step[0] == predicate) step[0] += 1 - cv.notify_all() + cond.notify_all() return return_value def test_two_function_executor_wait_for_both(): - cv = threading.Condition() + """Test two functions in executor.""" + cond = threading.Condition() step = [0] executor = zmake.multiproc.Executor() - executor.append(lambda: _lock_step(cv=cv, predicate=0, step=step)) - executor.append(lambda: _lock_step(cv=cv, predicate=1, step=step)) + 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(): - cv = threading.Condition() + """Test two functions in executor, when one fails.""" + cond = threading.Condition() step = [0] executor = zmake.multiproc.Executor() - executor.append(lambda: _lock_step(cv=cv, predicate=0, step=step, return_value=-1)) - executor.append(lambda: _lock_step(cv=cv, predicate=1, step=step)) + 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 |