connectors-ci: new checks to validate connector version (#25609)

* connectors-ci: new checks to validate connector version

* tmp ref attempt wip

* Revert "tmp ref attempt wip"

This reverts commit 0950d39.

* use semver and bypass increment check for specific files + no early fail

tools/ci_connector_ops/ci_connector_ops/pipelines/README.md

@@ -16,6 +16,7 @@ This documentation should be helpful for both local and CI use of the CLI. We in
 ### Install
 ```bash
 # Make sure that the current Python version is >= 3.10
+pyenv shell 3.10
 pip install "ci-connector-ops[pipelines] @ git+https://github.com/airbytehq/airbyte.git@master#subdirectory=tools/ci_connector_ops"
 cd airbyte
 airbyte-ci
@@ -108,6 +109,11 @@ Test connectors changed on the current branch:
 ```mermaid
 flowchart TD
     entrypoint[[For each selected connector]]
+    subgraph version ["Connector version checks"]
+        sem["Check version follows semantic versionning"]
+        incr["Check version is incremented"]
+        sem --> incr
+    end
     subgraph static ["Static code analysis"]
       qa[Run QA checks]
       fmt[Run code format checks]
@@ -126,8 +132,9 @@ flowchart TD
         build-->integration
         build-->cat
     end
-    entrypoint-->static
-    entrypoint-->tests
+    entrypoint-->version
+    version-->static
+    version-->tests
     report["Build test report"]
     tests-->report
     static-->report

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

@@ -107,25 +107,40 @@ def connectors(
         logger=logger,
     )
 
-    selected_connectors = get_all_released_connectors()
-    modified_connectors = get_modified_connectors(ctx.obj["modified_files"])
-    if modified:
-        selected_connectors = modified_connectors
-    else:
-        selected_connectors.update(modified_connectors)
+    all_connectors = get_all_released_connectors()
+
+    modified_connectors_and_files = get_modified_connectors(ctx.obj["modified_files"])
+    # We select all connectors by default
+    selected_connectors_and_files = {connector: modified_connectors_and_files.get(connector, []) for connector in all_connectors}
+
     if names:
-        selected_connectors = {connector for connector in selected_connectors if connector.technical_name in names}
+        selected_connectors_and_files = {
+            connector: selected_connectors_and_files[connector]
+            for connector in selected_connectors_and_files
+            if connector.technical_name in names
+        }
     if languages:
-        selected_connectors = {connector for connector in selected_connectors if connector.language in languages}
+        selected_connectors_and_files = {
+            connector: selected_connectors_and_files[connector]
+            for connector in selected_connectors_and_files
+            if connector.language in languages
+        }
     if release_stages:
-        selected_connectors = {connector for connector in selected_connectors if connector.release_stage in release_stages}
-
-    if not selected_connectors:
+        selected_connectors_and_files = {
+            connector: selected_connectors_and_files[connector]
+            for connector in selected_connectors_and_files
+            if connector.release_stage in release_stages
+        }
+    if modified:
+        selected_connectors_and_files = {
+            connector: modified_files for connector, modified_files in selected_connectors_and_files.items() if modified_files
+        }
+    if not selected_connectors_and_files:
         click.secho("No connector were selected according to your inputs. Please double check your filters.", fg="yellow")
         sys.exit(0)
 
-    ctx.obj["selected_connectors"] = selected_connectors
-    ctx.obj["selected_connectors_names"] = [c.technical_name for c in selected_connectors]
+    ctx.obj["selected_connectors_and_files"] = selected_connectors_and_files
+    ctx.obj["selected_connectors_names"] = [c.technical_name for c in selected_connectors_and_files.keys()]
 
 
 @connectors.command(cls=DaggerPipelineCommand, help="Test all the selected connectors.")
@@ -146,12 +161,13 @@ def test(
             ctx.obj["is_local"],
             ctx.obj["git_branch"],
             ctx.obj["git_revision"],
+            modified_files,
             ctx.obj["use_remote_secrets"],
             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"),
         )
-        for connector in ctx.obj["selected_connectors"]
+        for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
     ]
     try:
         anyio.run(run_connectors_pipelines, connectors_tests_contexts, run_connector_test_pipeline, "Test Pipeline", ctx.obj["concurrency"])
