Merge pull request #2 from aligheshlaghi97/feat/test

Feat/test

Author: aligheshlaghi97

docs/3. Handling CPU and IO Bound Tasks/ex_3_1.py

@@ -3,10 +3,16 @@
 import httpx
 
 
+async def fetch_data(url):
+    async with httpx.AsyncClient() as client:
+        response = await client.get(url)
+        return response
+
+
 async def main():
     t = time.time()
-    async with httpx.AsyncClient() as client:
-        response = await client.get('https://www.example.com/')
-        print(response)
+    response = await fetch_data('https://www.example.com/')
+    print(f"response is: {response}")
     print(f"it took {time.time() - t} s")
+
 asyncio.run(main())

docs/3. Handling CPU and IO Bound Tasks/ex_3_2.py

@@ -3,16 +3,21 @@
 import httpx
 
 
-async def main():
-    t = time.time()
+async def fetch_data(url):
     async with httpx.AsyncClient() as client:
-        task1 = asyncio.create_task(client.get('https://www.example.com/'))
-        task2 = asyncio.create_task(client.get('https://www.example.com/'))
-        task3 = asyncio.create_task(client.get('https://www.example.com/'))
+        task1 = asyncio.create_task(client.get(url))
+        task2 = asyncio.create_task(client.get(url))
+        task3 = asyncio.create_task(client.get(url))
         response1 = await task1
         response2 = await task2
         response3 = await task3
-        print(f'response1: {response1}, response2: {response2}, response3: {response3}')
+        return response1, response2, response3
+
+
+async def main():
+    t = time.time()
+    response1, response2, response3 = await fetch_data('https://www.example.com/')
+    print(f'response1: {response1}, response2: {response2}, response3: {response3}')
     print(f'It took {time.time() - t} s')
 
 asyncio.run(main())

docs/3. Handling CPU and IO Bound Tasks/ex_3_3.py

@@ -3,13 +3,17 @@
 import httpx
 
 
-async def main():
-    t = time.time()
-    url = 'https://www.example.com/'
+async def fetch_data(url):
     async with httpx.AsyncClient() as client:
         tasks = [client.get(url) for _ in range(3)]
         obj = await asyncio.gather(*tasks)
-        print(f'obj: {obj}, obj type: {type(obj)}')
+        return obj
+
+
+async def main():
+    t = time.time()
+    obj = await fetch_data('https://www.example.com/')
+    print(f'obj: {obj}, obj type: {type(obj)}')
     print(f'It took {time.time() - t} s')
 
 asyncio.run(main())

docs/3. Handling CPU and IO Bound Tasks/ex_3_4.py

@@ -2,10 +2,20 @@
 import httpx
 
 
-url = 'https://www.example.org/'
-t = time.time()
-response1 = httpx.get(url)
-response2 = httpx.get(url)
-response3 = httpx.get(url)
-print(f'It took {time.time() - t} s')
-print(f'response1: {response1}, response2: {response2}, response3: {response3}')
+def fetch_data(url):
+    response1 = httpx.get(url)
+    response2 = httpx.get(url)
+    response3 = httpx.get(url)
+    return response1, response2, response3
+
+
+def main():
+    url = 'https://www.example.org/'
+    t = time.time()
+    response1, response2, response3 = fetch_data(url)
+    print(f'It took {time.time() - t} s')
+    print(f'response1: {response1}, response2: {response2}, response3: {response3}')
+
+
+if __name__ == '__main__':
+    main()

docs/3. Handling CPU and IO Bound Tasks/ex_3_5.py

@@ -3,24 +3,42 @@
 from functools import partial
 
 
-def cpu_bound_task(a: int, n: int) -> float:
+def cpu_bound_task(a: int, n: int) -> int | float:
     _sum = 0
     for number in range(n):
         _sum += a
     return _sum
 
 
-t = time.time()
-value = cpu_bound_task(2, 100000000)
-print(f'value: {value}')
-value = cpu_bound_task(2, 100000000)
-print(f'It took without multiprocessing {time.time() - t} s')
-print(f'value: {value}')
+def without_multiprocessing(step: int, value: int) -> tuple[int, int, float]:
+    t = time.time()
+    value1 = cpu_bound_task(step, value)
+    value2 = cpu_bound_task(step, value)
+    time_taken = time.time() - t
 
-cpu_bound_partial = partial(cpu_bound_task, 2)
+    return value1, value2, time_taken
 
-with Pool(2) as p:
-    t = time.time()
-    value = p.map(cpu_bound_partial, [100000000, 100000000])
-    print(f'It took with multiprocessing {time.time() - t} s')
-    print(f'value: {value}')
+
+def with_multiprocessing(step: int, value: int) -> tuple((int, int, float)):
+    cpu_bound_partial = partial(cpu_bound_task, step)
+
+    with Pool(2) as p:
+        t = time.time()
+        values = p.map(cpu_bound_partial, [value, value])
+        time_taken = time.time() - t
+
+    return *values, time_taken
+
+
+def main() -> None:
+    value1, value2, time_taken_without_mp = without_multiprocessing(step=2, value=100000000)
+    print(f'Without multiprocessing, value1: {value1}, value2: {value2}')
+    print(f'time taken: {time_taken_without_mp} s')
+    print('======================================')
+    value1, value2, time_taken_with_mp = with_multiprocessing(step=2, value=100000000)
+    print(f'With multiprocessing, value1: {value1}, value2: {value2}')
+    print(f'time taken: {time_taken_with_mp} s')
+
+
+if __name__ == '__main__':
+    main()

tests/test_ch2.py

@@ -1,15 +1,22 @@
 import sys
 import os
