feat(standard-tests): run each acceptance-test-config.yml config with its section's command, and assert the check status and basic_read records

airbyte_cdk/test/models/outcome.py

@@ -15,11 +15,23 @@
 class ExpectedOutcome(Enum):
     """Enum to represent the expected outcome of a test scenario.
 
-    Class supports comparisons to a boolean or None.
+    The members mirror the `status` values of `acceptance-test-config.yml`:
+
+    - `EXPECT_SUCCESS` (`status: succeed`): the connector must report success. For `check`,
+      this means a `CONNECTION_STATUS` message with status `SUCCEEDED`.
+    - `EXPECT_EXCEPTION` (`status: failed`): the connector must report a graceful failure. For
+      `check`, this means a `CONNECTION_STATUS` message with status `FAILED`.
+    - `EXPECT_UNCAUGHT_ERROR` (`status: exception`): the connector must raise (a `TRACE` error
+      is emitted) instead of reporting a `CONNECTION_STATUS` message.
+    - `ALLOW_ANY` (no `status` declared): the CAT default of `succeed` applies to every command
+      the scenario runs (`ConnectorTestScenario.with_default_success`). As an outcome it is the
+      internal "no expectation yet" state of intermediate steps, such as the `discover` step of
+      a read test, which accept any outcome.
     """
 
     EXPECT_EXCEPTION = auto()
     EXPECT_SUCCESS = auto()
+    EXPECT_UNCAUGHT_ERROR = auto()
     ALLOW_ANY = auto()
 
     @classmethod
@@ -32,7 +44,7 @@ def from_status_str(cls, status: str | None) -> ExpectedOutcome:
             return {
                 "succeed": ExpectedOutcome.EXPECT_SUCCESS,
                 "failed": ExpectedOutcome.EXPECT_EXCEPTION,
-                "exception": ExpectedOutcome.EXPECT_EXCEPTION,  # same as 'failed'
+                "exception": ExpectedOutcome.EXPECT_UNCAUGHT_ERROR,
             }[status]
         except KeyError as ex:
             raise ValueError(
@@ -53,8 +65,20 @@ def from_expecting_exception_bool(cls, expecting_exception: bool | None) -> Expe
         )
 
     def expect_exception(self) -> bool:
-        """Return whether the expectation is that an exception should be raised."""
-        return self == ExpectedOutcome.EXPECT_EXCEPTION
+        """Return whether the expectation is that the connector run should not succeed.
+
+        This is true for both `status: failed` (a graceful, reported failure) and
+        `status: exception` (an uncaught error). Use `expect_uncaught_error()` to distinguish
+        the two.
+        """
+        return self in (
+            ExpectedOutcome.EXPECT_EXCEPTION,
+            ExpectedOutcome.EXPECT_UNCAUGHT_ERROR,
+        )
+
+    def expect_uncaught_error(self) -> bool:
+        """Return whether the expectation is that the connector raises (`status: exception`)."""
+        return self == ExpectedOutcome.EXPECT_UNCAUGHT_ERROR
 
     def expect_success(self) -> bool:
         """Return whether the expectation is that the test should succeed without exceptions."""

airbyte_cdk/test/models/scenario.py

@@ -13,15 +13,55 @@
 import tempfile
 from contextlib import contextmanager, suppress
 from pathlib import Path  # noqa: TC003  # Pydantic needs this (don't move to 'if typing' block)
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Literal, cast
 
 import yaml
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, field_validator
 
 from airbyte_cdk.test.models.outcome import ExpectedOutcome
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import Generator, Mapping
+
+
+ScenarioCommand = Literal["spec", "check", "discover", "read"]
+"""A connector command the standard tests run for a scenario. See `SECTION_COMMANDS`."""
+
+SECTION_COMMANDS: Mapping[str, frozenset[ScenarioCommand]] = MappingProxyType(
+    {
+        "spec": frozenset({"spec"}),
+        "connection": frozenset({"check"}),
+        "discovery": frozenset({"discover"}),
+        "basic_read": frozenset({"read"}),
+        "full_refresh": frozenset({"read"}),
+    }
+)
+"""The command that runs for a config listed under each section of `acceptance-test-config.yml`.
+
+Each section stands for the connector command the Connector Acceptance Tests ran its configs
+with, and the standard tests run that command for the config:
+
+- `spec`: the config must validate against the connector's spec (`test_config_matches_spec`).
+- `connection`: `check` must report the declared `status` (`test_check`).
+- `discovery`: `discover` must succeed and return streams (`test_discover`).
+- `basic_read`, `full_refresh`: a full-refresh `read` must succeed and return records
+  (`test_basic_read`, `test_fail_read_with_bad_catalog`). It reads the streams of the config's
+  `configured_catalog_path`, or every discovered stream when it declares none, minus its
+  `empty_streams`.
+
+A config listed under several sections runs the union of their commands. Configs listed only
+under `incremental` are not collected: without input state, an incremental `read` checks the same
+thing as a full-refresh one.
+"""
+
+SCENARIO_SECTIONS: tuple[str, ...] = tuple(SECTION_COMMANDS)
+"""Every section that `get_scenarios()` reads, in the order its entries are collected."""
+
+ALL_COMMANDS: frozenset[ScenarioCommand] = frozenset(
+    command for commands in SECTION_COMMANDS.values() for command in commands
+)
+"""Every command; what a scenario built by hand (no `sections`) runs."""
 
 
 class ConnectorTestScenario(BaseModel):
