refactor: make it possible to test span emission (#16)

pytest_mergify/__init__.py

@@ -1,25 +1,25 @@
 import os
-import typing
 
+import pytest
 import _pytest.main
 import _pytest.config
 import _pytest.config.argparsing
 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 pytest_opentelemetry.instrumentation
+from pytest_mergify import utils
 
 
-CIProviderT = typing.Literal["github_actions", "circleci"]
+import pytest_opentelemetry.instrumentation
 
 
 class InterceptingSpanProcessor(SpanProcessor):
@@ -37,43 +37,45 @@ def on_start(
             self.trace_id = span.context.trace_id
 
 
-def is_running_in_ci() -> bool:
-    return os.environ.get("CI") is not None
-
-
 class PytestMergify:
-    @staticmethod
-    def get_ci_provider() -> CIProviderT | None:
-        if os.getenv("GITHUB_ACTIONS") == "true":
-            return "github_actions"
-        if os.getenv("CIRCLECI") == "true":
-            return "circleci"
-        return None
+    __name__ = "PytestMergify"
+
+    exporter: export.SpanExporter
 
     def ci_supports_trace_interception(self) -> bool:
-        return self.get_ci_provider() == "github_actions"
+        return utils.get_ci_provider() == "github_actions"
 
+    # 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")
 
-        exporter: export.SpanExporter
+        span_processor: opentelemetry.sdk.trace.SpanProcessor
         if os.environ.get("PYTEST_MERGIFY_DEBUG"):
-            exporter = export.ConsoleSpanExporter()
+            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"
             )
-            exporter = OTLPSpanExporter(
+            self.exporter = OTLPSpanExporter(
                 endpoint=f"{url}/v1/ci/traces",
                 headers={"Authorization": f"Bearer {self.token}"},
                 compression=Compression.Gzip,
             )
+            span_processor = export.BatchSpanProcessor(self.exporter)
         else:
             return
 
         tracer_provider = TracerProvider()
 
-        span_processor = export.BatchSpanProcessor(exporter)
         tracer_provider.add_span_processor(span_processor)
 
         if self.ci_supports_trace_interception():
@@ -101,7 +103,7 @@ def pytest_terminal_summary(
                 "No trace id detected, this test run will not be attached to the CI job",
                 yellow=True,
             )
-        elif self.get_ci_provider() == "github_actions":
+        elif utils.get_ci_provider() == "github_actions":
             terminalreporter.write_line(
                 f"::notice title=Mergify CI::MERGIFY_TRACE_ID={self.interceptor.trace_id}",
             )
@@ -120,5 +122,4 @@ def pytest_addoption(parser: _pytest.config.argparsing.Parser) -> None:
 
 
 def pytest_configure(config: _pytest.config.Config) -> None:
-    if is_running_in_ci():
-        config.pluginmanager.register(PytestMergify())
+    config.pluginmanager.register(PytestMergify())

pytest_mergify/utils.py

@@ -0,0 +1,22 @@
+import os
+import typing
+
+CIProviderT = typing.Literal["github_actions", "circleci"]
+
+
+def get_ci_provider() -> CIProviderT | None:
+    if os.getenv("GITHUB_ACTIONS") == "true":
+        return "github_actions"
+    if os.getenv("CIRCLECI") == "true":
+        return "circleci"
+    return None
+
+
+def strtobool(string: str) -> bool:
+    if string.lower() in {"y", "yes", "t", "true", "on", "1"}:
+        return True
+
+    if string.lower() in {"n", "no", "f", "false", "off", "0"}:
+        return False
+
+    raise ValueError(f"Could not convert '{string}' to boolean")

tests/conftest.py

@@ -1,8 +1,7 @@
-import pytest
+import os
 
-pytest_plugins = ["pytester"]
 
-
-@pytest.fixture(autouse=True)
-def set_ci(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("CI", "1")
+# 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"

tests/test_plugin.py

@@ -5,23 +5,15 @@
 
 import pytest_mergify
 
+pytest_plugins = ["pytester"]
+
 
 def test_plugin_is_loaded(pytestconfig: _pytest.config.Config) -> None:
     plugin = pytestconfig.pluginmanager.get_plugin("pytest_mergify")
     assert plugin is pytest_mergify
 
-
-def test_no_ci(pytester: Pytester, monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.delenv("CI")
-    pytester.makepyfile(
-        """
-        def test_foo():
-            assert True
-        """
-    )
-    result = pytester.runpytest()
-    result.assert_outcomes(passed=1)
-    assert not any("Mergify CI" in line for line in result.stdout.lines)
+    plugin = pytestconfig.pluginmanager.get_plugin("PytestMergify")
+    assert isinstance(plugin, pytest_mergify.PytestMergify)
 
 
 def test_no_token(pytester: Pytester) -> None:

tests/test_spans.py

@@ -0,0 +1,12 @@
+import _pytest.pytester
+import _pytest.config
+
+
+def test_span(
+    pytestconfig: _pytest.config.Config,
+) -> None:
+    plugin = pytestconfig.pluginmanager.get_plugin("PytestMergify")
+    assert plugin is not None
+    assert plugin.exporter is not None
+    spans = plugin.exporter.get_finished_spans()
+    assert any(s.name == "pytestconfig setup" for s in spans)