+import time
 import asyncio
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, AsyncMock
 
 sys.path.append(os.path.abspath('docs/2. Getting Started with asyncio'))
 from ex_2_1 import async_sleep_for_one_second as async_sleep_ex_1
 from ex_2_2 import async_sleep_for_one_second as async_sleep_ex_2
+from ex_2_3 import nested as nested_ex_3, main as main_ex_3
+from ex_2_4 import sleep_coro as sleep_coro_ex_4, main as main_ex_4
+from ex_2_5 import sleep_coro as sleep_coro_ex_5, main as main_ex_5
+from ex_2_6 import sleep_coro as sleep_coro_ex_6, main as main_ex_6
+from ex_2_7 import sleep_coro as sleep_coro_ex_7, main as main_ex_7
+from ex_2_8 import sleep_coro as sleep_coro_ex_8, main as main_ex_8
 
 
-class TestAsyncSleep(unittest.TestCase):
+class TestEx1AsyncSleep(unittest.TestCase):
     @patch('asyncio.sleep', return_value=None)
     def test_async_sleep_for_one_second(self, mock_sleep):
         with patch('builtins.print') as mock_print:
@@ -23,7 +30,7 @@ def test_async_sleep_for_one_second(self, mock_sleep):
             mock_sleep.assert_called_once_with(1)
 
 
-class TestAsyncSleep(unittest.TestCase):
+class TestEx2AsyncSleep(unittest.TestCase):
     @patch('asyncio.sleep', return_value=None)
     def test_async_sleep_for_one_second(self, mock_sleep):
         with patch('builtins.print') as mock_print:
@@ -38,5 +45,118 @@ def test_async_sleep_for_one_second(self, mock_sleep):
                 mock_sleep.assert_called_once_with(1)
 
 
+class TestEx3AsyncNested(unittest.IsolatedAsyncioTestCase):
+    async def test_nested(self):
+        result = await nested_ex_3()
+        self.assertEqual(result, 42)
+
+    def test_main(self):
+        with patch('builtins.print') as mock_print:
+            asyncio.run(main_ex_3())
+            mock_print.assert_any_call(42)
+
+
+class TestEx4AsyncSleep(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_sleep_async_for_ten_second(self, mock_sleep):
+        with patch('builtins.print') as mock_print:
+            await sleep_coro_ex_4(10)
+
+            mock_print.assert_any_call('Started sleeping for 10 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 10 seconds!')
+
+            mock_sleep.assert_called_once_with(10)
+
+    async def test_main_async_sleep_for_one_second(self):
+        with patch('builtins.print') as mock_print:
+            await main_ex_4()
+            mock_print.assert_any_call('Started sleeping for 1 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 1 seconds!')
+
+
+class TestEx5AsyncSleep(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_sleep_async_for_five_second(self, mock_sleep):
+        with patch('builtins.print') as mock_print:
+            await sleep_coro_ex_5(5)
+
+            mock_print.assert_any_call('Started sleeping for 5 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 5 seconds!')
+
+            mock_sleep.assert_called_once_with(5)
+
+    async def test_main_async_sleep_two_concurrent_one_sec_sleep_took_about_one_sec(self):
+        with patch('builtins.print') as mock_print:
+            start_time = time.time()
+            await main_ex_5()
+            self.assertLess(time.time() - start_time, 1.1)
+            mock_print.assert_any_call('Started sleeping for 1 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 1 seconds!')
+
+
+class TestEx6AsyncSleep(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_sleep_async_for_three_second(self, mock_sleep):
+        with patch('builtins.print') as mock_print:
+            await sleep_coro_ex_6(3)
+
+            mock_print.assert_any_call('Started sleeping for 3 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 3 seconds!')
+
+            mock_sleep.assert_called_once_with(3)
+
+    async def test_main_async_sleep_for_one_second(self):
+        with patch('builtins.print') as mock_print:
+            await main_ex_6()
+            mock_print.assert_any_call('Started sleeping for 1 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 1 seconds!')
+
+
+class TestEx7AsyncSleep(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_sleep_async_for_seven_second(self, mock_sleep):
+        with patch('builtins.print') as mock_print:
+            result = await sleep_coro_ex_7(7)
+            self.assertEqual(result, 7)
+            mock_print.assert_any_call('Started sleeping for 7 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 7 seconds!')
+
+            mock_sleep.assert_called_once_with(7)
+
+    async def test_main_async_sleep_two_concurrent_two_and_three_sec_sleep_took_about_three_sec(self):
+        with patch('builtins.print') as mock_print:
+            start_time = time.time()
+            await main_ex_7()
+            self.assertLess(time.time() - start_time, 3.1)
+
+            mock_print.assert_any_call('Before running task group!')
+
+            mock_print.assert_any_call('Started sleeping for 2 seconds!')
+            mock_print.assert_any_call('Started sleeping for 3 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 2 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 3 seconds!')
+
+            mock_print.assert_any_call('After running task group!')
+            mock_print.assert_any_call('task1: 2, task2: 3')
+
+
+class TestEx8AsyncSleep(unittest.IsolatedAsyncioTestCase):
+    @patch('asyncio.sleep', return_value=None)
+    async def test_sleep_async_for_four_second(self, mock_sleep):
+        with patch('builtins.print') as mock_print:
+            await sleep_coro_ex_8(4)
+
+            mock_print.assert_any_call('Started sleeping for 4 seconds!')
+            mock_print.assert_any_call('Finished sleeping for 4 seconds!')
+
+            mock_sleep.assert_called_once_with(4)
+
+    async def test_main_async_sleep_for_two_second_expect_timeout(self):
+        with patch('builtins.print') as mock_print:
+            await main_ex_8()
+            mock_print.assert_any_call('Started sleeping for 2 seconds!')
+            mock_print.assert_any_call('timeout!')
+
+
 if __name__ == '__main__':
     unittest.main()