Gecko [ www.smanegeri1koba.sch.id ]

Name	Size	Permission
certs	[ DIR ]	drwxr-xr-x
testserver	[ DIR ]	drwxr-xr-x
__init__.py	469 B	-rw-r--r--
compat.py	480 B	-rw-r--r--
conftest.py	1.54 KB	-rw-r--r--
test_adapters.py	311 B	-rw-r--r--
test_help.py	839 B	-rw-r--r--
test_hooks.py	424 B	-rw-r--r--
test_lowlevel.py	14.98 KB	-rw-r--r--
test_packages.py	229 B	-rw-r--r--
test_requests.py	102.05 KB	-rw-r--r--
test_structures.py	2.23 KB	-rw-r--r--
test_testserver.py	5.42 KB	-rw-r--r--
test_utils.py	28.67 KB	-rw-r--r--
utils.py	366 B	-rw-r--r--

Code Editor : test_testserver.py

import socket
import threading
import time

import pytest
from tests.testserver.server import Server

import requests

class TestTestServer:
    def test_basic(self):
        """messages are sent and received properly"""
        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):
        """the server closes when leaving the context manager"""
        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):
        """the text_response_server sends the given text"""
        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):
        """the basic response server returns an empty http response"""
        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):
        """the server waits for the block_server event to be set before closing"""
        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()  # release server block

def test_multiple_requests(self):
        """multiple requests can be served"""
        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

# the (n+1)th request fails
            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):
        """can check the requests content"""
        # TODO: figure out why this sometimes fails when using pytest-xdist.
        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):
        """the basic response handler times out when receiving requests"""
        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):
        """a biggest timeout can be specified"""
        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):
        """the server thread exits even if an exception exits the context manager"""
        server = Server.basic_response_server()
        with pytest.raises(Exception):
            with server:
                raise Exception()

assert len(server.handler_results) == 0

# if the server thread fails to finish, the test suite will hang
        # and get killed by the jenkins timeout.

def test_server_finishes_when_no_connections(self):
        """the server thread exits even if there are no connections"""
        server = Server.basic_response_server()
        with server:
            pass

assert len(server.handler_results) == 0

# if the server thread fails to finish, the test suite will hang
        # and get killed by the jenkins timeout.

${this.title}

Code Editor : test_testserver.py