diff --git a/tests/test_testserver.py b/tests/test_testserver.py new file mode 100644 index 00000000..03aa590f --- /dev/null +++ b/tests/test_testserver.py @@ -0,0 +1,128 @@ +import threading +import socket +import time + +import pytest +import requests +from testserver.server import Server + +class TestTestServer: + def test_basic(self): + question = b"sucess?" + answer = b"yeah, success" + def handler(sock): + text = sock.recv(1000) + assert text == question + sock.send(answer) + + with Server(handler) as (host, port): + sock = socket.socket() + sock.connect((host, port)) + sock.send(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('http://{0}:{1}'.format(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('http://{0}:{1}'.format(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.send(b'send something') + time.sleep(2.5) + sock.send(b'still alive') + block_server.set() # release server block + + 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 = 'http://{0}:{1}'.format(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) + + def test_request_recovery(self): + server = Server.basic_response_server(requests_to_handle=2) + first_request = "put your hands up in the air" + second_request = "put your hand down in the floor" + + with server as address: + sock1 = socket.socket() + sock2 = socket.socket() + + sock1.connect(address) + sock1.send(first_request.encode()) + sock1.close() + + sock2.connect(address) + sock2.send(second_request.encode()) + 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.send(b"hehehe, not received") + sock.close() + + assert server.handler_results[0] == "" + + + def test_request_recovery_with_bigger_timeout(self): + server = Server.basic_response_server(request_timeout=3) + data = "bananadine" + + with server as address: + sock = socket.socket() + sock.connect(address) + time.sleep(1.5) + sock.send(data.encode()) + sock.close() + + assert server.handler_results[0] == data diff --git a/tests/test_utils.py b/tests/test_utils.py index 3990780a..5972c005 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,12 +1,8 @@ # coding: utf-8 import os -import threading -import socket -import time from io import BytesIO import pytest -import requests from requests import compat from requests.utils import ( address_in_network, dotted_netmask, @@ -234,124 +230,3 @@ def test_select_proxies(url, expected): proxies = {'http': 'http://http.proxy', 'http://some.host': 'http://some.host.proxy'} assert select_proxy(url, proxies) == expected - -class TestTestServer: - def test_basic(self): - question = b"sucess?" - answer = b"yeah, success" - def handler(sock): - text = sock.recv(1000) - assert text == question - sock.send(answer) - - with Server(handler) as (host, port): - sock = socket.socket() - sock.connect((host, port)) - sock.send(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('http://{0}:{1}'.format(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('http://{0}:{1}'.format(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.send(b'send something') - time.sleep(2.5) - sock.send(b'still alive') - block_server.set() # release server block - - 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 = 'http://{0}:{1}'.format(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) - - def test_request_recovery(self): - server = Server.basic_response_server(requests_to_handle=2) - first_request = "put your hands up in the air" - second_request = "put your hand down in the floor" - - with server as address: - sock1 = socket.socket() - sock2 = socket.socket() - - sock1.connect(address) - sock1.send(first_request.encode()) - sock1.close() - - sock2.connect(address) - sock2.send(second_request.encode()) - 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.send(b"hehehe, not received") - sock.close() - - assert server.handler_results[0] == "" - - - def test_request_recovery_with_bigger_timeout(self): - server = Server.basic_response_server(request_timeout=3) - data = "bananadine" - - with server as address: - sock = socket.socket() - sock.connect(address) - time.sleep(1.5) - sock.send(data.encode()) - sock.close() - - assert server.handler_results[0] == data