xref: /aosp_15_r20/external/cronet/build/android/pylib/base/test_collection.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b) 
1# Copyright 2013 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5
6import threading
7
8
9class TestCollection:
10  """A threadsafe collection of tests.
11
12  Args:
13    tests: List of tests to put in the collection.
14  """
15
16  def __init__(self, tests=None):
17    if not tests:
18      tests = []
19    self._lock = threading.Lock()
20    self._tests = []
21    self._tests_in_progress = 0
22    # Used to signal that an item is available or all items have been handled.
23    self._item_available_or_all_done = threading.Event()
24    for t in tests:
25      self.add(t)
26
27  def _pop(self):
28    """Pop a test from the collection.
29
30    Waits until a test is available or all tests have been handled.
31
32    Returns:
33      A test or None if all tests have been handled.
34    """
35    while True:
36      # Wait for a test to be available or all tests to have been handled.
37      self._item_available_or_all_done.wait()
38      with self._lock:
39        # Check which of the two conditions triggered the signal.
40        if self._tests_in_progress == 0:
41          return None
42        try:
43          return self._tests.pop(0)
44        except IndexError:
45          # Another thread beat us to the available test, wait again.
46          self._item_available_or_all_done.clear()
47
48  def add(self, test):
49    """Add a test to the collection.
50
51    Args:
52      test: A test to add.
53    """
54    with self._lock:
55      self._tests.append(test)
56      self._item_available_or_all_done.set()
57      self._tests_in_progress += 1
58
59  def test_completed(self):
60    """Indicate that a test has been fully handled."""
61    with self._lock:
62      self._tests_in_progress -= 1
63      if self._tests_in_progress == 0:
64        # All tests have been handled, signal all waiting threads.
65        self._item_available_or_all_done.set()
66
67  def __iter__(self):
68    """Iterate through tests in the collection until all have been handled."""
69    while True:
70      r = self._pop()
71      if r is None:
72        break
73      yield r
74
75  def __len__(self):
76    """Return the number of tests currently in the collection."""
77    return len(self._tests)
78
79  def test_names(self):
80    """Return a list of the names of the tests currently in the collection."""
81    with self._lock:
82      return list(t.test for t in self._tests)
83