1# Copyright 2019 the gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""The Python implementation of the GRPC helloworld.Greeter client."""
15
16from concurrent import futures
17import contextlib
18import datetime
19import logging
20import unittest
21
22# TODO(https://github.com/grpc/grpc/issues/29284)
23# isort: off
24import grpc
25from google.protobuf import duration_pb2
26from google.protobuf import timestamp_pb2
27from google.cloud import helloworld_pb2
28from google.cloud import helloworld_pb2_grpc
29# isort: on
30
31_HOST = 'localhost'
32_SERVER_ADDRESS = '{}:0'.format(_HOST)
33
34
35class Greeter(helloworld_pb2_grpc.GreeterServicer):
36
37    def SayHello(self, request, context):
38        request_in_flight = datetime.datetime.now() - \
39                            request.request_initiation.ToDatetime()
40        request_duration = duration_pb2.Duration()
41        request_duration.FromTimedelta(request_in_flight)
42        return helloworld_pb2.HelloReply(
43            message='Hello, %s!' % request.name,
44            request_duration=request_duration,
45        )
46
47
48@contextlib.contextmanager
49def _listening_server():
50    server = grpc.server(futures.ThreadPoolExecutor())
51    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
52    port = server.add_insecure_port(_SERVER_ADDRESS)
53    server.start()
54    try:
55        yield port
56    finally:
57        server.stop(0)
58
59
60class ImportTest(unittest.TestCase):
61
62    def test_import(self):
63        with _listening_server() as port:
64            with grpc.insecure_channel('{}:{}'.format(_HOST, port)) as channel:
65                stub = helloworld_pb2_grpc.GreeterStub(channel)
66                request_timestamp = timestamp_pb2.Timestamp()
67                request_timestamp.GetCurrentTime()
68                response = stub.SayHello(helloworld_pb2.HelloRequest(
69                    name='you',
70                    request_initiation=request_timestamp,
71                ),
72                                         wait_for_ready=True)
73                self.assertEqual(response.message, "Hello, you!")
74                self.assertGreater(response.request_duration.nanos, 0)
75
76
77if __name__ == '__main__':
78    logging.basicConfig()
79    unittest.main()
80