feat: add empty_streams filtering to standard test suites

- Add EmptyStreamConfig model and empty_streams field to ConnectorTestScenario
- Extend get_scenarios() to include basic_read category
- Implement catalog filtering in test_basic_read and test_docker_image_build_and_read methods
- Exclude streams declared as empty_streams from test execution to prevent failures in sandbox environments

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/test/models/scenario.py

@@ -24,6 +24,13 @@
     from collections.abc import Generator
 
 
+class EmptyStreamConfig(BaseModel):
+    """Configuration for streams that should be excluded from tests."""
+    
+    name: str
+    bypass_reason: str
+
+
 class ConnectorTestScenario(BaseModel):
     """Acceptance test scenario, as a Pydantic model.
 
@@ -50,6 +57,7 @@ class AcceptanceTestFileTypes(BaseModel):
     _id: str | None = None  # Used to override the default ID generation
 
     configured_catalog_path: Path | None = None
+    empty_streams: list[EmptyStreamConfig] | None = None
     timeout_seconds: int | None = None
     expect_records: AcceptanceTestExpectRecords | None = None
     file_types: AcceptanceTestFileTypes | None = None

airbyte_cdk/test/standard_tests/docker_base.py

@@ -73,7 +73,7 @@ def get_scenarios(
         This has to be a separate function because pytest does not allow
         parametrization of fixtures with arguments from the test class itself.
         """
-        categories = ["connection", "spec"]
+        categories = ["connection", "spec", "basic_read"]
         try:
             acceptance_test_config_path = cls.acceptance_test_config_path
         except FileNotFoundError as e:
@@ -336,6 +336,10 @@ def test_docker_image_build_and_read(
                 # If `read_from_streams` is a list, we filter the discovered streams.
                 streams_list = list(set(streams_list) & set(read_from_streams))
 
+            if scenario.empty_streams:
+                streams_to_exclude = {empty_stream.name for empty_stream in scenario.empty_streams}
+                streams_list = [stream for stream in streams_list if stream not in streams_to_exclude]
+
             configured_catalog: ConfiguredAirbyteCatalog = ConfiguredAirbyteCatalog(
                 streams=[
                     ConfiguredAirbyteStream(

airbyte_cdk/test/standard_tests/source_base.py

@@ -121,6 +121,10 @@ def test_basic_read(
             # Failed as expected; we're done.
             return
 
+        streams_to_exclude = set()
+        if scenario.empty_streams:
+            streams_to_exclude = {empty_stream.name for empty_stream in scenario.empty_streams}
+
         configured_catalog = ConfiguredAirbyteCatalog(
             streams=[
                 ConfiguredAirbyteStream(
@@ -129,6 +133,7 @@ def test_basic_read(
                     destination_sync_mode=DestinationSyncMode.append_dedup,
                 )
                 for stream in discover_result.catalog.catalog.streams  # type: ignore [reportOptionalMemberAccess, union-attr]
+                if stream.name not in streams_to_exclude
             ]
         )
         result = run_test_job(