diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index e3e196ce1f..d9ab96e78d 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -22,6 +22,7 @@ ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, DestinationSyncMode, + Status, SyncMode, ) from airbyte_cdk.models.connector_metadata import MetadataFile @@ -210,15 +211,16 @@ def test_docker_image_build_and_check( """Run `docker_image` acceptance tests. This test builds the connector image and runs the `check` command inside the container. + It validates that the connector emits exactly one `CONNECTION_STATUS` message whose + status matches the scenario's expected outcome (`SUCCEEDED` or `FAILED`). Scenarios + whose `acceptance-test-config.yml` entry has `status: "failed"` are exercised too, + so that `check` against bad configs is validated end-to-end rather than skipped. Note: - It is expected for docker image caches to be reused between test runs. - 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") @@ -233,11 +235,13 @@ def test_docker_image_build_and_check( no_verify=False, ) + expect_success = scenario.expected_outcome.expect_success() + container_config_path = "/secrets/config.json" with scenario.with_temp_config_file( connector_root=connector_root, ) as temp_config_file: - _ = run_docker_airbyte_command( + result = run_docker_airbyte_command( [ "docker", "run", @@ -249,7 +253,50 @@ def test_docker_image_build_and_check( "--config", container_config_path, ], - raise_if_errors=True, + # Only raise on trace errors when we expect a successful check. When the + # scenario expects the check to fail, the connector is supposed to exit + # cleanly with a `CONNECTION_STATUS: FAILED` message rather than raising. + raise_if_errors=expect_success, + ) + + self._assert_check_result_matches_expected_outcome(result, scenario) + + @staticmethod + def _assert_check_result_matches_expected_outcome( + result: EntrypointOutput, + scenario: ConnectorTestScenario, + ) -> None: + """Assert that the `check` output matches the scenario's expected outcome. + + The connector must emit exactly one `CONNECTION_STATUS` message. When the scenario + expects success, the status must be `SUCCEEDED`; when it expects an exception (i.e. + `acceptance-test-config.yml` declares `status: "failed"` or `status: "exception"`), + the status must be `FAILED`. When the expected outcome is `ALLOW_ANY`, only the + presence of a single `CONNECTION_STATUS` message is validated. + """ + status_messages = result.connection_status_messages + assert len(status_messages) == 1, ( + "Expected exactly one CONNECTION_STATUS message but got " + f"{len(status_messages)}:\n" + + "\n".join(str(msg) for msg in status_messages) + + result.get_formatted_error_message() + ) + + connection_status = status_messages[0].connectionStatus + assert connection_status is not None, ( + "Expected CONNECTION_STATUS message to have a connectionStatus payload. Got: \n" + + "\n".join(str(msg) for msg in status_messages) + ) + + if scenario.expected_outcome.expect_success(): + assert connection_status.status == Status.SUCCEEDED, ( + "Expected CONNECTION_STATUS to be SUCCEEDED but got " + f"{connection_status.status}. Message: {connection_status.message!r}" + ) + elif scenario.expected_outcome.expect_exception(): + assert connection_status.status == Status.FAILED, ( + "Expected CONNECTION_STATUS to be FAILED but got " + f"{connection_status.status}. Message: {connection_status.message!r}" ) @pytest.mark.skipif( diff --git a/unit_tests/test/test_docker_base.py b/unit_tests/test/test_docker_base.py new file mode 100644 index 0000000000..0c49c9ecfc --- /dev/null +++ b/unit_tests/test/test_docker_base.py @@ -0,0 +1,102 @@ +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +"""Unit tests for `DockerConnectorTestSuite._assert_check_result_matches_expected_outcome`. + +These tests cover the `CONNECTION_STATUS` validation added to +`test_docker_image_build_and_check` (see airbyte-internal-issues#16212). They +construct `EntrypointOutput` objects directly with synthetic messages rather +than invoking Docker, which lets us exercise the assertion logic in isolation +without requiring the Docker CLI. +""" + +from __future__ import annotations + +import json +from contextlib import AbstractContextManager, nullcontext + +import pytest + +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.models import ConnectorTestScenario +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite + + +def _connection_status_message(status: str, message: str | None = None) -> str: + payload: dict[str, object] = {"status": status} + if message is not None: + payload["message"] = message + return json.dumps({"type": "CONNECTION_STATUS", "connectionStatus": payload}) + + +_SUCCEEDED = _connection_status_message("SUCCEEDED") +_FAILED = _connection_status_message("FAILED", "bad credentials") + + +@pytest.mark.parametrize( + ("scenario_status", "messages", "expectation"), + [ + pytest.param( + "succeed", + [_SUCCEEDED], + nullcontext(), + id="succeed_scenario_with_succeeded_status_passes", + ), + pytest.param( + "succeed", + [_FAILED], + pytest.raises(AssertionError, match="SUCCEEDED"), + id="succeed_scenario_with_failed_status_raises__gap_1", + ), + pytest.param( + "failed", + [_FAILED], + nullcontext(), + id="failed_scenario_with_failed_status_passes__gap_2", + ), + pytest.param( + "failed", + [_SUCCEEDED], + pytest.raises(AssertionError, match="FAILED"), + id="failed_scenario_with_succeeded_status_raises", + ), + pytest.param( + "exception", + [_FAILED], + nullcontext(), + id="exception_scenario_with_failed_status_passes", + ), + pytest.param( + None, + [_SUCCEEDED], + nullcontext(), + id="no_expectation_accepts_succeeded", + ), + pytest.param( + None, + [_FAILED], + nullcontext(), + id="no_expectation_accepts_failed", + ), + pytest.param( + "succeed", + [], + pytest.raises(AssertionError, match="Expected exactly one CONNECTION_STATUS"), + id="missing_connection_status_message_raises", + ), + pytest.param( + "succeed", + [_SUCCEEDED, _SUCCEEDED], + pytest.raises(AssertionError, match="Expected exactly one CONNECTION_STATUS"), + id="multiple_connection_status_messages_raises", + ), + ], +) +def test_assert_check_result_matches_expected_outcome( + scenario_status: str | None, + messages: list[str], + expectation: AbstractContextManager[object], +) -> None: + output = EntrypointOutput(messages=messages) + scenario = ConnectorTestScenario(status=scenario_status) + + with expectation: + DockerConnectorTestSuite._assert_check_result_matches_expected_outcome(output, scenario)