fix(standard-tests): honor acceptance-test bypass_reason for basic_read

airbyte_cdk/test/standard_tests/docker_base.py

@@ -123,6 +123,37 @@ def acceptance_test_config(cls) -> Any:
             )
         return tests_config
 
+    @classmethod
+    def get_bypass_reason(cls, category: str) -> str | None:
+        """Return the `bypass_reason` declared for an acceptance-test category, if any.
+
+        A category is considered bypassed only when it declares a non-empty `bypass_reason`
+        and does not declare any tests. This mirrors the semantics of `bypass_reason` in
+        `acceptance-test-config.yml`, so connectors with a documented reason (for example, a
+        test account that has no records) are not blocked by the standard test suite.
+        """
+        try:
+            all_tests_config = cls.acceptance_test_config
+        except (FileNotFoundError, ValueError):
+            return None
+
+        category_config = all_tests_config["acceptance_tests"].get(category)
+        if not isinstance(category_config, dict) or category_config.get("tests"):
+            return None
+
+        bypass_reason = category_config.get("bypass_reason")
+        if not isinstance(bypass_reason, str) or not bypass_reason.strip():
+            return None
+
+        return bypass_reason
+
+    @classmethod
+    def skip_if_bypassed(cls, category: str) -> None:
+        """Skip the running test if the given acceptance-test category is bypassed."""
+        bypass_reason = cls.get_bypass_reason(category)
+        if bypass_reason:
+            pytest.skip(f"Bypassed by '{category}.bypass_reason': {bypass_reason}")
+
     @staticmethod
     def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTestScenario]:
         """
@@ -346,6 +377,8 @@ def test_docker_image_build_and_read(
         if self.is_destination_connector():
             pytest.skip("Skipping read test for destination connector.")
 
+        self.skip_if_bypassed("basic_read")
+
         if scenario.expected_outcome.expect_exception():
             pytest.skip("Skipping (expected to fail).")
 

airbyte_cdk/test/standard_tests/source_base.py

@@ -110,7 +110,12 @@ def test_basic_read(
         from the source and return records. It first runs a `discover` job to
         obtain the catalog of streams, and then it runs a `read` job to fetch
         records from those streams.
+
+        The test is skipped if the connector's `acceptance-test-config.yml` declares a
+        `bypass_reason` for the `basic_read` category.
         """
+        self.skip_if_bypassed("basic_read")
+
         discover_result = run_test_job(
             self.create_connector(scenario),
             "discover",

unit_tests/test/test_standard_tests.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 import pytest
+from _pytest.outcomes import Skipped
 
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
@@ -14,6 +15,8 @@
 from airbyte_cdk.test.standard_tests._job_runner import IConnector
 from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite
 from airbyte_cdk.test.standard_tests.pytest_hooks import _scenario_test_ids
+from airbyte_cdk.test.standard_tests.source_base import SourceTestSuiteBase
+from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG
 
 
 @pytest.mark.parametrize(
@@ -108,3 +111,96 @@ def test_dedup_scenarios_conflicting_statuses_raise() -> None:
 def test_scenario_test_ids(config_paths: list[Path], expected_ids: list[str]) -> None:
     scenarios = [ConnectorTestScenario(config_path=path) for path in config_paths]
     assert _scenario_test_ids(scenarios) == expected_ids
+
+
+BYPASSED_BASIC_READ_CONFIG = """
+acceptance_tests:
+  connection:
+    tests:
+      - config_path: "secrets/config.json"
+        status: "succeed"
+  basic_read:
+    bypass_reason: "Test account doesn't have records."
+"""
+ENABLED_BASIC_READ_CONFIG = """
+acceptance_tests:
+  basic_read:
+    tests:
+      - config_path: "secrets/config.json"
+"""
+
+
+def _test_suite_for_config(
+    tmp_path: Path,
+    acceptance_test_config: str,
+) -> type[SourceTestSuiteBase]:
+    """Build a source test suite rooted in a temp dir with the given acceptance test config."""
+    (tmp_path / ACCEPTANCE_TEST_CONFIG).write_text(acceptance_test_config)
+    return type(
+        "TestSuiteTemp",
+        (SourceTestSuiteBase,),
+        {"get_connector_root_dir": classmethod(lambda cls: tmp_path)},
+    )
+
+
+@pytest.mark.parametrize(
+    "acceptance_test_config, expected_reason",
+    [
+        pytest.param(
+            BYPASSED_BASIC_READ_CONFIG,
+            "Test account doesn't have records.",
+            id="bypass_reason_without_tests_is_honored",
+        ),
+        pytest.param(
+            ENABLED_BASIC_READ_CONFIG,
+            None,
+            id="declared_tests_are_not_bypassed",
+        ),
+        pytest.param(
+            """
+acceptance_tests:
+  basic_read:
+    bypass_reason: "  "
+    """,
+            None,
+            id="blank_bypass_reason_is_not_a_bypass",
+        ),
+        pytest.param(
+            """
+acceptance_tests:
+  basic_read:
+    bypass_reason: "Documented, but tests are declared too."
+    tests:
+      - config_path: "secrets/config.json"
+    """,
+            None,
+            id="tests_win_over_bypass_reason",
+        ),
+        pytest.param(
+            """
+acceptance_tests:
+  connection:
+    tests:
+      - config_path: "secrets/config.json"
+    """,
+            None,
+            id="missing_category_is_not_a_bypass",
+        ),
+    ],
+)
+def test_get_bypass_reason(
+    tmp_path: Path,
+    acceptance_test_config: str,
+    expected_reason: str | None,
+) -> None:
+    test_suite = _test_suite_for_config(tmp_path, acceptance_test_config)
+    assert test_suite.get_bypass_reason("basic_read") == expected_reason
+
+
+def test_basic_read_is_skipped_when_bypassed(tmp_path: Path) -> None:
+    """`test_basic_read` should skip (not fail) when `basic_read` declares a bypass reason."""
+    test_suite = _test_suite_for_config(tmp_path, BYPASSED_BASIC_READ_CONFIG)
+    with pytest.raises(Skipped, match="Test account doesn't have records."):
+        test_suite().test_basic_read(
+            scenario=ConnectorTestScenario(config_path=Path("secrets/config.json")),
+        )