tests: forward compatible fix for google-cloud-monitoring (#13379)

Towards #13356

This PR fixes the issue seen in PR
#13359 where
presubmits fail with

```
>   thunk = lambda m: self._channel.unary_unary(
        m,
        request_serializer,
        response_deserializer,
        _registered_method,
    )
E   TypeError: ChannelStub.unary_unary() takes from 2 to 4 positional arguments but 5 were given
```

The reason for the failures is that in
https://github.com/googleapis/gapic-generator-python/pull/2284/files
support for debug logging was added, which required using grpc
[intercept_channel](https://github.com/googleapis/gapic-generator-python/blob/e050f4eb3eddfe150f028a2a2bd901899afb965a/gapic/templates/%25namespace/%25name_%25version/%25sub/services/%25service/transports/grpc.py.j2#L281)
and there is a slight different in the [standard gRPC channel
](https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_channel.py#L2126
)and the
[intercept_channel](https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L688).
If gRPC is mocked, it needs to be compatible with
[intercept_channel](https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L688).

The issue, as seen in the stack trace, is that `unary_unary`
[here](https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L696-L701)
expects an additional parameter `_registered_method` in the method
signature of `unary_unary` however the stub code below for `unary_unary`
does not have this parameter.


https://github.com/googleapis/google-cloud-python/blob/9b2f079a87eb30c75bc25bd339471c3a35a4f1fc/packages/google-cloud-monitoring/tests/unit/test_query.py#L87

In addition, the grpc code
[here](https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L516-L525)
also expects `with_call` which needs to be added as well. Also see
https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L277-L284

---------

Co-authored-by: Victor Chudnovsky <[email protected]>

Author: parthea

packages/google-cloud-monitoring/tests/unit/test_query.py

@@ -73,6 +73,24 @@ def __call__(self, request, timeout=None, metadata=None, credentials=None):
         if response:
             return response
 
+    # This method is required; otherwise we see `AttributeError: 'MultiCallableStub' object has no attribute 'with_call'`
+    # See https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L315
+    # for the specific location in gRPC code which expects this method.
+    def with_call(
+        self,
+        request,
+        timeout=None,
+        metadata=None,
+        credentials=None,
+        wait_for_ready=None,
+        compression=None,
+    ):
+        self.channel_stub.requests.append((self.method, request))
+        mock_call = None  # Actual type is grpc.Call but using None since `with_call` is not needed for tests
+        response = self.channel_stub.responses.pop()
+        if response:
+            return (response, mock_call)
+
 
 class ChannelStub(grpc.Channel):
     """Stub for the grpc.Channel interface."""
@@ -84,7 +102,16 @@ def __init__(self, responses=[]):
         ]
         self.requests = []
 
-    def unary_unary(self, method, request_serializer=None, response_deserializer=None):
+    # The method signature should be compatible with this code
+    # https://github.com/grpc/grpc/blob/b53f4055a93fb98601c75dcefaa8f3665167e6cf/src/python/grpcio/grpc/_interceptor.py#L696-L701
+    # where arguments `method`, `request_serializer`, `response_deserializer` and `_registered_method` are supported
+    def unary_unary(
+        self,
+        method,
+        request_serializer=None,
+        response_deserializer=None,
+        _registered_method=None,
+    ):
         return MultiCallableStub(method, self)
 
     def subscribe(self, callback, try_to_connect=False):