From efb3270c670d734333da7e63e63666135b514ecf Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Mon, 16 Dec 2024 00:47:02 +0200 Subject: [PATCH] style: Rewrite asyncFetcher tests to a cleaner version --- tests/fetchers/async/test_async_httpx.py | 151 +++++++++++------------ 1 file changed, 71 insertions(+), 80 deletions(-) diff --git a/tests/fetchers/async/test_async_httpx.py b/tests/fetchers/async/test_async_httpx.py index ef09bb8..67cc037 100644 --- a/tests/fetchers/async/test_async_httpx.py +++ b/tests/fetchers/async/test_async_httpx.py @@ -1,92 +1,83 @@ -import asyncio -import unittest - +import pytest import pytest_httpbin from scrapling.fetchers import AsyncFetcher @pytest_httpbin.use_class_based_httpbin -class TestAsyncFetcher(unittest.TestCase): - def setUp(self): - self.fetcher = AsyncFetcher(auto_match=True) - url = self.httpbin.url - self.status_200 = f'{url}/status/200' - self.status_404 = f'{url}/status/404' - self.status_501 = f'{url}/status/501' - self.basic_url = f'{url}/get' - self.post_url = f'{url}/post' - self.put_url = f'{url}/put' - self.delete_url = f'{url}/delete' - self.html_url = f'{url}/html' - - async def async_test(self, coro): - return await coro - - def test_basic_get(self): +@pytest.mark.asyncio +class TestAsyncFetcher: + @pytest.fixture(scope="class") + def fetcher(self): + return AsyncFetcher(auto_match=True) + + @pytest.fixture(scope="class") + def urls(self, httpbin): + return { + 'status_200': f'{httpbin.url}/status/200', + 'status_404': f'{httpbin.url}/status/404', + 'status_501': f'{httpbin.url}/status/501', + 'basic_url': f'{httpbin.url}/get', + 'post_url': f'{httpbin.url}/post', + 'put_url': f'{httpbin.url}/put', + 'delete_url': f'{httpbin.url}/delete', + 'html_url': f'{httpbin.url}/html' + } + + async def test_basic_get(self, fetcher, urls): """Test doing basic get request with multiple statuses""" + assert (await fetcher.get(urls['status_200'])).status == 200 + assert (await fetcher.get(urls['status_404'])).status == 404 + assert (await fetcher.get(urls['status_501'])).status == 501 - async def run_tests(): - self.assertEqual((await self.fetcher.get(self.status_200)).status, 200) - self.assertEqual((await self.fetcher.get(self.status_404)).status, 404) - self.assertEqual((await self.fetcher.get(self.status_501)).status, 501) - - asyncio.run(run_tests()) - - def test_get_properties(self): + async def test_get_properties(self, fetcher, urls): """Test if different arguments with GET request breaks the code or not""" - - async def run_tests(): - self.assertEqual((await self.fetcher.get(self.status_200, stealthy_headers=True)).status, 200) - self.assertEqual((await self.fetcher.get(self.status_200, follow_redirects=True)).status, 200) - self.assertEqual((await self.fetcher.get(self.status_200, timeout=None)).status, 200) - self.assertEqual( - (await self.fetcher.get(self.status_200, stealthy_headers=True, follow_redirects=True, timeout=None)).status, - 200 - ) - - asyncio.run(run_tests()) - - def test_post_properties(self): + assert (await fetcher.get(urls['status_200'], stealthy_headers=True)).status == 200 + assert (await fetcher.get(urls['status_200'], follow_redirects=True)).status == 200 + assert (await fetcher.get(urls['status_200'], timeout=None)).status == 200 + assert (await fetcher.get( + urls['status_200'], + stealthy_headers=True, + follow_redirects=True, + timeout=None + )).status == 200 + + async def test_post_properties(self, fetcher, urls): """Test if different arguments with POST request breaks the code or not""" - - async def run_tests(): - self.assertEqual((await self.fetcher.post(self.post_url, data={'key': 'value'})).status, 200) - self.assertEqual((await self.fetcher.post(self.post_url, data={'key': 'value'}, stealthy_headers=True)).status, 200) - self.assertEqual((await self.fetcher.post(self.post_url, data={'key': 'value'}, follow_redirects=True)).status, 200) - self.assertEqual((await self.fetcher.post(self.post_url, data={'key': 'value'}, timeout=None)).status, 200) - self.assertEqual( - (await self.fetcher.post(self.post_url, data={'key': 'value'}, stealthy_headers=True, follow_redirects=True, timeout=None)).status, - 200 - ) - - asyncio.run(run_tests()) - - def test_put_properties(self): + assert (await fetcher.post(urls['post_url'], data={'key': 'value'})).status == 200 + assert (await fetcher.post(urls['post_url'], data={'key': 'value'}, stealthy_headers=True)).status == 200 + assert (await fetcher.post(urls['post_url'], data={'key': 'value'}, follow_redirects=True)).status == 200 + assert (await fetcher.post(urls['post_url'], data={'key': 'value'}, timeout=None)).status == 200 + assert (await fetcher.post( + urls['post_url'], + data={'key': 'value'}, + stealthy_headers=True, + follow_redirects=True, + timeout=None + )).status == 200 + + async def test_put_properties(self, fetcher, urls): """Test if different arguments with PUT request breaks the code or not""" - - async def run_tests(): - self.assertIn((await self.fetcher.put(self.put_url, data={'key': 'value'})).status, [200, 405]) - self.assertIn((await self.fetcher.put(self.put_url, data={'key': 'value'}, stealthy_headers=True)).status, [200, 405]) - self.assertIn((await self.fetcher.put(self.put_url, data={'key': 'value'}, follow_redirects=True)).status, [200, 405]) - self.assertIn((await self.fetcher.put(self.put_url, data={'key': 'value'}, timeout=None)).status, [200, 405]) - self.assertIn( - (await self.fetcher.put(self.put_url, data={'key': 'value'}, stealthy_headers=True, follow_redirects=True, timeout=None)).status, - [200, 405] - ) - - asyncio.run(run_tests()) - - def test_delete_properties(self): + assert (await fetcher.put(urls['put_url'], data={'key': 'value'})).status in [200, 405] + assert (await fetcher.put(urls['put_url'], data={'key': 'value'}, stealthy_headers=True)).status in [200, 405] + assert (await fetcher.put(urls['put_url'], data={'key': 'value'}, follow_redirects=True)).status in [200, 405] + assert (await fetcher.put(urls['put_url'], data={'key': 'value'}, timeout=None)).status in [200, 405] + assert (await fetcher.put( + urls['put_url'], + data={'key': 'value'}, + stealthy_headers=True, + follow_redirects=True, + timeout=None + )).status in [200, 405] + + async def test_delete_properties(self, fetcher, urls): """Test if different arguments with DELETE request breaks the code or not""" - - async def run_tests(): - self.assertEqual((await self.fetcher.delete(self.delete_url, stealthy_headers=True)).status, 200) - self.assertEqual((await self.fetcher.delete(self.delete_url, follow_redirects=True)).status, 200) - self.assertEqual((await self.fetcher.delete(self.delete_url, timeout=None)).status, 200) - self.assertEqual( - (await self.fetcher.delete(self.delete_url, stealthy_headers=True, follow_redirects=True, timeout=None)).status, - 200 - ) - - asyncio.run(run_tests()) + assert (await fetcher.delete(urls['delete_url'], stealthy_headers=True)).status == 200 + assert (await fetcher.delete(urls['delete_url'], follow_redirects=True)).status == 200 + assert (await fetcher.delete(urls['delete_url'], timeout=None)).status == 200 + assert (await fetcher.delete( + urls['delete_url'], + stealthy_headers=True, + follow_redirects=True, + timeout=None + )).status == 200