refactor: split tracer setup code in its own part (#25)

This will allow to replace the MergifyTracer dynamically in tests for
easier scenario testing.

pytest_mergify/__init__.py

@@ -1,4 +1,4 @@
-import os
+import typing
 
 import pytest
 import _pytest.main
@@ -7,128 +7,57 @@
 import _pytest.nodes
 import _pytest.terminal
 
-from opentelemetry import context
-import opentelemetry.sdk.trace
-from opentelemetry.sdk.trace import export
-from opentelemetry.sdk.trace import TracerProvider, SpanProcessor, Span
-from opentelemetry.exporter.otlp.proto.http import Compression
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
-    OTLPSpanExporter,
-)
-import opentelemetry.sdk.resources
-
 from pytest_mergify import utils
-import pytest_mergify.resources.ci as resources_ci
-import pytest_mergify.resources.github_actions as resources_gha
-
-import pytest_opentelemetry.instrumentation
-
-
-class InterceptingSpanProcessor(SpanProcessor):
-    trace_id: None | int
-
-    def __init__(self) -> None:
-        self.trace_id = None
-
-    def on_start(
-        self, span: Span, parent_context: context.Context | None = None
-    ) -> None:
-        if span.attributes is not None and any(
-            "pytest" in attr for attr in span.attributes
-        ):
-            self.trace_id = span.context.trace_id
+from pytest_mergify.tracer import MergifyTracer
 
 
 class PytestMergify:
     __name__ = "PytestMergify"
 
-    exporter: export.SpanExporter
-    repo_name: str | None
-
-    def ci_supports_trace_interception(self) -> bool:
-        return utils.get_ci_provider() == "github_actions"
+    mergify_tracer: MergifyTracer
 
     # Do this after pytest-opentelemetry has setup things
     @pytest.hookimpl(trylast=True)
     def pytest_configure(self, config: _pytest.config.Config) -> None:
-        self.token = os.environ.get("MERGIFY_TOKEN")
-        self.repo_name = utils.get_repository_name()
-
-        span_processor: opentelemetry.sdk.trace.SpanProcessor
-        if os.environ.get("PYTEST_MERGIFY_DEBUG"):
-            self.exporter = export.ConsoleSpanExporter()
-            span_processor = export.SimpleSpanProcessor(self.exporter)
-        elif utils.strtobool(os.environ.get("_PYTEST_MERGIFY_TEST", "false")):
-            from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
-                InMemorySpanExporter,
-            )
-
-            self.exporter = InMemorySpanExporter()
-            span_processor = export.SimpleSpanProcessor(self.exporter)
-        elif self.token:
-            url = config.getoption("--mergify-api-url") or os.environ.get(
-                "MERGIFY_API_URL", "https://api.mergify.com"
-            )
-            if self.repo_name is None:
-                return
-
-            self.exporter = OTLPSpanExporter(
-                endpoint=f"{url}/v1/repos/{self.repo_name}/ci/traces",
-                headers={"Authorization": f"Bearer {self.token}"},
-                compression=Compression.Gzip,
-            )
-            span_processor = export.BatchSpanProcessor(self.exporter)
+        api_url = config.getoption("--mergify-api-url")
+        if api_url is None:
+            self.reconfigure()
         else:
-            return
-
-        resources_gha.GitHubActionsResourceDetector().detect()
-        resource = opentelemetry.sdk.resources.get_aggregated_resources(
-            [
-                resources_ci.CIResourceDetector(),
-                resources_gha.GitHubActionsResourceDetector(),
-            ]
-        )
-
-        tracer_provider = TracerProvider(resource=resource)
-
-        tracer_provider.add_span_processor(span_processor)
-
-        if self.ci_supports_trace_interception():
-            self.interceptor = InterceptingSpanProcessor()
-            tracer_provider.add_span_processor(self.interceptor)
+            self.reconfigure(api_url=api_url)
 
