connectors-ci: pass PR id to airbyte-ci (#26504)

.github/workflows/connector_integration_test_single_dagger.yml

@@ -6,10 +6,14 @@ on:
       test-connectors-options:
         description: "Options to pass to the 'airbyte-ci connectors test' command"
         default: "--modified"
-  # pull_request:
-  #   paths:
-  #     - "airbyte-integrations/connectors/**"
-  #     - ".github/workflows/connector_integration_test_single_dagger.yml"
+  pull_request:
+    paths:
+      #- "airbyte-integrations/connectors/**"
+      #- ".github/workflows/connector_integration_test_single_dagger.yml"
+    types:
+      - opened
+      - synchronize
+      - ready_for_review
 jobs:
   connectors_ci:
     name: Connectors CI
@@ -89,3 +93,4 @@ jobs:
           CI_GIT_REVISION: ${{ github.event.pull_request.head.sha }}
           CI_CONTEXT: "pull_request"
           CI_PIPELINE_START_TIMESTAMP: ${{ steps.get-start-timestamp.outputs.start-timestamp }}
+          PULL_REQUEST_NUMBER: ${{ github.event.pull_request.number }}

tools/ci_connector_ops/ci_connector_ops/pipelines/bases.py

@@ -23,6 +23,7 @@
 from rich.style import Style
 from rich.table import Table
 from rich.text import Text
+from tabulate import tabulate
 
 if TYPE_CHECKING:
     from ci_connector_ops.pipelines.contexts import ConnectorContext, PipelineContext
@@ -73,6 +74,15 @@ def get_rich_style(self) -> Style:
         if self is StepStatus.SKIPPED:
             return Style(color="yellow")
 
+    def get_emoji(self) -> str:
+        """Match emoji used in the console output to the step status."""
+        if self is StepStatus.SUCCESS:
+            return "✅"
+        if self is StepStatus.FAILURE:
+            return "❌"
+        if self is StepStatus.SKIPPED:
+            return "🟡"
+
     def __str__(self) -> str:  # noqa D105
         return self.value
 
@@ -264,6 +274,7 @@ def to_json(self) -> str:
                 "git_branch": self.pipeline_context.git_branch,
                 "git_revision": self.pipeline_context.git_revision,
                 "ci_context": self.pipeline_context.ci_context,
+                "pull_request_url": self.pipeline_context.pull_request.html_url if self.pipeline_context.pull_request else None,
             }
         )
 
@@ -314,6 +325,10 @@ class ConnectorReport(Report):
     def should_be_saved(self) -> bool:  # noqa D102
         return self.pipeline_context.is_ci
 
