tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

test_collection.py (2315B)

      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 
      6 import threading
      7 
      8 
      9 class 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)