feat: add jenkins support (#147)

Fixes MRGFY-5591

Author: sileht

pytest_mergify/resources/jenkins.py

@@ -0,0 +1,46 @@
+import os
+from opentelemetry.sdk.resources import Resource, ResourceDetector
+
+from opentelemetry.semconv._incubating.attributes import cicd_attributes
+from opentelemetry.semconv._incubating.attributes import vcs_attributes
+
+from pytest_mergify import utils
+
+
+class RepsoitoryNameExtractionError(Exception):
+    pass
+
+
+class JenkinsResourceDetector(ResourceDetector):
+    """Detects OpenTelemetry Resource attributes for Jenkins."""
+
+    OPENTELEMETRY_JENKINS_MAPPING = {
+        cicd_attributes.CICD_PIPELINE_NAME: (str, "JOB_NAME"),
+        cicd_attributes.CICD_PIPELINE_RUN_ID: (str, "BUILD_ID"),
+        "cicd.pipeline.run.url": (str, "BUILD_URL"),
+        "cicd.pipeline.runner.name": (str, "NODE_NAME"),
+        vcs_attributes.VCS_REF_HEAD_NAME: (str, "GIT_BRANCH"),
+        vcs_attributes.VCS_REF_HEAD_REVISION: (str, "GIT_COMMIT"),
+        vcs_attributes.VCS_REPOSITORY_URL_FULL: (str, "GIT_URL"),
+    }
+
+    def detect(self) -> Resource:
+        if utils.get_ci_provider() != "jenkins":
+            return Resource({})
+
+        attributes = {}
+        for attribute_name, (
+            type_,
+            envvar,
+        ) in self.OPENTELEMETRY_JENKINS_MAPPING.items():
+            if envvar in os.environ:
+                try:
+                    attributes[attribute_name] = type_(os.environ[envvar])
+                except Exception:
+                    pass  # skip invalid conversions
+
+        repository_name = utils._get_repository_name_from_env_url("GIT_URL")
+        if repository_name:
+            attributes["vcs.repository.name"] = repository_name
+
+        return Resource(attributes)

pytest_mergify/tracer.py

@@ -18,6 +18,7 @@
 import pytest_mergify.resources.github_actions as resources_gha
 import pytest_mergify.resources.pytest as resources_pytest
 import pytest_mergify.resources.mergify as resources_mergify
+import pytest_mergify.resources.jenkins as resources_jenkins
 
 
 class SynchronousBatchSpanProcessor(export.SimpleSpanProcessor):
@@ -103,6 +104,7 @@ def __post_init__(self) -> None:
             [
                 resources_ci.CIResourceDetector(),
                 resources_gha.GitHubActionsResourceDetector(),
+                resources_jenkins.JenkinsResourceDetector(),
                 resources_pytest.PytestResourceDetector(),
                 resources_mergify.MergifyResourceDetector(),
             ]

pytest_mergify/utils.py

@@ -2,11 +2,14 @@
 import re
 import typing
 
-CIProviderT = typing.Literal["github_actions", "circleci", "pytest_mergify_suite"]
+CIProviderT = typing.Literal[
+    "github_actions", "circleci", "pytest_mergify_suite", "jenkins"
+]
 
 SUPPORTED_CIs: typing.Dict[str, CIProviderT] = {
     "GITHUB_ACTIONS": "github_actions",
     "CIRCLECI": "circleci",
+    "JENKINS_URL": "jenkins",
     "_PYTEST_MERGIFY_TEST": "pytest_mergify_suite",
 }
 
@@ -19,27 +22,42 @@ def is_in_ci() -> bool:
 
 def get_ci_provider() -> typing.Optional[CIProviderT]:
     for envvar, name in SUPPORTED_CIs.items():
-        if envvar in os.environ and strtobool(os.environ[envvar]):
-            return name
+        if envvar in os.environ:
+            try:
+                enabled = strtobool(os.environ[envvar])
+            except ValueError:
+                # Not a boolean, just check it's not empty
+                enabled = bool(os.environ[envvar].strip())
+            if enabled:
+                return name
+
+    return None
+
+
+def _get_repository_name_from_env_url(env: str) -> typing.Optional[str]:
+    repository_url = os.getenv(env)
+    if repository_url and (
+        match := re.match(
+            r"(https?://[\w.-]+/)?(?P<full_name>[\w.-]+/[\w.-]+)/?$",
+            repository_url,
+        )
+    ):
+        return match.group("full_name")
 
     return None
 
 
 def get_repository_name() -> typing.Optional[str]:
     provider = get_ci_provider()
 
+    if provider == "jenkins":
+        return _get_repository_name_from_env_url("GIT_URL")
+
     if provider == "github_actions":
         return os.getenv("GITHUB_REPOSITORY")
 
     if provider == "circleci":
-        repository_url = os.getenv("CIRCLE_REPOSITORY_URL")
-        if repository_url and (
-            match := re.match(
-                r"(https?://[\w.-]+/)?(?P<full_name>[\w.-]+/[\w.-]+)/?$",
-                repository_url,
-            )
-        ):
-            return match.group("full_name")
+        return _get_repository_name_from_env_url("CIRCLE_REPOSITORY_URL")
 
     if provider == "pytest_mergify_suite":
         return "Mergifyio/pytest-mergify"

tests/test_resources.py

@@ -76,3 +76,39 @@ def test_span_github_actions(
         span.resource.attributes["cicd.pipeline.runner.name"] == "self-hosted"
         for span in spans.values()
     )
+
+
+def test_span_jenkins(
+    monkeypatch: pytest.MonkeyPatch,
+    pytester_with_spans: conftest.PytesterWithSpanT,
+) -> None:
+    monkeypatch.setenv("GITHUB_ACTIONS", "false")
+    monkeypatch.setenv("JENKINS_URL", "https://jenkins.example.com")
+    monkeypatch.setenv(
+        "BUILD_URL", "https://jenkins.example.com/Mergifyio/pytest-mergify"
+    )
+    monkeypatch.setenv("BUILD_ID", "jenkins-job-name#5")
+    monkeypatch.setenv("JOB_NAME", "jenkins-job-name")
+    monkeypatch.setenv("GIT_URL", "https://github.com/Mergifyio/pytest-mergify")
+    monkeypatch.setenv("GIT_BRANCH", "main")
+    monkeypatch.setenv("GIT_COMMIT", "1860cf377dd5610e256ff52e47cf38816cc04549")
+    monkeypatch.setenv("NODE_NAME", "self-hosted")
+    result, spans = pytester_with_spans()
+    assert spans is not None
+    assert all(
+        span.resource.attributes["vcs.repository.name"] == "Mergifyio/pytest-mergify"
+        for span in spans.values()
+    )
+    assert all(
+        span.resource.attributes["vcs.repository.url.full"]
+        == "https://github.com/Mergifyio/pytest-mergify"
+        for span in spans.values()
+    )
+    assert all(
+        span.resource.attributes["cicd.pipeline.run.id"] == "jenkins-job-name#5"
+        for span in spans.values()
+    )
+    assert all(
+        span.resource.attributes["cicd.pipeline.runner.name"] == "self-hosted"
+        for span in spans.values()
+    )