+    @property
+    def should_be_commented_on_pr(self) -> bool:  # noqa D102
+        return self.pipeline_context.is_ci and self.pipeline_context.pull_request and self.pipeline_context.PRODUCTION
+
     def to_json(self) -> str:
         """Create a JSON representation of the connector test report.
 
@@ -341,6 +356,24 @@ def to_json(self) -> str:
             }
         )
 
+    def post_comment_on_pr(self) -> None:
+        icon_url = f"https://raw.githubusercontent.com/airbytehq/airbyte/{self.pipeline_context.git_revision}/{self.pipeline_context.connector.code_directory}/icon.svg"
+        global_status_emoji = "✅" if self.success else "❌"
+        commit_url = f"{self.pipeline_context.pull_request.html_url}/commits/{self.pipeline_context.git_revision}"
+        markdown_comment = f'## <img src="{icon_url}" width="40" height="40"> {self.pipeline_context.connector.technical_name} test report (commit [`{self.pipeline_context.git_revision[:10]}`]({commit_url})) - {global_status_emoji}\n\n'
+        markdown_comment += f"⏲️  Total pipeline duration: {round(self.run_duration)} seconds\n\n"
+        report_data = [
+            [step_result.step.title, step_result.status.get_emoji()]
+            for step_result in self.steps_results
+            if step_result.status is not StepStatus.SKIPPED
+        ]
+        markdown_comment += tabulate(report_data, headers=["Step", "Result"], tablefmt="pipe") + "\n\n"
+        markdown_comment += f"🔗 [View the logs here]({self.pipeline_context.gha_workflow_run_url})\n\n"
+        markdown_comment += "*Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.*\n"
+        markdown_comment += "**You can run the same pipeline locally on this branch with the [airbyte-ci](https://github.com/airbytehq/airbyte/blob/master/tools/ci_connector_ops/ci_connector_ops/pipelines/README.md) tool with the following command**\n"
+        markdown_comment += f"```bash\nairbyte-ci connectors --name={self.pipeline_context.connector.technical_name} test\n```\n\n"
+        self.pipeline_context.pull_request.create_issue_comment(markdown_comment)
+
     def print(self):
         """Print the test report to the console in a nice way."""
         connector_name = self.pipeline_context.connector.technical_name

tools/ci_connector_ops/ci_connector_ops/pipelines/commands/airbyte_ci.py

@@ -5,6 +5,7 @@
 """This module is the CLI entrypoint to the airbyte-ci commands."""
 
 import click
+from ci_connector_ops.pipelines import github
 from ci_connector_ops.pipelines.bases import CIContext
 from ci_connector_ops.pipelines.utils import (
     get_current_epoch_time,
@@ -31,6 +32,8 @@
 @click.option("--gha-workflow-run-id", help="[CI Only] The run id of the GitHub action workflow", default=None, type=str)
 @click.option("--ci-context", default=CIContext.MANUAL, envvar="CI_CONTEXT", type=click.Choice(CIContext))
 @click.option("--pipeline-start-timestamp", default=get_current_epoch_time, envvar="CI_PIPELINE_START_TIMESTAMP", type=int)
+@click.option("--pull-request-number", envvar="PULL_REQUEST_NUMBER", type=int)
+@click.option("--ci-github-access-token", envvar="CI_GITHUB_ACCESS_TOKEN", type=str)
 @click.pass_context
 def airbyte_ci(
     ctx: click.Context,
@@ -41,9 +44,12 @@ def airbyte_ci(
     gha_workflow_run_id: str,
     ci_context: str,
     pipeline_start_timestamp: int,
+    pull_request_number: int,
+    ci_github_access_token: str,
 ):  # noqa D103
     ctx.ensure_object(dict)
     ctx.obj["is_local"] = is_local
+    ctx.obj["is_ci"] = not is_local
     ctx.obj["git_branch"] = git_branch
     ctx.obj["git_revision"] = git_revision
     ctx.obj["gha_workflow_run_id"] = gha_workflow_run_id
@@ -56,6 +62,20 @@ def airbyte_ci(
     ctx.obj["modified_files_in_commit"] = get_modified_files_in_commit(git_branch, git_revision, is_local)
     ctx.obj["modified_files"] = ctx.obj["modified_files_in_commit"] if git_branch == "master" else ctx.obj["modified_files_in_branch"]
 
+    if pull_request_number and ci_github_access_token:
+        ctx.obj["pull_request"] = github.get_pull_request(pull_request_number, ci_github_access_token)
+    else:
+        ctx.obj["pull_request"] = None
+    if not is_local:
+        click.echo("Running airbyte-ci in CI mode.")
+        click.echo(f"CI Context: {ci_context}")
+        click.echo(f"Git Branch: {git_branch}")
+        click.echo(f"Git Revision: {git_revision}")
+        click.echo(f"GitHub Workflow Run ID: {gha_workflow_run_id}")
+        click.echo(f"GitHub Workflow Run URL: {ctx.obj['gha_workflow_run_url']}")
+        click.echo(f"Pull Request Number: {pull_request_number}")
+        click.echo(f"Pipeline Start Timestamp: {pipeline_start_timestamp}")
+
 
 airbyte_ci.add_command(connectors)
 airbyte_ci.add_command(metadata)

tools/ci_connector_ops/ci_connector_ops/pipelines/commands/groups/connectors.py

@@ -6,6 +6,7 @@
 
 import logging
 import os
+import sys
 from pathlib import Path
 from typing import Any, Dict, Tuple
 
@@ -144,6 +145,10 @@ def test(
     Args:
         ctx (click.Context): The click context.
     """
