From d8395b9812bbb4ea0d44f9a8139259e043b6e6ef Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 4 Aug 2026 09:43:46 -0700 Subject: [PATCH 1/3] [python] Reclaim test pipelines a killed CI run leaves running Pipelines from the runtime suites keep running on the shared CI instance long after their run ends, consuming compute until the daily cleanup job stops them. Teardown only runs while pytest is alive. Cancelling the workflow run, evicting a runner or timing out a job kills it outright, and every pipeline in flight survives. A new 'if: always()' step sweeps up what such a run left behind, matching the name prefix unique_pipeline_name stamps on every test pipeline so it only ever touches its own run. It stops and clears storage but never deletes, so a failed run stays readable. Three defects made teardown fail on its own as well: - Every stop and clear waited without a timeout, so one wedged pipeline hung teardown until the runner killed the job, leaking all the rest. reclaim_pipeline() now bounds each wait. - stop() raising skipped the clear that followed it. Every step now runs regardless, since a pipeline that never reports Stopped can still release its storage. - Pipelines from new_pipeline_with_suffix() were in no teardown at all. SharedTestPipeline now tracks everything it builds. Cleanup failures warn rather than fail a test: a slow stop says nothing about what the test asserted, and the sweep reports anything genuinely left running. Signed-off-by: Gerd Zellweger --- .../workflows/test-integration-runtime.yml | 12 ++ python/feldera/testutils.py | 36 ++++++ python/tests/README.md | 18 +++ python/tests/platform/helper.py | 24 ++-- python/tests/runtime_aggtest/aggtst_base.py | 10 +- python/tests/shared_test_pipeline.py | 33 ++--- python/tests/stop_ci_run_pipelines.py | 115 ++++++++++++++++++ python/tests/unit/test_reclaim_pipeline.py | 93 ++++++++++++++ 8 files changed, 316 insertions(+), 25 deletions(-) create mode 100644 python/tests/stop_ci_run_pipelines.py create mode 100644 python/tests/unit/test_reclaim_pipeline.py diff --git a/.github/workflows/test-integration-runtime.yml b/.github/workflows/test-integration-runtime.yml index a90e68fa9cb..d8b2a2c7a59 100644 --- a/.github/workflows/test-integration-runtime.yml +++ b/.github/workflows/test-integration-runtime.yml @@ -79,3 +79,15 @@ jobs: KAFKA_BOOTSTRAP_SERVERS: ${{ vars.CI_KAFKA_BOOTSTRAP }} SCHEMA_REGISTRY_URL: ${{ vars.CI_SCHEMA_REGISTRY }} RUN_ID: 1 # we use this to run a single variant of the Kafka tests in test_kafka_avro.py + + # Test teardown only runs while pytest is alive. A cancel sentinel firing + # for some other job, a runner eviction or a job timeout leaves tests + # running on the shared instance to burn compute. + # This step reclaims the pipelines this run named after itself + - name: Stop pipelines this run leaked + if: ${{ always() && vars.CI_DRY_RUN != 'true' }} + timeout-minutes: 10 + run: uv run --locked python -m tests.stop_ci_run_pipelines + working-directory: python + env: + PYTHONPATH: ${{ github.workspace }}/python diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index 010ffb373d1..59237138c64 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -287,6 +287,42 @@ def unique_pipeline_name(base_name: str) -> str: return name +# Teardown must not block on one wedged pipeline: the SDK otherwise polls for +# `Stopped` forever, and when the runner finally kills the job every pipeline +# the run started keeps consuming compute on the shared instance. +RECLAIM_TIMEOUT_SECONDS = 60.0 + + +def reclaim_pipeline(name: str, delete: bool = False) -> List[str]: + """Force-stop `name`, clear its storage, and optionally delete it. + + Every step runs even when the one before it raised: a pipeline that never + reports `Stopped` can still release its storage, and abandoning the rest on + the first error leaves them running on a shared instance. Returns one + message per failed step, empty when the pipeline is fully reclaimed, so the + caller decides whether a cleanup failure is worth failing a test over. + """ + failures = [] + + try: + TEST_CLIENT.stop_pipeline(name, force=True, timeout_s=RECLAIM_TIMEOUT_SECONDS) + except Exception as error: + failures.append(f"{name}: stop: {error}") + + try: + TEST_CLIENT.clear_storage(name, timeout_s=RECLAIM_TIMEOUT_SECONDS) + except Exception as error: + failures.append(f"{name}: clear storage: {error}") + + if delete: + try: + TEST_CLIENT.delete_pipeline(name) + except Exception as error: + failures.append(f"{name}: delete: {error}") + + return failures + + def enterprise_only(fn): fn._enterprise_only = True return unittest.skipUnless( diff --git a/python/tests/README.md b/python/tests/README.md index db9af0778ad..2b4c167c695 100644 --- a/python/tests/README.md +++ b/python/tests/README.md @@ -111,3 +111,21 @@ after the test completed. def test_some_property(pipeline_name): pass ``` + +## Cleaning up after a run + +Teardown only runs while the test process is alive. A cancelled workflow run, an +evicted runner or a job timeout kills pytest outright, and every pipeline that +was running at that moment keeps consuming compute on the shared CI instance. + +`tests/stop_ci_run_pipelines.py` sweeps up what such a run left behind. It +matches the prefix `unique_pipeline_name` stamps on every test pipeline. + +```bash +cd python +PYTHONPATH=`pwd` uv run python -m tests.stop_ci_run_pipelines --prefix _ +``` + +Whatever cleanup a test does itself should go through +`feldera.testutils.reclaim_pipeline`: it bounds every wait, and it clears +storage even when the stop failed instead of abandoning a running pipeline. diff --git a/python/tests/platform/helper.py b/python/tests/platform/helper.py index 048ae378c68..efaca7189bf 100644 --- a/python/tests/platform/helper.py +++ b/python/tests/platform/helper.py @@ -13,6 +13,7 @@ from __future__ import annotations import json +import logging import os import unittest from http import HTTPStatus @@ -21,7 +22,11 @@ import pytest import requests -from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS +from feldera.testutils import ( + FELDERA_TEST_NUM_HOSTS, + FELDERA_TEST_NUM_WORKERS, + reclaim_pipeline, +) from feldera.testutils_oidc import get_oidc_test_helper from tests import ( @@ -35,6 +40,8 @@ API_PREFIX = "/v0" +logger = logging.getLogger(__name__) + def _base_headers() -> Dict[str, str]: headers = { @@ -162,22 +169,25 @@ def pause_pipeline(name: str, wait: bool = True): TEST_CLIENT.pause_pipeline(name, wait=wait) -def stop_pipeline(name: str, force: bool = True, wait: bool = True): - TEST_CLIENT.stop_pipeline(name, force=force, wait=wait) +def stop_pipeline( + name: str, force: bool = True, wait: bool = True, timeout_s: float | None = None +): + TEST_CLIENT.stop_pipeline(name, force=force, wait=wait, timeout_s=timeout_s) -def clear_pipeline(name: str, wait: bool = True): +def clear_pipeline(name: str, wait: bool = True, timeout_s: float | None = None): r = post_no_body(f"{API_PREFIX}/pipelines/{name}/clear") if wait and r.status_code == HTTPStatus.ACCEPTED: - TEST_CLIENT.clear_storage(name) + TEST_CLIENT.clear_storage(name, timeout_s=timeout_s) return r def reset_pipeline(name: str): + """Force-stop `name` and clear its storage.""" if get_pipeline(name, "status").status_code != HTTPStatus.OK: return - stop_pipeline(name, force=True) - clear_pipeline(name) + for failure in reclaim_pipeline(name): + logger.warning("pipeline cleanup: %s", failure) def delete_pipeline(name: str): diff --git a/python/tests/runtime_aggtest/aggtst_base.py b/python/tests/runtime_aggtest/aggtst_base.py index df420cb5fce..abb5d0d4963 100644 --- a/python/tests/runtime_aggtest/aggtst_base.py +++ b/python/tests/runtime_aggtest/aggtst_base.py @@ -9,7 +9,11 @@ from feldera.enums import CompilationProfile from feldera.rest.errors import FelderaAPIError from feldera.runtime_config import Resources, RuntimeConfig -from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS +from feldera.testutils import ( + FELDERA_TEST_NUM_WORKERS, + FELDERA_TEST_NUM_HOSTS, + reclaim_pipeline, +) from tests import TEST_CLIENT, unique_pipeline_name JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None @@ -325,8 +329,8 @@ def run_pipeline( raise if pipeline is not None: - pipeline.stop(force=True) - pipeline.delete(True) + for failure in reclaim_pipeline(pipeline_name, delete=True): + print(f"WARNING: {failure}") def assert_expected_error(self, obj: SqlObject, actual_exception: Exception): """Validate the error produced by the failing pipeline with the expected error type""" diff --git a/python/tests/shared_test_pipeline.py b/python/tests/shared_test_pipeline.py index afcfdaaddec..3da91c98820 100644 --- a/python/tests/shared_test_pipeline.py +++ b/python/tests/shared_test_pipeline.py @@ -1,3 +1,4 @@ +import logging import unittest from feldera import Pipeline, PipelineBuilder @@ -5,10 +6,13 @@ from feldera.testutils import ( FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS, + reclaim_pipeline, unique_pipeline_name, ) from tests import TEST_CLIENT +logger = logging.getLogger(__name__) + def sql(text_or_iterable): """ @@ -59,30 +63,27 @@ def setUpClass(cls): ).create_or_replace() def setUp(self): - p = PipelineBuilder( - self.client, - unique_pipeline_name(self._testMethodName), - sql=self.ddl, - runtime_config=RuntimeConfig( - workers=FELDERA_TEST_NUM_WORKERS, - hosts=FELDERA_TEST_NUM_HOSTS, - logging="debug", - ), - ).create_or_replace() - self.p = p + self._owned_pipelines = [] + self.p = self._build_pipeline(self._testMethodName) def tearDown(self): - self.p.stop(force=True) - self.p.clear_storage() + """Force-stop and clear every pipeline the test owns.""" + for pipeline in self._owned_pipelines: + for failure in reclaim_pipeline(pipeline.name): + logger.warning("pipeline teardown: %s", failure) @property def pipeline(self) -> Pipeline: return self.p def new_pipeline_with_suffix(self, suffix: str) -> Pipeline: - return PipelineBuilder( + return self._build_pipeline(f"{self._testMethodName}_{suffix}") + + def _build_pipeline(self, base_name: str) -> Pipeline: + """Create a pipeline over the class DDL and hand it to `tearDown`.""" + pipeline = PipelineBuilder( self.client, - unique_pipeline_name(f"{self._testMethodName}_{suffix}"), + unique_pipeline_name(base_name), sql=self.ddl, runtime_config=RuntimeConfig( workers=FELDERA_TEST_NUM_WORKERS, @@ -90,3 +91,5 @@ def new_pipeline_with_suffix(self, suffix: str) -> Pipeline: logging="debug", ), ).create_or_replace() + self._owned_pipelines.append(pipeline) + return pipeline diff --git a/python/tests/stop_ci_run_pipelines.py b/python/tests/stop_ci_run_pipelines.py new file mode 100644 index 00000000000..018faa896c3 --- /dev/null +++ b/python/tests/stop_ci_run_pipelines.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Stop every pipeline the current CI run left behind on a shared instance. + +Per-test teardown only runs while the test process is alive. A cancelled +workflow run, an evicted runner or a job timeout kills pytest outright, and +every pipeline that was running at that moment keeps burning compute on the +shared instance until the daily sweep finds it. Run this as an `if: always()` +step at the end of a job that talks to a shared instance. + +The sweep matches on the prefix `feldera.testutils.unique_pipeline_name` +stamps on every test pipeline, so it only ever touches this run's own +pipelines. It never fails the job: a red cleanup step on an otherwise green +run tells nobody anything the warnings do not, and the daily sweep is still +the backstop for a runner that dies before this step runs. + +It stops pipelines and clears their storage; it never deletes them. Stopped +and cleared, a pipeline consumes nothing, and its record is what you read to +work out why the run failed. Deleting is the daily sweep's job. + +Run it as a module, not as a path: `tests/` holds a `platform` package that +shadows the standard library one for anything imported by a script living +there. + +Usage: + PYTHONPATH=$PWD uv run python -m tests.stop_ci_run_pipelines [--prefix P] +""" + +from __future__ import annotations + +import argparse +import os +import sys +from concurrent.futures import ThreadPoolExecutor + +from feldera.testutils import ( + BASE_URL, + TEST_CLIENT, + reclaim_pipeline, + unique_pipeline_name, +) + +# A cancelled job gets a short grace window before the runner kills it, so the +# sweep works on several pipelines at once rather than serially. +MAX_CONCURRENT_RECLAIMS = 8 + + +def ci_run_prefix() -> str: + """The prefix `unique_pipeline_name` gives every pipeline of this run.""" + return unique_pipeline_name("") + + +def warn(message: str) -> None: + if os.environ.get("GITHUB_ACTIONS"): + print(f"::warning::{message}", flush=True) + else: + print(f"WARNING: {message}", flush=True) + + +def is_reclaimed(deployment_status: str | None, storage_status: str | None) -> bool: + """True once a pipeline holds neither compute nor storage.""" + return deployment_status == "Stopped" and storage_status == "Cleared" + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--prefix", + default=None, + help="Pipeline name prefix to sweep. Defaults to this run's own prefix, " + "the first five characters of GITHUB_SHA plus FELDERA_TEST_TAG_SUFFIX.", + ) + args = parser.parse_args() + + prefix = args.prefix or ci_run_prefix() + print(f"Sweeping pipelines named '{prefix}*' on {BASE_URL}", flush=True) + + try: + pipelines = TEST_CLIENT.pipelines() + except Exception as error: + warn(f"could not list pipelines on {BASE_URL}: {error}") + return 0 + + to_reclaim = [ + pipeline.name + for pipeline in pipelines + if pipeline.name.startswith(prefix) + and not is_reclaimed(pipeline.deployment_status, pipeline.storage_status) + ] + + if not to_reclaim: + print("No leaked pipelines.", flush=True) + return 0 + + warn( + f"{len(to_reclaim)} pipeline(s) outlived their test and are being reclaimed: " + + ", ".join(sorted(to_reclaim)) + ) + + def reclaim(name: str) -> tuple[str, list[str]]: + return name, reclaim_pipeline(name) + + with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_RECLAIMS) as pool: + for name, failures in pool.map(reclaim, to_reclaim): + if failures: + warn("; ".join(failures)) + else: + print(f"Reclaimed {name}", flush=True) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/tests/unit/test_reclaim_pipeline.py b/python/tests/unit/test_reclaim_pipeline.py new file mode 100644 index 00000000000..48722755361 --- /dev/null +++ b/python/tests/unit/test_reclaim_pipeline.py @@ -0,0 +1,93 @@ +"""Unit tests for the CI pipeline sweep: reclaiming and target selection.""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from feldera import testutils +from tests import stop_ci_run_pipelines + + +@pytest.fixture() +def client() -> MagicMock: + """Stand in for the module-level TEST_CLIENT of both modules under test.""" + mock = MagicMock() + with ( + patch.object(testutils, "TEST_CLIENT", mock), + patch.object(stop_ci_run_pipelines, "TEST_CLIENT", mock), + ): + yield mock + + +class TestReclaimPipeline: + def test_reclaims_a_running_pipeline(self, client: MagicMock): + assert testutils.reclaim_pipeline("p") == [] + client.stop_pipeline.assert_called_once() + client.clear_storage.assert_called_once() + client.delete_pipeline.assert_not_called() + + def test_bounds_the_wait_on_every_step(self, client: MagicMock): + testutils.reclaim_pipeline("p", delete=True) + for call in (client.stop_pipeline, client.clear_storage): + assert call.call_args.kwargs["timeout_s"] is not None + + def test_clears_storage_even_when_stopping_fails(self, client: MagicMock): + client.stop_pipeline.side_effect = RuntimeError("still Stopping") + + failures = testutils.reclaim_pipeline("p", delete=True) + + client.clear_storage.assert_called_once() + client.delete_pipeline.assert_called_once() + assert failures == ["p: stop: still Stopping"] + + def test_reports_every_failed_step(self, client: MagicMock): + client.stop_pipeline.side_effect = RuntimeError("no") + client.clear_storage.side_effect = RuntimeError("nope") + client.delete_pipeline.side_effect = RuntimeError("never") + + assert testutils.reclaim_pipeline("p", delete=True) == [ + "p: stop: no", + "p: clear storage: nope", + "p: delete: never", + ] + + +def pipeline(name: str, deployment: str, storage: str) -> SimpleNamespace: + return SimpleNamespace( + name=name, deployment_status=deployment, storage_status=storage + ) + + +class TestSweep: + ALL = [ + pipeline("abcde_test_running", "Running", "InUse"), + pipeline("abcde_test_stopped_but_not_cleared", "Stopped", "InUse"), + pipeline("abcde_test_done", "Stopped", "Cleared"), + pipeline("fffff_other_run_running", "Running", "InUse"), + ] + + def reclaimed_names(self, client: MagicMock, argv: list[str]) -> list[str]: + client.pipelines.return_value = self.ALL + with patch("sys.argv", ["stop_ci_run_pipelines", "--prefix", "abcde_", *argv]): + assert stop_ci_run_pipelines.main() == 0 + return sorted(call.args[0] for call in client.stop_pipeline.call_args_list) + + def test_leaves_other_runs_alone(self, client: MagicMock): + assert "fffff_other_run_running" not in self.reclaimed_names(client, []) + + def test_reclaims_whatever_still_holds_compute_or_storage(self, client: MagicMock): + assert self.reclaimed_names(client, []) == [ + "abcde_test_running", + "abcde_test_stopped_but_not_cleared", + ] + + def test_never_deletes(self, client: MagicMock): + """Records survive the sweep: a failed run is read off its pipelines.""" + self.reclaimed_names(client, []) + client.delete_pipeline.assert_not_called() + + def test_survives_an_unreachable_instance(self, client: MagicMock): + client.pipelines.side_effect = RuntimeError("connection refused") + with patch("sys.argv", ["stop_ci_run_pipelines"]): + assert stop_ci_run_pipelines.main() == 0 From 43c23f6c165966342d0200de2f2302a92e72472b Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 4 Aug 2026 10:31:58 -0700 Subject: [PATCH 2/3] [python] stop_ci_run_pipelines: address review comments Review feedback: ci_run_prefix() wrapped one call for one call site, so the call site names the prefix itself. warn() now says why it prints a workflow command under Actions and plain text elsewhere: ::warning:: reaches the run's annotation summary, and nothing reads it outside a run. Drop tests/unit/test_reclaim_pipeline.py. It drove a mocked client and mostly restated the two functions it covered. Signed-off-by: Gerd Zellweger --- python/tests/stop_ci_run_pipelines.py | 10 +-- python/tests/unit/test_reclaim_pipeline.py | 93 ---------------------- 2 files changed, 4 insertions(+), 99 deletions(-) delete mode 100644 python/tests/unit/test_reclaim_pipeline.py diff --git a/python/tests/stop_ci_run_pipelines.py b/python/tests/stop_ci_run_pipelines.py index 018faa896c3..f4eada043e9 100644 --- a/python/tests/stop_ci_run_pipelines.py +++ b/python/tests/stop_ci_run_pipelines.py @@ -44,12 +44,9 @@ MAX_CONCURRENT_RECLAIMS = 8 -def ci_run_prefix() -> str: - """The prefix `unique_pipeline_name` gives every pipeline of this run.""" - return unique_pipeline_name("") - - def warn(message: str) -> None: + # `::warning::` reaches the run's annotation summary, which is where a + # cleanup that found something has to show up. Nothing reads it elsewhere. if os.environ.get("GITHUB_ACTIONS"): print(f"::warning::{message}", flush=True) else: @@ -73,7 +70,8 @@ def main() -> int: ) args = parser.parse_args() - prefix = args.prefix or ci_run_prefix() + # `unique_pipeline_name` stamps this prefix on every pipeline of this run. + prefix = args.prefix or unique_pipeline_name("") print(f"Sweeping pipelines named '{prefix}*' on {BASE_URL}", flush=True) try: diff --git a/python/tests/unit/test_reclaim_pipeline.py b/python/tests/unit/test_reclaim_pipeline.py deleted file mode 100644 index 48722755361..00000000000 --- a/python/tests/unit/test_reclaim_pipeline.py +++ /dev/null @@ -1,93 +0,0 @@ -"""Unit tests for the CI pipeline sweep: reclaiming and target selection.""" - -from types import SimpleNamespace -from unittest.mock import MagicMock, patch - -import pytest - -from feldera import testutils -from tests import stop_ci_run_pipelines - - -@pytest.fixture() -def client() -> MagicMock: - """Stand in for the module-level TEST_CLIENT of both modules under test.""" - mock = MagicMock() - with ( - patch.object(testutils, "TEST_CLIENT", mock), - patch.object(stop_ci_run_pipelines, "TEST_CLIENT", mock), - ): - yield mock - - -class TestReclaimPipeline: - def test_reclaims_a_running_pipeline(self, client: MagicMock): - assert testutils.reclaim_pipeline("p") == [] - client.stop_pipeline.assert_called_once() - client.clear_storage.assert_called_once() - client.delete_pipeline.assert_not_called() - - def test_bounds_the_wait_on_every_step(self, client: MagicMock): - testutils.reclaim_pipeline("p", delete=True) - for call in (client.stop_pipeline, client.clear_storage): - assert call.call_args.kwargs["timeout_s"] is not None - - def test_clears_storage_even_when_stopping_fails(self, client: MagicMock): - client.stop_pipeline.side_effect = RuntimeError("still Stopping") - - failures = testutils.reclaim_pipeline("p", delete=True) - - client.clear_storage.assert_called_once() - client.delete_pipeline.assert_called_once() - assert failures == ["p: stop: still Stopping"] - - def test_reports_every_failed_step(self, client: MagicMock): - client.stop_pipeline.side_effect = RuntimeError("no") - client.clear_storage.side_effect = RuntimeError("nope") - client.delete_pipeline.side_effect = RuntimeError("never") - - assert testutils.reclaim_pipeline("p", delete=True) == [ - "p: stop: no", - "p: clear storage: nope", - "p: delete: never", - ] - - -def pipeline(name: str, deployment: str, storage: str) -> SimpleNamespace: - return SimpleNamespace( - name=name, deployment_status=deployment, storage_status=storage - ) - - -class TestSweep: - ALL = [ - pipeline("abcde_test_running", "Running", "InUse"), - pipeline("abcde_test_stopped_but_not_cleared", "Stopped", "InUse"), - pipeline("abcde_test_done", "Stopped", "Cleared"), - pipeline("fffff_other_run_running", "Running", "InUse"), - ] - - def reclaimed_names(self, client: MagicMock, argv: list[str]) -> list[str]: - client.pipelines.return_value = self.ALL - with patch("sys.argv", ["stop_ci_run_pipelines", "--prefix", "abcde_", *argv]): - assert stop_ci_run_pipelines.main() == 0 - return sorted(call.args[0] for call in client.stop_pipeline.call_args_list) - - def test_leaves_other_runs_alone(self, client: MagicMock): - assert "fffff_other_run_running" not in self.reclaimed_names(client, []) - - def test_reclaims_whatever_still_holds_compute_or_storage(self, client: MagicMock): - assert self.reclaimed_names(client, []) == [ - "abcde_test_running", - "abcde_test_stopped_but_not_cleared", - ] - - def test_never_deletes(self, client: MagicMock): - """Records survive the sweep: a failed run is read off its pipelines.""" - self.reclaimed_names(client, []) - client.delete_pipeline.assert_not_called() - - def test_survives_an_unreachable_instance(self, client: MagicMock): - client.pipelines.side_effect = RuntimeError("connection refused") - with patch("sys.argv", ["stop_ci_run_pipelines"]): - assert stop_ci_run_pipelines.main() == 0 From a5cd1186b60ce1f2ae15a5429e5c71c3188b9918 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 4 Aug 2026 10:32:05 -0700 Subject: [PATCH 3/3] ci: refresh the OIDC token on requests the tests build themselves tests/platform/helper.py built its Authorization header from FELDERA_API_KEY, read once when the module was imported. The CI auth step exports a GitHub ID token under that name and the token expires well inside a suite, so every raw request issued after that got a 401. test_adaptive_joins waits through two compilations before it reads circuit_json_profile, which is where it failed. The header now comes from feldera_bearer_token(), which resolves the same credential the SDK client uses and re-mints an ID token shortly before it expires. That function replaces _feldera_credential(): both spelled out the same precedence, and what is left to decide is whether a caller wants the resolver or its result. The client gets the resolver under Actions, where the token can lapse between requests. A 401 also made reset_pipeline() return before it stopped anything, so a suite that ran past the expiry left its pipelines on the shared instance. Signed-off-by: Gerd Zellweger --- python/feldera/testutils.py | 26 +++++++++++++++++--------- python/tests/platform/TEST_AUTH.md | 9 +++++++-- python/tests/platform/helper.py | 12 ++++-------- 3 files changed, 28 insertions(+), 19 deletions(-) diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index 59237138c64..b53a044a1ff 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -99,18 +99,19 @@ def _token_expiry(token: str) -> float: return time.time() -def _feldera_credential(): - """Bearer credential for the test client. - - Under GitHub Actions the credential is an ID token that expires well inside - a long test run, so hand the SDK the callable rather than a token: it - re-resolves per request and retries once on 401. A configured OIDC login - flow takes precedence over both. +def feldera_bearer_token() -> Optional[str]: + """The bearer token to send with a request issued now. + + A configured OIDC login flow wins, then a GitHub Actions ID token, then the + static API key. None when nothing is configured, which is how a local + instance without authentication runs. Callers that build their own requests + resolve this per request: under Actions the token expires well inside a test + run, and one read at import time starts returning 401 partway through. """ if os.environ.get("OIDC_TEST_ISSUER"): return _get_effective_api_key() if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"): - return _github_oidc_token + return _github_oidc_token() return API_KEY @@ -155,9 +156,16 @@ def __init__(self): def _ensure(self): if self._client is None: + # Under Actions the token expires inside a run, so the SDK gets the + # resolver itself: it re-resolves per request and retries once on + # 401. Elsewhere the credential is fixed for the process. self._client = FelderaClient( connection_timeout=10, - api_key=_feldera_credential(), + api_key=( + feldera_bearer_token + if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL") + else feldera_bearer_token() + ), ) return self._client diff --git a/python/tests/platform/TEST_AUTH.md b/python/tests/platform/TEST_AUTH.md index 49e73e0dc41..f17712351e5 100644 --- a/python/tests/platform/TEST_AUTH.md +++ b/python/tests/platform/TEST_AUTH.md @@ -136,8 +136,13 @@ uv run pytest tests/platform -v The platform tests automatically select authentication in this order: 1. **OIDC Token** (if all OIDC environment variables are set) -2. **API Key** (if `FELDERA_API_KEY` is set) -3. **No Authentication** (fallback for development) +2. **GitHub Actions ID token** (if `ACTIONS_ID_TOKEN_REQUEST_URL` is set) +3. **API Key** (if `FELDERA_API_KEY` is set) +4. **No Authentication** (fallback for development) + +`feldera.testutils.feldera_bearer_token()` resolves this order. Call it per +request rather than caching what it returns: an ID token expires well inside a +long suite, and the function re-mints one shortly before its expiry. ## CI/CD Configuration diff --git a/python/tests/platform/helper.py b/python/tests/platform/helper.py index efaca7189bf..d9013e916e9 100644 --- a/python/tests/platform/helper.py +++ b/python/tests/platform/helper.py @@ -25,12 +25,11 @@ from feldera.testutils import ( FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS, + feldera_bearer_token, reclaim_pipeline, ) -from feldera.testutils_oidc import get_oidc_test_helper from tests import ( - API_KEY, BASE_URL, FELDERA_REQUESTS_VERIFY, TEST_CLIENT, @@ -48,13 +47,10 @@ def _base_headers() -> Dict[str, str]: "Accept": "application/json", } - # Try OIDC authentication first, then fall back to API_KEY - oidc_helper = get_oidc_test_helper() - if oidc_helper is not None: - token = oidc_helper.obtain_access_token() + # Resolved per request: a GitHub Actions ID token expires mid-suite. + token = feldera_bearer_token() + if token: headers["Authorization"] = f"Bearer {token}" - elif API_KEY: - headers["Authorization"] = f"Bearer {API_KEY}" return headers