Move end-to-end test HTTP server to a thread to support 3.14+

.github/workflows/ci.yml

@@ -53,8 +53,16 @@ jobs:
       - mypy
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
-        python: [ '3.9', '3.10', '3.11', '3.12', '3.13']
+        python:
+          - '3.9'
+          - '3.10'
+          - '3.11'
+          - '3.12'
+          - '3.13'
+          - '3.14'
+          - '3.15'
     steps:
       - uses: actions/checkout@v7.0.1
       - name: Setup Build Tools
@@ -64,6 +72,7 @@ jobs:
       - name: Set Up Python
         uses: actions/setup-python@v7.0.0
         with:
+          allow-prereleases: true
           python-version: ${{ matrix.python }}
       - name: Install dependencies
         run: |

setup.cfg

@@ -14,6 +14,8 @@ classifiers =
     Programming Language :: Python :: 3.11
     Programming Language :: Python :: 3.12
     Programming Language :: Python :: 3.13
+    Programming Language :: Python :: 3.14
+    Programming Language :: Python :: 3.15
     Topic :: Software Development :: Libraries :: Python Modules
 keywords =
     httpx

tests/conftest.py

@@ -2,12 +2,11 @@
 import os
 import re
 import copy
-import socket
-import contextlib
-import multiprocessing as mp
-from time import sleep
+import threading as th
+from contextlib import contextmanager
 from base64 import b64decode
 from http.server import HTTPServer, BaseHTTPRequestHandler
+from typing import Generator, cast
 
 import pytest
 import k5test  # type: ignore[import-untyped]
@@ -72,22 +71,28 @@ def _get_context(self):
         return gssapi.SecurityContext(creds=server_cred)
 
 
-def start_http_server(realm: k5test.K5Realm,
-                      host: str = 'localhost',
-                      port: int = 8080):
+@contextmanager
+def start_http_server(
+    realm: k5test.K5Realm,
+    host: str = 'localhost',
+    port: int = 0,
+) -> Generator[HTTPServer, None, None]:
     princ = f'HTTP/{host}@{realm.realm}'
     realm.addprinc(princ)
     realm.extract_keytab(princ, realm.keytab)
     realm.ccache = realm.env['KRB5CCNAME'] \
         = os.path.join(realm.tmpdir, 'service_ccache')
     realm.kinit(princ, flags=['-k', '-t', realm.keytab])
 
-    os.environ.update(realm.env)
-
-    with HTTPServer(server_address=(host, port),
-                    RequestHandlerClass=KrbRequestHandler) as httpd:
+    with HTTPServer(
+        server_address=(host, port),
+        RequestHandlerClass=KrbRequestHandler,
+    ) as httpd:
         httpd.krb5_realm = realm  # type: ignore[attr-defined]
-        httpd.serve_forever()
+        thread = th.Thread(target=httpd.serve_forever, daemon=True)
+        thread.start()
+        yield httpd
+        thread.join(timeout=5)
 
 
 @pytest.fixture(scope='session')
@@ -101,29 +106,15 @@ def krb_realm() -> k5test.K5Realm:
 
 
 @pytest.fixture(scope='session')
-def http_server_port() -> int:
-    with contextlib.closing(socket.socket()) as sock:
-        sock.bind(('127.0.0.1', 0))
-        return sock.getsockname()[1]
-
-
-@pytest.fixture(scope='session')
-def http_server(request, krb_realm: k5test.K5Realm, http_server_port: int):
-    ps = mp.Process(
-        target=start_http_server,
-        args=(krb_realm,),
-        kwargs={'port': http_server_port},
-    )
-    ps.start()
-
-    sleep(1)
-
-    @request.addfinalizer
-    def cleanup():
-        if ps.is_alive():
-            ps.terminate()
+def http_server(krb_realm: k5test.K5Realm) -> Generator[str, None, None]:
+    with start_http_server(krb_realm) as httpd:
+        host, port = cast(tuple[str, int], httpd.server_address)
+        yield f"http://{host}:{port}/"
+        httpd.shutdown()
 
 
 @pytest.fixture
-def http_creds(krb_realm: k5test.K5Realm):
+def http_creds(
+    krb_realm: k5test.K5Realm,
+) -> Generator[gssapi.Credentials, None, None]:
     yield gssapi.Credentials(usage='initiate', name=gssapi.Name('user'))

tests/test_end_to_end.py

@@ -1,17 +1,22 @@
 #!/usr/bin/env python
 """Tests for httpx_gssapi."""
-
+import gssapi
 import httpx
+import k5test  # type: ignore[import-untyped]
 import pytest
 
 import httpx_gssapi
 
 
-def test_end_to_end(http_server, http_creds, krb_realm, http_server_port):
+def test_end_to_end(
+    http_server: str,
+    http_creds: gssapi.Credentials,
+    krb_realm: k5test.K5Realm,
+) -> None:
     auth = httpx_gssapi.HTTPSPNEGOAuth(creds=http_creds)
     with httpx.Client(auth=auth, timeout=500) as client:
         for i in range(2):
-            resp = client.get(f'http://localhost:{http_server_port}/')
+            resp = client.get(http_server)
             assert resp.status_code == 200