### Example Dockerfile for aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md A basic Dockerfile to set up a Python 3.11 environment, install aioresponses and testing dependencies, and run pytest. ```dockerfile FROM python:3.11-slim WORKDIR /app # Install aioresponses RUN pip install aioresponses # Install test dependencies RUN pip install pytest pytest-asyncio COPY . . # Run tests CMD ["pytest"] ``` -------------------------------- ### Basic aioresponses Example Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Demonstrates how to mock a GET request and retrieve a JSON payload using aiohttp and aioresponses. ```python import asyncio import aiohttp from aioresponses import aioresponses async def main(): with aioresponses() as m: # Register a mock response m.get('http://example.com/api', status=200, payload={'message': 'Hello'}) # Make a request async with aiohttp.ClientSession() as session: resp = await session.get('http://example.com/api') data = await resp.json() print(data) # {'message': 'Hello'} asyncio.run(main()) ``` -------------------------------- ### Manually Start Mocking with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Manually start the aioresponses mocking by calling the start() method. This is an alternative to using the context manager. ```python m = aioresponses() m.start() # Begin patching m.get('http://example.com') # ... make requests ... m.stop() # End patching ``` -------------------------------- ### Pyright Configuration Example Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Example configuration for pyright, a static type checker, to ensure compatibility with aioresponses. ```json // pyrightconfig.json { "pythonVersion": "3.10", "typeCheckingMode": "strict", "include": ["src"], "exclude": ["**/node_modules", "**/__pycache__"] } ``` -------------------------------- ### Install aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/README.rst Install the aioresponses package using pip. ```bash $ pip install aioresponses ``` -------------------------------- ### Mypy Configuration Example Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Example configuration for mypy, a static type checker, to ensure compatibility with aioresponses. ```ini # mypy.ini [mypy] python_version = 3.10 warn_return_any = True warn_unused_configs = True disallow_untyped_defs = False ``` -------------------------------- ### start() Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/utility-methods.md Manually starts mocking by initiating the patching process. This method is typically called automatically when using aioresponses as a context manager, but can be invoked manually for finer control over the mocking lifecycle. ```APIDOC ## start() ### Description Manually start mocking (begin patching). Usually called automatically by context manager. ### Method ```python def start(self) -> None: ``` ### Behavior: - Initializes `self._responses` as empty list - Initializes `self._matches` as empty dictionary - Calls `self.patcher.start()` to begin patching - Sets patcher return value to `self._request_mock` ### Example: ```python from aioresponses import aioresponses m = aioresponses() m.start() try: m.get('http://example.com', status=200) # Mocking is active async with aiohttp.ClientSession() as session: resp = await session.get('http://example.com') finally: m.stop() ``` ### When to Use: - Manual lifecycle management instead of context manager - Fine-grained control over patching duration - Testing framework that doesn't support context managers ``` -------------------------------- ### Set up local development environment Source: https://github.com/pnuckowski/aioresponses/blob/master/CONTRIBUTING.rst Install the project locally into a virtual environment using `python setup.py develop`. ```shell mkvirtualenv aioresponses cd aioresponses/ python setup.py develop ``` -------------------------------- ### Manually Start Mocking Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/utility-methods.md Manually starts mocking by initiating the patching process. This is an alternative to using the context manager for manual lifecycle control. ```python from aioresponses import aioresponses m = aioresponses() m.start() try: m.get('http://example.com', status=200) # Mocking is active async with aiohttp.ClientSession() as session: resp = await session.get('http://example.com') finally: m.stop() ``` -------------------------------- ### Mock a GET request and assert response Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/types.md Demonstrates mocking a GET request using `aioresponses` and asserting the response status and JSON payload with `aiohttp.ClientSession`. This example shows how to use `aioresponses` to simulate HTTP responses for testing. ```python with aioresponses() as m: m.get('http://example.com', status=200, payload={'foo': 'bar'}) async with aiohttp.ClientSession() as session: resp = await session.get('http://example.com') assert resp.status == 200 data = await resp.json() assert data == {'foo': 'bar'} ``` -------------------------------- ### Basic Setup with Context Manager Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Use aioresponses as a context manager for mocking HTTP requests within a block of code. This is the most common way to use the library. ```python # Basic setup from aioresponses import aioresponses, CallbackResult import aiohttp # Context manager with aioresponses() as m: m.get('http://example.com', status=200, payload={}) async with aiohttp.ClientSession() as session: resp = await session.get('http://example.com') ``` -------------------------------- ### Start Patcher Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/utility-methods.md Initiates the patching process using unittest.mock.patch. This is typically done at the beginning of a test to enable mocking. ```python self.patcher = patch('aiohttp.client.ClientSession._request', side_effect=self._request_mock, autospec=True) ``` -------------------------------- ### docker-compose Example for aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md A docker-compose configuration to build and run tests for an application using aioresponses. ```yaml version: '3.9' services: test: build: . image: test-aioresponses command: pytest -v volumes: - .:/app ``` -------------------------------- ### Mocking aiohttp ClientSession Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Demonstrates the basic usage of aioresponses with aiohttp.ClientSession to mock HTTP GET requests. ```python import aiohttp from aioresponses import aioresponses # Works with aiohttp ClientSession async with aiohttp.ClientSession() as session: with aioresponses() as m: m.get('http://example.com') resp = await session.get('http://example.com') ``` -------------------------------- ### Tracking State with CallbackResult Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/callback-result.md Shows how to manage and return stateful resources using CallbackResult, handling POST for creation and GET for retrieval. ```python class StateTracker: def __init__(self): self.resources = {} def handle_request(self, url, **kwargs): method = kwargs.get('method', 'GET').upper() if method == 'POST': resource_id = str(uuid.uuid4()) self.resources[resource_id] = {'created': True} return CallbackResult( status=201, payload={'id': resource_id}, headers={'Location': f'/resources/{resource_id}'} ) elif method == 'GET': # Return all resources return CallbackResult( status=200, payload=list(self.resources.values()) ) else: return CallbackResult(status=405, body='Method not allowed') tracker = StateTracker() with aioresponses() as m: m.add('http://example.com/resources', callback=tracker.handle_request, repeat=True) ``` -------------------------------- ### Specify aiohttp Version in Requirements Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Example of how to specify the aiohttp version requirement in a requirements.txt or setup.cfg file. ```toml # requirements.txt or setup.cfg aiohttp>=3.8,<4.0 ``` -------------------------------- ### Register GET Response Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Use the `get` method to register a mocked response for a specific URL when an HTTP GET request is made. You can specify status codes and payloads. ```python def get(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ```python m.get('http://example.com', status=200, payload={'data': 'value'}) ``` -------------------------------- ### Enter aioresponses Context Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Use `__enter__` to start mocking HTTP requests within a `with` block. The mocking is active only inside this block. ```python def __enter__(self) -> 'aioresponses': """Start mocking and return self.""" ``` ```python with aioresponses() as m: m.get('http://example.com') # Mocking is active here ``` -------------------------------- ### __enter__ and __exit__ Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md The __enter__ and __exit__ methods enable aioresponses to be used as a context manager. __enter__ starts the mocking process, and __exit__ stops it and cleans up resources. ```APIDOC ## Context Manager Methods ### `__enter__()` Enter the context manager. Starts mocking. ```python def __enter__(self) -> 'aioresponses': """Start mocking and return self.""" ``` **Returns:** `aioresponses` instance **Example:** ```python with aioresponses() as m: m.get('http://example.com') # Mocking is active here ``` ### `__exit__(exc_type, exc_val, exc_tb)` Exit the context manager. Stops mocking and cleans up. ```python def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: """Stop mocking, close responses, and clean up.""" ``` ``` -------------------------------- ### Print aioresponses Version Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md This snippet shows how to print the currently installed aioresponses version. ```python import aioresponses print(aioresponses.__version__) # e.g., "0.7.9" ``` -------------------------------- ### Internal Usage Example of RequestMatch.build_response() Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md An example demonstrating the internal call to `build_response` to create a `ClientResponse` object. This method is typically used internally by `aioresponses.match()` and not directly in user test code. ```python # This is called internally by aioresponses # Not typically used directly in test code from aioresponses.compat import URL match = RequestMatch( 'http://example.com', method='GET', status=200, payload={'data': 'test'} ) # Internally called as: url = URL('http://example.com') result = await match.build_response(url, headers={}, data=None) # result is a ClientResponse object ``` -------------------------------- ### Mock Simple GET Request with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/patterns-and-usage.md Use this pattern to mock a single GET request. Ensure aiohttp and aioresponses are imported. The mock is active within the 'with' block. ```python import aiohttp from aioresponses import aioresponses async def test_simple_get(): with aioresponses() as m: m.get('http://api.example.com/users', status=200, payload={'id': 1, 'name': 'John'}) async with aiohttp.ClientSession() as session: resp = await session.get('http://api.example.com/users') data = await resp.json() assert resp.status == 200 assert data == {'id': 1, 'name': 'John'} ``` -------------------------------- ### Mocking GET Request with Regex Source: https://github.com/pnuckowski/aioresponses/blob/master/README.rst Mocks a GET request to a URL matching a regular expression. Use this when the exact URL might vary but follows a pattern. ```python import asyncio import aiohttp import re from aioresponses import aioresponses @aioresponses() def test_regexp_example(m): loop = asyncio.get_event_loop() session = aiohttp.ClientSession() pattern = re.compile(r'^http://example\.com/api\?foo=.*$') m.get(pattern, status=200) resp = loop.run_until_complete(session.get('http://example.com/api?foo=bar')) assert resp.status == 200 ``` -------------------------------- ### Combine strategies for complex scenarios Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Integrate multiple configuration options like `param`, `passthrough`, and `passthrough_unmatched` within a single `aioresponses` context for sophisticated mocking setups. ```python with aioresponses( param='mocks', passthrough=['http://internal-db.example.com'], passthrough_unmatched=True ) as mocks: # Mock critical public API mocks.post('http://api.example.com/auth', status=200) # Internal requests go to real DB # Other requests pass through ``` -------------------------------- ### Accessing Positional Arguments (args) Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-call.md Demonstrates how to retrieve positional arguments passed to a mocked GET request. Typically, `args` is an empty tuple. ```python with aioresponses() as m: m.get('http://example.com') async with aiohttp.ClientSession() as session: await session.get('http://example.com') key = ('GET', URL('http://example.com')) request_call = m.requests[key][0] print(request_call.args) # () — Usually empty ``` -------------------------------- ### HTTP Method Shortcuts Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md aioresponses provides convenient shortcuts for registering mocked responses for common HTTP methods like GET, POST, PUT, PATCH, DELETE, HEAD, and OPTIONS. ```APIDOC ## HTTP Method Shortcuts ### `get(url, **kwargs)` Register a GET response. ```python def get(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` #### Parameters - **url** (`Union[URL, str, Pattern]`) - URL string, yarl.URL, or compiled regex pattern - **kwargs** (`Any`) - Additional options (see `add()` method) **Example:** ```python m.get('http://example.com', status=200, payload={'data': 'value'}) ``` ### `post(url, **kwargs)` Register a POST response. ```python def post(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` **Example:** ```python m.post('http://example.com/api', status=201, payload={'id': 123}) ``` ### `put(url, **kwargs)` Register a PUT response. ```python def put(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ### `patch(url, **kwargs)` Register a PATCH response. ```python def patch(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ### `delete(url, **kwargs)` Register a DELETE response. ```python def delete(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ### `head(url, **kwargs)` Register a HEAD response. ```python def head(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ### `options(url, **kwargs)` Register an OPTIONS response. ```python def options(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` ``` -------------------------------- ### Internal Usage Example of aioresponses.match() Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Demonstrates how the match() method is called internally during request processing. This is not typically used directly in test code. ```python # This is called internally during request processing # Not typically used directly in test code from aioresponses import aioresponses from aioresponses.compat import URL with aioresponses() as m: m.get('http://example.com', status=200) # When this request is made: # await session.get('http://example.com') # Internally, match() is called like: url = URL('http://example.com') response = await m.match('GET', url, allow_redirects=True) # response is the mocked ClientResponse ``` -------------------------------- ### Register Mocked Response with URL Pattern Matching Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Use a regular expression pattern with `re.compile` to match multiple URLs for a mocked response. This example matches URLs for user IDs. ```python import re m.add(re.compile(r'^http://api\.example\.com/users/\d+$'), status=200, payload={'id': 1}) ``` -------------------------------- ### Check aiohttp Version at Runtime Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Use this snippet to conditionally execute code based on the installed aiohttp version. It requires the 'packaging' library. ```python from aioresponses.compat import AIOHTTP_VERSION from packaging.version import Version if AIOHTTP_VERSION >= Version('3.9.0'): # Use 3.9+ features pass else: # Use 3.8 features pass ``` -------------------------------- ### Activate aioresponses using a context manager Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Use `aioresponses` as a context manager to automatically start and stop mocking. Mocking is active within the `with` block and inactive outside of it. ```python with aioresponses(param='m') as m: # Mocking is active here pass # Mocking is stopped here ``` -------------------------------- ### Test Multiple Services with Different Methods Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Mock requests to multiple distinct services using different HTTP methods (GET and POST). This demonstrates mocking interactions across different microservices. ```python async def test_multi_service(): with aioresponses() as m: m.get('http://user-service.com/users', payload=[{'id': 1}]) m.get('http://post-service.com/posts', payload=[{'id': 1, 'title': 'Hello'}]) m.post('http://notification-service.com/send', status=202) async with aiohttp.ClientSession() as session: users = await session.get('http://user-service.com/users') posts = await session.get('http://post-service.com/posts') notification = await session.post('http://notification-service.com/send') ``` -------------------------------- ### Initialize aioresponses with Default and Custom Configurations Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Demonstrates creating an aioresponses instance with default settings and with specific configuration parameters like 'param', 'passthrough', and 'passthrough_unmatched'. ```python from aioresponses import aioresponses # No configuration (defaults) m = aioresponses() # With configuration m = aioresponses( param='mock_instance', passthrough=['http://backend.example.com'], passthrough_unmatched=False ) ``` -------------------------------- ### Absolute URL Redirects with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Demonstrates handling absolute URL redirects. The `aioresponses` mock is configured with an initial GET request that returns a 301 status and a 'Location' header pointing to a new URL. A second mock is set up for the target URL. When `allow_redirects=True`, the client session follows the redirect, and the response object reflects the final URL and status. ```python with aioresponses() as m: # Initial request m.get('http://example.com/', headers={'Location': 'http://new.example.com/'}, status=301) # Redirect target m.get('http://new.example.com/', status=200, body='Final content') async with aiohttp.ClientSession() as session: # Following redirects resp = await session.get('http://example.com/', allow_redirects=True) # resp points to the final URL assert resp.url == 'http://new.example.com/' assert resp.status == 200 # resp.history contains the redirect assert len(resp.history) == 1 assert resp.history[0].status == 301 ``` -------------------------------- ### Mock GET Request with Status and Body Source: https://github.com/pnuckowski/aioresponses/blob/master/README.rst Use aioresponses as a decorator to mock a GET request, specifying the status code and response body. ```python import aiohttp import asyncio from aioresponses import aioresponses @aioresponses() def test_request(mocked): loop = asyncio.get_event_loop() mocked.get('http://example.com', status=200, body='test') session = aiohttp.ClientSession() resp = loop.run_until_complete(session.get('http://example.com')) assert resp.status == 200 mocked.assert_called_once_with('http://example.com') ``` -------------------------------- ### Test API Client with GET Request Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Mock a GET request to a specific API endpoint and assert the JSON response. Ensure aiohttp and aioresponses are imported. ```python from aioresponses import aioresponses import aiohttp async def test_api_client(): with aioresponses() as m: m.get('http://api.example.com/users/1', payload={'id': 1, 'name': 'John'}) async with aiohttp.ClientSession() as session: resp = await session.get('http://api.example.com/users/1') user = await resp.json() assert user['name'] == 'John' ``` -------------------------------- ### Implementing Content Negotiation with CallbackResult Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/callback-result.md Demonstrates how to return different response formats (JSON or XML) based on the 'Accept' header in the request. ```python def content_negotiation_callback(url, **kwargs): headers = kwargs.get('headers', {}) accept = headers.get('Accept', 'application/json') data = {'id': 1, 'name': 'Example'} if 'xml' in accept: return CallbackResult( status=200, body='1Example', content_type='application/xml' ) else: return CallbackResult( status=200, payload=data, content_type='application/json' ) with aioresponses() as m: m.add('http://example.com/data', callback=content_negotiation_callback) ``` -------------------------------- ### Use Exact URLs or Regex for Mocking Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md For simple cases, use exact URLs. For more complex patterns, use regular expressions. Ensure imports like 're' are available. ```python import re m.get('http://example.com/api') # Simple m.get(re.compile(r'^http://api\..*$')) # Pattern ``` -------------------------------- ### Using aiohttp ClientSession with Base URL Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Demonstrates how to use aiohttp's ClientSession with a base URL, requiring aiohttp version 3.8.0 or later. ```python # Works with aiohttp 3.8.0+ async with aiohttp.ClientSession(base_url='http://example.com') as session: resp = await session.get('/api') ``` -------------------------------- ### Match URLs with Regular Expressions Source: https://github.com/pnuckowski/aioresponses/blob/master/README.rst Demonstrates how to use regular expressions to match URLs for mocking responses. ```python import asyncio import aiohttp import re from aioresponses import aioresponses ``` -------------------------------- ### GitLab CI Configuration Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md A GitLab CI pipeline configuration for testing with multiple Python versions. It installs aioresponses and runs pytest. ```yaml image: python:3.11 stages: - test test: stage: test script: - pip install aioresponses pytest pytest-asyncio - pytest -v parallel: matrix: - PYTHON_VERSION: ['3.10', '3.11', '3.12'] ``` -------------------------------- ### Configure Passthrough with Prefix Matching Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Demonstrates using a URL prefix with aioresponses' 'passthrough' option to allow all requests under a specific path to make real HTTP calls, while other paths remain mocked. ```python # Passthrough all requests under /internal with aioresponses(passthrough=['http://example.com/internal']) as m: m.get('http://example.com/api/public', status=200) async with aiohttp.ClientSession() as session: # Mocked resp = await session.get('http://example.com/api/public') # Real request # resp = await session.get('http://example.com/internal/config') ``` -------------------------------- ### Standalone asyncio Usage Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Demonstrates how to use aioresponses with asyncio directly, without relying on a testing framework. ```python import asyncio from aioresponses import aioresponses async def test(): with aioresponses() as m: m.get('http://example.com') # ... asyncio.run(test()) ``` -------------------------------- ### Register mocks with different configurations using .add() parameters Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md The `.add()` method and HTTP method shortcuts accept parameters to configure mocks. This allows registering URLs with specific statuses, timeouts, or repetition counts. ```python with aioresponses() as m: # Register with different configurations m.get('http://success.com', status=200) m.post('http://error.com', status=500) m.put('http://timeout.com', timeout=True) m.delete('http://api.com', repeat=3) # Use 3 times ``` -------------------------------- ### Manual management of aioresponses lifecycle Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Manually control the lifecycle of `aioresponses` by calling `start()` and `stop()`. This provides explicit control over when mocking is active. ```python m = aioresponses(passthrough=['http://backend.com']) m.start() # Mocking is active try: m.get('http://api.com') # ... test code ... finally: m.stop() # Mocking is stopped ``` -------------------------------- ### Passthrough Specific URL Prefixes Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/errors.md Allow requests to URLs starting with a specified prefix to bypass mocking and make real HTTP requests. ```python # Solution 4: Passthrough specific prefix with aioresponses(passthrough=['http://api.internal.com']) as m: m.get('http://example.com') # api.internal.com makes real requests, example.com is mocked ``` -------------------------------- ### Registering Exact URL Matches Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-match.md Demonstrates how to register the same match using a string URL or a yarl.URL object. Query parameters are normalized for comparison. ```python m.add('http://example.com/api') m.add(URL('http://example.com/api')) # Query parameters are normalized for comparison m.add('http://example.com/api?foo=bar&x=1') m.add('http://example.com/api?x=1&foo=bar') # Same match ``` -------------------------------- ### ClientSession Base URL Support Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Demonstrates using ClientSession with a base URL, available in aiohttp version 3.8.0 and later. Requires importing Version from packaging.version and AIOHTTP_VERSION from aioresponses.compat. ```python from packaging.version import Version from aioresponses.compat import AIOHTTP_VERSION if AIOHTTP_VERSION >= Version('3.8.0'): # Base URL is supported async with aiohttp.ClientSession(base_url='http://example.com') as session: resp = await session.get('/api') # Full URL: http://example.com/api ``` -------------------------------- ### Mocking HTTP Methods Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md aioresponses supports mocking for all standard HTTP methods: GET, POST, PUT, PATCH, DELETE, HEAD, and OPTIONS. ```python # HTTP methods m.get(url, ...) # GET m.post(url, ...) # POST m.put(url, ...) # PUT m.patch(url, ...) # PATCH m.delete(url, ...) # DELETE m.head(url, ...) # HEAD m.options(url, ...) # OPTIONS ``` -------------------------------- ### Handling Timeouts with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Configure a timeout for a mocked GET request and demonstrate catching the `asyncio.TimeoutError`. This ensures that timeouts are correctly raised and handled. ```python import asyncio with aioresponses() as m: m.get('http://example.com', timeout=True) async with aiohttp.ClientSession() as session: try: await session.get('http://example.com') except asyncio.TimeoutError: print("Timeout handled") ``` -------------------------------- ### Match Priority (FIFO) Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Demonstrates how multiple matches for the same URL are handled in a First-In, First-Out (FIFO) order. The first registered match is used for the first request, and subsequent matches are used for subsequent requests. ```python with aioresponses() as m: # Register multiple matches for same URL m.get('http://example.com', status=500) # Matched first m.get('http://example.com', status=200) # Matched second async with aiohttp.ClientSession() as session: resp1 = await session.get('http://example.com') assert resp1.status == 500 resp2 = await session.get('http://example.com') assert resp2.status == 200 ``` -------------------------------- ### Get Total Request Count Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-call.md Calculates the total number of HTTP requests made during a test session. Ensure aiohttp and aioresponses are imported. ```python with aioresponses() as m: m.get('http://example.com', repeat=True) async with aiohttp.ClientSession() as session: for i in range(10): await session.get('http://example.com') # Total number of calls across all URLs total_calls = sum( len(call_list) for call_list in m.requests.values() ) print(f"Total requests: {total_calls}") # 10 ``` -------------------------------- ### Registering Regex Pattern Matches Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-match.md Shows how to use compiled regex patterns for flexible URL matching, including matching any user ID, any path under /api, or specific search query patterns. ```python import re # Match any user ID m.add(re.compile(r'^http://api\.example\.com/users/\d+$')) # Match any path under /api m.add(re.compile(r'^http://api\.example\.com/.*$')) # Match with query parameters m.add(re.compile(r'^http://example\.com/search\?.*q=.*$')) ``` -------------------------------- ### Registering URLs and Handling Methods in aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Register mock URLs and ensure the HTTP method matches the request. Use regex for flexible URL matching and `passthrough` for real network requests. ```python # 1. Register the URL m.get('http://example.com') # 2. Ensure method matches m.post('http://example.com') # Not GET # 3. Handle query parameters m.get('http://example.com?x=1&y=2') # Order matters unless using regex # 4. Use regex for flexible matching import re m.get(re.compile(r'^http://example\.com.*$')) # 5. Use passthrough for specific URLs with aioresponses(passthrough=['http://real.example.com']) as m: # real.example.com makes actual requests ``` -------------------------------- ### Clone the aioresponses repository Source: https://github.com/pnuckowski/aioresponses/blob/master/CONTRIBUTING.rst Clone your forked repository locally to begin development. ```shell git clone git@github.com:your_name_here/aioresponses.git ``` -------------------------------- ### Basic Assertions with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/patterns-and-usage.md Assert that specific HTTP GET and POST requests were made using aioresponses. Note that assert_called_once() counts unique URL+method combinations. ```python async def test_assertions(): with aioresponses() as m: m.get('http://api.example.com', status=200) m.post('http://api.example.com/data', status=201) async with aiohttp.ClientSession() as session: await session.get('http://api.example.com') await session.post('http://api.example.com/data') # Assert requests were made m.assert_called() m.assert_called_once() # Only 2 requests, but... # assert_called_once() counts unique URL+method combinations m.assert_any_call('http://api.example.com') m.assert_any_call('http://api.example.com/data', method='POST') ``` -------------------------------- ### Optimizing Request Matching with Regex Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md Demonstrates using regular expressions with `aioresponses` for faster request matching when dealing with a large number of similar patterns. ```python import re with aioresponses() as m: # Much faster than 10000 exact matches m.get(re.compile(r'^http://example\.com/endpoint\d+$')) ``` -------------------------------- ### Using Static Callback Functions Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Define and use a static callback function to generate mock responses. ```python def my_callback(url, **kwargs): return CallbackResult(status=200, payload={'url': str(url)}) m.add('http://example.com', callback=my_callback) ``` -------------------------------- ### Asserting Requests with Specific Parameters Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/patterns-and-usage.md Verify that an HTTP GET request was made with the exact query parameters specified. This is useful for ensuring correct data filtering or pagination. ```python async def test_assert_with_params(): with aioresponses() as m: m.get('http://api.example.com/search', status=200) async with aiohttp.ClientSession() as session: await session.get('http://api.example.com/search', params={'q': 'python', 'limit': 10}) # Assert the exact parameters were passed m.assert_called_with( 'http://api.example.com/search', params={'q': 'python', 'limit': 10} ) ``` -------------------------------- ### Check Request Parameters Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-call.md Verify that specific query parameters were sent with a GET request. This is crucial for ensuring that your client code correctly constructs URLs with the intended parameters. ```python with aioresponses() as m: m.get('http://api.example.com/search') async with aiohttp.ClientSession() as session: await session.get( 'http://api.example.com/search', params={'q': 'python', 'limit': 10} ) from aioresponses.compat import URL key = ('GET', URL('http://api.example.com/search?limit=10&q=python')) call = m.requests[key][0] # Check parameters params = call.kwargs.get('params', {}) assert params.get('q') == 'python' assert params.get('limit') == 10 ``` -------------------------------- ### Mixed Configuration with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/configuration.md Demonstrates using aioresponses as a context manager with specific parameters like 'param', 'passthrough', and 'passthrough_unmatched'. This configuration allows for selective mocking and passthrough of requests. ```python with aioresponses( param='mocks', passthrough=['http://internal-api.example.com'], passthrough_unmatched=True ) as mocks: # Register some mocks mocks.get('http://app.example.com/api', status=200) # Behaviors: # - http://app.example.com/api: Uses mock (status 200) # - http://internal-api.example.com/*: Real request (passthrough prefix) # - Any other URL: Real request (passthrough_unmatched=True) ``` -------------------------------- ### Get All Calls to a Specific URL Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-call.md Retrieve all recorded calls made to a particular URL. This is useful for verifying that a specific endpoint was hit the expected number of times and with the correct parameters. ```python with aioresponses() as m: m.get('http://example.com', repeat=True) async with aiohttp.ClientSession() as session: await session.get('http://example.com?page=1') await session.get('http://example.com?page=2') await session.get('http://example.com?page=3') from aioresponses.compat import URL # Get all calls to a URL key = ('GET', URL('http://example.com?page=1')) all_calls = m.requests[key] print(f"Total calls to {key[1]}: {len(all_calls)}") # 1 print(all_calls[0].kwargs.get('params')) # {'page': '1'} ``` -------------------------------- ### Import aioresponses and RequestMatch Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Import the necessary components from the aioresponses library. ```python from aioresponses import aioresponses from aioresponses.core import RequestMatch ``` -------------------------------- ### Mock JSON Response as Context Manager Source: https://github.com/pnuckowski/aioresponses/blob/master/README.rst Use aioresponses as a context manager to mock a GET request and return a JSON payload. This is useful for testing JSON API responses. ```python import asyncio import aiohttp from aioresponses import aioresponses def test_ctx(): loop = asyncio.get_event_loop() session = aiohttp.ClientSession() with aioresponses() as m: m.get('http://test.example.com', payload=dict(foo='bar')) resp = loop.run_until_complete(session.get('http://test.example.com')) data = loop.run_until_complete(resp.json()) assert dict(foo='bar') == data m.assert_called_once_with('http://test.example.com') ``` -------------------------------- ### Register OPTIONS Response Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Use the `options` method to register a mocked response for a specific URL when an HTTP OPTIONS request is made. This is used to describe the communication options for the target resource. ```python def options(self, url: 'Union[URL, str, Pattern]', **kwargs: Any) -> None: ``` -------------------------------- ### Use Pytest Fixtures for Reusable Mocks Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Define pytest fixtures to manage the lifecycle of `aioresponses` mocks, making them easily reusable across multiple tests. Ensure `pytest` is installed. ```python import pytest from aioresponses import aioresponses @pytest.fixture def mock_api(): with aioresponses() as m: yield m async def test_something(mock_api): mock_api.get('http://api.example.com') ``` -------------------------------- ### Exception Handling in match() Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Shows how to configure a mock response to raise a specific exception when matched. This is useful for testing error handling in your application. ```python from aiohttp import ClientConnectionError with aioresponses() as m: m.get('http://example.com', exception=ClientConnectionError('Network error')) async with aiohttp.ClientSession() as session: try: resp = await session.get('http://example.com') except ClientConnectionError as e: print(f"Got expected error: {e}") ``` -------------------------------- ### GitHub Actions CI Configuration Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/compatibility.md A GitHub Actions workflow to test your project across multiple Python and aiohttp versions. It installs specific versions of aioresponses and aiohttp before running pytest. ```yaml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest strategy: matrix: python-version: ['3.10', '3.11', '3.12', '3.13'] aiohttp-version: ['3.8.0', '3.9.x'] steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | pip install aioresponses==0.7.9 pip install aiohttp==${{ matrix.aiohttp-version }} pip install pytest pytest-asyncio - name: Run tests run: pytest ``` -------------------------------- ### Simulating Rate Limiting with CallbackResult Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/callback-result.md Illustrates simulating rate limiting by returning a 429 status code after a certain number of requests, using a global counter. ```python request_count = 0 def rate_limit_callback(url, **kwargs): global request_count request_count += 1 if request_count <= 3: return CallbackResult(status=200, payload={'count': request_count}) else: return CallbackResult(status=429, body='Too many requests') with aioresponses() as m: m.add('http://example.com', callback=rate_limit_callback, repeat=True) ``` -------------------------------- ### Run linters and tests Source: https://github.com/pnuckowski/aioresponses/blob/master/CONTRIBUTING.rst Ensure your changes pass flake8 and the project tests, including testing with tox for multiple Python versions. ```shell flake8 aioresponses tests python setup.py test tox ``` -------------------------------- ### Assertion Checking with Exception Handling Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/errors.md This example demonstrates how to perform assertion checks on mocked requests using aioresponses and simultaneously handle potential AssertionErrors. This is useful for verifying that specific requests were made as expected during testing. ```python # With assertion checking with aioresponses() as m: m.post('http://api.example.com/users', status=201) try: # ... make request ... m.assert_called_once_with('http://api.example.com/users', method='POST') except AssertionError as e: print(f"Assertion failed: {e}") raise ``` -------------------------------- ### Mock Multiple HTTP Methods (CRUD) with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/patterns-and-usage.md This pattern mocks multiple HTTP methods (POST, GET, PUT, DELETE) for a single resource. It is useful for testing full CRUD functionality. The mocks are active within the 'with' block. ```python async def test_crud_operations(): with aioresponses() as m: # Create m.post('http://api.example.com/users', status=201, payload={'id': 1}) # Read m.get('http://api.example.com/users/1', status=200, payload={'id': 1, 'name': 'John'}) # Update m.put('http://api.example.com/users/1', status=200, payload={'id': 1, 'name': 'Jane'}) # Delete m.delete('http://api.example.com/users/1', status=204) async with aiohttp.ClientSession() as session: # POST resp = await session.post('http://api.example.com/users') assert resp.status == 201 # GET resp = await session.get('http://api.example.com/users/1') assert resp.status == 200 # PUT resp = await session.put('http://api.example.com/users/1') assert resp.status == 200 # DELETE resp = await session.delete('http://api.example.com/users/1') assert resp.status == 204 ``` -------------------------------- ### Import aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/aioresponses-class.md Import the necessary aioresponses class for mocking. ```python from aioresponses import aioresponses ``` -------------------------------- ### Disabling Redirects with aioresponses Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/match-method.md Shows how to disable automatic redirect following. The mock is set up for a redirect, but `allow_redirects=False` is passed to the client session's `get` method. The response object then reflects the initial redirect status code and headers, and `resp.history` remains empty. ```python with aioresponses() as m: m.get('http://example.com/', headers={'Location': 'http://new.example.com/'}, status=301) m.get('http://new.example.com/', status=200) async with aiohttp.ClientSession() as session: # Don't follow redirect resp = await session.get('http://example.com/', allow_redirects=False) assert resp.status == 301 assert resp.headers['Location'] == 'http://new.example.com/' assert len(resp.history) == 0 ``` -------------------------------- ### Configuring Response Options Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/README.md Customize mock responses using various parameters like status, body, payload, headers, content type, exceptions, timeouts, and repetition. ```python # Response options status=200 # HTTP status body='text' # Response body payload={'key': 'value'} # JSON response headers={'X-Custom': 'value'} # Custom headers content_type='application/json' # Content type exception=ClientConnectionError() # Raise exception timeout=True # Simulate timeout repeat=3 # Repeat N times repeat=True # Repeat infinitely callback=my_callback # Dynamic response reason='OK' # Reason phrase ``` -------------------------------- ### Create a Simple CallbackResult Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/callback-result.md Instantiate CallbackResult with a status code and a simple string body. ```python result = CallbackResult(status=200, body='OK') ``` -------------------------------- ### RequestMatch Constructor Source: https://github.com/pnuckowski/aioresponses/blob/master/_autodocs/api-reference/request-match.md The `__init__` method of the `RequestMatch` class is used to create a pattern for a mocked HTTP request. It accepts various parameters to define the URL, HTTP method, response status, body, headers, and other aspects of the mock response. ```APIDOC ## `__init__` ### Description Create a request match pattern. ### Method Constructor ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **url** (Union[URL, str, Pattern]) - Required - URL string, yarl.URL, or compiled regex pattern - **method** (str) - Optional - HTTP method (default: 'GET') - **status** (int) - Optional - HTTP status code (default: 200) - **body** (Union[str, bytes]) - Optional - Response body (default: '') - **payload** (Optional[Dict]) - Optional - JSON payload (overrides body) (default: None) - **exception** (Optional[Exception]) - Optional - Exception to raise (default: None) - **headers** (Optional[Union[CIMultiDict, dict]]) - Optional - Response headers (default: None) - **content_type** (str) - Optional - Content-Type header (default: 'application/json') - **response_class** (Optional[Type[ClientResponse]]) - Optional - Custom response class (default: None) - **timeout** (bool) - Optional - If True, raise asyncio.TimeoutError (default: False) - **repeat** (Union[bool, int]) - Optional - Repeat behavior: False/1 = once, int = n times, True = infinite (default: False) - **reason** (Optional[str]) - Optional - HTTP reason phrase (default: None) - **callback** (Optional[Callable]) - Optional - Function to generate dynamic response (default: None) ### Request Example None ### Response #### Success Response None #### Response Example None ```