feat(someip): add server/provider side (offer services, answer RPC, publish events)

python/packages/jumpstarter-driver-someip/README.md

@@ -156,6 +156,44 @@ with env() as client:
     someip.close_connection()
 ```
 
+### Server / Provider (act as an ECU)
+
+The same driver can also *provide* SOME/IP services — offer them via Service
+Discovery, answer RPC requests with canned responses, and publish events. This
+turns a Jumpstarter exporter into a simulated ECU, useful for exercising a
+device-under-test that is a SOME/IP *client* of another ECU.
+
+RPC handlers run inside the exporter process, so responses are configured
+declaratively rather than via a per-request callback: set a canned response for
+a `(service_id, method_id)` and the server serves it. This maps naturally onto
+getter-style SOME/IP methods; update the response to change what a client reads.
+
+```python
+from jumpstarter.common.utils import env
+
+with env() as client:
+    someip = client.someip
+
+    # Offer a service instance (starts the server on first use)
+    someip.offer_service(0x1801, instance_id=0x0001, major_version=1)
+
+    # Answer an RPC method with a fixed payload (E_OK by default)
+    someip.set_method_response(0x1801, 0x0005, b"\x01\x02\x03\x04")
+    # ...or return an error return code
+    someip.set_method_response(0x1801, 0x0006, b"", return_code=0x01)
+
+    # Publish events to subscribers of an event group
+    someip.register_event(0x1801, 0x8001, eventgroup_id=1)
+    someip.publish_event(0x1801, 0x8001, b"\x2d\x00")
+    # Field events are cached and served to new subscribers
+    someip.set_field(0x1801, 0x8002, b"\x01")
+
+    # Introspect / tear down
+    print(someip.list_offered_services())
+    someip.stop_offer_service(0x1801, 0x0001)
+    someip.stop_server()
+```
+
 ## API Reference
 
 ```{eval-rst}

python/packages/jumpstarter-driver-someip/examples/exporter.yaml

@@ -32,3 +32,24 @@ export:
       transport_mode: UDP
       remote_host: "192.168.100.10"
       remote_port: 30490
+---
+# Provider / server mode - offer services, answer RPC, and publish events
+# (act as a simulated ECU that a device-under-test's SOME/IP client talks to).
+# Bind to the interface the DUT shares; services are offered at runtime via the
+# offer_service / set_method_response / publish_event client verbs.
+apiVersion: jumpstarter.dev/v1alpha1
+kind: ExporterConfig
+metadata:
+  namespace: default
+  name: someip-server-exporter
+endpoint: ""
+token: ""
+export:
+  someip:
+    type: jumpstarter_driver_someip.driver.SomeIp
+    config:
+      host: "192.168.100.1"
+      port: 30490
+      transport_mode: UDP
+      multicast_group: "239.127.0.1"
+      multicast_port: 30490

python/packages/jumpstarter-driver-someip/jumpstarter_driver_someip/client.py

@@ -5,6 +5,7 @@
 from .common import (
     SomeIpEventNotification,
     SomeIpMessageResponse,
+    SomeIpOfferedService,
     SomeIpPayload,
     SomeIpServiceEntry,
 )
@@ -93,3 +94,75 @@ def close_connection(self) -> None:
     def reconnect(self) -> None:
         """Reconnect to the SOME/IP endpoint."""
         self.call("reconnect")
+
+    # --- Server / provider side ---
+
+    def start_server(self) -> None:
+        """Force-start the SOME/IP server (otherwise started on first offer)."""
+        self.call("start_server")
+
+    def offer_service(
+        self,
+        service_id: int,
+        instance_id: int = 0x0001,
+        major_version: int = 1,
+        minor_version: int = 0,
+    ) -> None:
+        """Offer a service instance for discovery (act as the providing ECU)."""
+        self.call("offer_service", service_id, instance_id, major_version, minor_version)
+
+    def stop_offer_service(
+        self,
+        service_id: int,
+        instance_id: int = 0x0001,
+        major_version: int = 1,
+        minor_version: int = 0,
+    ) -> None:
+        """Withdraw a previously offered service instance."""
+        self.call("stop_offer_service", service_id, instance_id, major_version, minor_version)
+
+    def list_offered_services(self) -> list[SomeIpOfferedService]:
+        """Return the set of services this server currently offers."""
+        result = self.call("list_offered_services")
+        return [SomeIpOfferedService.model_validate(v) for v in result]
+
+    def set_method_response(
+        self,
+        service_id: int,
+        method_id: int,
+        payload: bytes,
+        return_code: int = 0,
+    ) -> None:
+        """Configure the canned response the server returns for an RPC method."""
+        msg = SomeIpPayload(data=payload.hex())
+        self.call("set_method_response", service_id, method_id, msg, return_code)
+
+    def clear_method_response(self, service_id: int, method_id: int) -> None:
+        """Remove a configured RPC response."""
+        self.call("clear_method_response", service_id, method_id)
+
+    def register_event(self, service_id: int, event_id: int, eventgroup_id: int) -> None:
+        """Register an event for publishing under an event group."""
+        self.call("register_event", service_id, event_id, eventgroup_id)
+
+    def publish_event(self, service_id: int, event_id: int, payload: bytes) -> None:
+        """Publish an event notification to subscribers of its event group.
+
+        ``service_id`` is accepted for API symmetry but currently unused;
+        events are addressed by ``event_id`` only.
+        """
+        msg = SomeIpPayload(data=payload.hex())
+        self.call("publish_event", service_id, event_id, msg)
+
+    def set_field(self, service_id: int, event_id: int, payload: bytes) -> None:
+        """Set a field event value (served to new subscribers and notified).
+
+        ``service_id`` is accepted for API symmetry but currently unused;
+        fields are addressed by ``event_id`` only.
+        """
+        msg = SomeIpPayload(data=payload.hex())
+        self.call("set_field", service_id, event_id, msg)
+
+    def stop_server(self) -> None:
+        """Stop the SOME/IP server, withdrawing all offers."""
+        self.call("stop_server")

