test: assert reported check status in both standard-test paths

airbyte_cdk/test/standard_tests/_assertions.py

@@ -0,0 +1,53 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Shared assertions for Airbyte Standard Tests.
+
+These assertions are shared between the in-process test runner (`_job_runner.run_test_job`)
+and the Docker-based test suite (`docker_base.DockerConnectorTestSuite`), so that both paths
+enforce the same expectations.
+"""
+
+from __future__ import annotations
+
+from airbyte_cdk.models import Status
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
+from airbyte_cdk.test.models import ExpectedOutcome
+
+
+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 report `FAILED`.
+    - A scenario that does not declare a status is treated as expecting success, matching the
+      default of the `status` property in `acceptance-test-config.yml`.
+    """
+    connection_statuses = [
+        message.connectionStatus
+        for message in check_result.connection_status_messages
+        if message.connectionStatus is not None
+    ]
+    assert connection_statuses, (
+        f"`check` for connector '{connector_name}' emitted no CONNECTION_STATUS message. "
+        f"A `check` implementation should report its outcome as a CONNECTION_STATUS message "
+        f"instead of raising. Logs: {check_result.logs}"
+    )
+    reported_status = connection_statuses[-1].status
+    if expected_outcome.expect_exception():
+        assert reported_status == Status.FAILED, (
+            f"`check` for connector '{connector_name}' was expected to fail, but reported: "
+            f"{connection_statuses[-1]}"
+        )
+        return
+
+    # Both `EXPECT_SUCCESS` and `ALLOW_ANY` (no declared status) require a successful `check`.
+    assert reported_status == Status.SUCCEEDED, (
+        f"`check` for connector '{connector_name}' did not succeed: {connection_statuses[-1]}"
+    )

airbyte_cdk/test/standard_tests/_job_runner.py

@@ -13,12 +13,12 @@
 
 from airbyte_cdk.models import (
     ConfiguredAirbyteCatalog,
-    Status,
 )
 from airbyte_cdk.test import entrypoint_wrapper
 from airbyte_cdk.test.models import (
     ConnectorTestScenario,
 )
+from airbyte_cdk.test.standard_tests._assertions import assert_check_outcome
 
 
 @runtime_checkable