@@ -63,6 +103,25 @@ def __hash__(self) -> int:
     expect_records: AcceptanceTestExpectRecords | None = None
     file_types: AcceptanceTestFileTypes | None = None
     status: Literal["succeed", "failed", "exception"] | None = None
+    # CAT's `basic_read` opt-out from validating records against the stream schemas.
+    validate_schema: bool = True
+    # The connector's top-level `test_strictness_level` in `acceptance-test-config.yml`. Populated
+    # by `get_scenarios()`; at `high`, `validate_schema: false` is rejected, as in CAT.
+    test_strictness_level: Literal["low", "high"] | None = None
+    # The sections of `acceptance-test-config.yml` that list this config. Populated by
+    # `get_scenarios()`; empty for scenarios built by hand, which run every command.
+    sections: tuple[str, ...] = ()
+
+    @field_validator("sections")
+    @classmethod
+    def _sections_must_be_known(cls, sections: tuple[str, ...]) -> tuple[str, ...]:
+        unknown = [section for section in sections if section not in SECTION_COMMANDS]
+        if unknown:
+            raise ValueError(
+                f"Unknown `acceptance-test-config.yml` section(s) {unknown}; "
+                f"expected any of {list(SCENARIO_SECTIONS)}."
+            )
+        return sections
 
     def get_config_dict(
         self,
@@ -107,6 +166,36 @@ def expected_outcome(self) -> ExpectedOutcome:
         """
         return ExpectedOutcome.from_status_str(self.status)
 
+    @property
+    def commands(self) -> frozenset[ScenarioCommand]:
+        """The connector commands the standard tests run for this scenario.
+
+        A config runs the command of every `acceptance-test-config.yml` section that lists it
+        (`SECTION_COMMANDS`). A scenario built by hand, with no `sections`, runs every command.
+        """
+        if not self.sections:
+            return ALL_COMMANDS
+
+        return frozenset(
+            command for section in self.sections for command in SECTION_COMMANDS[section]
+        )
+
+    def runs(self, *commands: ScenarioCommand) -> bool:
+        """Whether any of `commands` runs for this scenario. See `commands`."""
+        return bool(self.commands & frozenset(commands))
+
+    @property
+    def is_basic_read_config(self) -> bool:
+        """Whether this config is listed under `basic_read` (or the scenario was built by hand).
+
+        Only the records of these configs are checked as in CAT's basic read test: every stream
+        not declared in `empty_streams` returns records, and every record matches its stream's
+        schema unless `validate_schema: false` is set. A config read only because it is listed
+        under `full_refresh` never had CAT's basic read checks, and has no `basic_read` entry to
+        declare those exceptions in, so its `read` only has to return some records.
+        """
+        return not self.sections or "basic_read" in self.sections
+
     @property
     def id(self) -> str:
         """Return a unique identifier for the test scenario.
@@ -188,6 +277,18 @@ def with_expecting_success(self) -> ConnectorTestScenario:
             status="succeed",
         )
 
+    def with_default_success(self) -> ConnectorTestScenario:
+        """Return the scenario, expecting success when it declares no `status`.
+
+        No `status` in `acceptance-test-config.yml` means `succeed`, the CAT default, for every
+        command a config runs. `ALLOW_ANY` is kept only for intermediate steps that defer their
+        expectation (see `without_expected_outcome`).
+        """
+        if self.status is not None:
+            return self
+
+        return self.with_expecting_success()
+
     @property
     def requires_creds(self) -> bool:
         """Return True if the scenario requires credentials to run."""

airbyte_cdk/test/standard_tests/_assertions.py

