File size: 4,077 Bytes
edc8b84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | from __future__ import annotations
import ssl
import typing
from .._exceptions import ReadError
from .base import (
SOCKET_OPTION,
AsyncNetworkBackend,
AsyncNetworkStream,
NetworkBackend,
NetworkStream,
)
class MockSSLObject:
def __init__(self, http2: bool):
self._http2 = http2
def selected_alpn_protocol(self) -> str:
return "h2" if self._http2 else "http/1.1"
class MockStream(NetworkStream):
def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
self._buffer = buffer
self._http2 = http2
self._closed = False
def read(self, max_bytes: int, timeout: float | None = None) -> bytes:
if self._closed:
raise ReadError("Connection closed")
if not self._buffer:
return b""
return self._buffer.pop(0)
def write(self, buffer: bytes, timeout: float | None = None) -> None:
pass
def close(self) -> None:
self._closed = True
def start_tls(
self,
ssl_context: ssl.SSLContext,
server_hostname: str | None = None,
timeout: float | None = None,
) -> NetworkStream:
return self
def get_extra_info(self, info: str) -> typing.Any:
return MockSSLObject(http2=self._http2) if info == "ssl_object" else None
def __repr__(self) -> str:
return "<httpcore.MockStream>"
class MockBackend(NetworkBackend):
def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
self._buffer = buffer
self._http2 = http2
def connect_tcp(
self,
host: str,
port: int,
timeout: float | None = None,
local_address: str | None = None,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
) -> NetworkStream:
return MockStream(list(self._buffer), http2=self._http2)
def connect_unix_socket(
self,
path: str,
timeout: float | None = None,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
) -> NetworkStream:
return MockStream(list(self._buffer), http2=self._http2)
def sleep(self, seconds: float) -> None:
pass
class AsyncMockStream(AsyncNetworkStream):
def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
self._buffer = buffer
self._http2 = http2
self._closed = False
async def read(self, max_bytes: int, timeout: float | None = None) -> bytes:
if self._closed:
raise ReadError("Connection closed")
if not self._buffer:
return b""
return self._buffer.pop(0)
async def write(self, buffer: bytes, timeout: float | None = None) -> None:
pass
async def aclose(self) -> None:
self._closed = True
async def start_tls(
self,
ssl_context: ssl.SSLContext,
server_hostname: str | None = None,
timeout: float | None = None,
) -> AsyncNetworkStream:
return self
def get_extra_info(self, info: str) -> typing.Any:
return MockSSLObject(http2=self._http2) if info == "ssl_object" else None
def __repr__(self) -> str:
return "<httpcore.AsyncMockStream>"
class AsyncMockBackend(AsyncNetworkBackend):
def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
self._buffer = buffer
self._http2 = http2
async def connect_tcp(
self,
host: str,
port: int,
timeout: float | None = None,
local_address: str | None = None,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
) -> AsyncNetworkStream:
return AsyncMockStream(list(self._buffer), http2=self._http2)
async def connect_unix_socket(
self,
path: str,
timeout: float | None = None,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
) -> AsyncNetworkStream:
return AsyncMockStream(list(self._buffer), http2=self._http2)
async def sleep(self, seconds: float) -> None:
pass
|