feat(standard-tests): lint airbyte_secret marking and secret leaks

airbyte_cdk/destinations/destination.py

@@ -21,6 +21,7 @@
     Type,
 )
 from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit
+from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets, update_secrets
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 logger = logging.getLogger("airbyte")
@@ -125,6 +126,9 @@ def run_cmd(self, parsed_args: argparse.Namespace) -> Iterable[AirbyteMessage]:
             yield AirbyteMessage(type=Type.SPEC, spec=spec)
             return
         config = self.read_config(config_path=parsed_args.config)
+        # Register the config's secrets with the log filter before anything can print them,
+        # as the source entrypoint does. A config validation error quotes config values.
+        update_secrets(get_secrets(spec.connectionSpecification, config))
         if self.check_config_against_spec or cmd == "check":
             try:
                 check_config_against_spec_or_exit(config, spec)

airbyte_cdk/entrypoint.py

@@ -238,18 +238,15 @@ def check(
         try:
             check_result = self.source.check(self.logger, config)
         except AirbyteTracedException as traced_exc:
-            yield traced_exc.as_airbyte_message()
+            yield traced_exc.as_sanitized_airbyte_message()
             # The platform uses the exit code to surface unexpected failures so we raise the exception if the failure type not a config error
             # If the failure is not exceptional, we'll emit a failed connection status message and return
             if traced_exc.failure_type != FailureType.config_error:
                 raise traced_exc
             else:
-                yield AirbyteMessage(
-                    type=Type.CONNECTION_STATUS,
-                    connectionStatus=AirbyteConnectionStatus(
-                        status=Status.FAILED, message=traced_exc.message
-                    ),
-                )
+                connection_status = traced_exc.as_connection_status_message()
+                if connection_status:
+                    yield connection_status
                 return
         if check_result.status == Status.SUCCEEDED:
             self.logger.info("Check succeeded")

airbyte_cdk/test/entrypoint_wrapper.py

@@ -89,13 +89,15 @@ def __init__(
         *,
         command: list[str] | None = None,
         message_file: Path | None = None,
+        stderr: str | 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._stderr = stderr
         self._messages: list[AirbyteMessage] | None = None
         self._message_file: Path | None = message_file
         if messages:
@@ -124,6 +126,14 @@ def _parse_message(message: str) -> AirbyteMessage:
                 type=Type.LOG, log=AirbyteLogMessage(level=Level.INFO, message=message)
             )
 
+    @property
+    def stderr(self) -> str | None:
+        """The raw stderr of the connector process, or `None` if it was not captured.
+
+        Only a run in a separate process, such as `run_docker_airbyte_command`, captures it.
+        """
+        return self._stderr
+
     @property
     def records_and_state_messages(
         self,

airbyte_cdk/test/standard_tests/_job_runner.py

@@ -41,8 +41,14 @@ def run_test_job(
     connector_root: Path,
     test_scenario: ConnectorTestScenario | None = None,
     catalog: ConfiguredAirbyteCatalog | dict[str, Any] | None = None,
+    inspect_output: Callable[[entrypoint_wrapper.EntrypointOutput], None] | None = None,
 ) -> entrypoint_wrapper.EntrypointOutput:
-    """Run a test scenario from provided CLI args and return the result."""
+    """Run a test scenario from provided CLI args and return the result.
+
+    `inspect_output`, if given, is called with the output before any assertion on it, so it
+    runs whatever the outcome. The secret leak check uses it, since the assertions below
+    print raw messages.
+    """
     # Use default (empty) scenario if not provided:
     test_scenario = test_scenario or ConnectorTestScenario()
 
@@ -104,6 +110,9 @@ def run_test_job(
         args=args,
         expected_outcome=test_scenario.expected_outcome,
     )
+    if inspect_output:
+        inspect_output(result)
+
     if result.errors and test_scenario.expected_outcome.expect_success():
         raise result.as_exception()