add validate cli command

airbyte/cli.py

@@ -151,10 +151,83 @@ def _resolve_destination_job(
     )
 
 
-@click.command()
-def validate() -> None:
-    """Validate the data."""
-    click.echo("Validating data...")
+@click.command(
+    help=(
+        "Validate the connector has a valid CLI and is able to run `spec`. "
+        "If 'config' is provided, we will also run a `check` on the connector "
+        "with the provided config."
+    ),
+)
+@click.option(
+    "--connector",
+    type=str,
+    help="The connector name or path.",
+)
+@click.option(
+    "--config",
+    type=Path,
+    required=False,
+    help="The path to a configuration file for the named source or destination.",
+)
+@click.option(
+    "--install",
+    is_flag=True,
+    default=False,
+    help=(
+        "Whether to install the connector if it is not available locally. "
+        "Defaults to False, meaning the connector is expected to be already be installed."
+    ),
+)
+def validate(
+    connector: str | None = None,
+    config: Path | None = None,
+    *,
+    install: bool = False,
+) -> None:
+    """Validate the connector."""
+    local_executable: Path | None = None
+    if not connector:
+        raise PyAirbyteInputError(
+            message="No connector provided.",
+        )
+    if connector.startswith(".") or "/" in connector:
+        # Treat the connector as a path.
+        local_executable = Path(connector)
+        if not local_executable.exists():
+            raise PyAirbyteInputError(
+                message=f"Connector executable not found: {connector}",
+            )
+        connector_name = local_executable.stem
+    else:
+        connector_name = connector
+
+    if not connector_name.startswith("source-") and not connector_name.startswith("destination-"):
+        raise PyAirbyteInputError(
+            message="Expected a connector name or path to executable.",
+            input_value=connector,
+        )
+
+    connector_obj: Source | Destination
+    if connector_name.startswith("source-"):
+        connector_obj = get_source(
+            name=connector_name,
+            local_executable=local_executable,
+            install_if_missing=install,
+        )
+    else:  # destination
+        connector_obj = get_destination(
+            name=connector_name,
+            local_executable=local_executable,
+            install_if_missing=install,
+        )
+
+    print("Getting `spec` output from connector...")
+    connector_obj.print_config_spec()
+
+    if config:
+        print("Running connector check...")
+        connector_obj.set_config(json.loads(config.read_text(encoding="utf-8")))
+        connector_obj.check()
 
 
 @click.command()