@@ -189,12 +205,13 @@ def build(ctx: click.Context) -> bool:
             ctx.obj["is_local"],
             ctx.obj["git_branch"],
             ctx.obj["git_revision"],
+            modified_files,
             ctx.obj["use_remote_secrets"],
             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"),
         )
-        for connector in ctx.obj["selected_connectors"]
+        for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
     ]
     anyio.run(run_connectors_pipelines, connectors_contexts, run_connector_build_pipeline, "Build Pipeline", ctx.obj["concurrency"])
 
@@ -246,10 +263,10 @@ def publish(
             abort=True,
         )
     if ctx.obj["modified"]:
-        selected_connectors = get_modified_connectors(get_modified_metadata_files(ctx.obj["modified_files"]))
-        selected_connectors_names = [connector.technical_name for connector in selected_connectors]
+        selected_connectors_and_files = get_modified_connectors(get_modified_metadata_files(ctx.obj["modified_files"]))
+        selected_connectors_names = [connector.technical_name for connector in selected_connectors_and_files.keys()]
     else:
-        selected_connectors = ctx.obj["selected_connectors"]
+        selected_connectors_and_files = ctx.obj["selected_connectors_and_files"]
         selected_connectors_names = ctx.obj["selected_connectors_names"]
 
     click.secho(f"Will publish the following connectors: {', '.join(selected_connectors_names)}.", fg="green")
@@ -263,12 +280,13 @@ def publish(
             ctx.obj["is_local"],
             ctx.obj["git_branch"],
             ctx.obj["git_revision"],
+            modified_files,
             ctx.obj["use_remote_secrets"],
             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"),
         )