python/packages/jumpstarter-driver-someip/jumpstarter_driver_someip/common.py

@@ -63,3 +63,12 @@ class SomeIpEventNotification(BaseModel):
     @classmethod
     def _validate_hex(cls, v: str) -> str:
         return _validate_hex_string(v)
+
+
+class SomeIpOfferedService(BaseModel):
+    """A service instance the server is currently offering (server-side introspection)."""
+
+    service_id: int = Field(ge=0, le=0xFFFF)
+    instance_id: int = Field(ge=0, le=0xFFFF)
+    major_version: int = Field(default=1, ge=0, le=0xFF)
+    minor_version: int = Field(default=0, ge=0, le=0xFFFFFFFF)

python/packages/jumpstarter-driver-someip/jumpstarter_driver_someip/conftest.py

@@ -18,6 +18,9 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from opensomeip.message import Message as OsipMessage
+from opensomeip.types import MessageId as OsipMessageId
+from opensomeip.types import RequestId as OsipRequestId
 
 # =========================================================================
 # Wire-protocol constants
@@ -388,6 +391,122 @@ def unregister_service(self, service_id: int, instance_id: int):
         ]
 
 
+# =========================================================================
+# StatefulOsipServer / LoopbackOsipClient - simulated-ECU loopback pair
+#
+# The server fake tracks offers, RPC handlers, and events like a real
+# opensomeip.SomeIpServer.  The loopback client dispatches RPC calls to the
+# server's registered handlers and discovers the server's offers, so tests
+# can exercise the full simulated-ECU workflow (offer + canned response +
+# client RPC) through the gRPC boundary with a single driver.
+# =========================================================================
+
+
+class StatefulOsipServer:
+    """A drop-in replacement for ``opensomeip.SomeIpServer`` that tracks
+    offers, RPC handlers, and events, and loops published events back to an
+    attached ``LoopbackOsipClient``.
+    """
+
+    def __init__(self, config=None) -> None:
+        self._started = False
+        self._config = config
+        self._offered: list = []
+        self.handlers: dict = {}
+        self.registered_events: dict = {}
+        self.fields: dict = {}
+        self._client: LoopbackOsipClient | None = None
+
+    def attach_client(self, client: LoopbackOsipClient) -> None:
+        self._client = client
+
+    def start(self):
+        self._started = True
+
+    def stop(self):
+        self._started = False
+        self._offered.clear()
+
+    def offer(self, service):
+        self._offered.append(service)
+
+    def stop_offer(self, service):
+        self._offered = [
+            s
+            for s in self._offered
+            if not (s.service_id == service.service_id and s.instance_id == service.instance_id)
+        ]
+
+    @property
+    def offered_services(self):
+        return list(self._offered)
+
+    def register_method(self, message_id, handler):
+        self.handlers[(message_id.service_id, message_id.method_id)] = handler
+
+    def register_event(self, event_id, eventgroup_id):
+        self.registered_events[event_id] = eventgroup_id
+
+    def publish_event(self, event_id, payload):
+        """Deliver the event to the attached client if it is subscribed."""
+        eventgroup_id = self.registered_events.get(event_id)
+        client = self._client
+        if client is not None and eventgroup_id in client._subscribed_eventgroups:
+            client.inject_event(0x0000, event_id, payload)
+
+    def set_field(self, event_id, payload):
+        self.fields[event_id] = payload
+        self.publish_event(event_id, payload)
+
+
+class LoopbackOsipClient(StatefulOsipClient):
+    """A ``StatefulOsipClient`` wired to a ``StatefulOsipServer``.
+
+    RPC calls dispatch to the server's registered method handlers (built as
+    real ``opensomeip`` request messages) and service discovery reflects the
+    server's current offers.
+    """
+
+    def __init__(self, server: StatefulOsipServer, config=None) -> None:
+        super().__init__(config)
+        self._server = server
+        # Discovery and RPC reflect the server side, not a static registry.
+        self._registered_services = []
+        self._rpc_responses = {}
+
+    def call(self, message_id, *, payload: bytes = b"", timeout: float = 5.0):
+        self._require_started()
+        sid = message_id.service_id
+        mid = message_id.method_id
+        self._rpc_history.append((sid, mid, payload))
+        handler = self._server.handlers.get((sid, mid))
+        if handler is None:
+            raise TimeoutError(f"no handler registered for service 0x{sid:04X} method 0x{mid:04X}")
+        request = OsipMessage(
+            message_id=OsipMessageId(sid, mid),
+            request_id=OsipRequestId(0x0001, len(self._rpc_history)),
+            payload=payload,
+        )
+        return handler(request)
+
+    def find(self, service, *, callback=None):
+        self._require_started()
+        for svc in self._server.offered_services:
+            if svc.service_id == service.service_id:
+                if service.instance_id == 0xFFFF or svc.instance_id == service.instance_id:
+                    if callback:
+                        callback(svc)
+
+
+@pytest.fixture
+def loopback_pair():
+    """Provide a wired (StatefulOsipServer, LoopbackOsipClient) pair."""
+    server = StatefulOsipServer()
+    client = LoopbackOsipClient(server)
+    server.attach_client(client)
+    return server, client
+
+
 @pytest.fixture(autouse=True)
 def _mock_get_ext():
     """Ensure the native-extension guard passes in all tests by default."""