• 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
17import logging
18from pathlib import Path
19import unittest
20from unittest import mock
21
22from pw_hdlc import rpc
23from pw_rpc import testing
24from pw_unit_test_proto import unit_test_pb2
25from pw_unit_test import run_tests, EventHandler, TestCase
26from pw_status import Status
27
28# The three suites (Passing, Failing, and DISABLED_Disabled) have these cases.
29_CASES = ('Zero', 'One', 'Two', 'DISABLED_Disabled')
30_FILE = 'pw_unit_test/test_rpc_server.cc'
31
32PASSING = tuple(TestCase('Passing', case, _FILE) for case in _CASES[:-1])
33FAILING = tuple(TestCase('Failing', case, _FILE) for case in _CASES[:-1])
34EXECUTED_TESTS = PASSING + FAILING
35
36DISABLED_SUITE = tuple(
37    TestCase('DISABLED_Disabled', case, _FILE) for case in _CASES
38)
39
40ALL_DISABLED_TESTS = (
41    TestCase('Passing', 'DISABLED_Disabled', _FILE),
42    TestCase('Failing', 'DISABLED_Disabled', _FILE),
43    *DISABLED_SUITE,
44)
45
46
47class RpcIntegrationTest(unittest.TestCase):
48    """Calls RPCs on an RPC server through a socket."""
49
50    test_server_command: tuple[str, ...] = ()
51    port: int
52
53    def setUp(self) -> None:
54        self._context = rpc.HdlcRpcLocalServerAndClient(
55            self.test_server_command, self.port, [unit_test_pb2]
56        )
57        self.rpcs = self._context.client.channel(1).rpcs
58        self.handler = mock.NonCallableMagicMock(spec=EventHandler)
59
60    def tearDown(self) -> None:
61        self._context.close()
62
63    def test_run_tests_default_handler(self) -> None:
64        with self.assertLogs(logging.getLogger('pw_unit_test'), 'INFO') as logs:
65            self.assertFalse(run_tests(self.rpcs))
66
67        for test in EXECUTED_TESTS:
68            self.assertTrue(any(str(test) in log for log in logs.output), test)
69
70    def test_run_tests_calls_test_case_start(self) -> None:
71        self.assertFalse(run_tests(self.rpcs, event_handlers=[self.handler]))
72
73        self.handler.test_case_start.assert_has_calls(
74            [mock.call(case) for case in EXECUTED_TESTS], any_order=True
75        )
76
77    def test_run_tests_calls_test_case_end(self) -> None:
78        self.assertFalse(run_tests(self.rpcs, event_handlers=[self.handler]))
79
80        calls = [
81            mock.call(
82                case,
83                unit_test_pb2.SUCCESS
84                if case.suite_name == 'Passing'
85                else unit_test_pb2.FAILURE,
86            )
87            for case in EXECUTED_TESTS
88        ]
89        self.handler.test_case_end.assert_has_calls(calls, any_order=True)
90
91    def test_run_tests_calls_test_case_disabled(self) -> None:
92        self.assertFalse(run_tests(self.rpcs, event_handlers=[self.handler]))
93
94        self.handler.test_case_disabled.assert_has_calls(
95            [mock.call(case) for case in ALL_DISABLED_TESTS], any_order=True
96        )
97
98    def test_passing_tests_only(self) -> None:
99        self.assertTrue(
100            run_tests(
101                self.rpcs,
102                test_suites=['Passing'],
103                event_handlers=[self.handler],
104            )
105        )
106        calls = [mock.call(case, unit_test_pb2.SUCCESS) for case in PASSING]
107        self.handler.test_case_end.assert_has_calls(calls, any_order=True)
108
109    def test_disabled_tests_only(self) -> None:
110        self.assertTrue(
111            run_tests(
112                self.rpcs,
113                test_suites=['DISABLED_Disabled'],
114                event_handlers=[self.handler],
115            )
116        )
117
118        self.handler.test_case_start.assert_not_called()
119        self.handler.test_case_end.assert_not_called()
120        self.handler.test_case_disabled.assert_has_calls(
121            [mock.call(case) for case in DISABLED_SUITE], any_order=True
122        )
123
124    def test_failing_tests(self) -> None:
125        self.assertFalse(
126            run_tests(
127                self.rpcs,
128                test_suites=['Failing'],
129                event_handlers=[self.handler],
130            )
131        )
132        calls = [mock.call(case, unit_test_pb2.FAILURE) for case in FAILING]
133        self.handler.test_case_end.assert_has_calls(calls, any_order=True)
134
135
136def _main(
137    test_server_command: list[str], port: int, unittest_args: list[str]
138) -> None:
139    RpcIntegrationTest.test_server_command = tuple(test_server_command)
140    RpcIntegrationTest.port = port
141    unittest.main(argv=unittest_args)
142
143
144if __name__ == '__main__':
145    _main(**vars(testing.parse_test_server_args()))
146