+    if ctx.obj["is_ci"] and ctx.obj["pull_request"] and ctx.obj["pull_request"].draft:
+        click.echo("Skipping connectors tests for draft pull request.")
+        sys.exit(0)
+
     click.secho(f"Will run the test pipeline for the following connectors: {', '.join(ctx.obj['selected_connectors_names'])}.", fg="green")
     if ctx.obj["selected_connectors_and_files"]:
         update_global_commit_status_check_for_tests(ctx.obj, "pending")
@@ -165,6 +170,7 @@ def test(
             gha_workflow_run_url=ctx.obj.get("gha_workflow_run_url"),
             pipeline_start_timestamp=ctx.obj.get("pipeline_start_timestamp"),
             ci_context=ctx.obj.get("ci_context"),
+            pull_request=ctx.obj.get("pull_request"),
         )
         for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
     ]
@@ -327,6 +333,7 @@ def publish(
             gha_workflow_run_url=ctx.obj.get("gha_workflow_run_url"),
             pipeline_start_timestamp=ctx.obj.get("pipeline_start_timestamp"),
             ci_context=ctx.obj.get("ci_context"),
+            pull_request=ctx.obj.get("pull_request"),
         )
         for connector, modified_files in selected_connectors_and_files.items()
     ]

tools/ci_connector_ops/ci_connector_ops/pipelines/contexts.py

@@ -22,6 +22,7 @@
 from ci_connector_ops.pipelines.utils import AIRBYTE_REPO_URL, METADATA_FILE_NAME, sanitize_gcs_credentials
 from ci_connector_ops.utils import Connector
 from dagger import Client, Directory, Secret
+from github import PullRequest
 
 
 class ContextState(Enum):
@@ -37,6 +38,8 @@ class ContextState(Enum):
 class PipelineContext:
     """The pipeline context is used to store configuration for a specific pipeline run."""
 
