feat(test): persist configs rotated during standard tests to secrets/updated_configurations

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
+`<config dir>/updated_configurations/<config stem>|<emitted_at_ms>.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 `<config dir>/updated_configurations/<config stem>|*.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,29 +118,53 @@ 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
         """
         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:
             return {}
 
         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.

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()
 

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 `<config dir>/updated_configurations/<stem>|<emitted_at_ms>.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

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,48 @@ 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):
+        # 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(
+        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 +337,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 +349,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 +447,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 +459,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 +506,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 +522,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,
             )