tests: improve fast test outputs, skip 'read' tests for destinations

Author: aaronsteers

airbyte_cdk/test/entrypoint_wrapper.py

@@ -50,6 +50,21 @@
 from airbyte_cdk.test.models.scenario import ExpectedOutcome
 
 
+@dataclass
+class AirbyteEntrypointException(Exception):
+    """Exception raised for errors in the AirbyteEntrypoint execution.
+
+    Used to provide details of an Airbyte connector execution failure in the output
+    captured in an `EntrypointOutput` object. Use `EntrypointOutput.as_exception()` to
+    convert it to an exception.
+
+    Example Usage:
+        output = EntrypointOutput(...)
+        if output.errors:
+            raise output.as_exception("An error occurred during the connector execution.")
+    """
+
+
 class EntrypointOutput:
     """A class to encapsulate the output of an Airbyte connector's execution.
 
@@ -67,13 +82,15 @@ def __init__(
         messages: list[str] | None = None,
         uncaught_exception: Optional[BaseException] = None,
         *,
+        command: list[str] | None = None,
         message_file: Path | None = None,
     ) -> None:
         if messages is None and message_file is None:
             raise ValueError("Either messages or message_file must be provided")
         if messages is not None and message_file is not None:
             raise ValueError("Only one of messages or message_file can be provided")
 
+        self._command = command
         self._messages: list[AirbyteMessage] | None = None
         self._message_file: Path | None = message_file
         if messages:
@@ -182,6 +199,40 @@ def analytics_messages(self) -> List[AirbyteMessage]:
     def errors(self) -> List[AirbyteMessage]:
         return self._get_trace_message_by_trace_type(TraceType.ERROR)
 
+    def get_formatted_error_message(self) -> str:
+        """Returns a human-readable error message with the contents.
+
+        If there are no errors, returns an empty string.
+        """
+        errors = self.errors
+        if not errors:
+            # If there are no errors, return an empty string.
+            return ""
+
+        result = "Failed to run airbyte command"
+        result += ": " + " ".join(self._command) if self._command else "."
+        result += "\n" + "\n".join(
+            [str(error.trace.error).replace("\\n", "\n") for error in errors if error.trace],
+        )
+        return result
+
+    def as_exception(self) -> AirbyteEntrypointException:
+        """Convert the output to an exception with a custom message.
+
+        This is useful for raising an exception in tests or other scenarios where you want to
+        provide a specific error message.
+        """
+        return AirbyteEntrypointException(self.get_formatted_error_message())
+
+    def raise_if_errors(
+        self,
+    ) -> None:
+        """Raise an exception if there are errors in the output."""
+        if not self.errors:
+            return None
+
+        raise self.as_exception()
+
     @property
     def catalog(self) -> AirbyteMessage:
         catalog = self.get_message_by_types([Type.CATALOG])

airbyte_cdk/test/standard_tests/_job_runner.py

@@ -21,26 +21,6 @@
 )
 
 
-def _errors_to_str(
-    entrypoint_output: entrypoint_wrapper.EntrypointOutput,
-) -> str:
-    """Convert errors from entrypoint output to a string."""
-    if not entrypoint_output.errors:
-        # If there are no errors, return an empty string.
-        return ""
-
-    return "\n" + "\n".join(
-        [
-            str(error.trace.error).replace(
-                "\\n",
-                "\n",
-            )
-            for error in entrypoint_output.errors
-            if error.trace
-        ],
-    )
-
-
 @runtime_checkable
 class IConnector(Protocol):
     """A connector that can be run in a test scenario.
@@ -125,9 +105,7 @@ def run_test_job(
         expected_outcome=test_scenario.expected_outcome,
     )
     if result.errors and test_scenario.expected_outcome.expect_success():
-        raise AssertionError(
-            f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
-        )
+        raise result.as_exception()
 
     if verb == "check":
         # Check is expected to fail gracefully without an exception.
@@ -137,7 +115,7 @@ def run_test_job(
             "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])
-            + _errors_to_str(result)
+            + result.get_formatted_error_message()
         )
         if test_scenario.expected_outcome.expect_exception():
             conn_status = result.connection_status_messages[0].connectionStatus
