From 66c36b716788ae049381364d96568be07374d6ed Mon Sep 17 00:00:00 2001 From: Codemageddon Date: Sun, 1 Nov 2026 16:13:19 +0100 Subject: [PATCH] Add Free-threaded python support --- .github/workflows/test.yaml | 7 +- CHANGELOG.md | 10 ++ CLAUDE.md | 10 +- docs/getting-started/installation.md | 3 + kubex/configuration/file_config.py | 49 +++++-- .../kubex_core/models/resource_config.py | 52 ++++++-- packages/kubex-core/pyproject.toml | 1 + packages/kubex-k8s-1-32/pyproject.toml | 1 + packages/kubex-k8s-1-33/pyproject.toml | 1 + packages/kubex-k8s-1-34/pyproject.toml | 1 + packages/kubex-k8s-1-35/pyproject.toml | 1 + packages/kubex-k8s-1-36/pyproject.toml | 1 + packages/kubex-k8s-1-37/pyproject.toml | 1 + packages/kubex-k8s-1-38/pyproject.toml | 1 + pyproject.toml | 1 + scripts/codegen/templates/pyproject.toml.j2 | 1 + .../golden/kubex-k8s-1-30/pyproject.toml | 1 + test/test_client/_servers.py | 41 +++--- test/test_client/test_httpx_websocket.py | 126 +++--------------- test/test_client/test_threads.py | 97 ++++++++++++++ test/test_configuration/test_file_config.py | 50 +++++++ test/test_free_threading.py | 50 +++++++ test/test_models/test_resource_config.py | 67 +++++++++- 23 files changed, 416 insertions(+), 157 deletions(-) create mode 100644 test/test_client/test_threads.py create mode 100644 test/test_free_threading.py diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 6ca4d065..835553b1 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -10,11 +10,16 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"] + # A "t" suffix is the free-threaded (no-GIL) build. + python-version: ["3.11", "3.12", "3.13", "3.14", "3.14t", "3.15", "3.15t"] steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v3 - name: Install dependencies run: uv lock --python ${{ matrix.python-version }} && uv sync --python ${{ matrix.python-version }} --all-extras + # Otherwise the free-threading tests would silently skip. + - name: Check the interpreter is free-threaded + if: endsWith(matrix.python-version, 't') + run: uv run python -c "import sys, sysconfig; sys.exit(0 if sysconfig.get_config_var('Py_GIL_DISABLED') else 'not a free-threaded build')" - name: Run pytest run: uv run pytest . diff --git a/CHANGELOG.md b/CHANGELOG.md index 0994d367..5cc9b562 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Free-threaded Python 3.14t and 3.15t support (beta). + +### Fixed + +- Loading a kubeconfig from several threads at once no longer leaves temp files with decoded + `*-data` credentials behind. +- A resource class first used from several threads at once builds its list model only once. + ## [0.1.0-beta.4] - 2026-10-28 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index a4fbc3b3..22216a79 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -196,6 +196,7 @@ test/ # Test suite ├── test_metadata_accessor.py # MetadataAccessor (get, list, patch) tests ├── test_invalid_objects.py # list()/watch(invalid=) parsing rules, is_initial_events_end, overload types (assert_type) ├── test_api_request_validation.py # Api-level name/namespace validation (nothing reaches the client) +├── test_free_threading.py # Free-threaded builds only (skipped elsewhere): importing kubex + httpx_ws/trio keeps the GIL off ├── test_models/ # Unit tests for kubex-core models, field types and generated-model behaviour │ └── test_eviction.py # Eviction model tests ├── test_patch/ # Unit tests for patch models @@ -217,7 +218,8 @@ test/ # Test suite ├── test_stream/ # Unit tests for _BaseChannelSession lifecycle + StreamSession channel multiplexer (shared by exec and attach) ├── test_client/ # Unit tests for BaseClient ABC, AioHttpClient, HttpxClient, and trust_env env-proxy + netrc resolver (test_trust_env.py); │ # conftest.py parametrizes backends (aiohttp/asyncio, httpx/asyncio, httpx/trio), _servers.py has local -│ # aiohttp/raw-TCP/TLS test servers (streaming, transport errors, tls_server_name, headers, WS keepalive) +│ # aiohttp/raw-TCP/TLS test servers (streaming, transport errors, tls_server_name, headers, WS keepalive); +│ # test_threads.py runs clients on several threads (one event loop each) sharing one ClientConfiguration ├── test_subresource_descriptors/ # Unit tests for descriptor-based subresource APIs └── test_timeout/ # Unit tests for HTTP timeout settings @@ -254,7 +256,7 @@ lychee.toml # Link checker configuration .github/workflows/ ├── lint.yaml # Pre-commit, ruff check, ruff format --check, mypy, codegen verify -├── test.yaml # pytest with all extras on every supported Python (3.11–3.15 matrix) +├── test.yaml # pytest with all extras on every supported Python (3.11–3.15 matrix, plus free-threaded 3.14t and 3.15t) ├── publish-test.yaml # Build + publish to Test PyPI on PRs (OIDC trusted publishing) └── publish.yaml # Build + publish to production PyPI on v* tag push (OIDC trusted publishing) ``` @@ -372,7 +374,9 @@ Five GitHub Actions workflows: 6. Verify generated packages — runs `python -m scripts.codegen verify` for each `packages/kubex-k8s-*` **Test** (`test.yaml`) — runs on push and pull_request: -1. `pytest .` with all extras, as a matrix over Python 3.11, 3.12, 3.13, 3.14, 3.15 (`fail-fast: false`) +1. `pytest .` with all extras, as a matrix over Python 3.11, 3.12, 3.13, 3.14, 3.14t, 3.15, 3.15t (`fail-fast: false`). + `t` marks a free-threaded build; a step first fails those jobs if the interpreter is not free-threaded, so + `test/test_free_threading.py` (GIL stays off after importing kubex) cannot skip silently **Publish Test** (`publish-test.yaml`) — runs on pull requests to `main`: 1. Appends `.devN` version suffix to all 9 packages diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index c68017b8..1cdb83cf 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -106,6 +106,9 @@ For most users, pick one HTTP client and one K8s version: Kubex is tested and supported on Python 3.11 through 3.15. +Free-threaded (no-GIL) builds of Python 3.14 and 3.15 (`3.14t`, `3.15t`) are tested too; support +for them is in beta. + ## Next steps Once installed, head to the [Quickstart](quickstart.md) to connect to your cluster and make your first request. diff --git a/kubex/configuration/file_config.py b/kubex/configuration/file_config.py index f9121e5a..0d1994e3 100644 --- a/kubex/configuration/file_config.py +++ b/kubex/configuration/file_config.py @@ -1,5 +1,7 @@ import atexit import os +import sys +import threading import warnings from base64 import b64decode, b64encode from pathlib import Path @@ -40,11 +42,29 @@ class _TempFile(NamedTuple): _temp_files: dict[str, _TempFile] = {} """Decoded `*-data` payload -> temp file holding it.""" _cleanup_registered = False +_temp_files_lock = threading.Lock() +"""Guards `_temp_files` and `_cleanup_registered`. + +Without it, threads loading the same kubeconfig at once each write their own +temp file and only the last one is tracked, so the others (decoded keys and +certificates) are never removed. +""" + +if sys.platform != "win32": + # Hold the lock across fork() so a child never inherits it locked by a + # thread that does not exist there. + os.register_at_fork( + before=_temp_files_lock.acquire, + after_in_parent=_temp_files_lock.release, + after_in_child=_temp_files_lock.release, + ) def _cleanup_temp_files() -> None: pid = os.getpid() - for temp_file in list(_temp_files.values()): + with _temp_files_lock: + temp_files = list(_temp_files.values()) + for temp_file in temp_files: if temp_file.owner_pid != pid: # Inherited across fork(): the creating process still uses it. continue @@ -473,16 +493,17 @@ def _decode_and_put_to_file(data: str) -> Path: only by the process that created it. """ global _cleanup_registered - cached = _temp_files.get(data) - if cached is not None and cached.path.exists(): - return cached.path - # Line breaks are tolerated (as by Go's decoder); other junk is an error. - decoded = b64decode("".join(data.split()), validate=True) - if not _cleanup_registered: - atexit.register(_cleanup_temp_files) - _cleanup_registered = True - with NamedTemporaryFile(delete=False) as f: - f.write(decoded) - path = Path(f.name) - _temp_files[data] = _TempFile(path, os.getpid()) - return path + with _temp_files_lock: + cached = _temp_files.get(data) + if cached is not None and cached.path.exists(): + return cached.path + # Line breaks are tolerated (as by Go's decoder); other junk is an error. + decoded = b64decode("".join(data.split()), validate=True) + if not _cleanup_registered: + atexit.register(_cleanup_temp_files) + _cleanup_registered = True + with NamedTemporaryFile(delete=False) as f: + f.write(decoded) + path = Path(f.name) + _temp_files[data] = _TempFile(path, os.getpid()) + return path diff --git a/packages/kubex-core/kubex_core/models/resource_config.py b/packages/kubex-core/kubex_core/models/resource_config.py index bf732880..9a23b014 100644 --- a/packages/kubex-core/kubex_core/models/resource_config.py +++ b/packages/kubex-core/kubex_core/models/resource_config.py @@ -1,5 +1,8 @@ from __future__ import annotations +import os +import sys +import threading from enum import Enum from functools import cached_property from typing import TYPE_CHECKING, Any, Generic, Literal, Self, Type, cast @@ -17,6 +20,24 @@ class Scope(Enum): NAMESPACE = "namespace" +_setup_lock = threading.RLock() +"""Makes the one-time setup in `ResourceConfig.__get__` happen once per class. + +Without it, threads touching a resource class for the first time at once each +build their own config copy and list model, and the last one stored wins. +Reentrant in case building a list model touches another resource's config. +""" + +if sys.platform != "win32": + # Hold the lock across fork() so a child never inherits it locked by a + # thread that does not exist there. + os.register_at_fork( + before=_setup_lock.acquire, + after_in_parent=_setup_lock.release, + after_in_child=_setup_lock.release, + ) + + class ResourceConfig(Generic[ResourceType]): """ResourceConfig is the configuration for a resource.""" @@ -50,18 +71,21 @@ def __get__(self, instance: Any, owner: Type[ResourceType]) -> Self: """ config = owner.__dict__.get("__RESOURCE_CONFIG__") if config is None: - # The list model is not copied: an inherited one is typed for the - # parent (e.g. `PodList` for `class MyPod(Pod)`), so `list()` - # would return parent instances. `_fill_from` builds one for - # `owner` instead. - config = ResourceConfig( - version=self._version, - kind=self._kind, - plural=self._plural, - scope=self._scope, - group=self._group, - ) - setattr(owner, "__RESOURCE_CONFIG__", config) + with _setup_lock: + config = owner.__dict__.get("__RESOURCE_CONFIG__") + if config is None: + # The list model is not copied: an inherited one is typed + # for the parent (e.g. `PodList` for `class MyPod(Pod)`), + # so `list()` would return parent instances. `_fill_from` + # builds one for `owner` instead. + config = ResourceConfig( + version=self._version, + kind=self._kind, + plural=self._plural, + scope=self._scope, + group=self._group, + ) + setattr(owner, "__RESOURCE_CONFIG__", config) return cast(Self, config._fill_from(owner)) def _fill_from(self, owner: Type[ResourceType]) -> Self: @@ -80,7 +104,9 @@ def _fill_from(self, owner: Type[ResourceType]) -> Self: else: return self if self._list_model is None: - self._list_model = create_list_model(owner, self) + with _setup_lock: + if self._list_model is None: + self._list_model = create_list_model(owner, self) if self._scope is None: self._scope = Scope.NAMESPACE if self._plural is None: diff --git a/packages/kubex-core/pyproject.toml b/packages/kubex-core/pyproject.toml index 531deded..4697b8a9 100644 --- a/packages/kubex-core/pyproject.toml +++ b/packages/kubex-core/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-32/pyproject.toml b/packages/kubex-k8s-1-32/pyproject.toml index eb0704e4..a45ec2b6 100644 --- a/packages/kubex-k8s-1-32/pyproject.toml +++ b/packages/kubex-k8s-1-32/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-33/pyproject.toml b/packages/kubex-k8s-1-33/pyproject.toml index c529a896..ab0d4ace 100644 --- a/packages/kubex-k8s-1-33/pyproject.toml +++ b/packages/kubex-k8s-1-33/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-34/pyproject.toml b/packages/kubex-k8s-1-34/pyproject.toml index 07c18f7d..d9410466 100644 --- a/packages/kubex-k8s-1-34/pyproject.toml +++ b/packages/kubex-k8s-1-34/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-35/pyproject.toml b/packages/kubex-k8s-1-35/pyproject.toml index 4667da49..c928f0e1 100644 --- a/packages/kubex-k8s-1-35/pyproject.toml +++ b/packages/kubex-k8s-1-35/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-36/pyproject.toml b/packages/kubex-k8s-1-36/pyproject.toml index 89656d38..f4918d80 100644 --- a/packages/kubex-k8s-1-36/pyproject.toml +++ b/packages/kubex-k8s-1-36/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-37/pyproject.toml b/packages/kubex-k8s-1-37/pyproject.toml index 7c21a9dc..2ff1e341 100644 --- a/packages/kubex-k8s-1-37/pyproject.toml +++ b/packages/kubex-k8s-1-37/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/packages/kubex-k8s-1-38/pyproject.toml b/packages/kubex-k8s-1-38/pyproject.toml index 6cb0fe16..36adf53f 100644 --- a/packages/kubex-k8s-1-38/pyproject.toml +++ b/packages/kubex-k8s-1-38/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/pyproject.toml b/pyproject.toml index ef9a79d7..94fb3280 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [project.urls] diff --git a/scripts/codegen/templates/pyproject.toml.j2 b/scripts/codegen/templates/pyproject.toml.j2 index 375f2666..c6fa8bb6 100644 --- a/scripts/codegen/templates/pyproject.toml.j2 +++ b/scripts/codegen/templates/pyproject.toml.j2 @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/scripts/codegen/tests/golden/kubex-k8s-1-30/pyproject.toml b/scripts/codegen/tests/golden/kubex-k8s-1-30/pyproject.toml index ba9e6b39..279dca3f 100644 --- a/scripts/codegen/tests/golden/kubex-k8s-1-30/pyproject.toml +++ b/scripts/codegen/tests/golden/kubex-k8s-1-30/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 2 - Beta", ] [build-system] diff --git a/test/test_client/_servers.py b/test/test_client/_servers.py index 24a8cf18..8a7976ac 100644 --- a/test/test_client/_servers.py +++ b/test/test_client/_servers.py @@ -43,25 +43,32 @@ def __enter__(self) -> ThreadedAioServer: errors: list[BaseException] = [] def _run() -> None: + loop: asyncio.AbstractEventLoop | None = None try: - loop = asyncio.new_event_loop() - self._loop = loop - asyncio.set_event_loop(loop) - runner = web.AppRunner(self.app, handle_signals=False) - loop.run_until_complete(runner.setup()) - site = web.TCPSite( - runner, "127.0.0.1", 0, ssl_context=self._ssl_context - ) - loop.run_until_complete(site.start()) - self._runner = runner - server: Any = site._server - self.port = server.sockets[0].getsockname()[1] - except BaseException as exc: # pragma: no cover - startup failure - errors.append(exc) + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + runner = web.AppRunner(self.app, handle_signals=False) + loop.run_until_complete(runner.setup()) + site = web.TCPSite( + runner, "127.0.0.1", 0, ssl_context=self._ssl_context + ) + loop.run_until_complete(site.start()) + server: Any = site._server + self.port = server.sockets[0].getsockname()[1] + self._loop = loop + self._runner = runner + except BaseException as exc: # pragma: no cover - startup failure + errors.append(exc) + ready.set() + return ready.set() - return - ready.set() - loop.run_forever() + loop.run_forever() + finally: + # Left open, the loop is closed by its finalizer later, which + # raises an unraisable-exception warning on free-threaded builds. + if loop is not None: + loop.close() self._thread = threading.Thread(target=_run, daemon=True) self._thread.start() diff --git a/test/test_client/test_httpx_websocket.py b/test/test_client/test_httpx_websocket.py index 4db11b22..cce5ae1a 100644 --- a/test/test_client/test_httpx_websocket.py +++ b/test/test_client/test_httpx_websocket.py @@ -14,76 +14,14 @@ from kubex.core.params import Timeout from kubex.core.request import Request +from ._servers import ThreadedAioServer + @pytest.fixture(params=["asyncio", "trio"]) def anyio_backend(request: pytest.FixtureRequest) -> str: return str(request.param) -class _ThreadedAioWsServer: - """Run an aiohttp echo WebSocket server on its own asyncio loop in a thread. - - This lets the test code run on either ``asyncio`` or ``trio`` while still - talking to a real aiohttp-served WebSocket endpoint over the wire. - """ - - def __init__(self, app: web.Application) -> None: - self.app = app - self._loop: asyncio.AbstractEventLoop | None = None - self._thread: threading.Thread | None = None - self._runner: web.AppRunner | None = None - self._site: web.TCPSite | None = None - self.port: int = 0 - - def start(self) -> None: - ready = threading.Event() - startup_error: list[BaseException] = [] - - def _run() -> None: - try: - loop = asyncio.new_event_loop() - self._loop = loop - asyncio.set_event_loop(loop) - runner = web.AppRunner(self.app) - loop.run_until_complete(runner.setup()) - site = web.TCPSite(runner, "127.0.0.1", 0) - loop.run_until_complete(site.start()) - self._runner = runner - self._site = site - server: Any = site._server - assert server is not None - sockets = server.sockets - assert sockets is not None - self.port = sockets[0].getsockname()[1] - except BaseException as exc: - startup_error.append(exc) - ready.set() - return - ready.set() - loop.run_forever() - - self._thread = threading.Thread(target=_run, daemon=True, name="aio-ws-server") - self._thread.start() - ready.wait() - if startup_error: - raise startup_error[0] - - def stop(self) -> None: - if self._loop is not None and self._runner is not None: - runner = self._runner - - async def _shutdown() -> None: - await runner.cleanup() - - try: - fut = asyncio.run_coroutine_threadsafe(_shutdown(), self._loop) - fut.result(timeout=5) - finally: - self._loop.call_soon_threadsafe(self._loop.stop) - if self._thread is not None: - self._thread.join(timeout=5) - - def _make_echo_app(server_protocols: list[str]) -> web.Application: async def echo_ws(request: web.Request) -> web.WebSocketResponse: ws = web.WebSocketResponse(protocols=server_protocols) @@ -103,32 +41,24 @@ async def echo_ws(request: web.Request) -> web.WebSocketResponse: @pytest.fixture -def ws_server() -> Generator[_ThreadedAioWsServer, None, None]: - server = _ThreadedAioWsServer(_make_echo_app(["v5.channel.k8s.io"])) - server.start() - try: +def ws_server() -> Generator[ThreadedAioServer, None, None]: + with ThreadedAioServer(_make_echo_app(["v5.channel.k8s.io"])) as server: yield server - finally: - server.stop() @pytest.fixture -def ws_server_no_proto() -> Generator[_ThreadedAioWsServer, None, None]: - server = _ThreadedAioWsServer(_make_echo_app([])) - server.start() - try: +def ws_server_no_proto() -> Generator[ThreadedAioServer, None, None]: + with ThreadedAioServer(_make_echo_app([])) as server: yield server - finally: - server.stop() -def _client_config(server: _ThreadedAioWsServer) -> ClientConfiguration: +def _client_config(server: ThreadedAioServer) -> ClientConfiguration: return ClientConfiguration(url=f"http://127.0.0.1:{server.port}") @pytest.mark.anyio async def test_connect_websocket_returns_websocket_connection( - ws_server: _ThreadedAioWsServer, + ws_server: ThreadedAioServer, ) -> None: config = _client_config(ws_server) async with HttpxClient(config) as client: @@ -142,7 +72,7 @@ async def test_connect_websocket_returns_websocket_connection( @pytest.mark.anyio async def test_connect_websocket_negotiates_subprotocol( - ws_server: _ThreadedAioWsServer, + ws_server: ThreadedAioServer, ) -> None: config = _client_config(ws_server) async with HttpxClient(config) as client: @@ -155,7 +85,7 @@ async def test_connect_websocket_negotiates_subprotocol( @pytest.mark.anyio async def test_connect_websocket_send_and_receive_bytes( - ws_server: _ThreadedAioWsServer, + ws_server: ThreadedAioServer, ) -> None: config = _client_config(ws_server) async with HttpxClient(config) as client: @@ -174,7 +104,7 @@ async def test_connect_websocket_send_and_receive_bytes( @pytest.mark.anyio async def test_connect_websocket_close_marks_closed( - ws_server: _ThreadedAioWsServer, + ws_server: ThreadedAioServer, ) -> None: config = _client_config(ws_server) async with HttpxClient(config) as client: @@ -187,7 +117,7 @@ async def test_connect_websocket_close_marks_closed( @pytest.mark.anyio async def test_connect_websocket_receive_after_remote_close_raises_stop( - ws_server: _ThreadedAioWsServer, + ws_server: ThreadedAioServer, ) -> None: config = _client_config(ws_server) async with HttpxClient(config) as client: @@ -204,7 +134,7 @@ async def test_connect_websocket_receive_after_remote_close_raises_stop( @pytest.mark.anyio async def test_connect_websocket_subprotocol_mismatch_raises( - ws_server_no_proto: _ThreadedAioWsServer, + ws_server_no_proto: ThreadedAioServer, ) -> None: config = _client_config(ws_server_no_proto) async with HttpxClient(config) as client: @@ -226,9 +156,7 @@ async def capture_ws(request: web.Request) -> web.WebSocketResponse: app = web.Application() app.router.add_get("/exec", capture_ws) - server = _ThreadedAioWsServer(app) - server.start() - try: + with ThreadedAioServer(app) as server: config = ClientConfiguration(url=f"http://127.0.0.1:{server.port}") async with HttpxClient(config) as client: request = Request( @@ -238,8 +166,6 @@ async def capture_ws(request: web.Request) -> web.WebSocketResponse: ) conn = await client.connect_websocket(request, ["v5.channel.k8s.io"]) await conn.close() - finally: - server.stop() assert ("command", "echo") in captured["query"] assert ("container", "main") in captured["query"] @@ -258,9 +184,7 @@ async def capture_ws(request: web.Request) -> web.WebSocketResponse: app = web.Application() app.router.add_get("/ws", capture_ws) - server = _ThreadedAioWsServer(app) - server.start() - try: + with ThreadedAioServer(app) as server: config = ClientConfiguration( url=f"http://127.0.0.1:{server.port}", token="secret-token", @@ -269,8 +193,6 @@ async def capture_ws(request: web.Request) -> web.WebSocketResponse: request = Request(method="GET", url="/ws") conn = await client.connect_websocket(request, ["v5.channel.k8s.io"]) await conn.close() - finally: - server.stop() assert captured["auth"] == "Bearer secret-token" @@ -289,9 +211,7 @@ async def closer(request: web.Request) -> web.WebSocketResponse: @pytest.mark.anyio async def test_receive_bytes_raises_on_abnormal_close_code() -> None: - server = _ThreadedAioWsServer(_make_abnormal_close_app(1011)) - server.start() - try: + with ThreadedAioServer(_make_abnormal_close_app(1011)) as server: config = ClientConfiguration(url=f"http://127.0.0.1:{server.port}") async with HttpxClient(config) as client: request = Request(method="GET", url="/ws") @@ -301,15 +221,11 @@ async def test_receive_bytes_raises_on_abnormal_close_code() -> None: await conn.receive_bytes() finally: await conn.close() - finally: - server.stop() @pytest.mark.anyio async def test_receive_bytes_treats_normal_close_as_eof() -> None: - server = _ThreadedAioWsServer(_make_abnormal_close_app(1000)) - server.start() - try: + with ThreadedAioServer(_make_abnormal_close_app(1000)) as server: config = ClientConfiguration(url=f"http://127.0.0.1:{server.port}") async with HttpxClient(config) as client: request = Request(method="GET", url="/ws") @@ -319,8 +235,6 @@ async def test_receive_bytes_treats_normal_close_as_eof() -> None: await conn.receive_bytes() finally: await conn.close() - finally: - server.stop() class _RawCloseNoCodeServer: @@ -438,16 +352,12 @@ async def hang_forever(request: web.Request) -> web.WebSocketResponse: app = web.Application() app.router.add_get("/ws", hang_forever) - server = _ThreadedAioWsServer(app) - server.start() - try: + with ThreadedAioServer(app) as server: config = ClientConfiguration(url=f"http://127.0.0.1:{server.port}") async with HttpxClient(config) as client: request = Request(method="GET", url="/ws", timeout=Timeout(total=0.1)) with pytest.raises(KubexClientException, match="handshake timed out"): await client.connect_websocket(request, ["v5.channel.k8s.io"]) - finally: - server.stop() @pytest.mark.anyio diff --git a/test/test_client/test_threads.py b/test/test_client/test_threads.py new file mode 100644 index 00000000..9fe7c79a --- /dev/null +++ b/test/test_client/test_threads.py @@ -0,0 +1,97 @@ +"""Clients on separate threads, each running its own event loop.""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any + +import anyio +import pytest +from aiohttp import web + +import kubex.configuration.configuration as configuration_module +from kubex.api import Api +from kubex.client import ClientChoice, create_client +from kubex.configuration import ClientConfiguration +from kubex.k8s.v1_35.core.v1.pod import Pod + +from ._servers import ThreadedAioServer + +_COMBOS = ( + ("asyncio", ClientChoice.AIOHTTP), + ("asyncio", ClientChoice.HTTPX), + ("trio", ClientChoice.HTTPX), +) + + +def _pod(name: str) -> dict[str, Any]: + return { + "apiVersion": "v1", + "kind": "Pod", + "metadata": {"name": name, "namespace": "default"}, + } + + +def test_clients_on_separate_threads_share_one_configuration( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + token_file = tmp_path / "token" + token_file.write_text("file-token") + # Re-read the token file on every request, from every thread at once. + monkeypatch.setattr(configuration_module, "TOKEN_REFRESH_INTERVAL", -1) + wrong_auth: list[str | None] = [] + + def check_auth(request: web.Request) -> None: + auth = request.headers.get("Authorization") + if auth != "Bearer file-token": + wrong_auth.append(auth) + + async def list_pods(request: web.Request) -> web.Response: + check_auth(request) + items = [_pod(f"pod-{i}") for i in range(3)] + return web.json_response( + {"apiVersion": "v1", "kind": "PodList", "metadata": {}, "items": items} + ) + + async def get_pod(request: web.Request) -> web.Response: + check_auth(request) + return web.json_response(_pod(request.match_info["name"])) + + app = web.Application() + app.router.add_get("/api/v1/namespaces/default/pods", list_pods) + app.router.add_get("/api/v1/namespaces/default/pods/{name}", get_pod) + threads = 2 * len(_COMBOS) + barrier = threading.Barrier(threads) + + with ThreadedAioServer(app) as server: + # The file's token takes precedence over the inline one, so every + # request from every thread must carry it. + config = ClientConfiguration( + url=server.url, + token="inline-token", + token_file=token_file, + try_refresh_token=True, + ) + + def worker(idx: int) -> None: + backend, choice = _COMBOS[idx % len(_COMBOS)] + + async def main() -> None: + async with await create_client(config, choice) as client: + api: Api[Pod] = Api(Pod, client=client, namespace="default") + for k in range(10): + listed = await api.list() + assert len(listed.items) == 3 + pod = await api.get(f"pod-{k}") + assert pod.metadata is not None + assert pod.metadata.name == f"pod-{k}" + + barrier.wait() + anyio.run(main, backend=backend) + + with ThreadPoolExecutor(threads) as pool: + list(pool.map(worker, range(threads))) + + assert wrong_auth == [] diff --git a/test/test_configuration/test_file_config.py b/test/test_configuration/test_file_config.py index d8631c11..c4f86cfe 100644 --- a/test/test_configuration/test_file_config.py +++ b/test/test_configuration/test_file_config.py @@ -1,16 +1,19 @@ from __future__ import annotations from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor from typing import Any import atexit import json import os import sys +import threading import time import warnings from base64 import b64encode, urlsafe_b64encode from pathlib import Path +from tempfile import NamedTemporaryFile from unittest.mock import patch import pytest @@ -590,6 +593,53 @@ def test_decode_and_put_to_file_recreates_deleted_file( assert second.read_bytes() == b"recreate-me" +def test_decode_and_put_to_file_concurrent_calls_share_one_file( + clean_temp_files: dict[str, _TempFile], monkeypatch: pytest.MonkeyPatch +) -> None: + # Threads loading the same kubeconfig at once must not each write a file: + # only the tracked one is removed at exit, the others would leak secrets. + def slow_named_temporary_file(*args: Any, **kwargs: Any) -> Any: + time.sleep(0.01) # widen the race window + return NamedTemporaryFile(*args, **kwargs) + + monkeypatch.setattr(file_config, "NamedTemporaryFile", slow_named_temporary_file) + encoded = b64encode(b"concurrent-key").decode() + threads = 8 + barrier = threading.Barrier(threads) + + def load(_: int) -> Path: + barrier.wait() + return _decode_and_put_to_file(encoded) + + with ThreadPoolExecutor(threads) as pool: + paths = set(pool.map(load, range(threads))) + assert paths == {clean_temp_files[encoded].path} + + +@pytest.mark.skipif(sys.platform == "win32", reason="no fork() on Windows") +def test_fork_while_another_thread_holds_temp_files_lock() -> None: + # The child must not inherit the lock held by a thread it does not have. + held = threading.Event() + + def hold() -> None: + with file_config._temp_files_lock: + held.set() + time.sleep(0.2) + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + with warnings.catch_warnings(): + # Forking with other threads running warns on Python 3.12+. + warnings.simplefilter("ignore", DeprecationWarning) + pid = os.fork() + if pid == 0: + os._exit(0 if file_config._temp_files_lock.acquire(timeout=1) else 1) + holder.join() + _, status = os.waitpid(pid, 0) + assert os.waitstatus_to_exitcode(status) == 0 + + def test_cleanup_temp_files_skips_files_created_by_another_process( clean_temp_files: dict[str, _TempFile], tmp_path: Path ) -> None: diff --git a/test/test_free_threading.py b/test/test_free_threading.py new file mode 100644 index 00000000..cd6950f0 --- /dev/null +++ b/test/test_free_threading.py @@ -0,0 +1,50 @@ +"""Checks that only mean something on a free-threaded (no-GIL) build.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import sysconfig + +import pytest + +pytestmark = pytest.mark.skipif( + not sysconfig.get_config_var("Py_GIL_DISABLED"), + reason="needs a free-threaded build", +) + +# kubex's entry points import every compiled dependency; httpx-ws and trio are +# only imported once used, so they are listed too. +_MODULES = ( + "kubex.api", + "kubex.client.aiohttp", + "kubex.client.httpx", + "kubex.configuration", + "kubex.k8s.v1_35.core.v1.pod", + "httpx_ws", + "trio", +) + +_SCRIPT = """ +import importlib, sys +for name in sys.argv[1:]: + importlib.import_module(name) + if sys._is_gil_enabled(): + sys.exit(f"importing {name} turned the GIL back on") +""" + + +def test_importing_kubex_keeps_the_gil_disabled() -> None: + # CPython turns the GIL back on, with only a warning, when it loads an + # extension module that does not declare free-threading support. Checked + # in a fresh interpreter: this one has already imported pytest's plugins. + env = {k: v for k, v in os.environ.items() if k != "PYTHON_GIL"} + result = subprocess.run( + [sys.executable, "-c", _SCRIPT, *_MODULES], + capture_output=True, + text=True, + env=env, + timeout=120, + ) + assert result.returncode == 0, result.stderr diff --git a/test/test_models/test_resource_config.py b/test/test_models/test_resource_config.py index 2ee0d552..ce583590 100644 --- a/test/test_models/test_resource_config.py +++ b/test/test_models/test_resource_config.py @@ -1,11 +1,18 @@ from __future__ import annotations import importlib +import os import pkgutil -from typing import ClassVar +import sys +import threading +import time +import warnings +from concurrent.futures import ThreadPoolExecutor +from typing import Any, ClassVar import pytest +import kubex_core.models.resource_config as resource_config_module from kubex_core.models.base_entity import BaseEntity from kubex_core.models.interfaces import NamespaceScopedEntity from kubex_core.models.metadata import ObjectMetadata @@ -175,6 +182,64 @@ class _Child(_Base): assert type(listed.items[0]) is _Child +def test_concurrent_first_access_sets_up_config_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Contended(NamespaceScopedEntity): + api_version: str = "example.io/v1" + kind: str = "Contended" + metadata: ObjectMetadata + + built: list[object] = [] + real_create_list_model = resource_config_module.create_list_model + + def slow_create_list_model(single_model: Any, config: Any) -> Any: + time.sleep(0.01) # widen the race window + list_model = real_create_list_model(single_model, config) + built.append(list_model) + return list_model + + monkeypatch.setattr( + resource_config_module, "create_list_model", slow_create_list_model + ) + threads = 8 + barrier = threading.Barrier(threads) + + def touch(_: int) -> tuple[object, object]: + barrier.wait() + config = _Contended.__RESOURCE_CONFIG__ + return config, config.list_model + + with ThreadPoolExecutor(threads) as pool: + results = set(pool.map(touch, range(threads))) + assert len(built) == 1 + assert results == {(_Contended.__RESOURCE_CONFIG__, built[0])} + + +@pytest.mark.skipif(sys.platform == "win32", reason="no fork() on Windows") +def test_fork_while_another_thread_holds_setup_lock() -> None: + # The child must not inherit the lock held by a thread it does not have. + held = threading.Event() + + def hold() -> None: + with resource_config_module._setup_lock: + held.set() + time.sleep(0.2) + + holder = threading.Thread(target=hold) + holder.start() + held.wait() + with warnings.catch_warnings(): + # Forking with other threads running warns on Python 3.12+. + warnings.simplefilter("ignore", DeprecationWarning) + pid = os.fork() + if pid == 0: + os._exit(0 if resource_config_module._setup_lock.acquire(timeout=1) else 1) + holder.join() + _, status = os.waitpid(pid, 0) + assert os.waitstatus_to_exitcode(status) == 0 + + def test_generated_list_model_reads_null_items() -> None: class _Thing(NamespaceScopedEntity): api_version: str = "example.io/v1"