feat(standard-tests): check primary keys, state messages and error traces in read tests

airbyte_cdk/test/models/scenario.py

@@ -105,6 +105,12 @@ def __hash__(self) -> int:
     status: Literal["succeed", "failed", "exception"] | None = None
     # CAT's `basic_read` opt-out from validating records against the stream schemas.
     validate_schema: bool = True
+    # CAT's `basic_read` opt-out from checking the primary key of every record.
+    validate_primary_keys_data_type: bool = True
+    # CAT's `basic_read` opt-out from checking that every stream emits per-stream state.
+    validate_state_messages: bool = True
+    # CAT's `basic_read` opt-out from requiring a well-formed error trace from a failing read.
+    expect_trace_message_on_failure: 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
@@ -189,8 +195,10 @@ 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
+        not declared in `empty_streams` returns records, every record matches its stream's
+        schema unless `validate_schema: false` is set, every record has a usable primary key
+        unless `validate_primary_keys_data_type: false` is set, and every stream emits state
+        unless `validate_state_messages: 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.
         """

airbyte_cdk/test/standard_tests/docker_base.py

@@ -261,6 +261,14 @@ def _run_docker_spec(connector_image: str, *, connector_name: str) -> ConnectorS
     "the check outcome is not asserted without it. The spec tests report the `spec` failure."
 )
 
+_READ_CHECK_OPT_OUTS = (
+    "validate_schema",
+    "validate_primary_keys_data_type",
+    "validate_state_messages",
+    "expect_trace_message_on_failure",
+)
+"""The `basic_read` opt-outs that a config keeps if any of its entries declares them `false`."""
+
 
 class DockerConnectorTestSuite:
     """Base class for connector test suites."""
@@ -327,7 +335,9 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe
         `discovery` and `basic_read` runs both `discover` and `read` (see
         `ConnectorTestScenario.commands`).
 
-        A `validate_schema: false` opt-out declared by any entry for the config is kept.
+        An opt-out (`validate_schema`, `validate_primary_keys_data_type`,
+        `validate_state_messages` or `expect_trace_message_on_failure` set to `false`) declared
+        by any entry for the config is kept.
 
         A config is read once, however many read sections (`basic_read`, `full_refresh`) list it.
         It keeps a `configured_catalog_path` only when every read entry for it declares that same
@@ -366,8 +376,11 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe
                         update={
                             "empty_streams": list(set(all_empty_streams)),
                             "status": existing_scenario.status or scenario.status,
-                            "validate_schema": existing_scenario.validate_schema
-                            and scenario.validate_schema,
+                            **{
+                                opt_out: getattr(existing_scenario, opt_out)
+                                and getattr(scenario, opt_out)
+                                for opt_out in _READ_CHECK_OPT_OUTS
+                            },
                             "sections": tuple(
                                 dict.fromkeys(existing_scenario.sections + scenario.sections)
                             ),

airbyte_cdk/test/standard_tests/source_base.py

@@ -16,6 +16,7 @@
     SyncMode,
     Type,
 )
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
 from airbyte_cdk.test.models import (
     ConnectorTestScenario,
 )
@@ -26,8 +27,10 @@
 )
 from airbyte_cdk.test.standard_tests._job_runner import run_test_job
 from airbyte_cdk.test.standard_tests._read_assertions import (
+    assert_error_trace_messages,
     assert_read_records,
     assert_schema_validation_opt_out_allowed,
+    make_invalid_config,
 )
 from airbyte_cdk.test.standard_tests._spec_lint import assert_spec_is_valid, get_single_spec
 from airbyte_cdk.test.standard_tests.connector_base import (
@@ -173,9 +176,16 @@ def test_basic_read(
         `get_read_streams`).
 
         A read expected to succeed must return records. For a config listed under `basic_read`,
-        the records are also checked as in CAT's basic read test: every stream it reads must
-        return at least one record, and every record must match its stream's JSON schema (opt out
-        with `validate_schema: false`, which a connector at `test_strictness_level: high` cannot).
+        the read is also checked as in CAT's basic read test:
+
+        - every stream it reads must return at least one record;
+        - every record must match its stream's JSON schema (opt out with
+          `validate_schema: false`, which a connector at `test_strictness_level: high` cannot);
+        - every record of a stream with a `source_defined_primary_key` must have a usable
+          primary key: no key field holds an object or an array, and not every key field is
+          missing or null (opt out with `validate_primary_keys_data_type: false`);
+        - every stream it reads must emit a `STREAM` (or `GLOBAL`) state message, and every
+          state message must carry `sourceStats` (opt out with `validate_state_messages: false`).
         """
         skip_unless_runs(scenario, "read")
         if scenario.is_basic_read_config:
@@ -231,6 +241,10 @@ def test_basic_read(
                 configured_catalog=configured_catalog,
                 require_records_per_stream=True,
                 validate_schema=scenario.validate_schema,
+                validate_primary_keys=scenario.validate_primary_keys_data_type,
+                state_messages=(
+                    result.state_messages if scenario.validate_state_messages else None
+                ),
             )
         elif next(result.records_iterator, None) is None:
             raise AssertionError("Expected records but got none.")
@@ -242,7 +256,8 @@ def test_fail_read_with_bad_catalog(
         """Standard test for `read` when passed a bad catalog file.
 
         Runs for configs listed under `basic_read` or `full_refresh` in
-        `acceptance-test-config.yml`.
+        `acceptance-test-config.yml`. The read must fail and emit a well-formed `ERROR` trace
+        message (see `assert_error_trace_messages`).
         """
         skip_unless_runs(scenario, "read")
         invalid_configured_catalog = ConfiguredAirbyteCatalog(
@@ -272,4 +287,83 @@ def test_fail_read_with_bad_catalog(
             catalog=asdict(invalid_configured_catalog),
         )
         assert result.errors, "Expected errors but got none."
-        assert result.trace_messages, "Expected trace messages but got none."
+        assert_error_trace_messages(result.get_message_iterator())
+
+    def test_fail_read_with_invalid_config(self) -> None:
+        """Standard test for `read` with a config that the connector's spec rejects.
+
+        Ported from CAT's `TestBasicRead.test_airbyte_trace_message_on_failure`: a read that
+        fails must report why in a well-formed `ERROR` trace message (see
+        `assert_error_trace_messages`). Where `test_fail_read_with_bad_catalog` breaks the
+        catalog, which the entrypoint rejects before any connector code runs, this test gives
+        the connector a config its own spec rejects: one that holds none of the spec's required
+        top-level properties and no value from any real config (`make_invalid_config`). No
+        credentials are needed, so it runs once per connector, and no request can be sent with
+        real credentials.
+
+        Skipped when the spec requires no top-level property (no config is invalid for sure),
+        or when any `basic_read` entry of `acceptance-test-config.yml` sets
+        `expect_trace_message_on_failure: false`, as CAT did.
+
+        A source that cannot even be created from the invalid config fails the same way
+        `source-declarative-manifest` reports it: as an uncaught error, turned into a trace
+        message.
+        """
+        opted_out = sorted(
+            scenario.id
+            for scenario in self.get_scenarios()
+            if not scenario.expect_trace_message_on_failure
+        )
+        if opted_out:
+            pytest.skip(
+                "`expect_trace_message_on_failure: false` is set for "
+                f"{', '.join(repr(scenario_id) for scenario_id in opted_out)} in "
+                "`acceptance-test-config.yml`."
+            )
+
+        connector_root = self.get_connector_root_dir()
+        spec_result = run_test_job(
+            self.create_connector(scenario=None),
+            "spec",
+            test_scenario=None,
+            connector_root=connector_root,
+        )
+        spec = get_single_spec(spec_result, connector_name=self.connector_name)
+        invalid_config = make_invalid_config(spec.connectionSpecification)
+        if invalid_config is None:
+            pytest.skip(
+                "The connector's spec requires no top-level property, so no config is "
+                "certain to be invalid."
+            )
+
+        scenario = ConnectorTestScenario(config_dict=invalid_config, status="failed")
+        catalog = ConfiguredAirbyteCatalog(
+            streams=[
+                ConfiguredAirbyteStream(
+                    stream=AirbyteStream(
+                        name="__AIRBYTE__stream_that_does_not_exist",
+                        json_schema={
+                            "type": "object",
+                            "properties": {"f1": {"type": "string"}},
+                        },
+                        supported_sync_modes=[SyncMode.full_refresh],
+                    ),
+                    sync_mode=SyncMode.full_refresh,
+                    destination_sync_mode=DestinationSyncMode.append,
+                )
+            ]
+        )
+        try:
+            connector = self.create_connector(scenario)
+        except Exception as error:  # noqa: BLE001  # Any error here is the read failing.
+            result = EntrypointOutput(messages=[], uncaught_exception=error)
+        else:
+            # Asserts that the read fails.
+            result = run_test_job(
+                connector,
+                "read",
+                connector_root=connector_root,
+                test_scenario=scenario,
+                catalog=catalog,
+            )
+        assert_error_trace_messages(result.get_message_iterator())

cdk-migrations.md

@@ -1,5 +1,41 @@
 # CDK Migration Guide
 
+## Upgrading to the next minor release (unreleased)
+
+This release of the CDK adds three checks of the Connector Acceptance Tests (CAT) basic read test to the Standard Tests. There is no change to how connectors run. The release is a minor one, since only the test harness changes, but the connector test suites of connectors that do not satisfy the new expectations start failing. The connector monorepo installs the `airbyte-cdk` CLI unpinned, so connector CI picks the new checks up with this release, without a per-connector CDK bump.
+
+### Every record has a usable primary key (`basic_read`)
+
+For every config listed under `basic_read` in `acceptance-test-config.yml` (and not declared `status: failed` or `status: exception`), the in-process `test_basic_read` now checks the primary key of every record of every stream that declares a `source_defined_primary_key`, as CAT's `validate_primary_keys_data_type` check did:
+
+- No key field may hold an object or an array. Strings, numbers and booleans are accepted, as in CAT.
+- At least one key field must be present and not null. In a composite key some fields may be null, as in CAT.
+
+The failure message names each stream, its primary key, the number of records with an invalid key, and for each distinct problem the record number and the path of the key field (for example ``record #3: key field `$.id` is an object``). Record values are never printed.
+
+### Every stream emits state (`basic_read`)
+
+For the same configs, `test_basic_read` now checks the state messages of the read, replacing CAT's `validate_state_messages` check:
+
+- Every stream the read covers must be named by at least one `STREAM` state message, or by the `stream_states` of a `GLOBAL` state message. CAT only compared the number of state messages with the number of streams, so the many state messages of one stream could hide another stream that emitted none.
+- No state message may use the deprecated `LEGACY` state type, or no type.
+- Every state message must carry `sourceStats.recordCount`.
+
+The CDK emits a final state message for every stream it completes, including full-refresh streams, and its entrypoint adds `sourceStats`, so a CDK connector meets the last two checks whenever it reads every stream to the end.
+
+### A failed read reports a well-formed error trace
+
+A new per-connector test, `SourceTestSuiteBase.test_fail_read_with_invalid_config`, ports CAT's `test_airbyte_trace_message_on_failure`. It runs `read` with a config that the connector's own spec rejects: a config that holds none of the spec's required top-level properties and no value from any real config, so it needs no credentials and cannot send a request with real ones. The read must fail and emit at least one `TRACE` message of type `ERROR`, and every `ERROR` trace message must be well-formed: it carries an `error` object with a non-blank `message` and a `failure_type`, and a positive `emitted_at`. The test is skipped when the spec requires no top-level property, and when any `basic_read` entry of `acceptance-test-config.yml` sets `expect_trace_message_on_failure: false`, as in CAT. `test_fail_read_with_bad_catalog` now checks its error trace messages the same way, instead of only requiring some trace message.
+
+Migration steps:
+
+- A record with an invalid primary key: fix the stream's `primary_key` (or the record) so that every record has a scalar key. To skip the check for one config while the fix is pending, set `validate_primary_keys_data_type: false` in its `basic_read` entry, with a YAML comment giving the reason. As in CAT, this is allowed at `test_strictness_level: high`.
+- A stream without state: a stream that ends early, or a source that emits only `LEGACY` state, has to be fixed. To skip the check for one config while the fix is pending, set `validate_state_messages: false` in its `basic_read` entry, with a YAML comment giving the reason.
+- A failed read without a well-formed error trace: make the connector raise (or emit) an `AirbyteTracedException` for invalid configs. To skip the check, set `expect_trace_message_on_failure: false` in any `basic_read` entry, with a YAML comment giving the reason; this skips the new test for the whole connector.
+- A config listed under several sections keeps an opt-out that any of its entries declares, as for `validate_schema`.
+
+Rationale: Destinations deduplicate on the primary key; a record whose key is an object, an array or entirely missing cannot be deduplicated. A stream that emits no state cannot be resumed, and the platform cannot tell whether it completed. A failed read without a well-formed error trace gives users and the platform no reason for the failure. CAT checked all three for every `basic_read` config, and the Standard Tests that replaced it did not.
+
 ## Upgrading to 7.36.0
 
 This release of the CDK adds checks to the Standard Tests (`airbyte-cdk connector test` and `airbyte-cdk image test`) and changes which commands they run for the configs listed in `acceptance-test-config.yml`. There is no change to how connectors run. The release is a minor one, since only the test harness changes, but the connector test suites of connectors that do not satisfy the new expectations start failing. The connector monorepo installs the `airbyte-cdk` CLI unpinned (`uv tool install --upgrade 'airbyte-cdk[dev]'`), so connector CI picks the new behaviour up with this release, without a per-connector CDK bump.

unit_tests/test/read_test_helpers.py

@@ -0,0 +1,38 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Message builders shared by the `read` Standard Test unit tests."""
+
+from __future__ import annotations
+
+from airbyte_cdk.models import (
+    AirbyteMessage,
+    AirbyteStateBlob,
+    AirbyteStateMessage,
+    AirbyteStateStats,
+    AirbyteStateType,
+    AirbyteStreamState,
+    StreamDescriptor,
+    Type,
+)
+
+
+def final_state(
+    stream: str, *, namespace: str | None = None, record_count: float | None = None
+) -> AirbyteMessage:
+    """The final state the CDK emits once a full-refresh stream completes.
+
+    With `record_count`, the message carries `sourceStats`, as the CDK entrypoint adds them;
+    without it, the entrypoint (when the message goes through one) sets them.
+    """
+    return AirbyteMessage(
+        type=Type.STATE,
+        state=AirbyteStateMessage(
+            type=AirbyteStateType.STREAM,
+            stream=AirbyteStreamState(
+                stream_descriptor=StreamDescriptor(name=stream, namespace=namespace),
+                stream_state=AirbyteStateBlob(__ab_no_cursor_state_message=True),
+            ),
+            sourceStats=(
+                AirbyteStateStats(recordCount=record_count) if record_count is not None else None
+            ),
+        ),
+    )

unit_tests/test/test_read_assertions.py

@@ -46,6 +46,7 @@
 )
 from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite
 from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG
+from unit_tests.test.read_test_helpers import final_state
 
 MSG_MISSING_STREAMS = "Every configured stream must return at least one record"
 MSG_SCHEMA = "Records do not match their stream's JSON schema"
@@ -565,6 +566,7 @@ def read(
         for configured_stream in catalog.streams:
             for data in self._records.get(configured_stream.stream.name, []):
                 yield _record(configured_stream.stream.name, data)
+            yield final_state(configured_stream.stream.name)
 
 
 def _run_basic_read(