testing notes in readme

README.md

@@ -14,6 +14,11 @@ See [USAGE.md](usage/USAGE.md)
 
 Python Version 3.8+, tested on Linux
 
+## Testing
+
+Install `dev` dependencies or just manually install `coverage` with `pip`. 
+Run the [`coverage.sh`](coverage.sh) shell script to execute all unit tests and receive the coverage report.
+
 ## Building from source
 
 Run `python -m build`

tests/test_helpers.py

@@ -0,0 +1,67 @@
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import MagicMock, AsyncMock, NonCallableMagicMock
+
+from asyncio_taskpool import helpers
+
+
+class HelpersTestCase(IsolatedAsyncioTestCase):
+
+    async def test_execute_optional(self):
+        f, args, kwargs = NonCallableMagicMock(), [1, 2], None
+        a = [f, args, kwargs]  # to avoid IDE nagging
+        self.assertIsNone(await helpers.execute_optional(*a))
+
+        expected_output = 'foo'
+        f = MagicMock(return_value=expected_output)
+        output = await helpers.execute_optional(f, args, kwargs)
+        self.assertEqual(expected_output, output)
+        f.assert_called_once_with(*args)
+
+        f.reset_mock()
+
+        kwargs = {'a': 100, 'b': 200}
+        output = await helpers.execute_optional(f, args, kwargs)
+        self.assertEqual(expected_output, output)
+        f.assert_called_once_with(*args, **kwargs)
+
+        f = AsyncMock(return_value=expected_output)
+        output = await helpers.execute_optional(f, args, kwargs)
+        self.assertEqual(expected_output, output)
+        f.assert_awaited_once_with(*args, **kwargs)
+
+    def test_star_function(self):
+        expected_output = 'bar'
+        f = MagicMock(return_value=expected_output)
+        a = (1, 2, 3)
+        stars = 0
+        output = helpers.star_function(f, a, stars)
+        self.assertEqual(expected_output, output)
+        f.assert_called_once_with(a)
+
+        f.reset_mock()
+
+        stars = 1
+        output = helpers.star_function(f, a, stars)
+        self.assertEqual(expected_output, output)
+        f.assert_called_once_with(*a)
+
+        f.reset_mock()
+
+        a = {'a': 1, 'b': 2}
+        stars = 2
+        output = helpers.star_function(f, a, stars)
+        self.assertEqual(expected_output, output)
+        f.assert_called_once_with(**a)
+
+        with self.assertRaises(ValueError):
+            helpers.star_function(f, a, 3)
+        with self.assertRaises(ValueError):
+            helpers.star_function(f, a, -1)
+        with self.assertRaises(ValueError):
+            helpers.star_function(f, a, 123456789)
+
+    async def test_join_queue(self):
+        mock_join = AsyncMock()
+        mock_queue = MagicMock(join=mock_join)
+        self.assertIsNone(await helpers.join_queue(mock_queue))
+        mock_join.assert_awaited_once_with()

usage/USAGE.md

@@ -126,9 +126,8 @@ async def main() -> None:
     # The second one will start (with ID 3), only once there is room in the pool,
     # which -- in this example -- will be the case after ID 2 ends;
     # until then the `starmap` method call **will block**!
-    # Once there is room in the pool again, the third one will start immediately (and receive ID 4).
-    # The last one will start (with ID 5) **only** once there is room in the pool **and** no more than one of these 
-    # last four tasks is running.
+    # Once there is room in the pool again, the third and fourth will each start (with IDs 4 and 5)
+    # **only** once there is room in the pool **and** no more than one of these last four tasks is running.
     args_list = [(0, 10), (10, 20), (20, 30), (30, 40)]
     print("Calling `starmap`...")
     await pool.starmap(other_work, args_list, num_tasks=2)