initial commit

Author: krylosov-aa

.python-version

@@ -0,0 +1 @@
+3.13

Makefile

@@ -0,0 +1,24 @@
+lint:
+	ruff format .
+	ruff check --fix .
+	mypy .
+	flake8 .
+
+test:
+	pytest --cov async_pytest_httpserver tests --cov-report=term-missing
+
+test_fastapi:
+	pytest examples/fastapi_example/tests
+
+test_starlette:
+	pytest examples/starlette_example/tests
+
+uv:
+	uv sync
+	source .venv/bin/activate
+
+build:
+	uv build
+
+publish:
+	uv publish

async_pytest_httpserver/__init__.py

@@ -0,0 +1,8 @@
+from .web_service_mock import MockData, WebServiceMock
+from .fixtures import external_service_mock
+
+__all__ = [
+    "MockData",
+    "WebServiceMock",
+    "external_service_mock",
+]

async_pytest_httpserver/fixtures.py

@@ -0,0 +1,30 @@
+from typing import Callable
+
+import pytest_asyncio
+from aiohttp import web
+
+from .web_service_mock import MockData, WebServiceMock
+
+
+@pytest_asyncio.fixture
+async def external_service_mock(aiohttp_server):
+    """
+    Mock server for an external service
+
+    Returns:
+    function for adding an API to the server
+    """
+
+    async def _create_mock() -> tuple[
+        str, Callable[[MockData], list[dict[str, any]]]
+    ]:
+        app = web.Application()
+        web_service = WebServiceMock()
+        # The route catches requests for any path with any method.
+        # The web_service is responsible for actual routing, allowing new
+        # APIs to be added at server runtime.
+        app.router.add_route("*", "/{tail:.+}", web_service.handle)
+        server = await aiohttp_server(app)
+        return f"http://{server.host}:{server.port}", web_service.add_mock_data
+
+    return _create_mock

async_pytest_httpserver/web_service_mock.py

@@ -0,0 +1,76 @@
+from copy import deepcopy
+from dataclasses import dataclass
+from inspect import isawaitable
+from typing import Awaitable, Callable
+
+from aiohttp import web
+
+
+@dataclass
+class MockData:
+    method: str  # the method we replace
+    path: str  # the API path we are replacing
+    response: (
+        web.Response
+        | Callable[[web.Request], web.Response | Awaitable[web.Response]]
+    )
+
+
+class WebServiceMock:
+    """
+    A mock web service with a single API handle.
+        Intended use:
+        1. Start aiohttp_server with a universal route to the handle
+        2. Add real APIs via add_mock_data
+    """
+
+    def __init__(self):
+        self._mock_data: list[MockData] = []
+        self._call_info = {}
+
+    async def handle(self, request: web.Request) -> web.Response:
+        """
+        The method searches for a mock among the registered MockData,
+        stores the request information, and returns a mock response.
+        """
+        for mock in self._mock_data:
+            if (
+                mock.method.lower() == request.method.lower()
+                and mock.path == request.path
+            ):
+                await self._save_request(mock.method, mock.path, request)
+                if isinstance(mock.response, web.Response):
+                    return deepcopy(mock.response)
+                else:
+                    response = mock.response(request)
+                    if isawaitable(response):
+                        return await response
+                    else:
+                        return response
+
+        raise Exception(
+            f"Mock with method={request.method} "
+            f"and url={request.path} not found"
+        )
+
+    def add_mock_data(self, mock_data: MockData) -> list[dict[str, any]]:
+        """Saves a new mock and returns a reference to the call history"""
+        self._mock_data.append(mock_data)
+
+        url_data = self._call_info.get(mock_data.path) or {}
+        method_data = url_data.get(mock_data.method) or []
+        url_data[mock_data.method] = method_data
+        self._call_info[mock_data.path] = url_data
+        return self._call_info[mock_data.path][mock_data.method]
+
+    async def _save_request(
+        self, method: str, path: str, request: web.Request
+    ) -> None:
+        data = {"headers": request.headers}
+        if request.can_read_body:
+            if request.content_type == "application/json":
+                data["json"] = await request.json()
+            if request.content_type == "text/plain":
+                data["text"] = await request.text()
+
+        self._call_info[path][method].append(data)

mypy.ini

@@ -0,0 +1,3 @@
+[mypy]
+strict = true
+pretty = true

pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "async-pytest-httpserver"
+version = "0.1.0"
+description = "Async mock HTTP server for pytest, built on top of aiohttp."
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "aiohttp>=3.13.2",
+    "pytest>=9.0.2",
+    "pytest-aiohttp>=1.1.0",
+    "pytest-asyncio>=1.3.0",
+]
+
+[dependency-groups]
+dev = [
+    "mypy>=1.19.0",
+    "pytest-cov>=7.0.0",
+    "ruff>=0.14.8",
+    "wemake-python-styleguide>=1.4.0",
+]
+
+[project.entry-points.pytest11]
+my_fixtures = "async_pytest_httpserver.fixtures"

ruff.toml

@@ -0,0 +1,4 @@
+line-length = 79
+exclude = [
+    ".venv"
+]

setup.cfg

@@ -0,0 +1,7 @@
+[flake8]
+exclude = .venv
+;ignore=WPS476, WPS300, WPS410, WPS412, WPS501, WPS229, WPS202, WPS110, WPS473
+;    WPS217, WPS430, WPS420
+;
+;per-file-ignores =
+;    examples/*: WPS110, WPS432, WPS204, WPS114, WPS226, WPS201, WPS213