add handling for empty config within a scenario

Author: aaronsteers

airbyte_cdk/test/standard_tests/_job_runner.py

@@ -62,6 +62,9 @@ def run_test_job(
     catalog: ConfiguredAirbyteCatalog | dict[str, Any] | None = None,
 ) -> entrypoint_wrapper.EntrypointOutput:
     """Run a test scenario from provided CLI args and return the result."""
+    # Use default (empty) scenario if not provided:
+    test_scenario = test_scenario or ConnectorTestScenario()
+
     if not connector:
         raise ValueError("Connector is required")
 
@@ -81,14 +84,14 @@ def run_test_job(
         )
 
     args: list[str] = [verb]
-    if test_scenario and test_scenario.config_path:
-        args += ["--config", str(test_scenario.config_path)]
-    elif test_scenario and test_scenario.config_dict:
+    config_dict = test_scenario.get_config_dict(empty_if_missing=True)
+    if config_dict and verb != "spec":
+        # Write the config to a temp json file and pass the path to the file as an argument.
         config_path = (
             Path(tempfile.gettempdir()) / "airbyte-test" / f"temp_config_{uuid.uuid4().hex}.json"
         )
         config_path.parent.mkdir(parents=True, exist_ok=True)
-        config_path.write_text(orjson.dumps(test_scenario.config_dict).decode())
+        config_path.write_text(orjson.dumps(config_dict).decode())
         args += ["--config", str(config_path)]
 
     catalog_path: Path | None = None
@@ -103,7 +106,7 @@ def run_test_job(
             )
             catalog_path.parent.mkdir(parents=True, exist_ok=True)
             catalog_path.write_text(orjson.dumps(catalog).decode())
-        elif test_scenario and test_scenario.configured_catalog_path:
+        elif test_scenario.configured_catalog_path:
             catalog_path = Path(test_scenario.configured_catalog_path)
 
         if catalog_path:
@@ -112,18 +115,12 @@ def run_test_job(
     # This is a bit of a hack because the source needs the catalog early.
     # Because it *also* can fail, we have to redundantly wrap it in a try/except block.
 
-    expect_exception = False
-    if test_scenario and test_scenario.expect_exception:
-        # If the test scenario expects an exception, we need to set the
-        # `expect_exception` flag to True.
-        expect_exception = True
-
     result: entrypoint_wrapper.EntrypointOutput = entrypoint_wrapper._run_command(  # noqa: SLF001  # Non-public API
         source=connector_obj,  # type: ignore [arg-type]
         args=args,
-        expecting_exception=expect_exception,
+        expecting_exception=test_scenario.expect_exception,
     )
-    if result.errors and not expect_exception:
+    if result.errors and not test_scenario.expect_exception:
         raise AssertionError(
             f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
         )
@@ -138,7 +135,7 @@ def run_test_job(
             + "\n".join([str(msg) for msg in result.connection_status_messages])
             + _errors_to_str(result)
         )
-        if expect_exception:
+        if test_scenario.expect_exception:
             conn_status = result.connection_status_messages[0].connectionStatus
             assert conn_status, (
                 "Expected CONNECTION_STATUS message to be present. Got: \n"
@@ -152,7 +149,7 @@ def run_test_job(
         return result
 
     # For all other verbs, we assert check that an exception is raised (or not).
-    if expect_exception:
+    if test_scenario.expect_exception:
         if not result.errors:
             raise AssertionError("Expected exception but got none.")
 

airbyte_cdk/test/standard_tests/declarative_sources.py

@@ -73,12 +73,12 @@ def create_connector(
 
         Subclasses should not need to override this method.
         """
+        scenario = scenario or ConnectorTestScenario()  # Use default (empty) scenario if None
         manifest_dict = yaml.safe_load(cls.manifest_yaml_path.read_text())
         config = {
             "__injected_manifest": manifest_dict,
         }
-        if scenario:
-            config.update(scenario.get_config_dict())
+        config.update(scenario.get_config_dict(empty_if_missing=True))
 
         if cls.components_py_path and cls.components_py_path.exists():
             os.environ["AIRBYTE_ENABLE_UNSAFE_CODE"] = "true"

airbyte_cdk/test/standard_tests/models/scenario.py

@@ -13,6 +13,7 @@
 from typing import Any, Literal, cast
 
 import yaml
+from numpy import empty
 from pydantic import BaseModel
 
 
@@ -43,18 +44,29 @@ class AcceptanceTestFileTypes(BaseModel):
     file_types: AcceptanceTestFileTypes | None = None
     status: Literal["succeed", "failed"] | None = None
 
-    def get_config_dict(self) -> dict[str, Any]:
+    def get_config_dict(
+        self,
+        *,
+        empty_if_missing: bool,
+    ) -> dict[str, Any]:
         """Return the config dictionary.
 
         If a config dictionary has already been loaded, return it. Otherwise, load
         the config file and return the dictionary.
+
+        If `self.config_dict` and `self.config_path` are both `None`:
+        - return an empty dictionary if `empty_if_missing` is True
+        - raise a ValueError if `empty_if_missing` is False
         """
-        if self.config_dict:
+        if self.config_dict is not None:
             return self.config_dict
 
-        if self.config_path:
+        if self.config_path is not None:
             return cast(dict[str, Any], yaml.safe_load(self.config_path.read_text()))
 
+        if empty_if_missing:
+            return {}
+
         raise ValueError("No config dictionary or path provided.")
 
     @property