chore: add uvicorn server for tests (#1092)

### Description

- Add uvicorn server for testing instead of making requests to external
services
- Adapted all tests that made requests to any external sites, for
example `crawlee.dev`
- Adapted almost all tests using `httpbin`. ~~Excludes:~~
- ~~test_send_request_with_proxy_disabled[curl, httpx] - proxy-py,
instead of a `ProxyError` error, returns a 400 response when accessing
`uvicorn`. The test does not make a real request to `httpbin.org`~~
   - ~~test_proxy_disabled[curl, httpx] - same~~
   - ~~test_http_2[httpx] - `uvicorn` only supports HTTP/1.1~~
- The `httpbin` fixture has been removed. The remaining tests make
requests to `https://apify.com/. This is sufficient for testing, but it
is much more stable.

### Issues

- Closes: #929
- Closes: #743

Author: Mantisus

pyproject.toml

@@ -110,6 +110,7 @@ dev = [
     "types-colorama~=0.4.15.20240106",
     "types-psutil~=7.0.0.20250218",
     "types-python-dateutil~=2.9.0.20240316",
+    "uvicorn[standard]~=0.34.0",
 ]
 
 [tool.hatch.build.targets.wheel]

tests/__init__.py

@@ -10,49 +10,49 @@
     from yarl import URL
 
 
-async def test_default_plugin_new_page_creation(httpbin: URL) -> None:
+async def test_default_plugin_new_page_creation(server_url: URL) -> None:
     async with BrowserPool() as browser_pool:
         page_1 = await browser_pool.new_page()
-        await page_1.page.goto(str(httpbin / 'get'))
+        await page_1.page.goto(str(server_url))
         assert page_1.browser_type == 'chromium'
-        assert page_1.page.url == str(httpbin / 'get')
+        assert page_1.page.url == str(server_url)
         assert '<html' in await page_1.page.content()  # there is some HTML content
         assert browser_pool.total_pages_count == 1
 
         page_2 = await browser_pool.new_page()
-        await page_2.page.goto(str(httpbin / 'status/200'))
+        await page_2.page.goto(str(server_url / 'status/200'))
         assert page_2.browser_type == 'chromium'
-        assert page_2.page.url == str(httpbin / 'status/200')
+        assert page_2.page.url == str(server_url / 'status/200')
         assert '<html' in await page_1.page.content()  # there is some HTML content
         assert browser_pool.total_pages_count == 2
 
         await page_1.page.close()
         await page_2.page.close()
 
 
-async def test_multiple_plugins_new_page_creation(httpbin: URL) -> None:
+async def test_multiple_plugins_new_page_creation(server_url: URL) -> None:
     plugin_chromium = PlaywrightBrowserPlugin(browser_type='chromium')
     plugin_firefox = PlaywrightBrowserPlugin(browser_type='firefox')
 
     async with BrowserPool([plugin_chromium, plugin_firefox]) as browser_pool:
         assert browser_pool.plugins == [plugin_chromium, plugin_firefox]
 
         page_1 = await browser_pool.new_page()
-        await page_1.page.goto(str(httpbin / 'get'))
+        await page_1.page.goto(str(server_url))
         assert page_1.browser_type == 'chromium'
-        assert page_1.page.url == str(httpbin / 'get')
+        assert page_1.page.url == str(server_url)
         assert '<html' in await page_1.page.content()  # there is some HTML content
 
         page_2 = await browser_pool.new_page()
-        await page_2.page.goto(str(httpbin / 'headers'))
+        await page_2.page.goto(str(server_url / 'headers'))
         assert page_2.browser_type == 'firefox'
-        assert page_2.page.url == str(httpbin / 'headers')
+        assert page_2.page.url == str(server_url / 'headers')
         assert '<html' in await page_2.page.content()  # there is some HTML content
 
         page_3 = await browser_pool.new_page()
-        await page_3.page.goto(str(httpbin / 'user-agent'))
+        await page_3.page.goto(str(server_url / 'user-agent'))
         assert page_3.browser_type == 'chromium'
-        assert page_3.page.url == str(httpbin / 'user-agent')
+        assert page_3.page.url == str(server_url / 'user-agent')
         assert '<html' in await page_3.page.content()  # there is some HTML content
 
         await page_1.page.close()
@@ -62,7 +62,7 @@ async def test_multiple_plugins_new_page_creation(httpbin: URL) -> None:
         assert browser_pool.total_pages_count == 3
 
 
-async def test_new_page_with_each_plugin(httpbin: URL) -> None:
+async def test_new_page_with_each_plugin(server_url: URL) -> None:
     plugin_chromium = PlaywrightBrowserPlugin(browser_type='chromium')
     plugin_firefox = PlaywrightBrowserPlugin(browser_type='firefox')
 
@@ -74,12 +74,12 @@ async def test_new_page_with_each_plugin(httpbin: URL) -> None:
         assert pages[0].browser_type == 'chromium'
         assert pages[1].browser_type == 'firefox'
 
-        await pages[0].page.goto(str(httpbin / 'get'))
-        assert pages[0].page.url == str(httpbin / 'get')
+        await pages[0].page.goto(str(server_url))
+        assert pages[0].page.url == str(server_url)
         assert '<html' in await pages[0].page.content()  # there is some HTML content
 
-        await pages[1].page.goto(str(httpbin / 'headers'))
-        assert pages[1].page.url == str(httpbin / 'headers')
+        await pages[1].page.goto(str(server_url / 'headers'))
+        assert pages[1].page.url == str(server_url / 'headers')
         assert '<html' in await pages[1].page.content()
 
         for page in pages:
@@ -88,16 +88,16 @@ async def test_new_page_with_each_plugin(httpbin: URL) -> None:
         assert browser_pool.total_pages_count == 2
 
 
-async def test_with_default_plugin_constructor(httpbin: URL) -> None:
+async def test_with_default_plugin_constructor(server_url: URL) -> None:
     async with BrowserPool.with_default_plugin(headless=True, browser_type='firefox') as browser_pool:
         assert len(browser_pool.plugins) == 1
         assert isinstance(browser_pool.plugins[0], PlaywrightBrowserPlugin)
 
         page = await browser_pool.new_page()
         assert page.browser_type == 'firefox'
 
-        await page.page.goto(str(httpbin / 'get'))
-        assert page.page.url == str(httpbin / 'get')
+        await page.page.goto(str(server_url))
+        assert page.page.url == str(server_url)
         assert '<html' in await page.page.content()  # there is some HTML content
 
         await page.page.close()
@@ -119,13 +119,13 @@ async def test_new_page_with_invalid_plugin() -> None:
             await browser_pool.new_page(browser_plugin=plugin_2)
 
 
-async def test_resource_management(httpbin: URL) -> None:
+async def test_resource_management(server_url: URL) -> None:
     playwright_plugin = PlaywrightBrowserPlugin(browser_type='chromium')
 
     async with BrowserPool([playwright_plugin]) as browser_pool:
         page = await browser_pool.new_page()
-        await page.page.goto(str(httpbin / 'get'))
-        assert page.page.url == str(httpbin / 'get')
+        await page.page.goto(str(server_url))
+        assert page.page.url == str(server_url)
         assert '<html' in await page.page.content()  # there is some HTML content
         assert browser_pool.total_pages_count == 1
 
@@ -153,10 +153,10 @@ async def test_methods_raise_error_when_not_active() -> None:
         assert browser_pool.active is True
 
 
-async def test_with_plugin_contains_page_options(httpbin: URL) -> None:
+async def test_with_plugin_contains_page_options(server_url: URL) -> None:
     plugin = PlaywrightBrowserPlugin(browser_new_context_options={'user_agent': 'My Best User-Agent'})
     async with BrowserPool(plugins=[plugin]) as browser_pool:
         test_page = await browser_pool.new_page()
-        await test_page.page.goto(str(httpbin / 'user-agent'))
+        await test_page.page.goto(str(server_url / 'user-agent'))
         assert 'My Best User-Agent' in await test_page.page.content()
         await test_page.page.close()

tests/unit/__init__.py

@@ -46,9 +46,9 @@ async def test_initial_state(browser: Browser) -> None:
     assert controller.has_free_capacity
 
 
-async def test_open_and_close_page(controller: PlaywrightBrowserController, httpbin: URL) -> None:
+async def test_open_and_close_page(controller: PlaywrightBrowserController, server_url: URL) -> None:
     page = await controller.new_page()
-    await page.goto(str(httpbin))
+    await page.goto(str(server_url))
 
     assert page in controller.pages
     assert controller.pages_count == 1

tests/unit/browsers/test_browser_pool.py

@@ -34,13 +34,13 @@ async def test_initial_state() -> None:
     assert plugin.max_open_pages_per_browser == 10
 
 
-async def test_new_browser(plugin: PlaywrightBrowserPlugin, httpbin: URL) -> None:
+async def test_new_browser(plugin: PlaywrightBrowserPlugin, server_url: URL) -> None:
     browser_controller = await plugin.new_browser()
 
     assert browser_controller.is_browser_connected
 
     page = await browser_controller.new_page()
-    await page.goto(str(httpbin))
+    await page.goto(str(server_url))
 
     await page.close()
     await browser_controller.close()

tests/unit/browsers/test_playwright_browser_controller.py

@@ -5,23 +5,26 @@
 
 import logging
 import os
-from typing import TYPE_CHECKING, Any, Callable, Optional, cast
+from typing import TYPE_CHECKING, Callable, Optional, cast
 
 import pytest
 from proxy import Proxy
-from yarl import URL
+from uvicorn.config import Config
 
 from crawlee import service_locator
 from crawlee.configuration import Configuration
 from crawlee.fingerprint_suite._browserforge_adapter import get_available_header_network
 from crawlee.proxy_configuration import ProxyInfo
 from crawlee.storage_clients import MemoryStorageClient
 from crawlee.storages import KeyValueStore, _creation_management
+from tests.unit.server import TestServer, app, serve_in_thread
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncGenerator
+    from collections.abc import AsyncGenerator, Iterator
     from pathlib import Path
 
+    from yarl import URL
+
 
 @pytest.fixture
 def prepare_test_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Callable[[], None]:
@@ -96,41 +99,6 @@ def _set_crawler_log_level(pytestconfig: pytest.Config, monkeypatch: pytest.Monk
         monkeypatch.setattr(_log_config, 'get_configured_log_level', lambda: getattr(logging, loglevel.upper()))
 
 
-@pytest.fixture
-def httpbin() -> URL:
-    class URLWrapper:
-        def __init__(self, url: URL) -> None:
-            self.url = url
-
-        def __getattr__(self, name: str) -> Any:
-            result = getattr(self.url, name)
-            return_type = getattr(result, '__annotations__', {}).get('return', None)
-
-            if return_type == 'URL':
-
-                def wrapper(*args: Any, **kwargs: Any) -> URLWrapper:
-                    return URLWrapper(result(*args, **kwargs))
-
-                return wrapper
-
-            return result
-
-        def with_path(
-            self, path: str, *, keep_query: bool = True, keep_fragment: bool = True, encoded: bool = False
-        ) -> URLWrapper:
-            return URLWrapper(
-                URL.with_path(self.url, path, keep_query=keep_query, keep_fragment=keep_fragment, encoded=encoded)
-            )
-
-        def __truediv__(self, other: Any) -> URLWrapper:
-            return self.with_path(other)
-
-        def __str__(self) -> str:
-            return str(self.url)
-
-    return cast('URL', URLWrapper(URL(os.environ.get('HTTPBIN_URL', 'https://httpbin.org'))))
-
-
 @pytest.fixture
 async def proxy_info(unused_tcp_port: int) -> ProxyInfo:
     username = 'user'
@@ -199,3 +167,32 @@ async def key_value_store() -> AsyncGenerator[KeyValueStore, None]:
     kvs = await KeyValueStore.open()
     yield kvs
     await kvs.drop()
+
+
+@pytest.fixture(scope='session')
+def http_server(unused_tcp_port_factory: Callable[[], int]) -> Iterator[TestServer]:
+    """Create and start an HTTP test server."""
+    config = Config(app=app, lifespan='off', loop='asyncio', port=unused_tcp_port_factory())
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+
+
+@pytest.fixture(scope='session')
+def server_url(http_server: TestServer) -> URL:
+    """Provide the base URL of the test server."""
+    return http_server.url
+
+
+# It is needed only in some tests, so we use the standard `scope=function`
+@pytest.fixture
+def redirect_http_server(unused_tcp_port_factory: Callable[[], int]) -> Iterator[TestServer]:
+    """Create and start an HTTP test server."""
+    config = Config(app=app, lifespan='off', loop='asyncio', port=unused_tcp_port_factory())
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+
+
+@pytest.fixture
+def redirect_server_url(redirect_http_server: TestServer) -> URL:
+    """Provide the base URL of the test server."""
+    return redirect_http_server.url