test(spyne): Added initial tests for spyne

Signed-off-by: Varsha GS <[email protected]>

Author: GSVarsha

tests/apps/spyne_app/__init__.py

@@ -0,0 +1,10 @@
+# (c) Copyright IBM Corp. 2025
+
+import os
+from tests.apps.spyne_app.app import spyne_server as server
+from tests.apps.utils import launch_background_thread
+
+app_thread = None
+
+if not os.environ.get('CASSANDRA_TEST') and app_thread is None:
+    app_thread = launch_background_thread(server.serve_forever, "Spyne")

tests/apps/spyne_app/app.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# (c) Copyright IBM Corp. 2025
+
+import logging
+
+from wsgiref.simple_server import make_server
+from spyne import Application, rpc, ServiceBase, Iterable, UnsignedInteger, \
+    String, Unicode, M, UnsignedInteger32
+
+from spyne.protocol.json import JsonDocument
+from spyne.protocol.http import HttpRpc
+from spyne.server.wsgi import WsgiApplication
+
+from spyne.error import ResourceNotFoundError
+
+from tests.helpers import testenv
+
+logging.basicConfig(level=logging.WARNING)
+logger = logging.getLogger(__name__)
+
+testenv["spyne_port"] = 10818
+testenv["spyne_server"] = ("http://127.0.0.1:" + str(testenv["spyne_port"]))
+
+class HelloWorldService(ServiceBase):
+    # @rpc(Unicode, _returns=Unicode)
+    # def get_resource(self, resource_id):
+    #     # Simulate checking for a resource
+    #     if resource_id != "existing_resource":
+    #         raise ResourceNotFoundError(
+    #             "Resource not found",
+    #             "The requested resource does not exist."
+    #         )
+    #     return f"Resource {resource_id} found."
+
+    @rpc(String, UnsignedInteger, _returns=Iterable(String))
+    def say_hello(ctx, name, times):
+        """
+        Docstrings for service methods do appear as documentation in the
+        interface documents. <b>What fun!</b>
+
+        :param name: The name to say hello to
+        :param times: The number of times to say hello
+
+        :returns: An array of 'Hello, <name>' strings, repeated <times> times.
+        """
+
+        for i in range(times):
+            yield 'Hello, %s' % name
+
+    @rpc(_returns=Unicode)
+    def hello(self):
+        return "<center><h1>🐍 Hello Stan! 🦄</h1></center>"
+    
+    @rpc(M(UnsignedInteger32))
+    def del_user(ctx, user_id):
+        raise ResourceNotFoundError(user_id)
+
+
+application = Application([HelloWorldService], 'spyne.examples.hello.http',
+    in_protocol=HttpRpc(validator='soft'),
+    out_protocol=JsonDocument(ignore_wrappers=True),
+)
+wsgi_app = WsgiApplication(application)
+spyne_server = make_server('127.0.0.1', testenv["spyne_port"], wsgi_app)
+
+if __name__ == '__main__':
+    # logging.info("listening to http://127.0.0.1:8000")
+    # logging.info("wsdl is at: http://localhost:8000/?wsdl")
+    spyne_server.request_queue_size = 20
+    spyne_server.serve_forever()

tests/frameworks/test_spyne.py

