当前位置：源码 > Python源码 >

Python源码示例：google.protobuf.service.RpcController()

示例1

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertTrue(srvc.GetRequestClass(service_descriptor.methods[1]) is
                    unittest_pb2.BarRequest)
    self.assertTrue(srvc.GetResponseClass(service_descriptor.methods[1]) is
                    unittest_pb2.BarResponse)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例2

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例3

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例4

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)
    
    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例5

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例6

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)
    
    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例7

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例8

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertTrue(srvc.GetRequestClass(service_descriptor.methods[1]) is
                    unittest_pb2.BarRequest)
    self.assertTrue(srvc.GetResponseClass(service_descriptor.methods[1]) is
                    unittest_pb2.BarResponse)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

示例9

def testService(self):
    class MockRpcChannel(service.RpcChannel):
      def CallMethod(self, method, controller, request, response, callback):
        self.method = method
        self.controller = controller
        self.request = request
        callback(response)

    class MockRpcController(service.RpcController):
      def SetFailed(self, msg):
        self.failure_message = msg

    self.callback_response = None

    class MyService(unittest_pb2.TestService):
      pass

    self.callback_response = None

    def MyCallback(response):
      self.callback_response = response

    rpc_controller = MockRpcController()
    channel = MockRpcChannel()
    srvc = MyService()
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual('Method Foo not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    rpc_controller.failure_message = None

    service_descriptor = unittest_pb2.TestService.GetDescriptor()
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual('Method Bar not implemented.',
                     rpc_controller.failure_message)
    self.assertEqual(None, self.callback_response)

    class MyServiceImpl(unittest_pb2.TestService):
      def Foo(self, rpc_controller, request, done):
        self.foo_called = True
      def Bar(self, rpc_controller, request, done):
        self.bar_called = True

    srvc = MyServiceImpl()
    rpc_controller.failure_message = None
    srvc.Foo(rpc_controller, unittest_pb2.FooRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.foo_called)

    rpc_controller.failure_message = None
    srvc.CallMethod(service_descriptor.methods[1], rpc_controller,
                    unittest_pb2.BarRequest(), MyCallback)
    self.assertEqual(None, rpc_controller.failure_message)
    self.assertEqual(True, srvc.bar_called)

Python源码示例：google.protobuf.service.RpcController()

微信关注