ci: test connector images against an SDM base image built from the current branch

.github/workflows/connector-tests.yml

@@ -192,14 +192,36 @@ jobs:
         run: |
           poetry run airbyte-cdk connector test
 
+      # Manifest-only connector images are built `FROM` the published `source-declarative-manifest`
+      # base image declared in metadata.yaml, so by default the container tests exercise the
+      # already-released CDK instead of this branch's CDK. Build an SDM image from the current
+      # branch (with a plain `docker build` so it lands in the local image store, where the
+      # connector image build can resolve it as its `FROM` base) and use it via `--connector-base-image`.
+      # Python connectors don't need this: the "Bump to Dev Branch CDK" step pins the connector's
+      # CDK dependency to this branch, which the container image build then installs.
+      - name: Build SDM image from current branch [Manifest-Only Connectors]
+        if: env.CONNECTOR_LANGUAGE == 'manifest-only'
+        working-directory: airbyte-python-cdk
+        env:
+          POETRY_DYNAMIC_VERSIONING_BYPASS: "0.0.0dev0"
+        run: |
+          poetry build -f wheel
+          docker build -t airbyte/source-declarative-manifest:dev .
+
+      # The FAST standard test suite also collects the docker image tests, so pass the
+      # branch-built SDM base image here too. Otherwise those tests build the connector image
+      # `FROM` the published metadata base image (running the released CDK instead of this
+      # branch's), and the later `image test` step can't reuse its layers.
       - name: Run FAST Standard Tests [Manifest-Only Connectors]
         if: env.CONNECTOR_LANGUAGE == 'manifest-only'
         working-directory: airbyte-python-cdk
         env:
           GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }}
           POETRY_DYNAMIC_VERSIONING_BYPASS: "0.0.0"
         run: |
-          poetry run airbyte-cdk connector test ${{ matrix.connector }}
+          poetry run airbyte-cdk connector test ${{ matrix.connector }} \
+            --pytest-arg=--connector-base-image \
+            --pytest-arg=airbyte/source-declarative-manifest:dev
 
       - name: Container Build and Test [All Connectors]
         if: steps.no_changes.outputs.status != 'cancelled'
@@ -208,4 +230,9 @@ jobs:
           GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }}
           POETRY_DYNAMIC_VERSIONING_BYPASS: "0.0.0"
         run: |
-          poetry run airbyte-cdk image test ${{ matrix.connector }}
+          if [ "$CONNECTOR_LANGUAGE" = "manifest-only" ]; then
+            poetry run airbyte-cdk image test ${{ matrix.connector }} \
+              --connector-base-image airbyte/source-declarative-manifest:dev
+          else
+            poetry run airbyte-cdk image test ${{ matrix.connector }}
+          fi

airbyte_cdk/cli/airbyte_cdk/_connector.py

@@ -148,7 +148,7 @@ def connector_test(
     click.echo("Connector test command executed.")
     connector_name, connector_directory = resolve_connector_name_and_directory(connector)
 
-    pytest_args = pytest_args or []
+    pytest_args = list(pytest_args or [])
     if collect_only:
         pytest_args.append("--collect-only")
 

airbyte_cdk/cli/airbyte_cdk/_image.py

@@ -42,12 +42,25 @@ def image_cli_group() -> None:
     type=click.Path(exists=True, file_okay=True, path_type=Path),
     help="Optional. Override the Dockerfile used for building the image.",
 )
+@click.option(
+    "--connector-base-image",
+    "base_image",
+    default=None,
+    help=(
+        "Optional. Build the connector `FROM` this base image instead of the `baseImage` "
+        "declared in metadata.yaml. Useful for testing a connector against a locally-built "
+        "`source-declarative-manifest` image (e.g. built from the current CDK branch). "
+        "The image must be visible to your default buildx builder; locally-built images "
+        "are only resolved by the default `docker` driver, not `docker-container` builders."
+    ),
+)
 def build(
     connector: str | None = None,
     *,
     tag: str = "dev",
     no_verify: bool = False,
     dockerfile: Path | None = None,
+    base_image: str | None = None,
 ) -> None:
     """Build a connector Docker image.
 
@@ -80,6 +93,7 @@ def build(
             tag=tag,
             no_verify=no_verify,
             dockerfile_override=dockerfile or None,
+            base_image_override=base_image or None,
         )
     except ConnectorImageBuildError as e:
         click.echo(
@@ -106,11 +120,25 @@ def build(
     default=False,
     help="Skip tests that require credentials (marked with 'requires_creds').",
 )
+@click.option(
+    "--connector-base-image",
+    "base_image",
+    default=None,
+    help=(
+        "Optional. When building the image to test, build it `FROM` this base image instead of "
+        "the `baseImage` declared in metadata.yaml. Useful for testing a manifest-only connector "
+        "against a locally-built `source-declarative-manifest` image (e.g. built from the current "
+        "CDK branch). The image must be visible to your default buildx builder; locally-built "
+        "images are only resolved by the default `docker` driver, not `docker-container` "
+        "builders. Ignored when `--image` is provided."
+    ),
+)
 def image_test(  # "image test" command
     connector: str | None = None,
     *,
     image: str | None = None,
     no_creds: bool = False,
+    base_image: str | None = None,
 ) -> None:
     """Test a connector Docker image.
 
