messages are sent and received properly question bsuccess answer byeah success def handlersock text sock recv1000 assert text question sock sendallanswer with serverhandler as host port sock socket socket sock connecthost port sock sendallquestion text sock recv1000 assert text answer sock close def testserverclosesself the textresponseserver sends the given text server server textresponseserver http1 1 200 okrn contentlength 6rn rnroflol with server as host port r requests getfhttp host port assert r statuscode 200 assert r text roflol assert r headerscontentlength 6 def testbasicresponseself the server waits for the blockserver event to be set before closing blockserver threading event with server basicresponseserverwaittocloseeventblockserver as host port sock socket socket sock connecthost port sock sendallbsend something time sleep2 5 sock sendallbstill alive blockserver set release server block def testmultiplerequestsself the n1th request fails can check the requests content todo figure out why this sometimes fails when using pytestxdist server server basicresponseserverrequeststohandle2 firstrequest bput your hands up in the air secondrequest bput your hand down in the floor with server as address sock1 socket socket sock2 socket socket sock1 connectaddress sock1 sendallfirstrequest sock1 close sock2 connectaddress sock2 sendallsecondrequest sock2 close assert server handlerresults0 firstrequest assert server handlerresults1 secondrequest def testrequestsaftertimeoutarenotreceivedself a biggest timeout can be specified server server basicresponseserverrequesttimeout3 data bbananadine with server as address sock socket socket sock connectaddress time sleep1 5 sock sendalldata sock close assert server handlerresults0 data def testserverfinishesonerrorself if the server thread fails to finish the test suite will hang and get killed by the jenkins timeout the server thread exits even if there are no connections server server basicresponseserver with server pass assert lenserver handlerresults 0 if the server thread fails to finish the test suite will hang and get killed by the jenkins timeout messages are sent and received properly the server closes when leaving the context manager the text_response_server sends the given text the basic response server returns an empty http response the server waits for the block_server event to be set before closing release server block multiple requests can be served the n 1 th request fails can check the requests content todo figure out why this sometimes fails when using pytest xdist the basic response handler times out when receiving requests a biggest timeout can be specified the server thread exits even if an exception exits the context manager if the server thread fails to finish the test suite will hang and get killed by the jenkins timeout the server thread exits even if there are no connections if the server thread fails to finish the test suite will hang and get killed by the jenkins timeout
import socket
import threading
import time

import pytest
from tests.testserver.server import Server

import requests


class TestTestServer:
    def test_basic(self):
        
        question = b"success?"
        answer = b"yeah, success"

        def handler(sock):
            text = sock.recv(1000)
            assert text == question
            sock.sendall(answer)

        with Server(handler) as (host, port):
            sock = socket.socket()
            sock.connect((host, port))
            sock.sendall(question)
            text = sock.recv(1000)
            assert text == answer
            sock.close()

    def test_server_closes(self):
        
        with Server.basic_response_server() as (host, port):
            sock = socket.socket()
            sock.connect((host, port))

            sock.close()

        with pytest.raises(socket.error):
            new_sock = socket.socket()
            new_sock.connect((host, port))

    def test_text_response(self):
        
        server = Server.text_response_server(
            "HTTP/1.1 200 OK\r\n" "Content-Length: 6\r\n" "\r\nroflol"
        )

        with server as (host, port):
            r = requests.get(f"http://{host}:{port}")

            assert r.status_code == 200
            assert r.text == "roflol"
            assert r.headers["Content-Length"] == "6"

    def test_basic_response(self):
        
        with Server.basic_response_server() as (host, port):
            r = requests.get(f"http://{host}:{port}")
            assert r.status_code == 200
            assert r.text == ""
            assert r.headers["Content-Length"] == "0"

    def test_basic_waiting_server(self):
        
        block_server = threading.Event()

        with Server.basic_response_server(wait_to_close_event=block_server) as (
            host,
            port,
        ):
            sock = socket.socket()
            sock.connect((host, port))
            sock.sendall(b"send something")
            time.sleep(2.5)
            sock.sendall(b"still alive")
            block_server.set()  

    def test_multiple_requests(self):
        
        requests_to_handle = 5

        server = Server.basic_response_server(requests_to_handle=requests_to_handle)

        with server as (host, port):
            server_url = f"http://{host}:{port}"
            for _ in range(requests_to_handle):
                r = requests.get(server_url)
                assert r.status_code == 200

            
            with pytest.raises(requests.exceptions.ConnectionError):
                r = requests.get(server_url)

    @pytest.mark.skip(reason="this fails non-deterministically under pytest-xdist")
    def test_request_recovery(self):
        
        
        server = Server.basic_response_server(requests_to_handle=2)
        first_request = b"put your hands up in the air"
        second_request = b"put your hand down in the floor"

        with server as address:
            sock1 = socket.socket()
            sock2 = socket.socket()

            sock1.connect(address)
            sock1.sendall(first_request)
            sock1.close()

            sock2.connect(address)
            sock2.sendall(second_request)
            sock2.close()

        assert server.handler_results[0] == first_request
        assert server.handler_results[1] == second_request

    def test_requests_after_timeout_are_not_received(self):
        
        server = Server.basic_response_server(request_timeout=1)

        with server as address:
            sock = socket.socket()
            sock.connect(address)
            time.sleep(1.5)
            sock.sendall(b"hehehe, not received")
            sock.close()

        assert server.handler_results[0] == b""

    def test_request_recovery_with_bigger_timeout(self):
        
        server = Server.basic_response_server(request_timeout=3)
        data = b"bananadine"

        with server as address:
            sock = socket.socket()
            sock.connect(address)
            time.sleep(1.5)
            sock.sendall(data)
            sock.close()

        assert server.handler_results[0] == data

    def test_server_finishes_on_error(self):
        
        server = Server.basic_response_server()
        with pytest.raises(Exception):
            with server:
                raise Exception()

        assert len(server.handler_results) == 0

        
        

    def test_server_finishes_when_no_connections(self):
        
        server = Server.basic_response_server()
        with server:
            pass

        assert len(server.handler_results) == 0