@@ -108,26 +108,14 @@ def run_test_job(
         raise result.as_exception()
 
     if verb == "check":
-        # Check is expected to fail gracefully without an exception.
-        # Instead, we assert that we have a CONNECTION_STATUS message with
-        # a failure status.
-        assert len(result.connection_status_messages) == 1, (
-            "Expected exactly one CONNECTION_STATUS message. Got "
-            f"{len(result.connection_status_messages)}:\n"
-            + "\n".join([str(msg) for msg in result.connection_status_messages])
-            + result.get_formatted_error_message()
+        # Check is expected to report its outcome as a CONNECTION_STATUS message, gracefully and
+        # without an exception. We use the same assertion as the Docker-based test suite so that
+        # both paths enforce the same expectations.
+        assert_check_outcome(
+            check_result=result,
+            expected_outcome=test_scenario.expected_outcome,
+            connector_name=connector_root.absolute().name,
         )
-        if test_scenario.expected_outcome.expect_exception():
-            conn_status = result.connection_status_messages[0].connectionStatus
-            assert conn_status, (
-                "Expected CONNECTION_STATUS message to be present. Got: \n"
-                + "\n".join([str(msg) for msg in result.connection_status_messages])
-            )
-            assert conn_status.status == Status.FAILED, (
-                "Expected CONNECTION_STATUS message to be FAILED. Got: \n"
-                + "\n".join([str(msg) for msg in result.connection_status_messages])
-            )
-
         return result
 
     # For all other verbs, we assert check that an exception is raised (or not).

airbyte_cdk/test/standard_tests/docker_base.py

@@ -22,12 +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, ExpectedOutcome
+from airbyte_cdk.test.models import ConnectorTestScenario
+from airbyte_cdk.test.standard_tests._assertions import assert_check_outcome
 from airbyte_cdk.utils.connector_paths import (
     ACCEPTANCE_TEST_CONFIG,
     find_connector_root,
@@ -38,44 +38,6 @@
 )
 
 
-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."""
 
@@ -314,7 +276,7 @@ def test_docker_image_build_and_check(
         # 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(
+        assert_check_outcome(
             check_result=check_result,
             expected_outcome=scenario.expected_outcome,
             connector_name=connector_root.absolute().name,

unit_tests/test/test_check_assertions.py

@@ -0,0 +1,182 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Unit tests for the shared `check` assertion used by the Standard Tests.
+
+The assertion is exercised directly (as the Docker-based suite calls it) and through the
+in-process job runner, since both paths must enforce the same expectations.
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Iterable, Mapping, MutableMapping
+
+import pytest
+
+from airbyte_cdk.models import (
+    AirbyteCatalog,
+    AirbyteConnectionStatus,
+    AirbyteMessage,
+    ConfiguredAirbyteCatalog,
+    ConnectorSpecification,
+    Status,
+)
+from airbyte_cdk.sources import Source
+from airbyte_cdk.test.entrypoint_wrapper import AirbyteEntrypointException, EntrypointOutput
+from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome
+from airbyte_cdk.test.standard_tests._assertions import assert_check_outcome
+from airbyte_cdk.test.standard_tests._job_runner import run_test_job
+
+
+def _check_output(*statuses: str) -> EntrypointOutput:
+    """Build an EntrypointOutput with one CONNECTION_STATUS message per given status."""
+    return EntrypointOutput(
+        messages=[
+            json.dumps({"type": "CONNECTION_STATUS", "connectionStatus": {"status": status}})
+            for status in statuses
+        ],
+        command=["docker", "run", "..."],
+    )
+
+
+OUTCOME_MATRIX = [
+    # Scenarios expecting success: only SUCCEEDED passes.
+    pytest.param(ExpectedOutcome.EXPECT_SUCCESS, "SUCCEEDED", True, id="success_succeeded"),
+    pytest.param(ExpectedOutcome.EXPECT_SUCCESS, "FAILED", False, id="success_failed"),
+    pytest.param(ExpectedOutcome.EXPECT_SUCCESS, None, False, id="success_no_status"),
+    # Scenarios expecting failure: only a reported FAILED status passes. A `check` that raises
+    # without reporting any status is itself a failure to report the outcome gracefully.
+    pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, "FAILED", True, id="failure_failed"),
+    pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, "SUCCEEDED", False, id="failure_succeeded"),
+    pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, None, False, id="failure_no_status"),
+    # Scenarios with no declared status default to expecting success, matching the default of
+    # the `status` property in `acceptance-test-config.yml`.
+    pytest.param(ExpectedOutcome.ALLOW_ANY, "SUCCEEDED", True, id="allow_any_succeeded"),
+    pytest.param(ExpectedOutcome.ALLOW_ANY, "FAILED", False, id="allow_any_failed"),
+    pytest.param(ExpectedOutcome.ALLOW_ANY, None, False, id="allow_any_no_status"),
+]
+
+
+@pytest.mark.parametrize("expected_outcome, status, should_pass", OUTCOME_MATRIX)
+def test_assert_check_outcome(
+    expected_outcome: ExpectedOutcome,
+    status: str | None,
+    should_pass: bool,
+) -> None:
+    check_result = _check_output(*([status] if status else []))
+    if should_pass:
+        assert_check_outcome(
+            check_result=check_result,
+            expected_outcome=expected_outcome,
+            connector_name="source-test",
+        )
+    else:
+        with pytest.raises(AssertionError):
+            assert_check_outcome(
+                check_result=check_result,
+                expected_outcome=expected_outcome,
+                connector_name="source-test",
+            )
+
+
+@pytest.mark.parametrize(
+    "expected_outcome, statuses, should_pass",
+    [
+        # The last CONNECTION_STATUS message wins.
+        pytest.param(
+            ExpectedOutcome.EXPECT_SUCCESS,
+            ["FAILED", "SUCCEEDED"],
+            True,
+            id="success_last_status_wins",
+        ),
+        pytest.param(
+            ExpectedOutcome.EXPECT_EXCEPTION,
+            ["FAILED", "SUCCEEDED"],
+            False,
+            id="failure_last_status_wins",
+        ),
+    ],
+)
+def test_assert_check_outcome_uses_last_status(
+    expected_outcome: ExpectedOutcome,
+    statuses: list[str],
+    should_pass: bool,
+) -> None:
+    check_result = _check_output(*statuses)
+    if should_pass:
+        assert_check_outcome(
+            check_result=check_result,
+            expected_outcome=expected_outcome,
+            connector_name="source-test",
+        )
+    else:
+        with pytest.raises(AssertionError):
+            assert_check_outcome(
+                check_result=check_result,
+                expected_outcome=expected_outcome,
+                connector_name="source-test",
+            )
+
+
+class _FakeSource(Source):
+    """A source whose `check` reports a fixed status, or raises if no status is given."""
+
+    def __init__(self, status: Status | None) -> None:
+        self._status = status
+
+    def spec(self, logger: logging.Logger) -> ConnectorSpecification:
+        return ConnectorSpecification(
+            connectionSpecification={"type": "object", "properties": {}},
+        )
+
+    def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
+        if self._status is None:
+            raise RuntimeError("Uncaught error during check.")
+
+        return AirbyteConnectionStatus(status=self._status)
+
+    def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog:
+        return AirbyteCatalog(streams=[])
+
+    def read(
+        self,
+        logger: logging.Logger,
+        config: Mapping[str, Any],
+        catalog: ConfiguredAirbyteCatalog,
+        state: MutableMapping[str, Any] | None = None,
+    ) -> Iterable[AirbyteMessage]:
+        yield from []
+
+
+@pytest.mark.parametrize("expected_outcome, status, should_pass", OUTCOME_MATRIX)
+def test_run_test_job_check_asserts_reported_status(
+    expected_outcome: ExpectedOutcome,
+    status: str | None,
+    should_pass: bool,
+    tmp_path: Path,
+) -> None:
+    """The in-process path must enforce the same expectations as the Docker path."""
+    scenario = ConnectorTestScenario(
+        config_dict={"dummy_setting": "dummy_value"},
+        status={
+            ExpectedOutcome.EXPECT_SUCCESS: "succeed",
+            ExpectedOutcome.EXPECT_EXCEPTION: "failed",
+            ExpectedOutcome.ALLOW_ANY: None,
+        }[expected_outcome],
+    )
+    source = _FakeSource(Status(status) if status else None)
+
+    def _run() -> None:
+        run_test_job(
+            source,
+            "check",
+            connector_root=tmp_path,
+            test_scenario=scenario,
+        )
+
+    if should_pass:
+        _run()
+    else:
+        # A `check` that raises instead of reporting a status surfaces as a traced exception
+        # before the status assertion is reached.
+        with pytest.raises((AssertionError, AirbyteEntrypointException)):
+            _run()