• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# Copyright 2020 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 creating pw_rpc client."""
16
17import unittest
18
19from pw_status import Status
20
21from pw_rpc.internal.packet_pb2 import PacketType, RpcPacket
22from pw_rpc import packets
23
24_TEST_IDS = packets.RpcIds(1, 2, 3, 4)
25_TEST_STATUS = 321
26_TEST_REQUEST = RpcPacket(
27    type=PacketType.REQUEST,
28    channel_id=_TEST_IDS.channel_id,
29    service_id=_TEST_IDS.service_id,
30    method_id=_TEST_IDS.method_id,
31    call_id=_TEST_IDS.call_id,
32    payload=RpcPacket(status=_TEST_STATUS).SerializeToString(),
33)
34_TEST_RESPONSE = RpcPacket(
35    type=PacketType.RESPONSE,
36    channel_id=_TEST_IDS.channel_id,
37    service_id=_TEST_IDS.service_id,
38    method_id=_TEST_IDS.method_id,
39    call_id=_TEST_IDS.call_id,
40    payload=RpcPacket(status=_TEST_STATUS).SerializeToString(),
41)
42
43
44class PacketsTest(unittest.TestCase):
45    """Tests for packet encoding and decoding."""
46
47    def test_encode_request(self):
48        data = packets.encode_request(_TEST_IDS, RpcPacket(status=_TEST_STATUS))
49        packet = RpcPacket()
50        packet.ParseFromString(data)
51
52        self.assertEqual(_TEST_REQUEST, packet)
53
54    def test_encode_response(self):
55        data = packets.encode_response(
56            _TEST_IDS, RpcPacket(status=_TEST_STATUS)
57        )
58        packet = RpcPacket()
59        packet.ParseFromString(data)
60
61        self.assertEqual(_TEST_RESPONSE, packet)
62
63    def test_encode_cancel(self):
64        data = packets.encode_cancel(packets.RpcIds(9, 8, 7, 6))
65
66        packet = RpcPacket()
67        packet.ParseFromString(data)
68
69        self.assertEqual(
70            packet,
71            RpcPacket(
72                type=PacketType.CLIENT_ERROR,
73                channel_id=9,
74                service_id=8,
75                method_id=7,
76                call_id=6,
77                status=Status.CANCELLED.value,
78            ),
79        )
80
81    def test_encode_client_error(self):
82        data = packets.encode_client_error(_TEST_REQUEST, Status.NOT_FOUND)
83
84        packet = RpcPacket()
85        packet.ParseFromString(data)
86
87        self.assertEqual(
88            packet,
89            RpcPacket(
90                type=PacketType.CLIENT_ERROR,
91                channel_id=_TEST_IDS.channel_id,
92                service_id=_TEST_IDS.service_id,
93                method_id=_TEST_IDS.method_id,
94                call_id=_TEST_IDS.call_id,
95                status=Status.NOT_FOUND.value,
96            ),
97        )
98
99    def test_decode(self):
100        self.assertEqual(
101            _TEST_REQUEST, packets.decode(_TEST_REQUEST.SerializeToString())
102        )
103
104    def test_for_server(self):
105        self.assertTrue(packets.for_server(_TEST_REQUEST))
106        self.assertFalse(packets.for_server(_TEST_RESPONSE))
107
108
109if __name__ == '__main__':
110    unittest.main()
111