-        for connector in selected_connectors
+        for connector, modified_files in selected_connectors_and_files.items()
     ]
     connectors_contexts = anyio.run(
         run_connectors_pipelines,

tools/ci_connector_ops/ci_connector_ops/pipelines/contexts.py

@@ -235,6 +235,7 @@ def __init__(
         is_local: bool,
         git_branch: bool,
         git_revision: bool,
+        modified_files: List[str],
         use_remote_secrets: bool = True,
         connector_acceptance_test_image: Optional[str] = DEFAULT_CONNECTOR_ACCEPTANCE_TEST_IMAGE,
         gha_workflow_run_url: Optional[str] = None,
@@ -259,7 +260,7 @@ def __init__(
         self.connector = connector
         self.use_remote_secrets = use_remote_secrets
         self.connector_acceptance_test_image = connector_acceptance_test_image
-
+        self.modified_files = modified_files
         self._secrets_dir = None
         self._updated_secrets_dir = None
 

tools/ci_connector_ops/ci_connector_ops/pipelines/tests/__init__.py

@@ -11,7 +11,7 @@
 from ci_connector_ops.pipelines.bases import ConnectorReport, StepResult
 from ci_connector_ops.pipelines.contexts import ConnectorContext
 from ci_connector_ops.pipelines.tests import java_connectors, python_connectors
-from ci_connector_ops.pipelines.tests.common import QaChecks
+from ci_connector_ops.pipelines.tests.common import QaChecks, VersionFollowsSemverCheck, VersionIncrementCheck
 from ci_connector_ops.utils import ConnectorLanguage
 
 LANGUAGE_MAPPING = {
@@ -28,6 +28,19 @@
 }
 
 
+async def run_version_checks(context: ConnectorContext) -> List[StepResult]:
+    """Run the version checks on a connector.
+
+    Args:
+        context (ConnectorContext): The current connector context.
+
+    Returns:
+        List[StepResult]: The results of the version checks steps.
+    """
+    context.logger.info("Run version checks.")
+    return [await VersionFollowsSemverCheck(context).run(), await VersionIncrementCheck(context).run()]
+
+
 async def run_qa_checks(context: ConnectorContext) -> List[StepResult]:
     """Run the QA checks on a connector.
 
@@ -89,6 +102,7 @@ async def run_connector_test_pipeline(context: ConnectorContext, semaphore: anyi
         async with context:
             async with asyncer.create_task_group() as task_group:
                 tasks = [
+                    task_group.soonify(run_version_checks)(context),
                     task_group.soonify(run_qa_checks)(context),
                     task_group.soonify(run_code_format_checks)(context),
                     task_group.soonify(run_all_tests)(context),

tools/ci_connector_ops/ci_connector_ops/pipelines/tests/common.py

@@ -4,15 +4,124 @@
 
 """This module groups steps made to run tests agnostic to a connector language."""
 
-from typing import Optional
+from abc import ABC, abstractmethod
+from functools import cached_property
+from typing import ClassVar, Optional
 
 import asyncer
+import requests
+import semver
+import yaml
 from ci_connector_ops.pipelines.actions import environments
 from ci_connector_ops.pipelines.bases import PytestStep, Step, StepResult, StepStatus
+from ci_connector_ops.pipelines.contexts import CIContext
+from ci_connector_ops.pipelines.utils import METADATA_FILE_NAME
 from ci_connector_ops.utils import DESTINATION_DEFINITIONS_FILE_PATH, SOURCE_DEFINITIONS_FILE_PATH
 from dagger import File
 
 
+class VersionCheck(Step, ABC):
+    """A step to validate the connector version was bumped if files were modified"""
+
+    GITHUB_URL_PREFIX_FOR_CONNECTORS = "https://raw.githubusercontent.com/airbytehq/airbyte/master/airbyte-integrations/connectors"
+    failure_message: ClassVar
+    should_run = True
+
+    @property
+    def github_master_metadata_url(self):
+        return f"{self.GITHUB_URL_PREFIX_FOR_CONNECTORS}/{self.context.connector.technical_name}/{METADATA_FILE_NAME}"
+
+    @cached_property
+    def master_metadata(self) -> dict:
+        response = requests.get(self.github_master_metadata_url)
+        response.raise_for_status()
+        return yaml.safe_load(response.text)
+
+    @property
+    def master_connector_version(self) -> semver.Version:
+        metadata = self.master_metadata
+        return semver.Version.parse(str(metadata["data"]["dockerImageTag"]))
+
+    @property
+    def current_connector_version(self) -> semver.Version:
+        return semver.Version.parse(str(self.context.metadata["dockerImageTag"]))
+
+    @property
+    def success_result(self) -> StepResult:
+        return StepResult(self, status=StepStatus.SUCCESS)
+
+    @property
+    def failure_result(self) -> StepResult:
+        return StepResult(self, status=StepStatus.FAILURE, stderr=self.failure_message)
+
+    @abstractmethod
+    def validate(self) -> StepResult:
+        raise NotImplementedError()
+
+    async def _run(self) -> StepResult:
+        if not self.should_run:
+            return StepResult(self, status=StepStatus.SKIPPED, stdout="No modified files required a version bump.")
+        if self.context.ci_context is CIContext.MASTER:
+            return StepResult(self, status=StepStatus.SKIPPED, stdout="Version check are not running in master context.")
+        try:
+            return self.validate()
+        except (requests.HTTPError, ValueError, TypeError) as e:
+            return StepResult(self, status=StepStatus.FAILURE, stderr=str(e))
+
+
+class VersionIncrementCheck(VersionCheck):
+
+    title = "Connector version increment check."
+
+    BYPASS_CHECK_FOR = [
+        METADATA_FILE_NAME,
+        "acceptance-test-config.yml",
+        "README.md",
+        "bootstrap.md",
+        ".dockerignore",
+        "unit_tests",
+        "integration_tests",
+        "src/test",
+        "src/test-integration",
+        "src/test-performance",
+        "build.gradle",
+    ]
+
+    @property
+    def failure_message(self) -> str:
+        return f"The dockerImageTag in {METADATA_FILE_NAME} was not incremented. The files you modified should lead to a version bump. Master version is {self.master_connector_version}, current version is {self.current_connector_version}"
+
+    @property
+    def should_run(self) -> bool:
+        for filename in self.context.modified_files:
+            relative_path = filename.replace(str(self.context.connector.code_directory) + "/", "")
+            if not any([relative_path.startswith(to_bypass) for to_bypass in self.BYPASS_CHECK_FOR]):
+                return True
+        return False
+
+    def validate(self) -> StepResult:
+        if not self.current_connector_version > self.master_connector_version:
+            return self.failure_result
+        return self.success_result
+
+
+class VersionFollowsSemverCheck(VersionCheck):
+
+    title = "Connector version semver check."
+
+    @property
+    def failure_message(self) -> str:
+        return f"The dockerImageTag in {METADATA_FILE_NAME} is not following semantic versioning or was decremented. Master version is {self.master_connector_version}, current version is {self.current_connector_version}"
+
+    def validate(self) -> StepResult:
+        try:
+            if not self.current_connector_version >= self.master_connector_version:
+                return self.failure_result
+        except ValueError:
+            return self.failure_result
+        return self.success_result
+
+
 class QaChecks(Step):
     """A step to run QA checks for a connector."""
 

tools/ci_connector_ops/ci_connector_ops/pipelines/utils.py

@@ -162,9 +162,14 @@ async def get_modified_files_in_branch_remote(
 
 
 def get_modified_files_in_branch_local(current_git_revision: str, diffed_branch: str = "master") -> Set[str]:
-    """Use git diff to spot the modified files on the local branch."""
+    """Use git diff and git status to spot the modified files on the local branch."""
     airbyte_repo = git.Repo()
     modified_files = airbyte_repo.git.diff("--diff-filter=MA", "--name-only", f"{diffed_branch}...{current_git_revision}").split("\n")
+    status_output = airbyte_repo.git.status("--porcelain")
+    for not_committed_change in status_output.split("\n"):
+        file_path = not_committed_change.strip().split(" ")[-1]
+        if file_path:
+            modified_files.append(file_path)
     return set(modified_files)
 
 
@@ -215,13 +220,17 @@ def get_modified_files_in_commit(current_git_branch: str, current_git_revision:
         return anyio.run(get_modified_files_in_commit_remote, current_git_branch, current_git_revision)
 
 
-def get_modified_connectors(modified_files: Set[Union[str, Path]]) -> Set[Connector]:
-    """Create a set of modified connectors according to the modified files on the branch."""
-    modified_connectors = []
+def get_modified_connectors(modified_files: Set[Union[str, Path]]) -> dict[Connector, List[str]]:
+    """Create a mapping of modified connectors (key) and modified files (value)."""
+    modified_connectors = {}
     for file_path in modified_files:
         if str(file_path).startswith(SOURCE_CONNECTOR_PATH_PREFIX) or str(file_path).startswith(DESTINATION_CONNECTOR_PATH_PREFIX):
-            modified_connectors.append(Connector(get_connector_name_from_path(str(file_path))))
-    return set(modified_connectors)
+            modified_connector = Connector(get_connector_name_from_path(str(file_path)))
+            if modified_connector in modified_connectors:
+                modified_connectors[modified_connector].append(file_path)
+            else:
+                modified_connectors[modified_connector] = [file_path]
+    return modified_connectors
 
 
 def get_modified_metadata_files(modified_files: Set[Union[str, Path]]) -> Set[Path]:

tools/ci_connector_ops/setup.py

@@ -42,7 +42,7 @@ def local_pkg(name: str) -> str:
     "pytablewriter~=0.64.2",
 ]
 
-PIPELINES_REQUIREMENTS = ["dagger-io==0.5.0", "asyncer", "anyio", "more-itertools", "docker"]
+PIPELINES_REQUIREMENTS = ["dagger-io==0.5.0", "asyncer", "anyio", "more-itertools", "docker", "requests", "semver"]
 
 setup(
     version="0.2.1",