wip (maybe revert): trying to duck-type IConnector to work with AirbyteEntrypoint

Author: aaronsteers

airbyte_cdk/test/standard_tests/docker_connectors.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import subprocess
+from collections.abc import Mapping
 from pathlib import Path
-from typing import Any, Mapping
+from typing import Any, TextIO
 
 from airbyte_cdk.connector import BaseConnector
+from airbyte_cdk.models import ConnectorSpecification
 from airbyte_cdk.test.standard_tests._job_runner import IConnector
 from airbyte_cdk.utils.docker import build_connector_image, run_docker_command
 
@@ -34,7 +37,7 @@ def read_config(config_path: str) -> Mapping[str, Any]:
                 f"The content of {config_path} is not an object and therefore is not a valid config. Please ensure the file represent a config."
             )
 
-    def spec(
+    def spec_(
         self,
         logger: logging.Logger,
     ) -> Any:
@@ -58,17 +61,22 @@ def check(
     def _run_cli(
         self,
         args: list[str],
+        *,
         logger: logging.Logger,
-    ) -> None:
+        stdout: TextIO | None = None,
+        stdin: TextIO | None = None,
+    ) -> subprocess.CompletedProcess[str]:
         """Run the CLI command."""
         logger.info(f"Running CLI connector: {self.connector_name} with args: {args}")
         base_cmd: list[str] = [
             self.connector_name,
             *args,
         ]
-        subprocess.run(
-            base_cmd,
-            check=True,
+        return subprocess.run(
+            args=base_cmd,
+            text=True,
+            stdout=stdout,
+            stdin=stdin,
         )
 
     def launch(
@@ -96,17 +104,37 @@ def __init__(
         logger: logging.Logger | None = None,
     ) -> None:
         self.docker_image = docker_image
+        self._config_file_path: Path | None = None
         super().__init__(
             connector_name=connector_name,
             logger=logger,
         )
 
+    def spec(self, logger: logging.Logger) -> ConnectorSpecification:
+        """
+        Returns the spec for this integration. The spec is a JSON-Schema object describing the required configurations (e.g: username and password)
+        required to run this integration. By default, this will be loaded from a "spec.yaml" or a "spec.json" in the package root.
+        """
+        self.launch(
+            ["spec"],
+            logger=logger,
+        )
+
+    def configure(self, config: Mapping[str, Any], temp_dir: str) -> Mapping[str, Any]:
+        """
+        Persist config in temporary directory to run the Source job
+        """
+        self._config_file_path = Path(temp_dir) / "config.json"
+        self._config_file_path.write_text(json.dumps(config))
+
+        return config
+
     def launch(
         self,
         args: list[str],
         *,
         logger: logging.Logger | None = None,
-    ) -> None:
+    ) -> subprocess.CompletedProcess[str]:
         """Run the connector."""
         _ = logger
         print(f"Running docker connector: {self.connector_name} with args: {args}")
@@ -116,7 +144,7 @@ def launch(
             "--rm",
             "--network=host",
         ]
-        run_docker_command(
+        return run_docker_command(
             cmd=[
                 *docker_base_cmd,
                 self.docker_image,