File manager - Edit - /home/u478019808/domains/bestandroidphones.store/public_html/static/img/logo/testcase.py.tar
Back
opt/gsutil/third_party/urllib3/dummyserver/testcase.py 0000644 00000031667 15025266626 0017360 0 ustar 00 from __future__ import annotations import contextlib import socket import ssl import threading import typing from test import LONG_TIMEOUT import hypercorn import pytest from dummyserver.app import hypercorn_app from dummyserver.asgi_proxy import ProxyApp from dummyserver.hypercornserver import run_hypercorn_in_thread from dummyserver.socketserver import DEFAULT_CERTS, HAS_IPV6, SocketServerThread from urllib3.connection import HTTPConnection from urllib3.util.ssltransport import SSLTransport from urllib3.util.url import parse_url def consume_socket( sock: SSLTransport | socket.socket, chunks: int = 65536, quit_event: threading.Event | None = None, ) -> bytearray: consumed = bytearray() sock.settimeout(LONG_TIMEOUT) while True: if quit_event and quit_event.is_set(): break try: b = sock.recv(chunks) except (TimeoutError, socket.timeout): continue assert isinstance(b, bytes) consumed += b if b.endswith(b"\r\n\r\n"): break return consumed class SocketDummyServerTestCase: """ A simple socket-based server is created for this class that is good for exactly one request. """ scheme = "http" host = "localhost" server_thread: typing.ClassVar[SocketServerThread] port: typing.ClassVar[int] tmpdir: typing.ClassVar[str] ca_path: typing.ClassVar[str] cert_combined_path: typing.ClassVar[str] cert_path: typing.ClassVar[str] key_path: typing.ClassVar[str] password_key_path: typing.ClassVar[str] server_context: typing.ClassVar[ssl.SSLContext] client_context: typing.ClassVar[ssl.SSLContext] proxy_server: typing.ClassVar[SocketDummyServerTestCase] @classmethod def _start_server( cls, socket_handler: typing.Callable[[socket.socket], None], quit_event: threading.Event | None = None, ) -> None: ready_event = threading.Event() cls.server_thread = SocketServerThread( socket_handler=socket_handler, ready_event=ready_event, host=cls.host, quit_event=quit_event, ) cls.server_thread.start() ready_event.wait(5) if not ready_event.is_set(): raise Exception("most likely failed to start server") cls.port = cls.server_thread.port @classmethod def start_response_handler( cls, response: bytes, num: int = 1, block_send: threading.Event | None = None, ) -> threading.Event: ready_event = threading.Event() quit_event = threading.Event() def socket_handler(listener: socket.socket) -> None: for _ in range(num): ready_event.set() listener.settimeout(LONG_TIMEOUT) while True: if quit_event.is_set(): return try: sock = listener.accept()[0] break except (TimeoutError, socket.timeout): continue consume_socket(sock, quit_event=quit_event) if quit_event.is_set(): sock.close() return if block_send: while not block_send.wait(LONG_TIMEOUT): if quit_event.is_set(): sock.close() return block_send.clear() sock.send(response) sock.close() cls._start_server(socket_handler, quit_event=quit_event) return ready_event @classmethod def start_basic_handler( cls, num: int = 1, block_send: threading.Event | None = None ) -> threading.Event: return cls.start_response_handler( b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", num, block_send, ) @staticmethod def quit_server_thread(server_thread: SocketServerThread) -> None: if server_thread.quit_event: server_thread.quit_event.set() # in principle the maximum time that the thread can take to notice # the quit_event is LONG_TIMEOUT and the thread should terminate # shortly after that, we give 5 seconds leeway just in case server_thread.join(LONG_TIMEOUT * 2 + 5.0) if server_thread.is_alive(): raise Exception("server_thread did not exit") @classmethod def teardown_class(cls) -> None: if hasattr(cls, "server_thread"): cls.quit_server_thread(cls.server_thread) def teardown_method(self) -> None: if hasattr(self, "server_thread"): self.quit_server_thread(self.server_thread) def assert_header_received( self, received_headers: typing.Iterable[bytes], header_name: str, expected_value: str | None = None, ) -> None: header_name_bytes = header_name.encode("ascii") if expected_value is None: expected_value_bytes = None else: expected_value_bytes = expected_value.encode("ascii") header_titles = [] for header in received_headers: key, value = header.split(b": ") header_titles.append(key) if key == header_name_bytes and expected_value_bytes is not None: assert value == expected_value_bytes assert header_name_bytes in header_titles class IPV4SocketDummyServerTestCase(SocketDummyServerTestCase): @classmethod def _start_server( cls, socket_handler: typing.Callable[[socket.socket], None], quit_event: threading.Event | None = None, ) -> None: ready_event = threading.Event() cls.server_thread = SocketServerThread( socket_handler=socket_handler, ready_event=ready_event, host=cls.host, quit_event=quit_event, ) cls.server_thread.USE_IPV6 = False cls.server_thread.start() ready_event.wait(5) if not ready_event.is_set(): raise Exception("most likely failed to start server") cls.port = cls.server_thread.port class HypercornDummyServerTestCase: host = "localhost" host_alt = "127.0.0.1" port: typing.ClassVar[int] base_url: typing.ClassVar[str] base_url_alt: typing.ClassVar[str] certs: typing.ClassVar[dict[str, typing.Any]] = {} _stack: typing.ClassVar[contextlib.ExitStack] @classmethod def setup_class(cls) -> None: with contextlib.ExitStack() as stack: config = hypercorn.Config() if cls.certs: config.certfile = cls.certs["certfile"] config.keyfile = cls.certs["keyfile"] config.verify_mode = cls.certs["cert_reqs"] config.ca_certs = cls.certs["ca_certs"] config.alpn_protocols = cls.certs["alpn_protocols"] config.bind = [f"{cls.host}:0"] stack.enter_context(run_hypercorn_in_thread(config, hypercorn_app)) cls._stack = stack.pop_all() cls.port = typing.cast(int, parse_url(config.bind[0]).port) @classmethod def teardown_class(cls) -> None: cls._stack.close() class HTTPSHypercornDummyServerTestCase(HypercornDummyServerTestCase): scheme = "https" host = "localhost" certs = DEFAULT_CERTS certs_dir = "" bad_ca_path = "" class HypercornDummyProxyTestCase: http_host: typing.ClassVar[str] = "localhost" http_host_alt: typing.ClassVar[str] = "127.0.0.1" http_port: typing.ClassVar[int] http_url: typing.ClassVar[str] http_url_alt: typing.ClassVar[str] https_host: typing.ClassVar[str] = "localhost" https_host_alt: typing.ClassVar[str] = "127.0.0.1" https_certs: typing.ClassVar[dict[str, typing.Any]] = DEFAULT_CERTS https_port: typing.ClassVar[int] https_url: typing.ClassVar[str] https_url_alt: typing.ClassVar[str] https_url_fqdn: typing.ClassVar[str] proxy_host: typing.ClassVar[str] = "localhost" proxy_host_alt: typing.ClassVar[str] = "127.0.0.1" proxy_port: typing.ClassVar[int] proxy_url: typing.ClassVar[str] https_proxy_port: typing.ClassVar[int] https_proxy_url: typing.ClassVar[str] certs_dir: typing.ClassVar[str] = "" bad_ca_path: typing.ClassVar[str] = "" server_thread: typing.ClassVar[threading.Thread] _stack: typing.ClassVar[contextlib.ExitStack] @classmethod def setup_class(cls) -> None: with contextlib.ExitStack() as stack: http_server_config = hypercorn.Config() http_server_config.bind = [f"{cls.http_host}:0"] stack.enter_context( run_hypercorn_in_thread(http_server_config, hypercorn_app) ) cls.http_port = typing.cast(int, parse_url(http_server_config.bind[0]).port) https_server_config = hypercorn.Config() https_server_config.certfile = cls.https_certs["certfile"] https_server_config.keyfile = cls.https_certs["keyfile"] https_server_config.verify_mode = cls.https_certs["cert_reqs"] https_server_config.ca_certs = cls.https_certs["ca_certs"] https_server_config.alpn_protocols = cls.https_certs["alpn_protocols"] https_server_config.bind = [f"{cls.https_host}:0"] stack.enter_context( run_hypercorn_in_thread(https_server_config, hypercorn_app) ) cls.https_port = typing.cast( int, parse_url(https_server_config.bind[0]).port ) http_proxy_config = hypercorn.Config() http_proxy_config.bind = [f"{cls.proxy_host}:0"] stack.enter_context(run_hypercorn_in_thread(http_proxy_config, ProxyApp())) cls.proxy_port = typing.cast(int, parse_url(http_proxy_config.bind[0]).port) https_proxy_config = hypercorn.Config() https_proxy_config.certfile = cls.https_certs["certfile"] https_proxy_config.keyfile = cls.https_certs["keyfile"] https_proxy_config.verify_mode = cls.https_certs["cert_reqs"] https_proxy_config.ca_certs = cls.https_certs["ca_certs"] https_proxy_config.alpn_protocols = cls.https_certs["alpn_protocols"] https_proxy_config.bind = [f"{cls.proxy_host}:0"] upstream_ca_certs = cls.https_certs.get("ca_certs") stack.enter_context( run_hypercorn_in_thread(https_proxy_config, ProxyApp(upstream_ca_certs)) ) cls.https_proxy_port = typing.cast( int, parse_url(https_proxy_config.bind[0]).port ) cls._stack = stack.pop_all() @classmethod def teardown_class(cls) -> None: cls._stack.close() @pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available") class IPv6HypercornDummyServerTestCase(HypercornDummyServerTestCase): host = "::1" @pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available") class IPv6HypercornDummyProxyTestCase(HypercornDummyProxyTestCase): http_host = "localhost" http_host_alt = "127.0.0.1" https_host = "localhost" https_host_alt = "127.0.0.1" https_certs = DEFAULT_CERTS proxy_host = "::1" proxy_host_alt = "127.0.0.1" class ConnectionMarker: """ Marks an HTTP(S)Connection's socket after a request was made. Helps a test server understand when a client finished a request, without implementing a complete HTTP server. """ MARK_FORMAT = b"$#MARK%04x*!" @classmethod @contextlib.contextmanager def mark( cls, monkeypatch: pytest.MonkeyPatch ) -> typing.Generator[None, None, None]: """ Mark connections under in that context. """ orig_request = HTTPConnection.request def call_and_mark( target: typing.Callable[..., None] ) -> typing.Callable[..., None]: def part( self: HTTPConnection, *args: typing.Any, **kwargs: typing.Any ) -> None: target(self, *args, **kwargs) self.sock.sendall(cls._get_socket_mark(self.sock, False)) return part with monkeypatch.context() as m: m.setattr(HTTPConnection, "request", call_and_mark(orig_request)) yield @classmethod def consume_request(cls, sock: socket.socket, chunks: int = 65536) -> bytearray: """ Consume a socket until after the HTTP request is sent. """ consumed = bytearray() mark = cls._get_socket_mark(sock, True) while True: b = sock.recv(chunks) if not b: break consumed += b if consumed.endswith(mark): break return consumed @classmethod def _get_socket_mark(cls, sock: socket.socket, server: bool) -> bytes: if server: port = sock.getpeername()[1] else: port = sock.getsockname()[1] return cls.MARK_FORMAT % (port,)
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.03 |
proxy
|
phpinfo
|
Settings