| # 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 tests for gdm_manager.""" |
| from parameterized import parameterized |
| import unittest |
| from unittest import mock |
| |
| import gazoo_device |
| from gazoo_device import errors as gdm_errors |
| |
| from local_agent import errors as agent_errors |
| from local_agent.translation_layer import gdm_manager |
| |
| |
| ####################### Fake data for unit test ############################# |
| _FAKE_DEVICE_TYPE = 'fake-device-type' |
| _FAKE_DEVICE_ID = 'fake-device-id' |
| _FAKE_DEVICE_ID2 = 'fake-device-id2' |
| _FAKE_DETECTION_RESULT = { |
| 'efr32-3453': { |
| 'persistent': {'device_type': 'efr32', 'serial_number': '000440173453'}, |
| }, |
| 'nrf52840-1234': { |
| 'persistent': {'device_type': 'nrf52849', 'serial_number': '1234'}, |
| }, |
| } |
| _FAKE_ERROR_MSG = 'error' |
| ############################################################################## |
| |
| |
| class GdmManagerTest(unittest.TestCase): |
| """Unit tests for gdm_manager module.""" |
| |
| def setUp(self): |
| super().setUp() |
| self.mgr = gdm_manager.GdmManager(mock.Mock()) |
| |
| @parameterized.expand([(True, ['pw_rpc_common']), (False, [])]) |
| @mock.patch.object(gazoo_device.Manager, 'create_and_close_device') |
| @mock.patch.object(gazoo_device.Manager, 'is_device_connected') |
| @mock.patch.object(gazoo_device.Manager, 'get_devices') |
| @mock.patch.object(gazoo_device.Manager, 'detect') |
| def test_detect_devices( |
| self, |
| has_capabilities_result, |
| expected_capability, |
| mock_detect, |
| mock_get_devices, |
| mock_is_connect, |
| mock_create_and_close_device): |
| """Verifies detect_devices on success.""" |
| mock_get_devices.return_value = _FAKE_DETECTION_RESULT |
| mock_is_connect.side_effect = [True, False] |
| fake_device = mock.Mock() |
| mock_create_and_close_device.return_value.iter.return_value = fake_device |
| mock_create_and_close_device.return_value.__enter__.return_value = fake_device |
| fake_device.matter_endpoints.get_supported_endpoints_and_clusters.return_value = {} |
| fake_device.has_capabilities.return_value = has_capabilities_result |
| |
| device_statuses = self.mgr.detect_devices() |
| |
| expected_statuses = [ |
| { |
| 'deviceId': 'efr32-3453', |
| 'serialNumber': '000440173453', |
| 'deviceType': 'efr32', |
| 'capabilities': expected_capability, |
| }, |
| ] |
| self.assertEqual(expected_statuses, device_statuses) |
| mock_detect.assert_called_once_with( |
| force_overwrite=True, log_directory='/tmp') |
| |
| @mock.patch.object(gazoo_device.Manager, 'create_device') |
| def test_create_devices_on_success(self, mock_create): |
| """Verifies create_devices on success.""" |
| identifiers = [_FAKE_DEVICE_ID, _FAKE_DEVICE_ID2] |
| self.mgr.create_devices(identifiers) |
| mock_create.assert_called_with( |
| identifier=_FAKE_DEVICE_ID2, log_directory=None) |
| self.assertEqual(2, mock_create.call_count) |
| |
| @mock.patch.object(gazoo_device.Manager, 'is_device_connected') |
| def test_check_device_connected_on_success( |
| self, mock_is_device_connected): |
| """Verifies check_device_connected succeeds when device connected.""" |
| mock_is_device_connected.return_value = True |
| self.mgr.check_device_connected(_FAKE_DEVICE_ID) |
| self.assertEqual(1, mock_is_device_connected.call_count) |
| |
| @mock.patch.object(gazoo_device.Manager, 'is_device_connected') |
| def test_check_device_connected_raises_when_disconnected( |
| self, mock_is_device_connected): |
| """Verifies check_device_connected raises when device disconnected.""" |
| mock_is_device_connected.return_value = False |
| with self.assertRaises(agent_errors.DeviceNotConnectedError): |
| self.mgr.check_device_connected(_FAKE_DEVICE_ID) |
| self.assertEqual(1, mock_is_device_connected.call_count) |
| |
| @mock.patch.object(gazoo_device.Manager, 'is_device_connected') |
| def test_check_device_connected_raises_unexpected_error( |
| self, mock_is_device_connected): |
| """ |
| Verifies check_device_connected raises when device operation has error. |
| """ |
| mock_is_device_connected.side_effect = ( |
| gdm_errors.DeviceError(_FAKE_ERROR_MSG)) |
| with self.assertRaisesRegex(gdm_errors.DeviceError, _FAKE_ERROR_MSG): |
| self.mgr.check_device_connected(_FAKE_DEVICE_ID) |
| self.assertEqual(1, mock_is_device_connected.call_count) |
| |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device') |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device_names') |
| def test_get_device_instance_on_success( |
| self, mock_get_open_device_names, mock_get_open_device): |
| """Verifies get_device_instance on success.""" |
| mock_get_open_device_names.return_value = [_FAKE_DEVICE_ID] |
| self.mgr.get_device_instance(_FAKE_DEVICE_ID) |
| self.assertEqual(1, mock_get_open_device_names.call_count) |
| mock_get_open_device.assert_called_once_with(_FAKE_DEVICE_ID) |
| |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device_names') |
| def test_get_device_instance_failure_because_device_not_open( |
| self, mock_get_open_device_names): |
| """Verifies get_device_instance on failure.""" |
| mock_get_open_device_names.return_value = [] |
| error_msg = f'{_FAKE_DEVICE_ID} is not open' |
| with self.assertRaisesRegex(agent_errors.DeviceNotOpenError, error_msg): |
| self.mgr.get_device_instance(_FAKE_DEVICE_ID) |
| |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device') |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device_names') |
| def test_get_device_type_on_success( |
| self, mock_get_names, mock_get_device): |
| """Verifies _get_device_type on success.""" |
| mock_get_names.return_value = {_FAKE_DEVICE_ID,} |
| mock_get_device.return_value.device_type = _FAKE_DEVICE_TYPE |
| dut_type = self.mgr.get_device_type(_FAKE_DEVICE_ID) |
| self.assertEqual(_FAKE_DEVICE_TYPE, dut_type) |
| self.assertEqual(1, mock_get_names.call_count) |
| self.assertEqual(1, mock_get_device.call_count) |
| |
| @mock.patch.object(gazoo_device.Manager, 'get_open_device_names') |
| def test_get_device_type_on_failure_not_open(self, mock_get_names): |
| """Verifies _get_device_type on failure not open.""" |
| mock_get_names.return_value = [] |
| error_msg = f'{_FAKE_DEVICE_ID} is not open' |
| with self.assertRaisesRegex(agent_errors.DeviceNotOpenError, error_msg): |
| self.mgr.get_device_type(_FAKE_DEVICE_ID) |
| |
| @mock.patch.object(gazoo_device.Manager, 'close_open_devices') |
| def test_close_open_devices_on_success(self, mock_close_open_devices): |
| """Verifies close_open_devices on success.""" |
| self.mgr.close_open_devices() |
| self.assertEqual(1, mock_close_open_devices.call_count) |
