• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2020 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://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, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Tests classes in pw_rpc.descriptors."""
15
16import unittest
17
18from google.protobuf.message_factory import MessageFactory
19
20from pw_protobuf_compiler import python_protos
21from pw_rpc import descriptors
22
23TEST_PROTO = """\
24syntax = "proto3";
25
26package pw.test1;
27
28message SomeMessage {
29  uint32 magic_number = 1;
30}
31
32message AnotherMessage {
33  enum Result {
34    FAILED = 0;
35    FAILED_MISERABLY = 1;
36    I_DONT_WANT_TO_TALK_ABOUT_IT = 2;
37  }
38
39  Result result = 1;
40  string payload = 2;
41}
42
43service PublicService {
44  rpc SomeUnary(SomeMessage) returns (AnotherMessage) {}
45  rpc SomeServerStreaming(SomeMessage) returns (stream AnotherMessage) {}
46  rpc SomeClientStreaming(stream SomeMessage) returns (AnotherMessage) {}
47  rpc SomeBidiStreaming(stream SomeMessage) returns (stream AnotherMessage) {}
48}
49"""
50
51
52class MethodTest(unittest.TestCase):
53    """Tests pw_rpc.Method."""
54    def setUp(self):
55        module, = python_protos.compile_and_import_strings([TEST_PROTO])
56        service = descriptors.Service.from_descriptor(
57            module.DESCRIPTOR.services_by_name['PublicService'])
58        self._method = service.methods['SomeUnary']
59
60    def test_get_request_with_both_message_and_kwargs(self):
61        with self.assertRaisesRegex(TypeError, r'either'):
62            self._method.get_request(self._method.request_type(),
63                                     {'magic_number': 1})
64
65    def test_get_request_with_wrong_type(self):
66        with self.assertRaisesRegex(TypeError, r'pw\.test1\.SomeMessage'):
67            self._method.get_request('a str!', {})
68
69    def test_get_request_with_different_message_type(self):
70        msg = self._method.response_type()
71        with self.assertRaisesRegex(TypeError, r'pw\.test1\.SomeMessage'):
72            self._method.get_request(msg, {})
73
74    def test_get_request_with_different_copy_of_same_message_class(self):
75        some_message_clone = MessageFactory(
76            self._method.request_type.DESCRIPTOR.file.pool).GetPrototype(
77                self._method.request_type.DESCRIPTOR)
78
79        msg = some_message_clone()
80
81        # Protobuf classes obtained with a MessageFactory may or may not be a
82        # unique type, but will always use the same descriptor instance.
83        self.assertIsInstance(msg, some_message_clone)
84        self.assertIs(msg.DESCRIPTOR, self._method.request_type.DESCRIPTOR)
85
86        result = self._method.get_request(msg, {})
87        self.assertIs(result, msg)
88
89
90if __name__ == '__main__':
91    unittest.main()
92