@@ -0,0 +1,149 @@
+# (c) Copyright IBM Corp. 2025
+
+import time
+import urllib3
+import pytest
+from typing import Generator
+
+from tests.apps import spyne_app
+from tests.helpers import testenv
+from instana.singletons import agent, tracer
+from instana.span.span import get_current_span
+from instana.util.ids import hex_id
+
+
+class TestSpyne:
+    @pytest.fixture(autouse=True)
+    def _resource(self) -> Generator[None, None, None]:
+        """Clear all spans before a test run"""
+        self.http = urllib3.PoolManager()
+        self.recorder = tracer.span_processor
+        self.recorder.clear_spans()
+        time.sleep(0.1)
+
+    def test_vanilla_requests(self) -> None:
+        response = self.http.request("GET", testenv["spyne_server"] + "/hello")
+        spans = self.recorder.queued_spans()
+
+        assert len(spans) == 1
+        assert get_current_span().is_recording() is False
+        assert response.status == 200
+
+    def test_get_request(self) -> None:
+        with tracer.start_as_current_span("test"):
+            response = self.http.request("GET", testenv["spyne_server"] + "/hello")
+
+        spans = self.recorder.queued_spans()
+
+        assert len(spans) == 3
+        assert get_current_span().is_recording() is False
+
+        spyne_span = spans[0]
+        urllib3_span = spans[1]
+        test_span = spans[2]
+
+        assert response
+        assert 200 == response.status
+
+        assert "X-INSTANA-T" in response.headers
+        assert int(response.headers["X-INSTANA-T"], 16)
+        assert response.headers["X-INSTANA-T"] == hex_id(spyne_span.t)
+
+        assert "X-INSTANA-S" in response.headers
+        assert int(response.headers["X-INSTANA-S"], 16)
+        assert response.headers["X-INSTANA-S"] == hex_id(spyne_span.s)
+
+        assert "X-INSTANA-L" in response.headers
+        assert response.headers["X-INSTANA-L"] == "1"
+
+        assert "Server-Timing" in response.headers
+        server_timing_value = f"intid;desc={hex_id(spyne_span.t)}"
+        assert response.headers["Server-Timing"] == server_timing_value
+
+        # Same traceId
+        assert test_span.t == urllib3_span.t
+        assert urllib3_span.t == spyne_span.t
+
+        # Parent relationships
+        assert urllib3_span.p == test_span.s
+        assert spyne_span.p == urllib3_span.s
+
+        assert spyne_span.sy is None
+        assert urllib3_span.sy is None
+        assert test_span.sy is None
+
+        # Error logging
+        assert test_span.ec is None
+        assert urllib3_span.ec is None
+        assert spyne_span.ec is None
+
+        # wsgi
+        assert "spyne" == spyne_span.n
+        assert (
+            "127.0.0.1:" + str(testenv["spyne_port"]) == spyne_span.data["http"]["host"]
+        )
+        assert "/hello" == spyne_span.data["http"]["url"]
+        assert "GET" == spyne_span.data["http"]["method"]
+        assert 200 == spyne_span.data["http"]["status"]
+        assert spyne_span.data["http"]["error"] is None
+        assert spyne_span.stack is None
+
+    def test_secret_scrubbing(self) -> None:
+        with tracer.start_as_current_span("test"):
+            response = self.http.request("GET", testenv["spyne_server"] + "/say_hello?name=World&times=4&secret=sshhh")
+
+        spans = self.recorder.queued_spans()
+
+        assert len(spans) == 3
+        assert get_current_span().is_recording() is False
+
+        spyne_span = spans[0]
+        urllib3_span = spans[1]
+        test_span = spans[2]
+
+        assert response
+        assert 200 == response.status
+
+        assert "X-INSTANA-T" in response.headers
+        assert int(response.headers["X-INSTANA-T"], 16)
+        assert response.headers["X-INSTANA-T"] == hex_id(spyne_span.t)
+
+        assert "X-INSTANA-S" in response.headers
+        assert int(response.headers["X-INSTANA-S"], 16)
+        assert response.headers["X-INSTANA-S"] == hex_id(spyne_span.s)
+
+        assert "X-INSTANA-L" in response.headers
+        assert response.headers["X-INSTANA-L"] == "1"
+
+        assert "Server-Timing" in response.headers
+        server_timing_value = f"intid;desc={hex_id(spyne_span.t)}"
+        assert response.headers["Server-Timing"] == server_timing_value
+
+        # Same traceId
+        assert test_span.t == urllib3_span.t
+        assert urllib3_span.t == spyne_span.t
+
+        # Parent relationships
+        assert urllib3_span.p == test_span.s
+        assert spyne_span.p == urllib3_span.s
+
+        assert spyne_span.sy is None
+        assert urllib3_span.sy is None
+        assert test_span.sy is None
+
+        # Error logging
+        assert test_span.ec is None
+        assert urllib3_span.ec is None
+        assert spyne_span.ec is None
+
+        # wsgi
+        assert "spyne" == spyne_span.n
+        assert (
+            "127.0.0.1:" + str(testenv["spyne_port"]) == spyne_span.data["http"]["host"]
+        )
+        assert "/say_hello" == spyne_span.data["http"]["url"]
+        assert spyne_span.data["http"]["params"] == "name=World&times=4&secret=<redacted>"
+        assert "GET" == spyne_span.data["http"]["method"]
+        assert 200 == spyne_span.data["http"]["status"]
+        assert spyne_span.data["http"]["error"] is None
+        assert spyne_span.stack is None