test: ex_3_2 to ex_3_5 covered and finished ch3.

Author: aligheshlaghi97

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_ch3.py

@@ -3,10 +3,14 @@
 import time
 import asyncio
 import unittest
-from unittest.mock import patch, AsyncMock
+from unittest.mock import patch, AsyncMock, MagicMock
 
 sys.path.append(os.path.abspath('docs/3. Handling CPU and IO Bound Tasks'))
 from ex_3_1 import fetch_data as fetch_data_ex_1, main as main_ex_1
+from ex_3_2 import fetch_data as fetch_data_ex_2, main as main_ex_2
+from ex_3_3 import fetch_data as fetch_data_ex_3, main as main_ex_3
+from ex_3_4 import fetch_data as fetch_data_ex_4, main as main_ex_4
+from ex_3_5 import cpu_bound_task, without_multiprocessing, with_multiprocessing, main as main_ex_5
 
 
 class TestEx1AsyncHttpx(unittest.IsolatedAsyncioTestCase):
@@ -47,5 +51,129 @@ async def test_main_success(self, mock_fetch):
             mocked_print.assert_any_call(f"response is: {mock_fetch.return_value}")
 
 
+class TestEx2AsyncHttpx(unittest.IsolatedAsyncioTestCase):
+    @patch('httpx.AsyncClient.get')
+    async def test_fetch_data_success(self, mock_get):
+        """Test successful fetching of data."""
+        mock_response = mock_get.return_value
+        mock_response.status_code = 200
+        mock_response.text = 'Success'
+
+        url = 'https://www.example.com/'
+        response1, response2, response3 = await fetch_data_ex_2(url)
+        mock_get.assert_called_with(url)
+        assert mock_get.call_count == 3
+
+        self.assertEqual(response1.status_code, 200)
+        self.assertEqual(response2.status_code, 200)
+        self.assertEqual(response3.status_code, 200)
+        self.assertEqual(response1.text, 'Success')
+        self.assertEqual(response2.text, 'Success')
+        self.assertEqual(response3.text, 'Success')
+
+    @patch('ex_3_2.fetch_data', new_callable=AsyncMock)
+    async def test_main_success(self, mock_fetch):
+        """Test main function when fetch_data is successful."""
+        mock_fetch.return_value.status_code = 200
+        mock_fetch.return_value = '<Response [200 OK]>', '<Response [200 OK]>', '<Response [200 OK]>'
+
+        with patch('builtins.print') as mocked_print:
+            await main_ex_2()
+            mocked_print.assert_any_call(f"response1: {mock_fetch.return_value[0]}, "
+                                         f"response2: {mock_fetch.return_value[1]}, "
+                                         f"response3: {mock_fetch.return_value[2]}")
+
+
+class TestEx3AsyncHttpx(unittest.IsolatedAsyncioTestCase):
+    @patch('httpx.AsyncClient.get')
+    async def test_fetch_data_success(self, mock_get):
+        """Test successful fetching of data."""
+        mock_response = mock_get.return_value
+        mock_response.status_code = 200
+        mock_response.text = 'Success'
+
+        url = 'https://www.example.com/'
+        response1, response2, response3 = await fetch_data_ex_3(url)
+        mock_get.assert_called_with(url)
+        assert mock_get.call_count == 3
+
+        self.assertEqual(response1.status_code, 200)
+        self.assertEqual(response2.status_code, 200)
+        self.assertEqual(response3.status_code, 200)
+        self.assertEqual(response1.text, 'Success')
+        self.assertEqual(response2.text, 'Success')
+        self.assertEqual(response3.text, 'Success')
+
+    @patch('ex_3_3.fetch_data', new_callable=AsyncMock)
+    async def test_main_success(self, mock_fetch):
+        """Test main function when fetch_data is successful."""
+        mock_fetch.return_value.status_code = 200
+        mock_fetch.return_value = ['<Response [200 OK]>', '<Response [200 OK]>', '<Response [200 OK]>']
+
+        with patch('builtins.print') as mocked_print:
+            await main_ex_3()
+            mocked_print.assert_any_call(f'obj: {mock_fetch.return_value}, obj type: {type(mock_fetch.return_value)}')
+
+
+class TestEx4AsyncHttpx(unittest.TestCase):
+    @patch('httpx.get')
+    def test_fetch_data_success(self, mock_get):
+        """Test successful fetching of data."""
+        mock_response = mock_get.return_value
+        mock_response.status_code = 200
+        mock_response.text = 'Success'
+
+        url = 'https://www.example.com/'
+        response1, response2, response3 = fetch_data_ex_4(url)
+        mock_get.assert_called_with(url)
+        assert mock_get.call_count == 3
+
+        self.assertEqual(response1.status_code, 200)
+        self.assertEqual(response2.status_code, 200)
+        self.assertEqual(response3.status_code, 200)
+        self.assertEqual(response1.text, 'Success')
+        self.assertEqual(response2.text, 'Success')
+        self.assertEqual(response3.text, 'Success')
+
+    @patch('ex_3_4.fetch_data', new_callable=MagicMock)
+    def test_main_success(self, mock_fetch):
+        """Test main function when fetch_data is successful."""
+        mock_fetch.return_value.status_code = 200
+        mock_fetch.return_value = '<Response [200 OK]>', '<Response [200 OK]>', '<Response [200 OK]>'
+
+        with patch('builtins.print') as mocked_print:
+            main_ex_4()
+            mocked_print.assert_any_call(f"response1: {mock_fetch.return_value[0]}, "
+                                         f"response2: {mock_fetch.return_value[1]}, "
+                                         f"response3: {mock_fetch.return_value[2]}")
+
+
+class TestMultiprocessingPerformance(unittest.TestCase):
+
+    def setUp(self):
+        self.step = 2
+        self.value = 100000000
+
+    def test_with_and_without_mp(self):
+        value1, value2, time_taken = without_multiprocessing(self.step, self.value)
+        self.assertEqual(value1, value2)
+        self.assertEqual(value1, self.value * self.step)
+
+        value1_mp, value2_mp, time_taken_mp = with_multiprocessing(self.step, self.value)
+        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!')
+
+    def test_main(self):
+        with patch('builtins.print') as mocked_print:
+            main_ex_5()
+            value = self.value * self.step
+            mocked_print.assert_any_call(f'Without multiprocessing, value1: {value}, value2: {value}')
+            mocked_print.assert_any_call(f'======================================')
+            mocked_print.assert_any_call(f'With multiprocessing, value1: {value}, value2: {value}')
+
+
 if __name__ == '__main__':
     unittest.main()