diff --git a/airbyte_cdk/test/standard_tests/_assertions.py b/airbyte_cdk/test/standard_tests/_assertions.py new file mode 100644 index 000000000..6f7bbef28 --- /dev/null +++ b/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]}" + ) diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py index ddd293ffa..b0895b22b 100644 --- a/airbyte_cdk/test/standard_tests/_job_runner.py +++ b/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). diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 84cfab2b0..548711ed3 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/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, diff --git a/unit_tests/test/test_check_assertions.py b/unit_tests/test/test_check_assertions.py new file mode 100644 index 000000000..87ea08a6d --- /dev/null +++ b/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() diff --git a/unit_tests/test/test_docker_base_check_outcome.py b/unit_tests/test/test_docker_base_check_outcome.py deleted file mode 100644 index 8874881e8..000000000 --- a/unit_tests/test/test_docker_base_check_outcome.py +++ /dev/null @@ -1,76 +0,0 @@ -# Copyright (c) 2025 Airbyte, Inc., all rights reserved. -"""Unit tests for `_assert_check_outcome` in `airbyte_cdk.test.standard_tests.docker_base`.""" - -import json - -import pytest - -from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput -from airbyte_cdk.test.models import ExpectedOutcome -from airbyte_cdk.test.standard_tests.docker_base import _assert_check_outcome - - -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", "..."], - ) - - -@pytest.mark.parametrize( - "expected_outcome, statuses, should_pass", - [ - # 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, [], False, id="success_no_status"), - # Scenarios expecting failure: a SUCCEEDED status is a regression and must fail; - # a FAILED status or no status at all (uncaught error) both count as failing as expected. - 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, [], True, id="failure_no_status"), - # ALLOW_ANY scenarios: either reported outcome passes, but `check` must still - # emit a CONNECTION_STATUS message. - pytest.param(ExpectedOutcome.ALLOW_ANY, ["SUCCEEDED"], True, id="allow_any_succeeded"), - pytest.param(ExpectedOutcome.ALLOW_ANY, ["FAILED"], True, id="allow_any_failed"), - pytest.param(ExpectedOutcome.ALLOW_ANY, [], False, id="allow_any_no_status"), - # 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( - 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", - )