From 9a33321438e6dbb76f0b4624387493373a36d806 Mon Sep 17 00:00:00 2001 From: foamyguy Date: Fri, 24 Oct 2025 17:10:46 -0500 Subject: [PATCH 1/3] use local server on more tests --- tests/files_test.py | 30 +++++++++++++++++++++++++++++- tests/local_test_server.py | 9 +++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/tests/files_test.py b/tests/files_test.py index c263cd3..449d768 100644 --- a/tests/files_test.py +++ b/tests/files_test.py @@ -5,12 +5,36 @@ """Post Files Tests""" # pylint: disable=line-too-long +import functools import re +import socketserver +import threading +import time from unittest import mock import mocket import pytest import requests as python_requests +from local_test_server import LocalTestServerHandler + + +def uses_local_server(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + with socketserver.TCPServer(("127.0.0.1", 5000), LocalTestServerHandler) as server: + server_thread = threading.Thread(target=server.serve_forever) + server_thread.daemon = True + server_thread.start() + time.sleep(2) # Give the server some time to start + + result = func(*args, **kwargs) + + server.shutdown() + server.server_close() + time.sleep(2) + return result + + return wrapper @pytest.fixture @@ -20,7 +44,8 @@ def log_stream(): @pytest.fixture def post_url(): - return "https://httpbin.org/post" + # return "https://httpbin.org/post" + return "http://127.0.0.1:5000/post" @pytest.fixture @@ -63,6 +88,7 @@ def get_actual_request_data(log_stream): return boundary, content_length, actual_request_post +@uses_local_server def test_post_file_as_data( # pylint: disable=unused-argument requests, sock, log_stream, post_url, request_logging ): @@ -85,6 +111,7 @@ def test_post_file_as_data( # pylint: disable=unused-argument assert sent.endswith(actual_request_post) +@uses_local_server def test_post_files_text( # pylint: disable=unused-argument sock, requests, log_stream, post_url, request_logging ): @@ -120,6 +147,7 @@ def test_post_files_text( # pylint: disable=unused-argument assert sent.endswith(actual_request_post) +@uses_local_server def test_post_files_file( # pylint: disable=unused-argument sock, requests, log_stream, post_url, request_logging ): diff --git a/tests/local_test_server.py b/tests/local_test_server.py index 73b52b3..b5b2be1 100644 --- a/tests/local_test_server.py +++ b/tests/local_test_server.py @@ -6,6 +6,15 @@ class LocalTestServerHandler(SimpleHTTPRequestHandler): + def do_POST(self): + if self.path == "/post": + resp_body = json.dumps({"url": "http://localhost:5000/post"}).encode("utf-8") + self.send_response(200) + self.send_header("Content-type", "application/json") + self.send_header("Content-Length", str(len(resp_body))) + self.end_headers() + self.wfile.write(resp_body) + def do_GET(self): if self.path == "/get": resp_body = json.dumps({"url": "http://localhost:5000/get"}).encode("utf-8") From 21d788c02e4296c0d12e7da30f54846c2fc47a34 Mon Sep 17 00:00:00 2001 From: foamyguy Date: Sat, 1 Nov 2025 10:40:16 -0500 Subject: [PATCH 2/3] move decorator to local_test_server.py. Use decorator in real_call_test. Add tests for https & redirect. --- tests/files_test.py | 27 ++---------------------- tests/local_test_server.py | 27 ++++++++++++++++++++++++ tests/real_call_test.py | 42 +++++++++++++++++++++----------------- 3 files changed, 52 insertions(+), 44 deletions(-) diff --git a/tests/files_test.py b/tests/files_test.py index 449d768..5a81877 100644 --- a/tests/files_test.py +++ b/tests/files_test.py @@ -5,36 +5,13 @@ """Post Files Tests""" # pylint: disable=line-too-long -import functools import re -import socketserver -import threading -import time from unittest import mock import mocket import pytest import requests as python_requests -from local_test_server import LocalTestServerHandler - - -def uses_local_server(func): - @functools.wraps(func) - def wrapper(*args, **kwargs): - with socketserver.TCPServer(("127.0.0.1", 5000), LocalTestServerHandler) as server: - server_thread = threading.Thread(target=server.serve_forever) - server_thread.daemon = True - server_thread.start() - time.sleep(2) # Give the server some time to start - - result = func(*args, **kwargs) - - server.shutdown() - server.server_close() - time.sleep(2) - return result - - return wrapper +from local_test_server import uses_local_server @pytest.fixture @@ -44,7 +21,6 @@ def log_stream(): @pytest.fixture def post_url(): - # return "https://httpbin.org/post" return "http://127.0.0.1:5000/post" @@ -192,6 +168,7 @@ def test_post_files_file( # pylint: disable=unused-argument assert sent.endswith(actual_request_post) +@uses_local_server def test_post_files_complex( # pylint: disable=unused-argument sock, requests, log_stream, post_url, request_logging ): diff --git a/tests/local_test_server.py b/tests/local_test_server.py index b5b2be1..04f47ce 100644 --- a/tests/local_test_server.py +++ b/tests/local_test_server.py @@ -1,10 +1,37 @@ # SPDX-FileCopyrightText: 2025 Tim Cocks # # SPDX-License-Identifier: MIT +import functools import json +import socketserver +import threading +import time from http.server import SimpleHTTPRequestHandler +def uses_local_server(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + with ReusableAddressTCPServer(("127.0.0.1", 5000), LocalTestServerHandler) as server: + server_thread = threading.Thread(target=server.serve_forever) + server_thread.daemon = True + server_thread.start() + time.sleep(2) # Give the server some time to start + + result = func(*args, **kwargs) + + server.shutdown() + server.server_close() + return result + + return wrapper + + +class ReusableAddressTCPServer(socketserver.TCPServer): + # Enable SO_REUSEADDR + allow_reuse_address = True + + class LocalTestServerHandler(SimpleHTTPRequestHandler): def do_POST(self): if self.path == "/post": diff --git a/tests/real_call_test.py b/tests/real_call_test.py index b982f00..60a3450 100644 --- a/tests/real_call_test.py +++ b/tests/real_call_test.py @@ -12,11 +12,12 @@ import adafruit_connection_manager import pytest -from local_test_server import LocalTestServerHandler +from local_test_server import uses_local_server import adafruit_requests +@uses_local_server def test_gets(): path_index = 0 status_code_index = 1 @@ -28,25 +29,28 @@ def test_gets(): ("status/204", 204, "", None), ] - with socketserver.TCPServer(("127.0.0.1", 5000), LocalTestServerHandler) as server: - server_thread = threading.Thread(target=server.serve_forever) - server_thread.daemon = True - server_thread.start() + for case in cases: + requests = adafruit_requests.Session(socket, ssl.create_default_context()) + with requests.get(f"http://127.0.0.1:5000/{case[path_index]}") as response: + assert response.status_code == case[status_code_index] + if case[text_result_index] is not None: + assert response.text == case[text_result_index] + if case[json_keys_index] is not None: + for key, value in case[json_keys_index].items(): + assert response.json()[key] == value - time.sleep(2) # Give the server some time to start + adafruit_connection_manager.connection_manager_close_all(release_references=True) - for case in cases: - requests = adafruit_requests.Session(socket, ssl.create_default_context()) - with requests.get(f"http://127.0.0.1:5000/{case[path_index]}") as response: - assert response.status_code == case[status_code_index] - if case[text_result_index] is not None: - assert response.text == case[text_result_index] - if case[json_keys_index] is not None: - for key, value in case[json_keys_index].items(): - assert response.json()[key] == value - adafruit_connection_manager.connection_manager_close_all(release_references=True) +def test_http_to_https_redirect(): + url = "http://www.adafruit.com/api/quotes.php" + requests = adafruit_requests.Session(socket, ssl.create_default_context()) + with requests.get(url) as response: + assert response.status_code == 200 - server.shutdown() - server.server_close() - time.sleep(2) + +def test_https_direct(): + url = "https://www.adafruit.com/api/quotes.php" + requests = adafruit_requests.Session(socket, ssl.create_default_context()) + with requests.get(url) as response: + assert response.status_code == 200 From a7ec3bae806b0de97275d65902189ed43f950b39 Mon Sep 17 00:00:00 2001 From: foamyguy Date: Mon, 3 Nov 2025 12:42:53 -0600 Subject: [PATCH 3/3] test allow_redirects --- tests/real_call_test.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/real_call_test.py b/tests/real_call_test.py index 60a3450..fad0178 100644 --- a/tests/real_call_test.py +++ b/tests/real_call_test.py @@ -42,11 +42,18 @@ def test_gets(): adafruit_connection_manager.connection_manager_close_all(release_references=True) -def test_http_to_https_redirect(): +@pytest.mark.parametrize( + ("allow_redirects", "status_code"), + ( + (True, 200), + (False, 301), + ), +) +def test_http_to_https_redirect(allow_redirects, status_code): url = "http://www.adafruit.com/api/quotes.php" requests = adafruit_requests.Session(socket, ssl.create_default_context()) - with requests.get(url) as response: - assert response.status_code == 200 + with requests.get(url, allow_redirects=allow_redirects) as response: + assert response.status_code == status_code def test_https_direct():