From bbd2ddb0052683205a1e8308cc58eb0b18c40de0 Mon Sep 17 00:00:00 2001 From: Serhii Lazebnyi Date: Mon, 5 Oct 2026 15:47:42 +0100 Subject: [PATCH 1/2] Persist configs rotated during standard tests to secrets/updated_configurations --- airbyte_cdk/test/models/scenario.py | 81 +++++++- .../test/standard_tests/_job_runner.py | 8 + .../test/standard_tests/_updated_configs.py | 134 +++++++++++++ .../test/standard_tests/docker_base.py | 59 +++++- unit_tests/test/test_updated_configs.py | 185 ++++++++++++++++++ 5 files changed, 455 insertions(+), 12 deletions(-) create mode 100644 airbyte_cdk/test/standard_tests/_updated_configs.py create mode 100644 unit_tests/test/test_updated_configs.py diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index ddd99ef74d..e4c5b27e52 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -24,6 +24,49 @@ from collections.abc import Generator +UPDATED_CONFIGURATIONS_DIRNAME = "updated_configurations" +"""Subdirectory (next to a config file) holding configs rotated during a test run. + +Connectors that rotate credentials (for example, single-use OAuth refresh tokens) +emit the new config as a `CONTROL`/`CONNECTOR_CONFIG` message. The standard test +harness persists each such config as +`/updated_configurations/|.json`. Later +tests prefer the newest of these files over the original config, and CI pushes +them back to the secrets store with `airbyte-ops secrets push`. +""" + + +def _updated_config_sort_key(path: Path) -> tuple[int, float]: + """Order updated config files by their `emitted_at` suffix, then by mtime.""" + _, _, suffix = path.stem.partition("|") + try: + emitted_at = int(suffix) + except ValueError: + emitted_at = -1 + return emitted_at, path.stat().st_mtime + + +def find_latest_updated_config_file(config_path: Path) -> Path | None: + """Return the newest persisted update for `config_path`, if any. + + Looks for `/updated_configurations/|*.json`. Returns + `None` when no update has been persisted. + """ + updated_dir = config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + if not updated_dir.is_dir(): + return None + + candidates = [ + path + for path in updated_dir.glob(f"{config_path.stem}|*{config_path.suffix}") + if path.is_file() + ] + if not candidates: + return None + + return max(candidates, key=_updated_config_sort_key) + + class ConnectorTestScenario(BaseModel): """Acceptance test scenario, as a Pydantic model. @@ -75,6 +118,11 @@ def get_config_dict( If a config dictionary has already been loaded, return it. Otherwise, load the config file and return the dictionary. + When loading from `self.config_path`, the newest persisted update under + `updated_configurations/` (see `find_latest_updated_config_file`) takes + precedence, so credentials rotated by an earlier test in the same run are + reused instead of the now-stale original. + If `self.config_dict` and `self.config_path` are both `None`: - return an empty dictionary if `empty_if_missing` is True - raise a ValueError if `empty_if_missing` is False @@ -82,15 +130,11 @@ def get_config_dict( if self.config_dict is not None: return self.config_dict - if self.config_path is not None: - config_path = self.config_path - if not config_path.is_absolute(): - # We usually receive a relative path here. Let's resolve it. - config_path = (connector_root / self.config_path).resolve().absolute() - + effective_config_path = self.get_effective_config_path(connector_root) + if effective_config_path is not None: return cast( dict[str, Any], - yaml.safe_load(config_path.read_text()), + yaml.safe_load(effective_config_path.read_text()), ) if empty_if_missing: @@ -98,6 +142,29 @@ def get_config_dict( raise ValueError("No config dictionary or path provided.") + def resolve_config_path(self, connector_root: Path) -> Path | None: + """Return `self.config_path` as an absolute path, or `None` if unset. + + Relative paths (the usual case) are resolved against `connector_root`. + """ + if self.config_path is None: + return None + + config_path = self.config_path + if not config_path.is_absolute(): + # We usually receive a relative path here. Let's resolve it. + config_path = (connector_root / self.config_path).resolve().absolute() + + return config_path + + def get_effective_config_path(self, connector_root: Path) -> Path | None: + """Return the config file to read: the newest persisted update, else the original.""" + config_path = self.resolve_config_path(connector_root) + if config_path is None: + return None + + return find_latest_updated_config_file(config_path) or config_path + @property def expected_outcome(self) -> ExpectedOutcome: """Whether the test scenario expects an exception to be raised. diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py index ddd293ffa2..5deba289c7 100644 --- a/airbyte_cdk/test/standard_tests/_job_runner.py +++ b/airbyte_cdk/test/standard_tests/_job_runner.py @@ -19,6 +19,7 @@ from airbyte_cdk.test.models import ( ConnectorTestScenario, ) +from airbyte_cdk.test.standard_tests._updated_configs import persist_config_updates @runtime_checkable @@ -104,6 +105,13 @@ def run_test_job( args=args, expected_outcome=test_scenario.expected_outcome, ) + # Persist rotated credentials before any assertion can short-circuit the job, so + # single-use tokens survive a failing test and can be pushed back to the secrets store. + persist_config_updates( + result, + scenario=test_scenario, + connector_root=connector_root, + ) if result.errors and test_scenario.expected_outcome.expect_success(): raise result.as_exception() diff --git a/airbyte_cdk/test/standard_tests/_updated_configs.py b/airbyte_cdk/test/standard_tests/_updated_configs.py new file mode 100644 index 0000000000..264c876014 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_updated_configs.py @@ -0,0 +1,134 @@ +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +"""Persist connector configs that were rotated while a standard test was running. + +Connectors that rotate credentials during a run (for example, single-use OAuth +refresh tokens) report the new config to the platform as a +`CONTROL`/`CONNECTOR_CONFIG` message. In production the platform stores that +config; in tests nothing would, and the next run would start from a stale, +already-consumed credential. + +This module captures those messages from an `EntrypointOutput` and writes each +new config to `/updated_configurations/|.json` +next to the scenario's config file. Two consumers rely on that location: + +- `ConnectorTestScenario.get_config_dict` prefers the newest update, so later + tests in the same session use the rotated credential. +- CI runs `airbyte-ops secrets push` after the tests to upload the files back + to Google Secret Manager as new secret versions. + +Persistence never fails a test: errors are logged and swallowed. +""" + +from __future__ import annotations + +import json +import logging +import time +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from airbyte_cdk.models import OrchestratorType, Type +from airbyte_cdk.test.models.scenario import ( + UPDATED_CONFIGURATIONS_DIRNAME, + find_latest_updated_config_file, +) + +if TYPE_CHECKING: + from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput + from airbyte_cdk.test.models import ConnectorTestScenario + +logger = logging.getLogger("airbyte") + + +def extract_config_updates(output: EntrypointOutput) -> list[tuple[dict[str, Any], int]]: + """Return `(config, emitted_at_ms)` for each CONNECTOR_CONFIG control message, in order.""" + updates: list[tuple[dict[str, Any], int]] = [] + for message in output.get_message_by_types([Type.CONTROL], safe_iterator=True): + control = message.control + if ( + control is None + or control.type != OrchestratorType.CONNECTOR_CONFIG + or control.connectorConfig is None + ): + continue + updates.append((dict(control.connectorConfig.config), int(control.emitted_at))) + return updates + + +def persist_updated_config( + new_config: dict[str, Any], + *, + config_path: Path, + emitted_at: int, +) -> Path | None: + """Write `new_config` as an update to `config_path` if it differs from the current one. + + "Current" is the newest already-persisted update, or the original file when + there is none. Returns the written path, or `None` when nothing changed. + """ + current_path = find_latest_updated_config_file(config_path) or config_path + current_config: Any = None + if current_path.is_file(): + try: + current_config = json.loads(current_path.read_text()) + except json.JSONDecodeError: + current_config = None + + if new_config == current_config: + return None + + updated_dir = config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + updated_dir.mkdir(parents=True, exist_ok=True) + target = updated_dir / f"{config_path.stem}|{emitted_at}{config_path.suffix}" + target.write_text(json.dumps(new_config)) + target.chmod(0o600) + logger.info("Persisted updated connector config to %s", target) + return target + + +def persist_config_updates( + output: EntrypointOutput, + *, + scenario: ConnectorTestScenario | None, + connector_root: Path, + in_place_updates: list[dict[str, Any]] | None = None, +) -> list[Path]: + """Persist every config rotation observed during a test job. + + Collects CONNECTOR_CONFIG control messages from `output`, plus any + `in_place_updates` (configs a containerized connector wrote back into its + mounted config file instead of, or in addition to, emitting a message), and + writes each one that differs from the current config via + `persist_updated_config`. + + Does nothing when the scenario has no `config_path` (inline `config_dict` + scenarios have no file to update). Never raises. + """ + if scenario is None: + return [] + config_path = scenario.resolve_config_path(connector_root) + if config_path is None: + return [] + + persisted: list[Path] = [] + try: + updates = extract_config_updates(output) + for in_place_update in in_place_updates or []: + updates.append((in_place_update, int(time.time() * 1000))) + + for new_config, emitted_at in updates: + written = persist_updated_config( + new_config, + config_path=config_path, + emitted_at=emitted_at, + ) + if written is not None: + persisted.append(written) + except Exception: + logger.warning( + "Failed to persist updated connector config for %s", + config_path, + exc_info=True, + ) + + return persisted diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 84cfab2b00..dd5f960db2 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -4,6 +4,7 @@ from __future__ import annotations import inspect +import json import shutil import sys import tempfile @@ -28,6 +29,7 @@ from airbyte_cdk.models.connector_metadata import MetadataFile from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome +from airbyte_cdk.test.standard_tests._updated_configs import persist_config_updates from airbyte_cdk.utils.connector_paths import ( ACCEPTANCE_TEST_CONFIG, find_connector_root, @@ -76,6 +78,46 @@ def _assert_check_outcome( ) +def _run_docker_airbyte_command_with_config_updates( + cmd: list[str], + *, + scenario: ConnectorTestScenario, + connector_root: Path, + temp_config_file: Path, + raise_if_errors: bool = True, +) -> EntrypointOutput: + """Run a containerized Airbyte command and persist any credential rotation it performed. + + Rotated configs are captured both from `CONTROL`/`CONNECTOR_CONFIG` messages and + from in-place edits to the mounted `temp_config_file`, then raised to + `updated_configurations/` via `persist_config_updates`. Persistence happens before + errors are raised so single-use tokens survive a failing command. When + `raise_if_errors` is False, the caller is responsible for checking the result. + """ + config_before = temp_config_file.read_text() + result = run_docker_airbyte_command(cmd, raise_if_errors=False) + + in_place_updates: list[dict[str, Any]] = [] + try: + config_after = temp_config_file.read_text() + if config_after != config_before: + parsed = json.loads(config_after) + if isinstance(parsed, dict): + in_place_updates.append(parsed) + except (OSError, ValueError): + pass + + persist_config_updates( + result, + scenario=scenario, + connector_root=connector_root, + in_place_updates=in_place_updates, + ) + if raise_if_errors: + result.raise_if_errors() + return result + + class DockerConnectorTestSuite: """Base class for connector test suites.""" @@ -293,7 +335,7 @@ def test_docker_image_build_and_check( with scenario.with_temp_config_file( connector_root=connector_root, ) as temp_config_file: - check_result = run_docker_airbyte_command( + check_result = _run_docker_airbyte_command_with_config_updates( [ "docker", "run", @@ -305,6 +347,9 @@ def test_docker_image_build_and_check( "--config", container_config_path, ], + scenario=scenario, + connector_root=connector_root, + temp_config_file=temp_config_file, # For expected-failure scenarios, a non-zero exit or trace error is an # acceptable way for `check` to fail; don't raise before we assert on it. raise_if_errors=not scenario.expected_outcome.expect_exception(), @@ -400,7 +445,7 @@ def test_docker_image_build_and_read( ) as temp_dir_str, ): temp_dir = Path(temp_dir_str) - discover_result = run_docker_airbyte_command( + discover_result = _run_docker_airbyte_command_with_config_updates( [ "docker", "run", @@ -412,7 +457,9 @@ def test_docker_image_build_and_read( "--config", container_config_path, ], - raise_if_errors=True, + scenario=scenario, + connector_root=connector_root, + temp_config_file=temp_config_file, ) catalog_message = discover_result.catalog # Get catalog message @@ -457,7 +504,7 @@ def test_docker_image_build_and_read( configured_catalog_path.write_text( orjson.dumps(asdict(configured_catalog)).decode("utf-8") ) - read_result: EntrypointOutput = run_docker_airbyte_command( + read_result: EntrypointOutput = _run_docker_airbyte_command_with_config_updates( [ "docker", "run", @@ -473,5 +520,7 @@ def test_docker_image_build_and_read( "--catalog", container_catalog_path, ], - raise_if_errors=True, + scenario=scenario, + connector_root=connector_root, + temp_config_file=temp_config_file, ) diff --git a/unit_tests/test/test_updated_configs.py b/unit_tests/test/test_updated_configs.py new file mode 100644 index 0000000000..f5ddfb29b3 --- /dev/null +++ b/unit_tests/test/test_updated_configs.py @@ -0,0 +1,185 @@ +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +"""Unit tests for persisting configs rotated during standard tests.""" + +import json +from pathlib import Path + +import orjson + +from airbyte_cdk.models import ( + AirbyteControlConnectorConfigMessage, + AirbyteControlMessage, + AirbyteMessage, + AirbyteMessageSerializer, + OrchestratorType, + Type, +) +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.models import ConnectorTestScenario +from airbyte_cdk.test.models.scenario import ( + UPDATED_CONFIGURATIONS_DIRNAME, + find_latest_updated_config_file, +) +from airbyte_cdk.test.standard_tests._updated_configs import ( + extract_config_updates, + persist_config_updates, + persist_updated_config, +) + + +def _control_message(config: dict, emitted_at: float) -> str: + message = AirbyteMessage( + type=Type.CONTROL, + control=AirbyteControlMessage( + type=OrchestratorType.CONNECTOR_CONFIG, + emitted_at=emitted_at, + connectorConfig=AirbyteControlConnectorConfigMessage(config=config), + ), + ) + return orjson.dumps(AirbyteMessageSerializer.dump(message)).decode() + + +def _record_message() -> str: + return json.dumps( + { + "type": "RECORD", + "record": {"stream": "s", "data": {"a": 1}, "emitted_at": 1}, + } + ) + + +def _write_scenario_config(connector_root: Path, name: str = "config.json") -> Path: + secrets_dir = connector_root / "secrets" + secrets_dir.mkdir(parents=True, exist_ok=True) + config_path = secrets_dir / name + config_path.write_text(json.dumps({"refresh_token": "original"})) + return config_path + + +def test_find_latest_updated_config_file_prefers_highest_emitted_at(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + assert find_latest_updated_config_file(config_path) is None + + updated_dir = config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + updated_dir.mkdir() + (updated_dir / "config|2000.json").write_text("{}") + (updated_dir / "config|1000.json").write_text("{}") + (updated_dir / "config_oauth|9000.json").write_text("{}") # different stem, ignored + + assert find_latest_updated_config_file(config_path) == updated_dir / "config|2000.json" + + +def test_scenario_get_config_dict_prefers_latest_update(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + scenario = ConnectorTestScenario(config_path=Path("secrets/config.json")) + + assert scenario.get_config_dict(connector_root=tmp_path, empty_if_missing=False) == { + "refresh_token": "original" + } + + updated_dir = config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + updated_dir.mkdir() + (updated_dir / "config|1700.json").write_text(json.dumps({"refresh_token": "rotated"})) + + assert scenario.get_effective_config_path(tmp_path) == updated_dir / "config|1700.json" + assert scenario.get_config_dict(connector_root=tmp_path, empty_if_missing=False) == { + "refresh_token": "rotated" + } + # Identity and creds detection still derive from the original path. + assert scenario.id == "config" + assert scenario.requires_creds + + +def test_extract_config_updates_only_returns_connector_config_controls() -> None: + output = EntrypointOutput( + messages=[ + _record_message(), + _control_message({"refresh_token": "first"}, 1000.0), + _control_message({"refresh_token": "second"}, 2000.0), + ] + ) + + assert extract_config_updates(output) == [ + ({"refresh_token": "first"}, 1000), + ({"refresh_token": "second"}, 2000), + ] + + +def test_persist_config_updates_writes_each_new_config(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + scenario = ConnectorTestScenario(config_path=Path("secrets/config.json")) + output = EntrypointOutput( + messages=[ + _control_message({"refresh_token": "first"}, 1000.0), + # Same as previous update: must not be persisted twice. + _control_message({"refresh_token": "first"}, 1500.0), + _control_message({"refresh_token": "second"}, 2000.0), + ] + ) + + persisted = persist_config_updates(output, scenario=scenario, connector_root=tmp_path) + + updated_dir = config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + assert persisted == [updated_dir / "config|1000.json", updated_dir / "config|2000.json"] + assert json.loads((updated_dir / "config|2000.json").read_text()) == {"refresh_token": "second"} + assert (updated_dir / "config|2000.json").stat().st_mode & 0o777 == 0o600 + # The original secret file is left untouched. + assert json.loads(config_path.read_text()) == {"refresh_token": "original"} + + +def test_persist_config_updates_skips_config_equal_to_original(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + scenario = ConnectorTestScenario(config_path=Path("secrets/config.json")) + output = EntrypointOutput(messages=[_control_message({"refresh_token": "original"}, 1000.0)]) + + assert persist_config_updates(output, scenario=scenario, connector_root=tmp_path) == [] + assert not (config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME).exists() + + +def test_persist_config_updates_handles_in_place_updates(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + scenario = ConnectorTestScenario(config_path=Path("secrets/config.json")) + output = EntrypointOutput(messages=[_record_message()]) + + persisted = persist_config_updates( + output, + scenario=scenario, + connector_root=tmp_path, + in_place_updates=[{"refresh_token": "rewritten"}], + ) + + assert len(persisted) == 1 + assert persisted[0].parent == config_path.parent / UPDATED_CONFIGURATIONS_DIRNAME + assert persisted[0].name.startswith("config|") + assert json.loads(persisted[0].read_text()) == {"refresh_token": "rewritten"} + + +def test_persist_config_updates_is_noop_without_config_path(tmp_path: Path) -> None: + output = EntrypointOutput(messages=[_control_message({"refresh_token": "x"}, 1000.0)]) + + assert persist_config_updates(output, scenario=None, connector_root=tmp_path) == [] + assert ( + persist_config_updates( + output, + scenario=ConnectorTestScenario(config_dict={"refresh_token": "inline"}), + connector_root=tmp_path, + ) + == [] + ) + + +def test_persist_updated_config_compares_against_latest_update(tmp_path: Path) -> None: + config_path = _write_scenario_config(tmp_path) + first = persist_updated_config({"refresh_token": "a"}, config_path=config_path, emitted_at=1000) + assert first is not None + + # Equal to the latest persisted update, not the original: nothing to write. + assert ( + persist_updated_config({"refresh_token": "a"}, config_path=config_path, emitted_at=2000) + is None + ) + # Differs from the latest update: written even though it matches nothing else. + assert ( + persist_updated_config({"refresh_token": "b"}, config_path=config_path, emitted_at=3000) + is not None + ) From 56df78cbbdebf9fe7b35efa22aa0ed9e88c3b1e2 Mon Sep 17 00:00:00 2001 From: Serhii Lazebnyi Date: Mon, 5 Oct 2026 16:07:45 +0100 Subject: [PATCH 2/2] Explain the ignored errors when re-reading the mounted config file --- airbyte_cdk/test/standard_tests/docker_base.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index dd5f960db2..e080e47d9d 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -105,6 +105,8 @@ def _run_docker_airbyte_command_with_config_updates( if isinstance(parsed, dict): in_place_updates.append(parsed) except (OSError, ValueError): + # The container may have deleted the file or left non-JSON behind; the CONTROL + # messages (if any) are still persisted below, so there is nothing more to do here. pass persist_config_updates(