From 8fd40839ee748b100fde875826d8dd349d853334 Mon Sep 17 00:00:00 2001 From: Daniil Fajnberg Date: Tue, 8 Feb 2022 16:40:13 +0100 Subject: [PATCH] full unittest coverage for the `helpers` module --- tests/test_helpers.py | 67 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 tests/test_helpers.py diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..6f2581a --- /dev/null +++ b/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()