Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return response
except Exception:
response = None
raise
finally:
self.record(request, response, pattern=pattern)
def close(self) -> None:
if self._assert_all_called:
self.assert_all_called()
async def aclose(self) -> None:
self.close()
class SyncMockTransport(BaseMockTransport, SyncHTTPTransport):
def request( # type: ignore
self,
method: bytes,
url: URL,
headers: Headers = None,
stream: SyncByteStream = None,
timeout: TimeoutDict = None,
) -> Response:
return self._sync_request(method, url, headers, stream, timeout)
class AsyncMockTransport(BaseMockTransport, AsyncHTTPTransport):
async def request( # type: ignore
self,
method: bytes,
url: URL,
import inspect
import warnings
from functools import partial, wraps
from typing import Any, Callable, Dict, List, Optional, Union
import asynctest
from .transports import BaseMockTransport
__all__ = ["MockTransport", "HTTPXMock"]
class MockTransport(BaseMockTransport):
_patches: List[asynctest.mock._patch] = []
transports: List["MockTransport"] = []
targets = [
"httpcore._sync.connection.SyncHTTPConnection.request",
"httpcore._sync.connection_pool.SyncConnectionPool.request",
"httpcore._sync.http_proxy.SyncHTTPProxy.request",
"httpcore._async.connection.AsyncHTTPConnection.request",
"httpcore._async.connection_pool.AsyncConnectionPool.request",
"httpcore._async.http_proxy.AsyncHTTPProxy.request",
"httpx._transports.asgi.ASGITransport.request",
"httpx._transports.wsgi.WSGITransport.request",
"httpx._transports.urllib3.URLLib3Transport.request",
]
def __init__(
self,
self.close()
class SyncMockTransport(BaseMockTransport, SyncHTTPTransport):
def request( # type: ignore
self,
method: bytes,
url: URL,
headers: Headers = None,
stream: SyncByteStream = None,
timeout: TimeoutDict = None,
) -> Response:
return self._sync_request(method, url, headers, stream, timeout)
class AsyncMockTransport(BaseMockTransport, AsyncHTTPTransport):
async def request( # type: ignore
self,
method: bytes,
url: URL,
headers: Headers = None,
stream: AsyncByteStream = None,
timeout: TimeoutDict = None,
) -> Response:
return await self._async_request(method, url, headers, stream, timeout)