Feat/test

docs/6. Exploring Python 3.12 Features/ex_6_2.py

@@ -7,7 +7,7 @@ async def light_coro():
 
 
 async def main():
-    print('Before running task group!')
+    print('Before running gather!')
     tasks_list = [light_coro() for _ in range(1000000)]
     time0 = time.time()
 

requirements.txt

@@ -1,2 +1,5 @@
 pytest~=8.3.3
 httpx~=0.27.2
+uvicorn~=0.30.2
+gunicorn~=19.9.0
+starlette==0.37.2

tests/test_ch3.py

@@ -163,8 +163,8 @@ def test_with_and_without_mp(self):
         self.assertEqual(value1_mp, value2_mp)
         self.assertEqual(value1_mp, self.value * self.step)
         self.assertEqual(value2_mp, self.value * self.step)
-        self.assertLessEqual(time_taken_mp * 1.7, time_taken,
-                             msg='time taken without mp is less than 1.7 times of with mp!')
+        self.assertLessEqual(time_taken_mp * 1.5, time_taken,
+                             msg='time taken without mp is less than 1.5 times of with mp!')
 
     def test_main(self):
         with patch('builtins.print') as mocked_print:

tests/test_ch4.py

@@ -0,0 +1,142 @@
+import sys
+import os
+import time
+import asyncio
+import unittest
+from unittest.mock import patch, AsyncMock, MagicMock
+
+sys.path.append(os.path.abspath('docs/4. Synchronization and Coordination'))
+import ex_4_1 as ex1
+import ex_4_2 as ex2
+import ex_4_3 as ex3
+import ex_4_4 as ex4
+import ex_4_5 as ex5
+
+
+class TestEx1AsyncTask(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_task(self, mock_sleep):
+        """Test task function."""
+        ex1.value = 0
+
+        asyncio.run(ex1.task())
+        self.assertEqual(ex1.value, 1)
+
+        asyncio.run(ex1.task())
+        self.assertEqual(ex1.value, 2)
+
+        mock_sleep.assert_called_with(0.01)
+
+    def test_main_success(self):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex1.main())
+            mocked_print.assert_any_call(1)
+
+
+class TestEx2AsyncTask(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_task(self, mock_sleep):
+        """Test task function."""
+        ex2.value = 0
+        lock = asyncio.Lock()
+        asyncio.run(ex2.task(lock))
+        self.assertEqual(ex2.value, 1)
+
+        asyncio.run(ex2.task(lock))
+        self.assertEqual(ex2.value, 2)
+
+        mock_sleep.assert_called_with(0.00)
+
+    def test_main_success(self):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex2.main())
+            mocked_print.assert_any_call(10000)
+
+
+class TestEx3Semaphore(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_limited_resource(self, mock_sleep):
+        """Test limited resource function."""
+        ex2.value = 0
+        lock = asyncio.Lock()
+        asyncio.run(ex2.task(lock))
+        self.assertEqual(ex2.value, 1)
+
+        asyncio.run(ex2.task(lock))
+        self.assertEqual(ex2.value, 2)
+
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex3.main())
+            mocked_print.assert_any_call("Accessing limited resource")
+            mocked_print.assert_any_call("Finished using limited resource")
+
+        mock_sleep.assert_called_with(1)
+
+    def test_main_success(self):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            time_start = time.time()
+            asyncio.run(ex3.main())
+            time_taken = time.time() - time_start
+
+            self.assertLess(time_taken, 3)
+            self.assertLess(2, time_taken)
+
+            mocked_print.assert_any_call("Accessing limited resource")
+            mocked_print.assert_any_call("Finished using limited resource")
+
+
+class TestEx4Barrier(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_example_barrier(self, mock_sleep):
+        """Test example barrier function."""
+        with patch('builtins.print') as mock_print:
+            asyncio.run(ex4.example_barrier())
+            mock_print.assert_any_call("barrier passed")
+            calls = mock_print.call_args_list
+
+            self.assertEqual(len(calls), 4)
+
+            printed_outputs = [call[0][0] for call in calls]
+            expected_output = [
+                '[filling, waiters:0/3]',
+                '[filling, waiters:0/3]',
+                'barrier passed',
+                '[filling, waiters:0/3]',
+            ]
+            for out_idx in range(3):
+                self.assertIn(expected_output[out_idx], str(printed_outputs[out_idx]))
+
+            mock_sleep.assert_called_with(0)
+
+
+class TestEx5EventWaiter(unittest.IsolatedAsyncioTestCase):
+    async def test_waiter(self):
+        """Test waiter(event) function."""
+        event = asyncio.Event()
+        with patch('builtins.print') as mock_print:
+            time_start = time.time()
+            waiter_task = asyncio.create_task(ex5.waiter(event))
+            event.set()
+            await waiter_task
+            time_taken = time.time() - time_start
+            self.assertLess(time_taken, 0.01)
+
+            mock_print.assert_any_call("waiting for it ...")
+            mock_print.assert_any_call("... got it!")
+
+    def test_main(self):
+        with patch('builtins.print') as mock_print:
+            time_start = time.time()
+            asyncio.run(ex5.main())
+            time_taken = time.time() - time_start
+            self.assertGreater(time_taken, 1)
+
+            mock_print.assert_any_call("waiting for it ...")
+            mock_print.assert_any_call("... got it!")
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_ch5.py

@@ -0,0 +1,173 @@
+import sys
+import os
+import time
+import asyncio
+import unittest
+from unittest.mock import patch, AsyncMock, MagicMock
+
+sys.path.append(os.path.abspath('docs/5. Advanced Techniques'))
+import ex_5_1 as ex1
+import ex_5_2 as ex2
+import ex_5_3 as ex3
+import ex_5_4 as ex4
+import ex_5_5 as ex5
+import ex_5_6 as ex6
+
+
+class TestEx1ExceptionPropagate(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_shorter_task(self, mock_sleep):
+        """Test shorter task function."""
+        with patch('builtins.print') as mocked_print:
+            with patch('ex_5_1.Exception') as mock_exception, self.assertRaises(Exception):
+                asyncio.run(ex1.shorter_task())
+                mock_exception.assert_called_once_with("Some exception happened!")
+            mocked_print.assert_called_once_with("Executing the task to raise an exception!")
+            mock_sleep.assert_called_once_with(0.1)
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_longer_task(self, mock_sleep):
+        """Test longer task function."""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex1.longer_task())
+            mocked_print.assert_any_call("Executing the task which will complete!")
+            mock_sleep.assert_called_once_with(1)
+            mocked_print.assert_any_call("longer_task is done!")
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_main(self, mock_sleep):
+        """Test main function."""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex1.main())
+            mocked_print.assert_any_call("Main coroutine started!")
+            mocked_print.assert_any_call("Executing the task to raise an exception!")
+            mocked_print.assert_any_call("Executing the task which will complete!")
+            mock_sleep.assert_any_call(1)
+            mock_sleep.assert_any_call(0.1)
+            mocked_print.assert_any_call("longer_task is done!")
+            mocked_print.assert_any_call("Exception: Some exception happened!")
+            mocked_print.assert_any_call("Main coroutine done!")
+
+
+class TestEx2ExceptionHandler(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_shorter_task(self, mock_sleep):
+        """Test shorter task function."""
+        with patch('builtins.print') as mocked_print:
+            with patch('ex_5_2.Exception') as mock_exception, self.assertRaises(Exception):
+                asyncio.run(ex2.shorter_task())
+                mock_exception.assert_called_once_with("Some exception happened!")
+            mocked_print.assert_called_once_with("Executing the task to raise an exception!")
+            mock_sleep.assert_called_once_with(0.01)
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_longer_task(self, mock_sleep):
+        """Test longer task function."""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex2.longer_task())
+            mocked_print.assert_any_call("Executing the task which will complete!")
+            mock_sleep.assert_called_once_with(1)
+            mocked_print.assert_any_call("longer_task is done!")
+
+    def test_exception_handler(self):
+        """Test exception handler."""
+        with patch('builtins.print') as mocked_print:
+            context = {"exception": "Test exception!"}
+            ex2.exception_handler(None, context)
+            mocked_print.assert_called_once_with("Exception: Test exception!")
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_main(self, mock_sleep):
+        """Test main function."""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex2.main())
+            mocked_print.assert_any_call("Main coroutine started!")
+            mocked_print.assert_any_call("Executing the task to raise an exception!")
+            mocked_print.assert_any_call("Executing the task which will complete!")
+            mock_sleep.assert_any_call(1)
+            mock_sleep.assert_any_call(0.01)
+            mocked_print.assert_any_call("longer_task is done!")
+            mocked_print.assert_any_call("Exception: Some exception happened!")
+            mocked_print.assert_any_call("Main coroutine done!")
+
+
+class TestEx3CancelTask(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_cancel_me(self, mock_sleep):
+        """Test cancel_me function."""
+        asyncio.run(ex3.cancel_me())
+        mock_sleep.assert_called_once_with(1)
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_main(self, mock_sleep):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex3.main())
+            mocked_print.assert_called_once_with("main(): cancel_me is cancelled now")
+            mock_sleep.assert_any_call(0.01)
+
+
+class TestEx4ChainTasks(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_task1(self, mock_sleep):
+        """Test task1 function."""
+        with patch('builtins.print') as mocked_print:
+            result = await ex4.task1()
+            mock_sleep.assert_called_once_with(1)
+            mocked_print.assert_called_once_with(">task1()")
+            self.assertEqual(result, 1)
+
+    @patch('asyncio.sleep', return_value=None)
+    async def test_task2(self, mock_sleep):
+        """Test task2 function."""
+        with patch('builtins.print') as mocked_print:
+            value = 1
+            await ex4.task2(value)
+            mock_sleep.assert_called_once_with(1)
+            mocked_print.assert_called_once_with(f">task2() got {value}")
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_main(self, mock_sleep):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex4.main())
+            mocked_print.assert_any_call(">task1()")
+            mocked_print.assert_any_call(f">task2() got 1")
+
+            mocked_print.assert_any_call("Main: chain is done")
+            mock_sleep.assert_any_call(1)
+            self.assertTrue(ex4.event.is_set())
+
+
+class TestEx5ProducerConsumer(unittest.TestCase):
+    @patch('asyncio.sleep', return_value=None)
+    def test_producer(self, mock_sleep):
+        """Test producer function"""
+        channel = asyncio.Queue()
+        asyncio.run(ex5.producer(channel))
+        self.assertEqual(channel.qsize(), 5)
+        mock_sleep.assert_any_call(1)
+
+    @patch('asyncio.sleep', return_value=None)
+    def test_main(self, mock_sleep):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex5.main())
+            mocked_print.assert_any_call("Done!")
+            for num in range(5):
+                mocked_print.assert_any_call(f"Got number {num}")
+            mock_sleep.assert_any_call(1)
+
+
+class TestEx6Future(unittest.TestCase):
+    def test_main(self):
+        """Test main function"""
+        with patch('builtins.print') as mocked_print:
+            asyncio.run(ex6.main())
+            mocked_print.assert_any_call("future status is done: False")
+            mocked_print.assert_any_call("future status is done: True, future result: 10")
+            mocked_print.assert_any_call("future result after being awaited: 10")
+
+
+if __name__ == '__main__':
+    unittest.main()