-        self.tracer = tracer_provider.get_tracer("pytest-mergify")
-        # Replace tracer of pytest-opentelemetry
-        pytest_opentelemetry.instrumentation.tracer = self.tracer
+    def reconfigure(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.mergify_tracer = MergifyTracer(*args, **kwargs)
 
     def pytest_terminal_summary(
         self, terminalreporter: _pytest.terminal.TerminalReporter
     ) -> None:
         terminalreporter.section("Mergify CI")
 
-        if self.token is None:
+        if self.mergify_tracer.token is None:
             terminalreporter.write_line(
                 "No token configured for Mergify; test results will not be uploaded",
                 yellow=True,
             )
             return
 
-        if self.interceptor.trace_id is None:
-            terminalreporter.write_line(
-                "No trace id detected, this test run will not be attached to the CI job",
-                yellow=True,
-            )
-        elif utils.get_ci_provider() == "github_actions":
-            terminalreporter.write_line(
-                f"::notice title=Mergify CI::MERGIFY_TRACE_ID={self.interceptor.trace_id}",
-            )
+        if self.mergify_tracer.interceptor is None:
+            terminalreporter.write_line("Nothing to do")
+        else:
+            if self.mergify_tracer.interceptor.trace_id is None:
+                terminalreporter.write_line(
+                    "No trace id detected, this test run will not be attached to the CI job",
+                    yellow=True,
+                )
+            elif utils.get_ci_provider() == "github_actions":
+                terminalreporter.write_line(
+                    f"::notice title=Mergify CI::MERGIFY_TRACE_ID={self.mergify_tracer.interceptor.trace_id}",
+                )
 
 
 def pytest_addoption(parser: _pytest.config.argparsing.Parser) -> None:
     group = parser.getgroup("pytest-mergify", "Mergify support for pytest")
     group.addoption(
         "--mergify-api-url",
-        default=None,
         help=(
             "URL of the Mergify API "
             "(or set via MERGIFY_API_URL environment variable)",

pytest_mergify/tracer.py

@@ -0,0 +1,103 @@
+import dataclasses
+import os
+
+import opentelemetry.sdk.resources
+from opentelemetry.sdk.trace import export
+from opentelemetry import context
+from opentelemetry.sdk.trace import TracerProvider, SpanProcessor, Span
+from opentelemetry.exporter.otlp.proto.http import Compression
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
+    OTLPSpanExporter,
+)
+
+from pytest_mergify import utils
+
+import pytest_opentelemetry.instrumentation
+import pytest_mergify.resources.ci as resources_ci
+import pytest_mergify.resources.github_actions as resources_gha
+
+
+class InterceptingSpanProcessor(SpanProcessor):
+    trace_id: None | int
+
+    def __init__(self) -> None:
+        self.trace_id = None
+
+    def on_start(
+        self, span: Span, parent_context: context.Context | None = None
+    ) -> None:
+        if span.attributes is not None and any(
+            "pytest" in attr for attr in span.attributes
+        ):
+            self.trace_id = span.context.trace_id
+
+
+@dataclasses.dataclass
+class MergifyTracer:
+    token: str | None = dataclasses.field(
+        default_factory=lambda: os.environ.get("MERGIFY_TOKEN")
+    )
+    repo_name: str | None = dataclasses.field(default_factory=utils.get_repository_name)
+    interceptor: InterceptingSpanProcessor | None = None
+    api_url: str = dataclasses.field(
+        default_factory=lambda: os.environ.get(
+            "MERGIFY_API_URL", "https://api.mergify.com"
+        )
+    )
+    exporter: export.SpanExporter | None = dataclasses.field(init=False, default=None)
+    tracer: opentelemetry.trace.Tracer | None = dataclasses.field(
+        init=False, default=None
+    )
+    tracer_provider: opentelemetry.sdk.trace.TracerProvider | None = dataclasses.field(
+        init=False, default=None
+    )
+
+    def __post_init__(self) -> None:
+        span_processor: SpanProcessor
+
+        if os.environ.get("PYTEST_MERGIFY_DEBUG"):
+            self.exporter = export.ConsoleSpanExporter()
+            span_processor = export.SimpleSpanProcessor(self.exporter)
+        elif utils.strtobool(os.environ.get("_PYTEST_MERGIFY_TEST", "false")):
+            from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+                InMemorySpanExporter,
+            )
+
+            self.exporter = InMemorySpanExporter()
+            span_processor = export.SimpleSpanProcessor(self.exporter)
+        elif self.token:
+            if self.repo_name is None:
+                return
+
+            self.exporter = OTLPSpanExporter(
+                endpoint=f"{self.api_url}/v1/repos/{self.repo_name}/ci/traces",
+                headers={"Authorization": f"Bearer {self.token}"},
+                compression=Compression.Gzip,
+            )
+            span_processor = export.BatchSpanProcessor(self.exporter)
+        else:
+            return
+
+        resources_gha.GitHubActionsResourceDetector().detect()
+        resource = opentelemetry.sdk.resources.get_aggregated_resources(
+            [
+                resources_ci.CIResourceDetector(),
+                resources_gha.GitHubActionsResourceDetector(),
+            ]
+        )
+
+        self.tracer_provider = TracerProvider(resource=resource)
+
+        self.tracer_provider.add_span_processor(span_processor)
+
+        if self.ci_supports_trace_interception():
+            self.interceptor = InterceptingSpanProcessor()
+            self.tracer_provider.add_span_processor(self.interceptor)
+
+        self.tracer = self.tracer_provider.get_tracer("pytest-mergify")
+
+        # Replace tracer of pytest-opentelemetry
+        pytest_opentelemetry.instrumentation.tracer = self.tracer
+
+    def ci_supports_trace_interception(self) -> bool:
+        return utils.get_ci_provider() == "github_actions"

tests/conftest.py

@@ -1,8 +1,46 @@
-import os
+import collections.abc
+import typing
 
+import pytest
 
-# Set this before we call any part of our plugin
-def pytest_cmdline_main() -> None:
-    os.environ["CI"] = "1"
-    os.environ["_PYTEST_MERGIFY_TEST"] = "1"
-    os.environ["MERGIFY_API_URL"] = "https://localhost/v1/ci/traces"
+import _pytest.config
+
+from pytest_mergify import tracer
+
+pytest_plugins = ["pytester"]
+
+
+ReconfigureT = typing.Callable[[dict[str, str]], None]
+
+
+@pytest.fixture
+def reconfigure_mergify_tracer(
+    pytestconfig: _pytest.config.Config,
+    monkeypatch: pytest.MonkeyPatch,
+) -> collections.abc.Generator[ReconfigureT, None, None]:
+    # Always override API
+    monkeypatch.setenv("MERGIFY_API_URL", "http://localhost:9999")
+
+    plugin = pytestconfig.pluginmanager.get_plugin("PytestMergify")
+    assert plugin is not None
+    old_tracer: tracer.MergifyTracer = plugin.mergify_tracer
+
+    def _reconfigure(env: dict[str, str]) -> None:
+        # Set environment variables
+        for key, value in env.items():
+            monkeypatch.setenv(key, value)
+        plugin.reconfigure()
+
+    yield _reconfigure
+    if plugin.mergify_tracer.tracer_provider is not None:
+        plugin.mergify_tracer.tracer_provider.shutdown()
+    plugin.mergify_tracer = old_tracer
+
+
+@pytest.fixture
+def reconfigure_mergify_tracer_gha(
+    reconfigure_mergify_tracer: ReconfigureT,
+) -> None:
+    reconfigure_mergify_tracer(
+        {"GITHUB_ACTIONS": "true", "GITHUB_REPOSITORY": "Mergifyio/pytest-mergify"}
+    )

tests/test_plugin.py

@@ -5,7 +5,7 @@
 
 import pytest_mergify
 
-pytest_plugins = ["pytester"]
+from tests import conftest
 
 
 def test_plugin_is_loaded(pytestconfig: _pytest.config.Config) -> None:
@@ -31,9 +31,17 @@ def test_foo():
     )
 
 
