python/google/protobuf/internal/service_reflection_test.py

   1 #! /usr/bin/python
   2 #
   3 # Protocol Buffers - Google's data interchange format
   4 # Copyright 2008 Google Inc.  All rights reserved.
   5 # https://developers.google.com/protocol-buffers/
   6 #
   7 # Redistribution and use in source and binary forms, with or without
   8 # modification, are permitted provided that the following conditions are
   9 # met:
  10 #
  11 #     * Redistributions of source code must retain the above copyright
  12 # notice, this list of conditions and the following disclaimer.
  13 #     * Redistributions in binary form must reproduce the above
  14 # copyright notice, this list of conditions and the following disclaimer
  15 # in the documentation and/or other materials provided with the
  16 # distribution.
  17 #     * Neither the name of Google Inc. nor the names of its
  18 # contributors may be used to endorse or promote products derived from
  19 # this software without specific prior written permission.
  20 #
  21 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32
  33 """Tests for google.protobuf.internal.service_reflection."""
  34
  35 __author__ = 'petar@google.com (Petar Petrov)'
  36
  37 from google.apputils import basetest
  38 from google.protobuf import unittest_pb2
  39 from google.protobuf import service_reflection
  40 from google.protobuf import service
  41
  42
  43 class FooUnitTest(basetest.TestCase):
  44
  45   def testService(self):
  46     class MockRpcChannel(service.RpcChannel):
  47       def CallMethod(self, method, controller, request, response, callback):
  48         self.method = method
  49         self.controller = controller
  50         self.request = request
  51         callback(response)
  52
  53     class MockRpcController(service.RpcController):
  54       def SetFailed(self, msg):
  55         self.failure_message = msg
  56
  57     self.callback_response = None
  58
  59     class MyService(unittest_pb2.TestService):
  60       pass
  61
  62     self.callback_response = None
  63
  64     def MyCallback(response):
  65       self.callback_response = response
  66
  67     rpc_controller = MockRpcController()
  68     channel = MockRpcChannel()
  69     srvc = MyService()
  70     srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
  71     self.assertEqual('Method Foo not implemented.',
  72                      rpc_controller.failure_message)
  73     self.assertEqual(None, self.callback_response)
  74
  75     rpc_controller.failure_message = None
  76
  77     service_descriptor = unittest_pb2.TestService.GetDescriptor()
  78     srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
  79                     unittest_pb2.BarRequest(), MyCallback)
  80     self.assertEqual('Method Bar not implemented.',
  81                      rpc_controller.failure_message)
  82     self.assertEqual(None, self.callback_response)
  83
  84     class MyServiceImpl(unittest_pb2.TestService):
  85       def Foo(self, rpc_controller, request, done):
  86         self.foo_called = True
  87       def Bar(self, rpc_controller, request, done):
  88         self.bar_called = True
  89
  90     srvc = MyServiceImpl()
  91     rpc_controller.failure_message = None
  92     srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
  93     self.assertEqual(None, rpc_controller.failure_message)
  94     self.assertEqual(True, srvc.foo_called)
  95
  96     rpc_controller.failure_message = None
  97     srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
  98                     unittest_pb2.BarRequest(), MyCallback)
  99     self.assertEqual(None, rpc_controller.failure_message)
 100     self.assertEqual(True, srvc.bar_called)
 101
 102   def testServiceStub(self):
 103     class MockRpcChannel(service.RpcChannel):
 104       def CallMethod(self, method, controller, request,
 105                      response_class, callback):
 106         self.method = method
 107         self.controller = controller
 108         self.request = request
 109         callback(response_class())
 110
 111     self.callback_response = None
 112
 113     def MyCallback(response):
 114       self.callback_response = response
 115
 116     channel = MockRpcChannel()
 117     stub = unittest_pb2.TestService_Stub(channel)
 118     rpc_controller = 'controller'
 119     request = 'request'
 120
 121     # GetDescriptor now static, still works as instance method for compatability
 122     self.assertEqual(unittest_pb2.TestService_Stub.GetDescriptor(),
 123                      stub.GetDescriptor())
 124
 125     # Invoke method.
 126     stub.Foo(rpc_controller, request, MyCallback)
 127
 128     self.assertTrue(isinstance(self.callback_response,
 129                                unittest_pb2.FooResponse))
 130     self.assertEqual(request, channel.request)
 131     self.assertEqual(rpc_controller, channel.controller)
 132     self.assertEqual(stub.GetDescriptor().methods[0], channel.method)
 133
 134
 135 if __name__ == '__main__':
 136   basetest.main()