| # Copyright 2022 Google LLC |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| """Unit test for Matter RPC verification tool.""" |
| import os |
| import shutil |
| import sys |
| import time |
| import unittest |
| from unittest import mock |
| |
| from freezegun import freeze_time |
| |
| from mobly import asserts |
| from mobly import config_parser |
| import pw_hdlc.rpc |
| import pw_rpc |
| import serial |
| import serial.tools.list_ports |
| import simple_term_menu |
| |
| import rpc_tool |
| |
| _FAKE_DEVICE_ADDRESS = "fake-device-address" |
| _FAKE_DESCRIPTION = "fake-description" |
| _FAKE_TIME_NOW = "2022-06-01" |
| _FAKE_INDEX = 0 |
| _FAKE_MATTER_DATA = 1 |
| |
| |
| class MatterDeviceTest(unittest.TestCase): |
| """Unit tests for MatterDevice.""" |
| |
| def setUp(self): |
| super().setUp() |
| serial_patcher = mock.patch.object(serial, "Serial") |
| serial_patcher.start() |
| self.addCleanup(serial_patcher.stop) |
| impl_patcher = mock.patch.object(pw_rpc.callback_client, "Impl") |
| impl_patcher.start() |
| self.addCleanup(impl_patcher.stop) |
| client_patcher = mock.patch.object(pw_hdlc.rpc, "HdlcRpcClient") |
| fake_client_class = client_patcher.start() |
| self.fake_client = fake_client_class.return_value |
| self.addCleanup(client_patcher.stop) |
| self.uut = rpc_tool.MatterDevice( |
| address=_FAKE_DEVICE_ADDRESS, |
| device_type=rpc_tool.MatterDeviceType.NoneOfTheAbove) |
| |
| def test_attribute_service(self): |
| """Verifies attribute_service on success.""" |
| self.assertEqual( |
| self.fake_client.rpcs.return_value.chip.rpc.Attributes, |
| self.uut.attribute_service) |
| |
| def test_descriptor_service(self): |
| """Verifies descriptor_service on success.""" |
| self.assertEqual( |
| self.fake_client.rpcs.return_value.chip.rpc.Descriptor, |
| self.uut.descriptor_service) |
| |
| def test_device_service(self): |
| """Verifies device_service on success.""" |
| self.assertEqual( |
| self.fake_client.rpcs.return_value.chip.rpc.Device, |
| self.uut.device_service) |
| |
| def test_close(self): |
| """Verifies close method on success.""" |
| self.uut.close() |
| |
| self.uut.serial_inst.close.assert_called_once() |
| |
| @mock.patch.object(rpc_tool, "open") |
| def test_write_hdlc_log(self, mock_open): |
| """Verifies write_hdlc_log method on success.""" |
| mock_output = mock.Mock() |
| mock_open.return_value.__enter__.return_value = mock_output |
| |
| self.uut.write_hdlc_log(b"") |
| |
| mock_output.write.assert_called_once_with("\n") |
| mock_output.flush.assert_called_once() |
| |
| def test_get_endpoint_id_by_device_type_none_type(self): |
| """Verifies _get_endpoint_id_by_device_type with NoneOfTheAbove.""" |
| self.assertEqual(-1, |
| self.uut._get_endpoint_id_by_device_type( |
| rpc_tool.MatterDeviceType.NoneOfTheAbove)) |
| |
| def test_get_endpoint_id_by_device_type_normal_type(self): |
| """Verifies _get_endpoint_id_by_device_type with normal type.""" |
| fake_endpoints = [mock.Mock(endpoint=_FAKE_MATTER_DATA)] |
| fake_type_inst = [mock.Mock(device_type=_FAKE_MATTER_DATA)] |
| fake_descriptor_service = mock.Mock() |
| fake_descriptor_service.PartsList.return_value = ( |
| True, fake_endpoints) |
| fake_descriptor_service.DeviceTypeList.return_value = ( |
| True, fake_type_inst) |
| self.fake_client.rpcs.return_value.chip.rpc.Descriptor = ( |
| fake_descriptor_service) |
| |
| fake_type = mock.Mock(value=_FAKE_MATTER_DATA) |
| self.assertEqual( |
| _FAKE_MATTER_DATA, |
| self.uut._get_endpoint_id_by_device_type(fake_type)) |
| |
| fake_type_not_exist = mock.Mock(value=0) |
| with self.assertRaisesRegex(RuntimeError, "No endpoint corresponding"): |
| self.uut._get_endpoint_id_by_device_type(fake_type_not_exist) |
| |
| def test_app_type(self): |
| """Verifies app_type property.""" |
| self.assertEqual(rpc_tool.MatterAppType.NONE, self.uut.app_type) |
| |
| def test_turn_light_on_or_off_on_failure(self): |
| """Verifies turn_light_on_or_off on failure.""" |
| with self.assertRaisesRegex(RuntimeError, "not a light"): |
| self.uut.turn_light_on_or_off(True) |
| |
| def test_turn_light_on_or_off_on_success(self): |
| """Verifies turn_light_on_or_off on success.""" |
| self.uut._app_type = rpc_tool.MatterAppType.LIGHT |
| self.uut._endpoint_id = _FAKE_MATTER_DATA |
| |
| self.uut.turn_light_on_or_off(True) |
| |
| (self.fake_client.rpcs.return_value.chip.rpc.Attributes. |
| Write.assert_called_once()) |
| |
| def test_get_light_state_on_failure(self): |
| """Verifies get_light_state on failure.""" |
| with self.assertRaisesRegex(RuntimeError, "not a light"): |
| self.uut.get_light_state() |
| |
| def test_get_light_state_on_success(self): |
| """Verifies get_light_state on success.""" |
| self.uut._app_type = rpc_tool.MatterAppType.LIGHT |
| self.uut._endpoint_id = _FAKE_MATTER_DATA |
| self.fake_client.rpcs.return_value.chip.rpc.Attributes.Read.return_value = ( |
| True, mock.Mock(data_bool=True)) |
| |
| self.assertTrue(self.uut.get_light_state()) |
| |
| |
| class MatterRpcToolTest(unittest.TestCase): |
| """Unit tests for RPC tool module.""" |
| |
| def setUp(self): |
| super().setUp() |
| mock_configs = config_parser.TestRunConfig() |
| self.uut = rpc_tool.MatterRpcTest(mock_configs) |
| mock_matter_device = mock.MagicMock(spec=rpc_tool.MatterDevice) |
| matter_device_class_patcher = mock.patch.object( |
| rpc_tool, |
| "MatterDevice", |
| return_value=mock_matter_device) |
| matter_device_class_patcher.start() |
| self.addCleanup(matter_device_class_patcher.stop) |
| self.uut.dut = mock_matter_device |
| |
| @mock.patch.object( |
| rpc_tool.MatterRpcTest, |
| "_device_type_selection", |
| return_value=rpc_tool.MatterDeviceType.NoneOfTheAbove) |
| @mock.patch.object( |
| rpc_tool.MatterRpcTest, |
| "_device_selection", |
| return_value=_FAKE_DEVICE_ADDRESS) |
| def test_setup_class_select_device( |
| self, mock_device_selection, mock_type_selection): |
| """Verifies setup class selects a valid device.""" |
| self.uut.setup_class() |
| |
| self.assertIsNotNone(self.uut.dut) |
| |
| @mock.patch.object( |
| rpc_tool.MatterRpcTest, |
| "_device_type_selection", |
| return_value=rpc_tool.MatterDeviceType.NoneOfTheAbove) |
| @mock.patch.object(sys, "exit") |
| @mock.patch.object( |
| rpc_tool.MatterRpcTest, "_device_selection", return_value=None) |
| def test_setup_class_exit( |
| self, mock_device_selection, mock_exit, mock_type_selection): |
| """Verifies setup class exits without selecting device.""" |
| self.uut.setup_class() |
| |
| mock_exit.assert_called_once() |
| |
| @mock.patch.object(rpc_tool.MatterRpcTest, "_generate_test_result") |
| def test_teardown_class(self, mock_generate_test_result): |
| """Verifies teardown_class on success.""" |
| self.uut.teardown_class() |
| |
| self.uut.dut.close.assert_called_once() |
| mock_generate_test_result.assert_called_once() |
| |
| @mock.patch.object(time, "sleep") |
| def test_teardown_test_on_success(self, mock_sleep): |
| """Verifies teardown_test on success.""" |
| self.uut.teardown_test() |
| |
| mock_sleep.assert_called_once() |
| self.uut.dut.device_service.GetDeviceInfo.assert_called_once() |
| |
| @mock.patch.object(time, "sleep") |
| @mock.patch.object(time, "time", |
| side_effect=[0, rpc_tool._BOOTUP_TIMEOUT-1, rpc_tool._BOOTUP_TIMEOUT+1]) |
| def test_teardown_test_on_failure(self, mock_time, mock_sleep): |
| """Verifies teardown_test on failure.""" |
| fake_rpc = mock.Mock(method='') |
| self.uut.dut.device_service.GetDeviceInfo.side_effect = ( |
| pw_rpc.callback_client.errors.RpcTimeout(rpc=fake_rpc, timeout=0)) |
| |
| with self.assertRaisesRegex(RuntimeError, "fails to bootup"): |
| self.uut.teardown_test() |
| |
| def test_rpc_firmware_version_test(self): |
| """Verifies firmware_version test on success.""" |
| mock_device_info = mock.Mock(software_version=_FAKE_MATTER_DATA) |
| self.uut.dut.device_service.GetDeviceInfo.return_value = ( |
| True, mock_device_info) |
| |
| self.uut.test_rpc_firmware_version() |
| |
| def test_rpc_reboot_test(self): |
| """Verifies rpc_reboot test on success.""" |
| self.uut.dut.device_service.Reboot.return_value = True, None |
| |
| self.uut.test_rpc_reboot() |
| |
| def test_rpc_factory_reset_test(self): |
| """Verifies rpc_factory_reset test on success.""" |
| self.uut.dut.device_service.FactoryReset.return_value = True, None |
| |
| self.uut.test_rpc_factory_reset() |
| |
| def test_rpc_descriptor_cluster_test(self): |
| """Verifies test_rpc_descriptor_cluster on success.""" |
| mock_endpoints = [mock.Mock(endpoint=_FAKE_MATTER_DATA)] |
| mock_device_type = [mock.Mock(device_type=_FAKE_MATTER_DATA)] |
| self.uut.dut.descriptor_service.PartsList.return_value = ( |
| True, mock_endpoints) |
| self.uut.dut.descriptor_service.DeviceTypeList.return_value = ( |
| True, mock_device_type) |
| |
| self.uut.test_rpc_descriptor_cluster() |
| |
| def test_rpc_attribute_service_test(self): |
| """Verifies test_rpc_attribute_service on success.""" |
| mock_endpoints = [mock.Mock(endpoint=_FAKE_MATTER_DATA)] |
| mock_clusters = [mock.Mock(cluster_id=_FAKE_MATTER_DATA)] |
| mock_data = mock.Mock(data_uint32=_FAKE_MATTER_DATA) |
| self.uut.dut.descriptor_service.PartsList.return_value = ( |
| True, mock_endpoints) |
| self.uut.dut.descriptor_service.ServerList.return_value = ( |
| True, mock_clusters) |
| self.uut.dut.attribute_service.Read.return_value = ( |
| True, mock_data) |
| |
| self.uut.test_rpc_attribute_service() |
| |
| def test_light_app_type_test(self): |
| """Verifies test_light_app_type on success.""" |
| self.uut.dut.app_type = rpc_tool.MatterAppType.LIGHT |
| self.uut.dut.get_light_state.side_effect = [True, False] |
| |
| self.uut.test_light_app_type() |
| |
| @mock.patch.object(asserts, "skip") |
| def test_light_app_type_test_skip(self, mock_skip): |
| """Verifies test_light_app_type skipping.""" |
| self.uut.test_light_app_type() |
| |
| mock_skip.assert_called_once() |
| |
| @mock.patch.object(simple_term_menu.TerminalMenu, "show") |
| def test_device_type_selection(self, mock_menu_show): |
| """Verifies _device_type_selection on success.""" |
| mock_menu_show.return_value = _FAKE_INDEX |
| expected_selected_type = list(rpc_tool.MatterDeviceType)[0] |
| |
| self.assertEqual( |
| expected_selected_type, |
| self.uut._device_type_selection()) |
| |
| @mock.patch.object(simple_term_menu.TerminalMenu, "show") |
| @mock.patch.object(serial.tools.list_ports, "comports") |
| def test_device_selection(self, mock_comports, mock_menu_show): |
| """Verifies device_selection on success.""" |
| fake_device = mock.Mock( |
| device=_FAKE_DEVICE_ADDRESS, description=_FAKE_DESCRIPTION) |
| mock_comports.return_value = [fake_device] |
| mock_menu_show.return_value = _FAKE_INDEX |
| |
| self.assertEqual(_FAKE_DEVICE_ADDRESS, self.uut._device_selection()) |
| |
| @freeze_time(_FAKE_TIME_NOW) |
| @mock.patch.object(os.path, "exists", return_value=True) |
| @mock.patch.object(shutil, "rmtree") |
| @mock.patch.object(shutil, "move") |
| @mock.patch.object(shutil, "make_archive") |
| @mock.patch.object(shutil, "copytree") |
| def test_generate_test_result( |
| self, mock_copy, mock_make, mock_move, mock_rm, mock_exists): |
| """Verifies generate_test_result on success.""" |
| self.uut.fw_version = None |
| fake_test_dir = ( |
| f"{rpc_tool._TEST_RESULT_DIR_PREFIX}-unknown-{_FAKE_TIME_NOW}" |
| " 00:00:00") |
| |
| self.uut._generate_test_result() |
| |
| mock_copy.assert_called_once_with( |
| rpc_tool._MOBLY_LOG_DIRECTORY, fake_test_dir) |
| mock_make.assert_called_once_with(fake_test_dir, "zip", fake_test_dir) |
| mock_move.assert_called_once_with( |
| rpc_tool._PW_HDLC_LOG, f"{fake_test_dir}/{rpc_tool._PW_HDLC_LOG}") |
| mock_rm.assert_called_once_with(fake_test_dir) |
| mock_exists.assert_called_once_with(rpc_tool._PW_HDLC_LOG) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(failfast=True) |
