summaryrefslogtreecommitdiff
path: root/zephyr/zmake/tests/test_multiproc_executor.py
diff options
context:
space:
mode:
Diffstat (limited to 'zephyr/zmake/tests/test_multiproc_executor.py')
-rw-r--r--zephyr/zmake/tests/test_multiproc_executor.py30
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