@@ -157,6 +185,7 @@ def image_test(  # "image test" command
                 metadata=metadata,
                 tag=tag,
                 no_verify=True,
+                base_image_override=base_image or None,
             )
         except ConnectorImageBuildError as e:
             click.echo(

airbyte_cdk/test/standard_tests/docker_base.py

@@ -22,11 +22,12 @@
     ConfiguredAirbyteCatalog,
     ConfiguredAirbyteStream,
     DestinationSyncMode,
+    Status,
     SyncMode,
 )
 from airbyte_cdk.models.connector_metadata import MetadataFile
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
-from airbyte_cdk.test.models import ConnectorTestScenario
+from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome
 from airbyte_cdk.utils.connector_paths import (
     ACCEPTANCE_TEST_CONFIG,
     find_connector_root,
@@ -37,6 +38,44 @@
 )
 
 
+def _assert_check_outcome(
+    *,
+    check_result: EntrypointOutput,
+    expected_outcome: ExpectedOutcome,
+    connector_name: str,
+) -> None:
+    """Assert that the reported CONNECTION_STATUS matches the scenario's expected outcome.
+
+    A failing `check` reports `status: FAILED` in a `CONNECTION_STATUS` message and still
+    exits 0, so exit-code checks alone do not catch it. We therefore assert the reported
+    status explicitly, in both directions:
+    - A scenario expecting success must report `SUCCEEDED`.
+    - A scenario expecting failure must not report `SUCCEEDED` (a graceful `FAILED` status
+      or an uncaught error with no status message both count as the expected failure).
+    - `ALLOW_ANY` scenarios accept either outcome.
+    """
+    connection_statuses = [
+        message.connectionStatus
+        for message in check_result.connection_status_messages
+        if message.connectionStatus is not None
+    ]
+    if expected_outcome.expect_exception():
+        assert not connection_statuses or connection_statuses[-1].status != Status.SUCCEEDED, (
+            f"`check` for connector '{connector_name}' was expected to fail, but reported: "
+            f"{connection_statuses[-1]}"
+        )
+        return
+
+    assert connection_statuses, (
+        f"`check` for connector '{connector_name}' emitted no CONNECTION_STATUS message. "
+        f"Logs: {check_result.logs}"
+    )
+    if expected_outcome.expect_success():
+        assert connection_statuses[-1].status == Status.SUCCEEDED, (
+            f"`check` for connector '{connector_name}' did not succeed: {connection_statuses[-1]}"
+        )
+
+
 class DockerConnectorTestSuite:
     """Base class for connector test suites."""
 
@@ -94,6 +133,9 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe
         take the union of any defined empty_streams, to have high confidence that runnning a read with the
         config will not error on the lack of data in the empty streams or lack of permissions to read them.
 
+        We also carry over any explicitly declared `status`. Only the `connection` section declares one,
+        so without this an entry from another section (e.g. `spec`) would win and silently downgrade the
+        scenario to `ALLOW_ANY`, which accepts a failing `check`.
         """
         deduped_scenarios: list[ConnectorTestScenario] = []
 
@@ -105,8 +147,21 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe
                     all_empty_streams = (existing_scenario.empty_streams or []) + (
                         scenario.empty_streams or []
                     )
+                    if (
+                        existing_scenario.status is not None
+                        and scenario.status is not None
+                        and existing_scenario.status != scenario.status
+                    ):
+                        raise ValueError(
+                            f"Conflicting expected statuses declared for config "
+                            f"'{scenario.config_path}': '{existing_scenario.status}' and "
+                            f"'{scenario.status}'."
+                        )
                     merged_scenario = existing_scenario.model_copy(
-                        update={"empty_streams": list(set(all_empty_streams))}
+                        update={
+                            "empty_streams": list(set(all_empty_streams)),
+                            "status": existing_scenario.status or scenario.status,
+                        }
                     )
                     deduped_scenarios.remove(existing_scenario)
                     deduped_scenarios.append(merged_scenario)
@@ -170,6 +225,7 @@ def get_scenarios(
     def test_docker_image_build_and_spec(
         self,
         connector_image_override: str | None,
+        connector_base_image_override: str | None,
     ) -> None:
         """Run `docker_image` acceptance tests."""
         connector_root = self.get_connector_root_dir().absolute()
@@ -184,6 +240,7 @@ def test_docker_image_build_and_spec(
                 metadata=metadata,
                 tag=tag,
                 no_verify=False,
+                base_image_override=connector_base_image_override,
             )
 
         _ = run_docker_airbyte_command(
@@ -206,6 +263,7 @@ def test_docker_image_build_and_check(
         self,
         scenario: ConnectorTestScenario,
         connector_image_override: str | None,
+        connector_base_image_override: str | None,
     ) -> None:
         """Run `docker_image` acceptance tests.
 
