Merge pull request #214 from FoamyGuy/local_test_server

use a local test server instead of httpbin

Author: FoamyGuy

tests/local_test_server.py

@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: 2025 Tim Cocks
+#
+# SPDX-License-Identifier: MIT
+import json
+from http.server import SimpleHTTPRequestHandler
+
+
+class LocalTestServerHandler(SimpleHTTPRequestHandler):
+    def do_GET(self):
+        if self.path == "/get":
+            resp_body = json.dumps({"url": "http://localhost:5000/get"}).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)
+        if self.path.startswith("/status"):
+            try:
+                requested_status = int(self.path.split("/")[2])
+            except ValueError:
+                resp_body = json.dumps({"error": "requested status code must be int"}).encode(
+                    "utf-8"
+                )
+                self.send_response(400)
+                self.send_header("Content-type", "application/json")
+                self.send_header("Content-Length", str(len(resp_body)))
+                self.end_headers()
+                self.wfile.write(resp_body)
+                return
+
+            if requested_status != 204:
+                self.send_response(requested_status)
+                self.send_header("Content-type", "text/html")
+                self.send_header("Content-Length", "0")
+            else:
+                self.send_response(requested_status)
+                self.send_header("Content-type", "text/html")
+            self.end_headers()

tests/real_call_test.py

@@ -5,30 +5,48 @@
 """Real call Tests"""
 
 import socket
+import socketserver
 import ssl
+import threading
+import time
 
 import adafruit_connection_manager
 import pytest
+from local_test_server import LocalTestServerHandler
 
 import adafruit_requests
 
 
-@pytest.mark.parametrize(
-    ("path", "status_code", "text_result", "json_keys"),
-    (
-        ("get", 200, None, {"url": "https://httpbin.org/get"}),
+def test_gets():
+    path_index = 0
+    status_code_index = 1
+    text_result_index = 2
+    json_keys_index = 3
+    cases = [
+        ("get", 200, None, {"url": "http://localhost:5000/get"}),
         ("status/200", 200, "", None),
         ("status/204", 204, "", None),
-    ),
-)
-def test_gets(path, status_code, text_result, json_keys):
-    requests = adafruit_requests.Session(socket, ssl.create_default_context())
-    with requests.get(f"https://httpbin.org/{path}") as response:
-        assert response.status_code == status_code
-        if text_result is not None:
-            assert response.text == text_result
-        if json_keys is not None:
-            for key, value in json_keys.items():
-                assert response.json()[key] == value
-
-    adafruit_connection_manager.connection_manager_close_all(release_references=True)
+    ]
+
+    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
+
+        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)
+
+        server.shutdown()
+        server.server_close()
+        time.sleep(2)