dummy server using for unit testing waiteventtimeout 5 def init self handlernone hostlocalhost port0 requeststohandle1 waittocloseeventnone super init self handler handler or consumesocketcontent self handlerresults self host host self port port self requeststohandle requeststohandle self waittocloseevent waittocloseevent self readyevent threading event self stopevent threading event classmethod def textresponseservercls text requesttimeout0 5 kwargs def textresponsehandlersock requestcontent consumesocketcontentsock timeoutrequesttimeout sock sendtext encodeutf8 return requestcontent return servertextresponsehandler kwargs classmethod def basicresponseservercls kwargs return cls textresponseserver http1 1 200 okrn contentlength 0rnrn kwargs def runself try self serversock self createsocketandbind in case self port 0 self port self serversock getsockname1 self readyevent set self handlerequests if self waittocloseevent self waittocloseevent waitself waiteventtimeout finally self readyevent set just in case of exception self closeserversockignoreerrors self stopevent set def createsocketandbindself sock socket socket sock bindself host self port sock listen return sock def closeserversockignoreerrorsself try self serversock close except oserror pass def handlerequestsself for in rangeself requeststohandle sock self acceptconnection if not sock break handlerresult self handlersock self handlerresults appendhandlerresult sock close def acceptconnectionself try ready select select self serversock self waiteventtimeout if not ready return none return self serversock accept0 except oserror return none def enterself self start if not self readyevent waitself waiteventtimeout raise runtimeerrortimeout waiting for server to be ready return self host self port def exitself exctype excvalue traceback if exctype is none self stopevent waitself waiteventtimeout else if self waittocloseevent avoid server from waiting for event timeouts if an exception is found in the main thread self waittocloseevent set ensure server thread doesn t get stuck waiting for connections self closeserversockignoreerrors self join return false allow exceptions to propagate dummy server using for unit testing in case self port 0 just in case of exception avoid server from waiting for event timeouts if an exception is found in the main thread ensure server thread doesn t get stuck waiting for connections allow exceptions to propagate
import select
import socket
import threading


def consume_socket_content(sock, timeout=0.5):
    chunks = 65536
    content = b""

    while True:
        more_to_read = select.select([sock], [], [], timeout)[0]
        if not more_to_read:
            break

        new_content = sock.recv(chunks)
        if not new_content:
            break

        content += new_content

    return content


class Server(threading.Thread):
    

    WAIT_EVENT_TIMEOUT = 5

    def __init__(
        self,
        handler=None,
        host="localhost",
        port=0,
        requests_to_handle=1,
        wait_to_close_event=None,
    ):
        super().__init__()

        self.handler = handler or consume_socket_content
        self.handler_results = []

        self.host = host
        self.port = port
        self.requests_to_handle = requests_to_handle

        self.wait_to_close_event = wait_to_close_event
        self.ready_event = threading.Event()
        self.stop_event = threading.Event()

    @classmethod
    def text_response_server(cls, text, request_timeout=0.5, **kwargs):
        def text_response_handler(sock):
            request_content = consume_socket_content(sock, timeout=request_timeout)
            sock.send(text.encode("utf-8"))

            return request_content

        return Server(text_response_handler, **kwargs)

    @classmethod
    def basic_response_server(cls, **kwargs):
        return cls.text_response_server(
            "HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n\r\n", **kwargs
        )

    def run(self):
        try:
            self.server_sock = self._create_socket_and_bind()
            
            self.port = self.server_sock.getsockname()[1]
            self.ready_event.set()
            self._handle_requests()

            if self.wait_to_close_event:
                self.wait_to_close_event.wait(self.WAIT_EVENT_TIMEOUT)
        finally:
            self.ready_event.set()  
            self._close_server_sock_ignore_errors()
            self.stop_event.set()

    def _create_socket_and_bind(self):
        sock = socket.socket()
        sock.bind((self.host, self.port))
        sock.listen()
        return sock

    def _close_server_sock_ignore_errors(self):
        try:
            self.server_sock.close()
        except OSError:
            pass

    def _handle_requests(self):
        for _ in range(self.requests_to_handle):
            sock = self._accept_connection()
            if not sock:
                break

            handler_result = self.handler(sock)

            self.handler_results.append(handler_result)
            sock.close()

    def _accept_connection(self):
        try:
            ready, _, _ = select.select(
                [self.server_sock], [], [], self.WAIT_EVENT_TIMEOUT
            )
            if not ready:
                return None

            return self.server_sock.accept()[0]
        except OSError:
            return None

    def __enter__(self):
        self.start()
        if not self.ready_event.wait(self.WAIT_EVENT_TIMEOUT):
            raise RuntimeError("Timeout waiting for server to be ready.")
        return self.host, self.port

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is None:
            self.stop_event.wait(self.WAIT_EVENT_TIMEOUT)
        else:
            if self.wait_to_close_event:
                
                
                self.wait_to_close_event.set()

        
        self._close_server_sock_ignore_errors()
        self.join()
        return False