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