Teach our integrations utils about http tests

And add a test for http otlp exporter user-agent. The http server
requires to bind to a random port for each test because as it won't
unbind it until process exit.
And SO_REUSEPORT does not seem to help either.

Author: xrmx

tests/integration/test_integration.py

@@ -23,11 +23,16 @@
 )
 
 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()
@@ -184,7 +189,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

@@ -16,10 +16,12 @@
 
 import base64
 import inspect
+import random
 import subprocess
 import os
 import tempfile
 import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
 from typing import Callable, Mapping, Optional
 
@@ -33,7 +35,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 +44,7 @@ class ElasticIntegrationTestCase(unittest.TestCase):
 
     A basic TestCase would look like:
 
-    class MyTestCase(ElasticIntegrationTestCase):
+    class MyTestCase(ElasticIntegrationGRPCTestCase):
         @classmethod
         def requirements(cls):
             requirements = super().requirements()
@@ -274,3 +276,101 @@ def run_script(
                     ex.stderr.decode() if ex.stderr else None,
                     proc.returncode,
                 )
+
+
+class HttpSink(ot.HttpSink):
+    """Backport of 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):
+        if self.httpd:
+            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.
+    """
+
+    def set_http_port(self):
+        self._http_port = int(random.uniform(44318, 45000))
+        return self._http_port
+
+    def get_http_port(self):
+        return self._http_port
+
+    def setUp(self):
+        self.handler = ot.AccumulatingHandler()
+        # pick a new port each time because otherwise the next test will fail to bind to the same port, even with SO_REUSEPORT :(
+        port = self.set_http_port()
+        self.sink = HttpSink(self.handler, port=port)
+        self.sink.start()
+
+    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:
+            otlp_endpoint = f"http://localhost:{self.get_http_port()}"
+            environment_variables = {
+                "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
+                "OTEL_EXPORTER_OTLP_ENDPOINT": otlp_endpoint,
+            }
+        return super().run_script(script_func, environment_variables, wrapper_script, on_start)