move with_http_server to selfdrive/test/helpers (#31597)

* move

* fix
old-commit-hash: 9d1cafd0fc

selfdrive/athena/tests/helpers.py

@@ -1,7 +1,5 @@
 import http.server
-import threading
 import socket
-from functools import wraps
 
 
 class MockResponse:
@@ -65,25 +63,3 @@ class HTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
     self.rfile.read(length)
     self.send_response(201, "Created")
     self.end_headers()
-
-
-def with_http_server(func, handler=http.server.BaseHTTPRequestHandler, setup=None):
-  @wraps(func)
-  def inner(*args, **kwargs):
-    host = '127.0.0.1'
-    server = http.server.HTTPServer((host, 0), handler)
-    port = server.server_port
-    t = threading.Thread(target=server.serve_forever)
-    t.start()
-
-    if setup is not None:
-      setup(host, port)
-
-    try:
-      return func(*args, f'http://{host}:{port}', **kwargs)
-    finally:
-      server.shutdown()
-      server.server_close()
-      t.join()
-
-  return inner

selfdrive/athena/tests/test_athenad.py

@@ -23,9 +23,9 @@ from openpilot.common.params import Params
 from openpilot.common.timeout import Timeout
 from openpilot.selfdrive.athena import athenad
 from openpilot.selfdrive.athena.athenad import MAX_RETRY_COUNT, dispatcher
-from openpilot.selfdrive.athena.tests.helpers import MockWebsocket, MockApi, EchoSocket, with_http_server
+from openpilot.selfdrive.athena.tests.helpers import HTTPRequestHandler, MockWebsocket, MockApi, EchoSocket
+from openpilot.selfdrive.test.helpers import with_http_server
 from openpilot.system.hardware.hw import Paths
-from openpilot.selfdrive.athena.tests.helpers import HTTPRequestHandler
 
 
 def seed_athena_server(host, port):

selfdrive/test/helpers.py

@@ -1,4 +1,6 @@
+import http.server
 import os
+import threading
 import time
 
 from functools import wraps
@@ -76,3 +78,25 @@ def read_segment_list(segment_list_path):
     seg_list = f.read().splitlines()
 
   return [(platform[2:], segment) for platform, segment in zip(seg_list[::2], seg_list[1::2], strict=True)]
+
+
+def with_http_server(func, handler=http.server.BaseHTTPRequestHandler, setup=None):
+  @wraps(func)
+  def inner(*args, **kwargs):
+    host = '127.0.0.1'
+    server = http.server.HTTPServer((host, 0), handler)
+    port = server.server_port
+    t = threading.Thread(target=server.serve_forever)
+    t.start()
+
+    if setup is not None:
+      setup(host, port)
+
+    try:
+      return func(*args, f'http://{host}:{port}', **kwargs)
+    finally:
+      server.shutdown()
+      server.server_close()
+      t.join()
+
+  return inner

tools/lib/tests/test_caching.py

@@ -7,7 +7,7 @@ import socket
 import unittest
 
 from parameterized import parameterized
-from openpilot.selfdrive.athena.tests.helpers import with_http_server
+from openpilot.selfdrive.test.helpers import with_http_server
 from openpilot.system.hardware.hw import Paths
 from openpilot.tools.lib.url_file import URLFile