• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# Copyright 2021 The Pigweed Authors
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may not
5# use this file except in compliance with the License. You may obtain a copy of
6# the License at
7#
8#     https://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations under
14# the License.
15"""Tests using the callback client for pw_rpc."""
16
17from typing import List, Tuple
18import unittest
19
20import pw_hdlc.rpc
21from pw_rpc import benchmark_pb2, testing
22from pw_status import Status
23
24ITERATIONS = 50
25
26
27class RpcIntegrationTest(unittest.TestCase):
28    """Calls RPCs on an RPC server through a socket."""
29
30    test_server_command: Tuple[str, ...] = ()
31    port: int
32
33    def setUp(self) -> None:
34        self._context = pw_hdlc.rpc.HdlcRpcLocalServerAndClient(
35            self.test_server_command, self.port, [benchmark_pb2]
36        )
37        self.rpcs = self._context.client.channel(1).rpcs
38
39    def tearDown(self) -> None:
40        self._context.close()
41
42    def test_unary(self) -> None:
43        for i in range(ITERATIONS):
44            payload = f'O_o #{i}'.encode()
45            status, reply = self.rpcs.pw.rpc.Benchmark.UnaryEcho(
46                payload=payload
47            )
48            self.assertIs(status, Status.OK)
49            self.assertEqual(reply.payload, payload)
50
51    def test_bidirectional(self) -> None:
52        with self.rpcs.pw.rpc.Benchmark.BidirectionalEcho.invoke() as call:
53            responses = call.get_responses()
54
55            for i in range(ITERATIONS):
56                payload = f'O_o #{i}'.encode()
57                call.send(benchmark_pb2.Payload(payload=payload))
58
59                self.assertEqual(next(responses).payload, payload)
60
61    def test_bidirectional_call_twice(self) -> None:
62        rpc = self.rpcs.pw.rpc.Benchmark.BidirectionalEcho
63
64        for _ in range(ITERATIONS):
65            first_call = rpc.invoke()
66            first_call.send(payload=b'abc')
67            self.assertEqual(
68                next(iter(first_call)), rpc.response(payload=b'abc')
69            )
70            self.assertFalse(first_call.completed())
71
72            second_call = rpc.invoke()
73            second_call.send(payload=b'123')
74            self.assertEqual(
75                next(iter(second_call)), rpc.response(payload=b'123')
76            )
77
78            self.assertIs(first_call.error, Status.CANCELLED)
79            self.assertEqual(
80                first_call.responses, [rpc.response(payload=b'abc')]
81            )
82
83            self.assertFalse(second_call.completed())
84            self.assertEqual(
85                second_call.responses, [rpc.response(payload=b'123')]
86            )
87
88
89def _main(
90    test_server_command: List[str], port: int, unittest_args: List[str]
91) -> None:
92    RpcIntegrationTest.test_server_command = tuple(test_server_command)
93    RpcIntegrationTest.port = port
94    unittest.main(argv=unittest_args)
95
96
97if __name__ == '__main__':
98    _main(**vars(testing.parse_test_server_args()))
99