@@ -0,0 +1,110 @@
+# Copyright (c) 2026 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 typing import TYPE_CHECKING, Any
+
+import jsonschema
+
+from airbyte_cdk.models import Status
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
+from airbyte_cdk.test.models import ExpectedOutcome
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+def assert_check_outcome(
+    *,
+    check_result: EntrypointOutput,
+    expected_outcome: ExpectedOutcome,
+    connector_name: str,
+) -> None:
+    """Assert that the outcome of `check` 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, following the `status` semantics of `acceptance-test-config.yml`:
+
+    - `succeed` (`EXPECT_SUCCESS`): a `CONNECTION_STATUS` message with status `SUCCEEDED`.
+    - `failed` (`EXPECT_EXCEPTION`): a `CONNECTION_STATUS` message with status `FAILED`.
+    - `exception` (`EXPECT_UNCAUGHT_ERROR`): `check` raises, so a `TRACE` error must be present
+      and no `SUCCEEDED` status may be reported. A `CONNECTION_STATUS` message is not required.
+    - no `status` (`ALLOW_ANY`): treated as `succeed`, matching the CAT default. A config that
+      `check` is expected to reject must declare `status: failed` (or `exception`).
+
+    When more than one `CONNECTION_STATUS` message is present, the last one wins.
+    """
+    connection_statuses = [
+        message.connectionStatus
+        for message in check_result.connection_status_messages
+        if message.connectionStatus is not None
+    ]
+    if expected_outcome.expect_uncaught_error():
+        assert check_result.errors, (
+            f"`check` for connector '{connector_name}' was expected to raise "
+            f"(`status: exception`), but emitted no TRACE error. "
+            f"Reported statuses: {connection_statuses}"
+        )
+        assert not connection_statuses or connection_statuses[-1].status != Status.SUCCEEDED, (
+            f"`check` for connector '{connector_name}' was expected to raise "
+            f"(`status: exception`), but reported: {connection_statuses[-1]}"
+        )
+        return
+
+    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. If raising is the expected outcome for this config, declare "
+        f"`status: exception` for it in `acceptance-test-config.yml`.\n"
+        + (f"Errors: {check_result.get_formatted_error_message()}\n" if check_result.errors else "")
+        + f"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]}"
+    )
+
+
+def assert_config_matches_spec(
+    *,
+    config: Mapping[str, Any],
+    spec_result: EntrypointOutput,
+    connector_name: str,
+    scenario_id: str,
+) -> None:
+    """Assert that a config validates against the `connectionSpecification` the connector reports.
+
+    This is the CAT `test_config_match_spec` check for configs listed under `spec` in
+    `acceptance-test-config.yml`: the connector's own spec must accept the config. The config's
+    values never appear in the failure message, since they may be secrets.
+    """
+    specs = [message.spec for message in spec_result.spec_messages if message.spec is not None]
+    assert len(specs) == 1, (
+        f"Expected exactly one SPEC message from connector '{connector_name}', got {len(specs)}."
+        + (f"\nErrors: {spec_result.get_formatted_error_message()}" if spec_result.errors else "")
+    )
+    try:
+        jsonschema.validate(instance=config, schema=specs[0].connectionSpecification)
+    except jsonschema.ValidationError as exc:
+        # `exc.message` embeds the offending value; keep it (and the chained exception, whose
+        # text includes the instance) out of the test output.
+        message = exc.message.replace(repr(exc.instance), "<config value>")
+        raise AssertionError(
+            f"Config '{scenario_id}' does not match the spec of connector '{connector_name}' "
+            f"at `{exc.json_path}`: {message}"
+        ) from None

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
@@ -113,32 +113,25 @@ def run_test_job(
     if inspect_output:
         inspect_output(result)
 
-    if result.errors and test_scenario.expected_outcome.expect_success():
-        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()
+        # Unless the scenario declares `status: exception`, `check` is expected to report its
+        # outcome as a CONNECTION_STATUS message, gracefully and without an exception. Any
+        # uncaught error is surfaced as-is (mirroring `raise_if_errors` on the Docker path), and
+        # the reported status is then validated with the same assertion as the Docker-based test
+        # suite so that both paths enforce the same expectations.
+        if result.errors and not test_scenario.expected_outcome.expect_exception():
+            raise result.as_exception()
+
+        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
 
+    if result.errors and test_scenario.expected_outcome.expect_success():
+        raise result.as_exception()
+
     # For all other verbs, we assert check that an exception is raised (or not).
     if test_scenario.expected_outcome.expect_exception():
         if not result.errors: