codekingpro/portable-devtools

1from __future__ import annotations2 3import ssl4import typing5 6from .._exceptions import ReadError7from .base import (8    SOCKET_OPTION,9    AsyncNetworkBackend,10    AsyncNetworkStream,11    NetworkBackend,12    NetworkStream,13)14 15 16class MockSSLObject:17    def __init__(self, http2: bool):18        self._http2 = http219 20    def selected_alpn_protocol(self) -> str:21        return "h2" if self._http2 else "http/1.1"22 23 24class MockStream(NetworkStream):25    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:26        self._buffer = buffer27        self._http2 = http228        self._closed = False29 30    def read(self, max_bytes: int, timeout: float | None = None) -> bytes:31        if self._closed:32            raise ReadError("Connection closed")33        if not self._buffer:34            return b""35        return self._buffer.pop(0)36 37    def write(self, buffer: bytes, timeout: float | None = None) -> None:38        pass39 40    def close(self) -> None:41        self._closed = True42 43    def start_tls(44        self,45        ssl_context: ssl.SSLContext,46        server_hostname: str | None = None,47        timeout: float | None = None,48    ) -> NetworkStream:49        return self50 51    def get_extra_info(self, info: str) -> typing.Any:52        return MockSSLObject(http2=self._http2) if info == "ssl_object" else None53 54    def __repr__(self) -> str:55        return "<httpcore.MockStream>"56 57 58class MockBackend(NetworkBackend):59    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:60        self._buffer = buffer61        self._http2 = http262 63    def connect_tcp(64        self,65        host: str,66        port: int,67        timeout: float | None = None,68        local_address: str | None = None,69        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,70    ) -> NetworkStream:71        return MockStream(list(self._buffer), http2=self._http2)72 73    def connect_unix_socket(74        self,75        path: str,76        timeout: float | None = None,77        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,78    ) -> NetworkStream:79        return MockStream(list(self._buffer), http2=self._http2)80 81    def sleep(self, seconds: float) -> None:82        pass83 84 85class AsyncMockStream(AsyncNetworkStream):86    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:87        self._buffer = buffer88        self._http2 = http289        self._closed = False90 91    async def read(self, max_bytes: int, timeout: float | None = None) -> bytes:92        if self._closed:93            raise ReadError("Connection closed")94        if not self._buffer:95            return b""96        return self._buffer.pop(0)97 98    async def write(self, buffer: bytes, timeout: float | None = None) -> None:99        pass100 101    async def aclose(self) -> None:102        self._closed = True103 104    async def start_tls(105        self,106        ssl_context: ssl.SSLContext,107        server_hostname: str | None = None,108        timeout: float | None = None,109    ) -> AsyncNetworkStream:110        return self111 112    def get_extra_info(self, info: str) -> typing.Any:113        return MockSSLObject(http2=self._http2) if info == "ssl_object" else None114 115    def __repr__(self) -> str:116        return "<httpcore.AsyncMockStream>"117 118 119class AsyncMockBackend(AsyncNetworkBackend):120    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:121        self._buffer = buffer122        self._http2 = http2123 124    async def connect_tcp(125        self,126        host: str,127        port: int,128        timeout: float | None = None,129        local_address: str | None = None,130        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,131    ) -> AsyncNetworkStream:132        return AsyncMockStream(list(self._buffer), http2=self._http2)133 134    async def connect_unix_socket(135        self,136        path: str,137        timeout: float | None = None,138        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,139    ) -> AsyncNetworkStream:140        return AsyncMockStream(list(self._buffer), http2=self._http2)141 142    async def sleep(self, seconds: float) -> None:143        pass144