Teach our integrations utils about http tests

And add a test for http otlp exporter user-agent.

Author: xrmx

tests/integration/test_integration.py

@@ -19,11 +19,16 @@
 import pytest
 
 from elasticotel.distro import version
-from .utils import ElasticIntegrationTestCase, OTEL_INSTRUMENTATION_VERSION, ROOT_DIR
+from .utils import (
+    ElasticIntegrationGRPCTestCase,
+    ElasticIntegrationHTTPTestCase,
+    OTEL_INSTRUMENTATION_VERSION,
+    ROOT_DIR,
+)
 
 
 @pytest.mark.integration
-class IntegrationTestCase(ElasticIntegrationTestCase):
+class GRPCIntegrationTestCase(ElasticIntegrationGRPCTestCase):
     @classmethod
     def requirements(cls):
         requirements = super().requirements()
@@ -176,7 +181,47 @@ def test_script():
 
 
 @pytest.mark.integration
-class OperatorTestCase(ElasticIntegrationTestCase):
+class HTTPIntegrationTestCase(ElasticIntegrationHTTPTestCase):
+    @classmethod
+    def requirements(cls):
+        requirements = super().requirements()
+        return requirements + [f"opentelemetry-instrumentation-sqlite3=={OTEL_INSTRUMENTATION_VERSION}"]
+
+    def test_edot_user_agent_is_used_in_otlp_http_exporter(self):
+        def test_script():
+            import sqlite3
+
+            from opentelemetry._events import Event, get_event_logger
+
+            connection = sqlite3.connect(":memory:")
+            cursor = connection.cursor()
+            cursor.execute("CREATE TABLE movie(title, year, score)")
+
+            event = Event(name="test.event", attributes={}, body={"key": "value"})
+            event_logger = get_event_logger(__name__)
+            event_logger.emit(event)
+
+        stdout, stderr, returncode = self.run_script(test_script, wrapper_script="opentelemetry-instrument")
+
+        telemetry = self.get_telemetry()
+        (metrics_headers, logs_headers, traces_headers) = (
+            telemetry["metrics_headers"],
+            telemetry["logs_headers"],
+            telemetry["traces_headers"],
+        )
+
+        assert metrics_headers
+        assert traces_headers
+        assert logs_headers
+
+        edot_user_agent = "elastic-otlp-http-python/" + version.__version__
+        self.assertIn(edot_user_agent, metrics_headers[0]["User-Agent"])
+        self.assertIn(edot_user_agent, traces_headers[0]["User-Agent"])
+        self.assertIn(edot_user_agent, logs_headers[0]["User-Agent"])
+
+
+@pytest.mark.integration
+class OperatorTestCase(ElasticIntegrationHTTPTestCase):
     @staticmethod
     def _read_operator_requirements():
         requirements = []

tests/integration/utils.py

@@ -20,6 +20,7 @@
 import os
 import tempfile
 import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
 from typing import Callable, Mapping, Optional
 
@@ -33,7 +34,7 @@
 ROOT_DIR = os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
 
 
-class ElasticIntegrationTestCase(unittest.TestCase):
+class ElasticIntegrationGRPCTestCase(unittest.TestCase):
     """This is an experimental reimplementation of OtelTest using unittest
 
     The idea is to do integration testing by creating a separate virtualenv for each TestCase inheriting
@@ -42,7 +43,7 @@ class ElasticIntegrationTestCase(unittest.TestCase):
 
     A basic TestCase would look like:
 
-    class MyTestCase(ElasticIntegrationTestCase):
+    class MyTestCase(ElasticIntegrationGRPCTestCase):
         @classmethod
         def requirements(cls):
             requirements = super().requirements()
@@ -65,6 +66,8 @@ def test_one_span_generated(self):
     Each TestCase costs around 10 seconds for settings up the virtualenv so split tests accordingly.
     """
 
+    sink_class = ot.GrpcSink
+
     @classmethod
     def requirements(cls) -> list:
         """Installs requirements in a virtualenv
@@ -96,7 +99,7 @@ def tearDownClass(cls):
 
     def setUp(self):
         self.handler = ot.AccumulatingHandler()
-        self.sink = ot.GrpcSink(self.handler)
+        self.sink = self.sink_class(self.handler)
         self.sink.start()
 
     def tearDown(self):
@@ -274,3 +277,84 @@ def run_script(
                     ex.stderr.decode() if ex.stderr else None,
                     proc.returncode,
                 )
+
+
+class HttpSink(ot.HttpSink):
+    """Backports teardown fixes, drop if we ever rebase on top of oteltest 0.37.0+"""
+
+    # This code is copyright Pablo Collins under Apache-2.0
+
+    def __init__(self, listener, port=4318, daemon=True):
+        self.httpd = None
+        super().__init__(listener, port, daemon)
+
+    def run_server(self):
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(this):
+                # /v1/traces
+                content_length = int(this.headers["Content-Length"])
+                post_data = this.rfile.read(content_length)
+
+                otlp_handler_func = self.handlers.get(this.path)
+                if otlp_handler_func:
+                    otlp_handler_func(post_data, {k: v for k, v in this.headers.items()})
+
+                this.send_response(200)
+                this.send_header("Content-type", "text/html")
+                this.end_headers()
+
+                this.wfile.write("OK".encode("utf-8"))
+
+        self.httpd = HTTPServer(("", self.port), Handler)
+        self.httpd.serve_forever()
+
+    def stop(self):
+        self.httpd.shutdown()
+        super().stop()
+
+
+class ElasticIntegrationHTTPTestCase(ElasticIntegrationGRPCTestCase):
+    """This is an experimental reimplementation of OtelTest using unittest
+
+    The idea is to do integration testing by creating a separate virtualenv for each TestCase inheriting
+    from ElasticIntegrationTestCase, run a script, collect OTLP http/protobuf calls and make the received data
+    available in order to add assertions.
+
+    A basic TestCase would look like:
+
+    class MyTestCase(ElasticIntegrationHTTPTestCase):
+        @classmethod
+        def requirements(cls):
+            requirements = super().requirements()
+            return requirements + [f"my-library"]
+
+        def script(self):
+            import sqlite3
+
+            connection = sqlite3.connect(":memory:")
+            cursor = connection.cursor()
+            cursor.execute("CREATE TABLE movie(title, year, score)")
+
+        def test_one_span_generated(self):
+            stdout, stderr, returncode = self.run_script(self.script, wrapper_script="opentelemetry-instrument")
+
+            telemetry.get_telemetry()
+            assert len(telemetry["traces"], 1)
+
+
+    Each TestCase costs around 10 seconds for settings up the virtualenv so split tests accordingly.
+    """
+
+    sink_class = HttpSink
+
+    def run_script(
+        self,
+        script_func: Callable[[], None],
+        environment_variables: Optional[Mapping[str, str]] = None,
+        wrapper_script: Optional[str] = None,
+        on_start: Optional[Callable[[], Optional[float]]] = None,
+    ):
+        # Add a sane default so that callers don't need to remember to set that
+        if environment_variables is None:
+            environment_variables = {"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"}
+        return super().run_script(script_func, environment_variables, wrapper_script, on_start)