+ test cases

Author: caffeine-addictt

tests/__init__.py

@@ -0,0 +1 @@
+# Need this file, else it can't import src.thread

tests/test_parallelprocessing.py

@@ -0,0 +1,80 @@
+import time
+import numpy
+import pytest
+from src.thread import ParallelProcessing, exceptions
+
+
+# >>>>>>>>>> Dummy Functions <<<<<<<<<< #
+def _dummy_dataProcessor(data_in: int, delay: float = 0) -> int:
+  time.sleep(delay)
+  return data_in
+
+def _dummy_raiseException(x: Exception, delay: float = 0):
+  time.sleep(delay)
+  raise x
+
+
+
+
+# >>>>>>>>>> General Use <<<<<<<<<< #
+def test_threadsScaleDown():
+  """This test is for testing if threads scale down `max_threads` when the dataset is lesser than the thread count"""
+  dataset = numpy.arange(0, 2).tolist()
+  new = ParallelProcessing(
+    function = _dummy_dataProcessor,
+    dataset = dataset,
+    max_threads = 4,
+    kwargs = { 'delay': 2 }
+  )
+  new.start()
+  assert len(new._threads) == 2
+
+def test_threadsProcessing():
+  """This test is for testing if threads correctly order data in the `dataset` arrangement"""
+  dataset = numpy.arange(0, 500).tolist()
+  new = ParallelProcessing(
+    function = _dummy_dataProcessor,
+    dataset = dataset,
+    args = [0.001]
+  )
+  new.start()
+  assert new.get_return_values() == dataset
+
+
+
+
+# >>>>>>>>>> Raising Exceptions <<<<<<<<<< #
+def test_raises_notInitializedError():
+  """This test should raise ThreadNotInitializedError"""
+  dataset = numpy.arange(0, 8).tolist()
+  new = ParallelProcessing(
+    function = _dummy_dataProcessor,
+    dataset = dataset
+  )
+
+  with pytest.raises(exceptions.ThreadNotInitializedError):
+    new.results
+
+def test_raises_StillRunningError():
+  """This test should raise ThreadStillRunningError"""
+  dataset = numpy.arange(0, 8).tolist()
+  new = ParallelProcessing(
+    function = _dummy_dataProcessor,
+    dataset = dataset,
+    args = [1]
+  )
+  new.start()
+  with pytest.raises(exceptions.ThreadStillRunningError):
+    new.results
+
+def test_raises_RunTimeError():
+  """This test should raise a RunTimeError"""
+  dataset = [RuntimeError()] * 8
+  new = ParallelProcessing(
+    function = _dummy_raiseException,
+    dataset = dataset,
+    args = [0.01]
+  )
+  with pytest.raises(RuntimeError):
+    new.start()
+    new.join()

tests/test_placeholder.py

@@ -0,0 +1,116 @@
+import time
+import pytest
+from src.thread import Thread, exceptions
+
+
+# >>>>>>>>>> Dummy Functions <<<<<<<<<< #
+def _dummy_target_raiseToPower(x: float, power: float, delay: float = 0):
+  time.sleep(delay)
+  return x**power
+
+def _dummy_raiseException(x: Exception, delay: float = 0):
+  time.sleep(delay)
+  raise x
+
+
+
+
+# >>>>>>>>>> General Use <<<<<<<<<< #
+def test_threadCreation():
+  """This test is for testing parsing of args and kwargs and `.join()` method"""
+  new = Thread(
+    target = _dummy_target_raiseToPower,
+    args = [4],
+    kwargs = { 'power': 2 }
+  )
+  new.start()
+  s = new.join()
+  assert s.status == 'Thread terminated'
+  assert new.result == 16
+
+def test_threadingThreadParsing():
+  """This test is for testing parsing arguments to `threading.Thead`"""
+  new = Thread(
+    target = _dummy_target_raiseToPower,
+    args = [4, 2, 5],
+    name = 'testingThread',
+    daemon = True
+  )
+  new.start()
+  assert new._thread and new._thread.name == 'testingThread'
+
+def test_suppressAll():
+  """This test is for testing that errors are suppressed properly"""
+  new = Thread(
+    target = _dummy_raiseException,
+    args = [ValueError()],
+    suppress_errors = True,
+    daemon = True
+  )
+  new.start()
+  new.join()
+  assert len(new.errors) == 1
+  assert isinstance(new.errors[0], ValueError)
+
+def test_ignoreSpecificError():
+  """This test is for testing that specific errors are ignored properly"""
+  new = Thread(
+    target = _dummy_raiseException,
+    args = [ValueError()],
+    ignore_errors = [ValueError],
+  )
+  new.start()
+  new.join()
+  assert len(new.errors) == 0
+
+
+
+
+# >>>>>>>>>> Raising Exceptions <<<<<<<<<< #
+def test_raises_notInitializedError():
+  """This test should raise ThreadNotInitializedError"""
+  new = Thread(
+    target = _dummy_target_raiseToPower,
+    args = [4, 2]
+  )
+
+  with pytest.raises(exceptions.ThreadNotInitializedError):
+    new.result
+
+def test_raises_stillRunningError():
+  """This test should raise ThreadStillRunningError"""
+  new = Thread(
+    target = _dummy_target_raiseToPower,
+    args = [4, 2, 5]
+  )
+  new.start()
+
+  with pytest.raises(exceptions.ThreadStillRunningError):
+    new.result
+
+def test_raises_ignoreSpecificError():
+  """This test is for testing that non-specified errors are not ignored"""
+  new = Thread(
+    target = _dummy_raiseException,
+    args = [FileExistsError()],
+    ignore_errors = [ValueError],
+    suppress_errors = False
+  )
+  with pytest.raises(FileExistsError):
+    new.start()
+    new.join()
+
+def test_raises_HookError():
+  """This test should raise """
+  new = Thread(
+    target = _dummy_target_raiseToPower,
+    args = [4, 2]
+  )
+
+  def newhook(x: int):
+    raise RuntimeError()
+  new.add_hook(newhook)
+
+  with pytest.raises(exceptions.HookRuntimeError):
+    new.start()
+    new.join()