@@ -161,7 +139,8 @@ def run_test_job(
 
     if test_scenario.expected_outcome.expect_success():
         assert not result.errors, (
-            f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
+            f"Test job failed with {len(result.errors)} error(s): \n"
+            + result.get_formatted_error_message()
         )
 
     return result

airbyte_cdk/test/standard_tests/connector_base.py

@@ -44,8 +44,7 @@ def connector(cls) -> type[IConnector] | Callable[[], IConnector] | None:
         This assumes a python connector and should be overridden by subclasses to provide the
         specific connector class to be tested.
         """
-        connector_root = cls.get_connector_root_dir()
-        connector_name = connector_root.absolute().name
+        connector_name = cls.connector_name
 
         expected_module_name = connector_name.replace("-", "_").lower()
         expected_class_name = connector_name.replace("-", "_").title().replace("_", "")

airbyte_cdk/test/standard_tests/docker_base.py

@@ -25,19 +25,18 @@
     DestinationSyncMode,
     SyncMode,
 )
-from airbyte_cdk.models.airbyte_protocol_serializers import (
-    AirbyteCatalogSerializer,
-    AirbyteStreamSerializer,
-)
 from airbyte_cdk.models.connector_metadata import MetadataFile
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
 from airbyte_cdk.test.models import ConnectorTestScenario
-from airbyte_cdk.test.utils.reading import catalog
 from airbyte_cdk.utils.connector_paths import (
     ACCEPTANCE_TEST_CONFIG,
     find_connector_root,
 )
-from airbyte_cdk.utils.docker import build_connector_image, run_docker_command
+from airbyte_cdk.utils.docker import (
+    build_connector_image,
+    run_docker_airbyte_command,
+    run_docker_command,
+)
 
 
 class DockerConnectorTestSuite:
@@ -55,6 +54,17 @@ def get_connector_root_dir(cls) -> Path:
         """Get the root directory of the connector."""
         return find_connector_root([cls.get_test_class_dir(), Path.cwd()])
 
+    @classproperty
+    def connector_name(self) -> str:
+        """Get the name of the connector."""
+        connector_root = self.get_connector_root_dir()
+        return connector_root.absolute().name
+
+    @classmethod
+    def is_destination_connector(cls) -> bool:
+        """Check if the connector is a destination."""
+        return cls.connector_name.startswith("destination-")
+
     @classproperty
     def acceptance_test_config_path(cls) -> Path:
         """Get the path to the acceptance test config file."""
@@ -145,23 +155,16 @@ def test_docker_image_build_and_spec(
                 no_verify=False,
             )
 
-        try:
-            result: CompletedProcess[str] = run_docker_command(
-                [
-                    "docker",
-                    "run",
-                    "--rm",
-                    connector_image,
-                    "spec",
-                ],
-                check=True,  # Raise an error if the command fails
-                capture_stderr=True,
-                capture_stdout=True,
-            )
-        except SubprocessError as ex:
-            raise AssertionError(
-                f"Failed to run `spec` command in docker image {connector_image!r}. Error: {ex!s}"
-            ) from None
+        _ = run_docker_airbyte_command(
+            [
+                "docker",
+                "run",
+                "--rm",
+                connector_image,
+                "spec",
+            ],
+            check=True,
+        )
 
     @pytest.mark.skipif(
         shutil.which("docker") is None,
@@ -203,7 +206,7 @@ def test_docker_image_build_and_check(
         with scenario.with_temp_config_file(
             connector_root=connector_root,
         ) as temp_config_file:
-            _ = run_docker_command(
+            _ = run_docker_airbyte_command(
                 [
                     "docker",
                     "run",
@@ -215,9 +218,7 @@ def test_docker_image_build_and_check(
                     "--config",
                     container_config_path,
                 ],
-                check=True,  # Raise an error if the command fails
-                capture_stderr=True,
-                capture_stdout=True,
+                check=True,
             )
 
     @pytest.mark.skipif(
@@ -242,6 +243,9 @@ def test_docker_image_build_and_read(
             the local docker image cache using `docker image prune -a` command.
           - If the --connector-image arg is provided, it will be used instead of building the image.
         """
+        if self.is_destination_connector():
+            pytest.skip("Skipping read test for destination connector.")
+
         if scenario.expected_outcome.expect_exception():
             pytest.skip("Skipping (expected to fail).")
 
@@ -295,7 +299,7 @@ def test_docker_image_build_and_read(
             ) as temp_dir_str,
         ):
             temp_dir = Path(temp_dir_str)
-            discover_result = run_docker_command(
+            discover_result = run_docker_airbyte_command(
                 [
                     "docker",
                     "run",
@@ -307,20 +311,12 @@ def test_docker_image_build_and_read(
                     "--config",
                     container_config_path,
                 ],
-                check=True,  # Raise an error if the command fails
-                capture_stderr=True,
-                capture_stdout=True,
+                check=True,
             )
-            parsed_output = EntrypointOutput(messages=discover_result.stdout.splitlines())
-            try:
-                catalog_message = parsed_output.catalog  # Get catalog message
-                assert catalog_message.catalog is not None, "Catalog message missing catalog."
-                discovered_catalog: AirbyteCatalog = parsed_output.catalog.catalog
-            except Exception as ex:
-                raise AssertionError(
-                    f"Failed to load discovered catalog from {discover_result.stdout}. "
-                    f"Error: {ex!s}"
-                ) from None
+
+            catalog_message = discover_result.catalog  # Get catalog message
+            assert catalog_message.catalog is not None, "Catalog message missing catalog."
+            discovered_catalog: AirbyteCatalog = catalog_message.catalog
             if not discovered_catalog.streams:
                 raise ValueError(
                     f"Discovered catalog for connector '{connector_name}' is empty. "
@@ -355,7 +351,7 @@ def test_docker_image_build_and_read(
             configured_catalog_path.write_text(
                 orjson.dumps(asdict(configured_catalog)).decode("utf-8")
             )
-            read_result: CompletedProcess[str] = run_docker_command(
+            read_result: EntrypointOutput = run_docker_airbyte_command(
                 [
                     "docker",
                     "run",
@@ -371,18 +367,5 @@ def test_docker_image_build_and_read(
                     "--catalog",
                     container_catalog_path,
                 ],
-                check=False,
-                capture_stderr=True,
-                capture_stdout=True,
+                check=True,
             )
-            if read_result.returncode != 0:
-                raise AssertionError(
-                    f"Failed to run `read` command in docker image {connector_image!r}. "
-                    "\n-----------------"
-                    f"EXIT CODE: {read_result.returncode}\n"
-                    "STDERR:\n"
-                    f"{read_result.stderr}\n"
-                    f"STDOUT:\n"
-                    f"{read_result.stdout}\n"
-                    "\n-----------------"
-                ) from None