-def test_with_token_gha(pytester: Pytester, monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("MERGIFY_TOKEN", "foobar")
-    monkeypatch.setenv("GITHUB_ACTIONS", "true")
+def test_with_token_gha(
+    pytester: Pytester, reconfigure_mergify_tracer: conftest.ReconfigureT
+) -> None:
+    reconfigure_mergify_tracer(
+        {
+            "CI": "1",
+            "GITHUB_REPOSITORY": "Mergifyio/pytest-mergify",
+            "MERGIFY_TOKEN": "foobar",
+            "GITHUB_ACTIONS": "true",
+        },
+    )
     pytester.makepyfile(
         """
         def test_foo():
@@ -51,7 +59,28 @@ def test_foo():
         pytest.fail("No trace id found")
 
 
-def test_repo_name(pytestconfig: _pytest.config.Config) -> None:
+def test_repo_name_github_actions(
+    pytestconfig: _pytest.config.Config,
+    reconfigure_mergify_tracer_gha: None,
+) -> None:
     plugin = pytestconfig.pluginmanager.get_plugin("PytestMergify")
     assert plugin is not None
-    assert plugin.repo_name == "Mergifyio/pytest-mergify"
+    assert plugin.mergify_tracer.repo_name == "Mergifyio/pytest-mergify"
+
+
+def test_with_token_no_ci_provider(
+    pytester: Pytester,
+    reconfigure_mergify_tracer: conftest.ReconfigureT,
+) -> None:
+    reconfigure_mergify_tracer(
+        {"MERGIFY_TOKEN": "x", "CI": "1", "GITHUB_ACTIONS": "false"}
+    )
+    pytester.makepyfile(
+        """
+        def test_foo():
+            assert True
+        """
+    )
+    result = pytester.runpytest_subprocess()
+    result.assert_outcomes(passed=1)
+    assert "Nothing to do" in result.stdout.lines