Make FakeServicerContext Inherit from grpcio ServicerContext

Author: sjoerdvandenbos-prodrive

django_grpc_testtools/context.py

@@ -1,11 +1,12 @@
 from typing import NoReturn
 from grpc import RpcError, StatusCode
 from collections.abc import Sequence, Mapping
+from grpc import ServicerContext
 
 MetadataType = Sequence[tuple[str, str]]
 
 
-class FakeServicerContext:
+class FakeServicerContext(ServicerContext):
     """
     Implementation of basic RpcContext methods that stores data
     for validation in tests
@@ -50,4 +51,71 @@ def set_invocation_metadata(self, items: MetadataType) -> None:
         """
         Helper to emulate request metadata
         """
-        self._invocation_metadata = items
+        self._invocation_metadata = items
+
+    def set_code(self, code: StatusCode) -> None:
+        self.abort_status = code
+
+    def set_details(self, details: str) -> None:
+        """Sets the value to be used as detail string upon RPC completion.
+
+        This method need not be called by method implementations if they have
+        no details to transmit.
+
+        Args:
+          details: A UTF-8-encodable string to be sent to the client upon
+            termination of the RPC.
+        """
+        self.abort_message = details
+
+    def code(self) -> StatusCode:
+        """Accesses the value to be used as status code upon RPC completion.
+
+        Returns:
+          The StatusCode value for the RPC.
+        """
+        return self.abort_status
+
+    def details(self) -> str:
+        """Accesses the value to be used as detail string upon RPC completion.
+
+        Returns:
+          The details string of the RPC.
+        """
+        return self.abort_message
+
+    def peer(self):
+        raise NotImplementedError()
+
+    def peer_identities(self):
+        raise NotImplementedError()
+
+    def peer_identity_key(self):
+        raise NotImplementedError()
+
+    def auth_context(self):
+        raise NotImplementedError()
+
+    def set_compression(self, compression):
+        raise NotImplementedError()
+
+    def send_initial_metadata(self, initial_metadata):
+        raise NotImplementedError()
+
+    def abort_with_status(self, status):
+        raise NotImplementedError()
+
+    def disable_next_message_compression(self):
+        raise NotImplementedError()
+
+    def is_active(self):
+        raise NotImplementedError()
+
+    def time_remaining(self):
+        raise NotImplementedError()
+
+    def cancel(self):
+        raise NotImplementedError()
+
+    def add_callback(self, callback):
+        raise NotImplementedError()

tests/test_fake_servicer_context.py

@@ -1,10 +1,11 @@
-from grpc import RpcError, StatusCode
+from grpc import RpcError, StatusCode, ServicerContext
 import pytest
 
 from django_grpc_testtools.context import FakeServicerContext
 
 
 def test_fake_servicer_context():
     context = FakeServicerContext()
+    assert isinstance(context, ServicerContext)
     with pytest.raises(RpcError):
         context.abort(StatusCode.UNAVAILABLE, 'test')