+    PRODUCTION = bool(os.environ.get("PRODUCTION", False))  # Set this to True to enable production mode (e.g. to send PR comments)
+
     DEFAULT_EXCLUDED_FILES = (
         [".git"]
         + glob("**/build", recursive=True)
@@ -63,6 +66,7 @@ def __init__(
         is_ci_optional: bool = False,
         slack_webhook: Optional[str] = None,
         reporting_slack_channel: Optional[str] = None,
+        pull_request: PullRequest = None,
     ):
         """Initialize a pipeline context.
 
@@ -77,6 +81,7 @@ def __init__(
             is_ci_optional (bool, optional): Whether the CI is optional. Defaults to False.
             slack_webhook (Optional[str], optional): Slack webhook to send messages to. Defaults to None.
             reporting_slack_channel (Optional[str], optional): Slack channel to send messages to. Defaults to None.
+            pull_request (PullRequest, optional): The pull request object if the pipeline was triggered by a pull request. Defaults to None.
         """
         self.pipeline_name = pipeline_name
         self.is_local = is_local
@@ -90,6 +95,7 @@ def __init__(
         self.is_ci_optional = is_ci_optional
         self.slack_webhook = slack_webhook
         self.reporting_slack_channel = reporting_slack_channel
+        self.pull_request = pull_request
         self.logger = logging.getLogger(self.pipeline_name)
         self.dagger_client = None
         self._report = None
@@ -268,6 +274,7 @@ def __init__(
         ci_context: Optional[str] = None,
         slack_webhook: Optional[str] = None,
         reporting_slack_channel: Optional[str] = None,
+        pull_request: PullRequest = None,
     ):
         """Initialize a connector context.
 
@@ -285,6 +292,7 @@ def __init__(
             ci_context (Optional[str], optional): Pull requests, workflow dispatch or nightly build. Defaults to None.
             slack_webhook (Optional[str], optional): The slack webhook to send messages to. Defaults to None.
             reporting_slack_channel (Optional[str], optional): The slack channel to send messages to. Defaults to None.
+            pull_request (PullRequest, optional): The pull request object if the pipeline was triggered by a pull request. Defaults to None.
         """
 
         self.pipeline_name = pipeline_name
@@ -306,6 +314,7 @@ def __init__(
             ci_context=ci_context,
             slack_webhook=slack_webhook,
             reporting_slack_channel=reporting_slack_channel,
+            pull_request=pull_request,
         )
 
     @property
@@ -406,6 +415,8 @@ async def __aexit__(
             )
             if report_upload_exit_code != 0:
                 self.logger.error("Uploading the report to S3 failed.")
+        if self.report.should_be_commented_on_pr:
+            self.report.post_comment_on_pr()
         await asyncify(update_commit_status_check)(**self.github_commit_status)
         if self.should_send_slack_message:
             await asyncify(send_message_to_webhook)(self.create_slack_message(), self.reporting_slack_channel, self.slack_webhook)
@@ -436,6 +447,7 @@ def __init__(
         gha_workflow_run_url: Optional[str] = None,
         pipeline_start_timestamp: Optional[int] = None,
         ci_context: Optional[str] = None,
+        pull_request: PullRequest = None,
     ):
         self.pre_release = pre_release
         self.spec_cache_bucket_name = spec_cache_bucket_name

tools/ci_connector_ops/ci_connector_ops/pipelines/github.py

@@ -15,7 +15,7 @@
 if TYPE_CHECKING:
     from logging import Logger
 
-from github import Github
+from github import Github, PullRequest
 
 AIRBYTE_GITHUB_REPO = "airbytehq/airbyte"
 GITHUB_GLOBAL_CONTEXT_FOR_TESTS = "Connectors CI tests"
@@ -76,6 +76,20 @@ def update_commit_status_check(
     safe_log(logger, f"Created {state} status for commit {sha} on Github in {context} context with desc: {description}.")
 
 
+def get_pull_request(pull_request_number: int, github_access_token: str) -> PullRequest:
+    """Get a pull request object from its number.
+
+    Args:
+        pull_request_number (str): The number of the pull request to get.
+        github_access_token (str): The GitHub access token to use to authenticate.
+    Returns:
+        PullRequest: The pull request object.
+    """
+    github_client = Github(github_access_token)
+    airbyte_repo = github_client.get_repo(AIRBYTE_GITHUB_REPO)
+    return airbyte_repo.get_pull(pull_request_number)
+
+
 def update_global_commit_status_check_for_tests(click_context: dict, github_state: str, logger: Logger = None):
     update_commit_status_check(
         click_context["git_revision"],

tools/ci_connector_ops/ci_connector_ops/pipelines/pipelines/metadata.py

@@ -10,7 +10,7 @@
 from ci_connector_ops.pipelines.actions.environments import with_pip_packages, with_poetry_module, with_python_base
 from ci_connector_ops.pipelines.bases import Report, Step, StepResult
 from ci_connector_ops.pipelines.contexts import PipelineContext
-from ci_connector_ops.pipelines.utils import DAGGER_CONFIG, METADATA_FILE_NAME, execute_concurrently, METADATA_ICON_FILE_NAME
+from ci_connector_ops.pipelines.utils import DAGGER_CONFIG, METADATA_FILE_NAME, METADATA_ICON_FILE_NAME, execute_concurrently
 
 METADATA_DIR = "airbyte-ci/connectors/metadata_service"
 METADATA_LIB_MODULE_PATH = "lib"

tools/ci_connector_ops/setup.py

@@ -51,6 +51,7 @@ def local_pkg(name: str) -> str:
     "requests",
     "semver",
     "airbyte-protocol-models",
+    "tabulate",
 ]
 
 setup(