• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#! /usr/bin/python
2#
3# Protocol Buffers - Google's data interchange format
4# Copyright 2008 Google Inc.  All rights reserved.
5# http://code.google.com/p/protobuf/
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
37import unittest
38from google.protobuf import unittest_pb2
39from google.protobuf import service_reflection
40from google.protobuf import service
41
42
43class FooUnitTest(unittest.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
135if __name__ == '__main__':
136  unittest.main()
137