Repository: ansys/pypim (language: python)

Issue:
Connecting to a grpc service
A simple code such as:

```py
channel = instance.build_grpc_channel()
```

Should build a usable channel, taking in account the instance spec (address, headers)


Discussion / hints:

Fix patch:
diff --git a/src/ansys/platform/instancemanagement/interceptor.py b/src/ansys/platform/instancemanagement/interceptor.py
new file mode 100644
index 0000000..e1c0523
--- /dev/null
+++ b/src/ansys/platform/instancemanagement/interceptor.py
@@ -0,0 +1,86 @@
+"""Interceptor that adds headers to outgoing requests.
+
+This is a direct adaptation of the official gRPC example, but supporting multiple headers.
+https://github.com/grpc/grpc/tree/master/examples/python/interceptors/headers
+"""
+
+import collections
+from typing import Sequence, Tuple
+
+import grpc
+
+
+class _GenericClientInterceptor(
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+):
+    def __init__(self, interceptor_function):
+        self._fn = interceptor_function
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        new_details, new_request_iterator, postprocess = self._fn(
+            client_call_details, iter((request,)), False, False
+        )
+        response = continuation(new_details, next(new_request_iterator))
+        return postprocess(response) if postprocess else response
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        new_details, new_request_iterator, postprocess = self._fn(
+            client_call_details, iter((request,)), False, True
+        )
+        response_it = continuation(new_details, next(new_request_iterator))
+        return postprocess(response_it) if postprocess else response_it
+
+    def intercept_stream_unary(self, continuation, client_call_details, request_iterator):
+        new_details, new_request_iterator, postprocess = self._fn(
+            client_call_details, request_iterator, True, False
+        )
+        response = continuation(new_details, new_request_iterator)
+        return postprocess(response) if postprocess else response
+
+    def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
+        new_details, new_request_iterator, postprocess = self._fn(
+            client_call_details, request_iterator, True, True
+        )
+        response_it = continuation(new_details, new_request_iterator)
+        return postprocess(response_it) if postprocess else response_it
+
+
+class _ClientCallDetails(
+    collections.namedtuple("_ClientCallDetails", ("method", "timeout", "metadata", "credentials")),
+    grpc.ClientCallDetails,
+):
+    pass
+
+
+def header_adder_interceptor(headers: Sequence[Tuple[str, str]]):
+    """Interceptor adding a set of headers to the requests.
+
+    Args:
+        headers (Sequence[Tuple[str, str]]): List of metadata to inject
+    """
+
+    def intercept_call(client_call_details, request_iterator, *_):
+        metadata = []
+        if client_call_details.metadata is not None:
+            metadata = list(client_call_details.metadata)
+
+        for header, value in headers:
+            metadata.append(
+                (
+                    header,
+                    value,
+                )
+            )
+
+        client_call_details = _ClientCallDetails(
+            client_call_details.method,
+            client_call_details.timeout,
+            metadata,
+            client_call_details.credentials,
+        )
+        return client_call_details, request_iterator, None
+
+    return _GenericClientInterceptor(intercept_call)
diff --git a/src/ansys/platform/instancemanagement/service.py b/src/ansys/platform/instancemanagement/service.py
index 81cdaf9..0afa935 100644
--- a/src/ansys/platform/instancemanagement/service.py
+++ b/src/ansys/platform/instancemanagement/service.py
@@ -6,6 +6,9 @@
 from ansys.api.platform.instancemanagement.v1.product_instance_manager_pb2 import (
     Service as ServiceV1,
 )
+import grpc
+
+from ansys.platform.instancemanagement.interceptor import header_adder_interceptor
 
 
 @dataclass(frozen=True)
@@ -35,6 +38,25 @@ class Service:
     every communication with the service.
     """
 
+    def build_grpc_channel(
+        self,
+        **kwargs,
+    ) -> grpc.Channel:
+        """Build a gRPC channel communicating with the product instance.
+
+        Args:
+            kwargs: Optional named arguments for gRPC construction.
+            They will be passed to `grpc.insecure_channel`
+
+        Returns:
+            grpc.Channel: gRPC channel ready to be used for communicating with
+            the service.
+        """
+        headers = self.headers.items()
+        interceptor = header_adder_interceptor(headers)
+        channel = grpc.insecure_channel(self.uri, **kwargs)
+        return grpc.intercept_channel(channel, interceptor)
+
     @staticmethod
     def _from_pim_v1(service: ServiceV1):
         """Build a Definition from the PIM API v1 protobuf object.


Test patch:
diff --git a/requirements/requirements_tests.txt b/requirements/requirements_tests.txt
index 4f6d703..f6de369 100644
--- a/requirements/requirements_tests.txt
+++ b/requirements/requirements_tests.txt
@@ -1,4 +1,5 @@
 pytest>=7.1.0
 pytest-cov>=3.0.0
 ansys-api-platform-instancemanagement>=1.0.0b1
-grpcio-testing
\ No newline at end of file
+grpcio-testing
+grpcio-health-checking
diff --git a/tests/test_service.py b/tests/test_service.py
index 3f23334..a013099 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -1,6 +1,9 @@
 from ansys.api.platform.instancemanagement.v1.product_instance_manager_pb2 import (
     Service as ServiceV1,
 )
+import grpc
+import grpc_health.v1.health_pb2 as health_pb2
+import grpc_health.v1.health_pb2_grpc as health_pb2_grpc
 import pytest
 
 from ansys.platform.instancemanagement import Service
@@ -26,3 +29,50 @@ def test_from_pim_v1_proto():
 def test_from_pim_v1_proto_value_error(invalid_service):
     with pytest.raises(ValueError):
         Service._from_pim_v1(invalid_service)
+
+
+@pytest.mark.parametrize("headers", [{}, {"a": "b"}, {"my-token": "value", "identity": "thing"}])
+def test_build_channel(testing_pool, headers):
+    # Arrange
+    # A very basic server implementing health check
+    # with a Service object representing a connection to it
+    server = grpc.server(testing_pool)
+
+    received_metadata = []
+    received_requests = []
+
+    class HealthServicer(health_pb2_grpc.HealthServicer):
+        def Check(self, request, context):
+            received_metadata.append(context.invocation_metadata())
+            received_requests.append(request)
+            return health_pb2.HealthCheckResponse()
+
+    health_pb2_grpc.add_HealthServicer_to_server(HealthServicer(), server)
+    port = server.add_insecure_port("127.0.0.1:0")
+    server.start()
+
+    service = Service(uri=f"127.0.0.1:{port}", headers=headers)
+
+    # Act
+    # Build a grpc channel from the service,
+    # and send a message
+    with service.build_grpc_channel() as channel:
+        stub = health_pb2_grpc.HealthStub(channel)
+        request = health_pb2.HealthCheckRequest(service="hello world")
+        stub.Check(request)
+
+    server.stop(grace=0.1)
+
+    # Assert
+    # The message was received
+    # it's intact and the headers were injected.
+    assert len(received_metadata) == 1
+    assert len(received_requests) == 1
+
+    # note: no assertion on length, gRPC itself injects metadata
+    metadata_dict = dict(received_metadata[0])
+    for header, value in headers.items():
+        assert header in metadata_dict.keys()
+        assert metadata_dict[header] == value
+
+    assert received_requests[0] == health_pb2.HealthCheckRequest(service="hello world")