@@ -216,9 +274,6 @@ def test_docker_image_build_and_check(
           - In the rare case that image caches need to be cleared, please clear
             the local docker image cache using `docker image prune -a` command.
         """
-        if scenario.expected_outcome.expect_exception():
-            pytest.skip("Skipping test_docker_image_build_and_check (expected to fail).")
-
         tag = "dev-latest"
         connector_root = self.get_connector_root_dir()
         metadata = MetadataFile.from_file(connector_root / "metadata.yaml")
@@ -231,13 +286,14 @@ def test_docker_image_build_and_check(
                 metadata=metadata,
                 tag=tag,
                 no_verify=False,
+                base_image_override=connector_base_image_override,
             )
 
         container_config_path = "/secrets/config.json"
         with scenario.with_temp_config_file(
             connector_root=connector_root,
         ) as temp_config_file:
-            _ = run_docker_airbyte_command(
+            check_result = run_docker_airbyte_command(
                 [
                     "docker",
                     "run",
@@ -249,9 +305,21 @@ def test_docker_image_build_and_check(
                     "--config",
                     container_config_path,
                 ],
-                raise_if_errors=True,
+                # 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(),
             )
 
+        # This makes the image test exercise the connector's actual `check` outcome inside the
+        # container, in both directions (e.g. it fails if bundled custom components are rejected
+        # by the CDK baked into the base image, and it fails if a `check` that is expected to
+        # fail starts succeeding).
+        _assert_check_outcome(
+            check_result=check_result,
+            expected_outcome=scenario.expected_outcome,
+            connector_name=connector_root.absolute().name,
+        )
+
     @pytest.mark.skipif(
         shutil.which("docker") is None,
         reason="docker CLI not found in PATH, skipping docker image tests",
@@ -261,6 +329,7 @@ def test_docker_image_build_and_read(
         self,
         scenario: ConnectorTestScenario,
         connector_image_override: str | None,
+        connector_base_image_override: str | None,
         read_from_streams: Literal["all", "none", "default"] | list[str],
         read_scenarios: Literal["all", "none", "default"] | list[str],
     ) -> None:
@@ -315,6 +384,7 @@ def test_docker_image_build_and_read(
                 metadata=metadata,
                 tag=tag,
                 no_verify=False,
+                base_image_override=connector_base_image_override,
             )
 
         container_config_path = "/secrets/config.json"

airbyte_cdk/test/standard_tests/pytest_hooks.py

@@ -17,13 +17,27 @@
 
 import pytest
 
+from airbyte_cdk.test.models.scenario import ConnectorTestScenario
+
 
 @pytest.fixture
 def connector_image_override(request: pytest.FixtureRequest) -> str | None:
     """Return the value of --connector-image, or None if not set."""
     return cast(str | None, request.config.getoption("--connector-image"))
 
 
+@pytest.fixture
+def connector_base_image_override(request: pytest.FixtureRequest) -> str | None:
+    """Return the value of --connector-base-image, or None if not set.
+
+    When set, the connector image is built `FROM` this base image instead of the
+    `baseImage` declared in metadata.yaml. This lets the image tests exercise a
+    locally-built `source-declarative-manifest` image (e.g. one built from the current
+    CDK branch) rather than the published base image's CDK.
+    """
+    return cast(str | None, request.config.getoption("--connector-base-image"))
+
+
 @pytest.fixture
 def read_from_streams(
     request: pytest.FixtureRequest,
@@ -116,6 +130,18 @@ def pytest_addoption(parser: pytest.Parser) -> None:
         default=None,
         help="Use this pre-built connector Docker image instead of building one.",
     )
+    parser.addoption(
+        "--connector-base-image",
+        action="store",
+        default=None,
+        help=(
+            "Build the connector image `FROM` this base image instead of the `baseImage` "
+            "declared in metadata.yaml. Useful for testing against a locally-built "
+            "`source-declarative-manifest` image built from the current CDK branch. "
+            "The image must be visible to your default buildx builder; locally-built images "
+            "are only resolved by the default `docker` driver, not `docker-container` builders."
+        ),
+    )
     parser.addoption(
         "--read-from-streams",
         action="store",
@@ -185,5 +211,24 @@ class TestMyConnector(ConnectorTestSuiteBase):
         metafunc.parametrize(
             "scenario",
             parametrized_scenarios,
-            ids=[str(scenario) for scenario in scenarios],
+            ids=_scenario_test_ids(scenarios),
         )
+
+
+def _scenario_test_ids(scenarios: list[ConnectorTestScenario]) -> list[str]:
+    """Return unique, human-readable pytest IDs for the given scenarios.
+
+    Scenario IDs are derived from the config file stem, so configs in different directories
+    (e.g. `integration_tests/config.json` and `secrets/config.json`) collide. Where they do,
+    we qualify the ID with the config's parent directory so failures name the config that
+    failed. Note this only affects the displayed test ID, not `ConnectorTestScenario.id`,
+    which scenario selection (`--read-scenarios`) matches against.
+    """
+    labels = [str(scenario) for scenario in scenarios]
+    duplicated = {label for label in labels if labels.count(label) > 1}
+    return [
+        f"{scenario.config_path.parent.name}/{label}"
+        if label in duplicated and scenario.config_path is not None
+        else label
+        for scenario, label in zip(scenarios, labels)
+    ]