xref: /aosp_15_r20/external/autotest/server/hosts/cros_host_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7) 
1#!/usr/bin/python3
2# pylint: disable=missing-docstring
3
4import unittest
5from unittest import mock
6
7import common
8
9from autotest_lib.client.common_lib import error
10from autotest_lib.server.cros.servo import servo
11from autotest_lib.server.hosts import cros_host
12from autotest_lib.server.hosts import servo_constants
13from autotest_lib.server.hosts import host_info
14
15CROSSYSTEM_RESULT = '''
16fwb_tries              = 0                              # Fake comment
17fw_vboot2              = 1                              # Fake comment
18fwid                   = Google_Reef.9933.0.0           # Fake comment
19fwupdate_tries         = 0                              #
20fw_tried               = B                              #
21fw_try_count           = 0                              #
22'''
23
24NON_UNI_LSB_RELEASE_OUTPUT = '''
25CHROMEOS_RELEASE_BOARD=reef
26'''
27
28UNI_LSB_RELEASE_OUTPUT = '''
29CHROMEOS_RELEASE_BOARD=coral
30CHROMEOS_RELEASE_UNIBUILD=1
31'''
32
33SERVO_STATE_PREFIX = servo_constants.SERVO_STATE_LABEL_PREFIX
34
35
36class MockCmd(object):
37    """Simple mock command with base command and results"""
38
39    def __init__(self, cmd, exit_status, stdout):
40        self.cmd = cmd
41        self.stdout = stdout
42        self.exit_status = exit_status
43
44
45class MockHost(cros_host.CrosHost):
46    """Simple host for running mock'd host commands"""
47
48    def __init__(self, *args):
49        self._mock_cmds = {c.cmd: c for c in args}
50        self.hostname = 'MockHost'
51
52    def run(self, command, **kwargs):
53        """Finds the matching result by command value"""
54        mock_cmd = self._mock_cmds[command]
55        file_out = kwargs.get('stdout_tee', None)
56        if file_out:
57            file_out.write(mock_cmd.stdout)
58        return mock_cmd
59
60
61class GetPlatformModelTests(unittest.TestCase):
62    """Unit tests for CrosHost.get_platform_model"""
63
64    def test_cros_config_succeeds(self):
65        host = MockHost(
66                MockCmd('cat /etc/lsb-release', 0, UNI_LSB_RELEASE_OUTPUT),
67                MockCmd('cros_config / name', 0, 'coral'))
68        self.assertEqual(host.get_platform(), 'coral')
69
70    def test_cros_config_resorts_to_fallback(self):
71        host = MockHost(
72                MockCmd('cat /etc/lsb-release', 0, UNI_LSB_RELEASE_OUTPUT),
73                MockCmd('cros_config / name', 1, ''),
74                MockCmd('mosys platform model', 0, 'coral'))
75        self.assertEqual(host.get_platform(), 'coral')
76
77    def test_cros_config_fails(self):
78        host = MockHost(
79                MockCmd('cat /etc/lsb-release', 0, UNI_LSB_RELEASE_OUTPUT),
80                MockCmd('cros_config / name', 1, ''),
81                MockCmd('mosys platform model', 1, ''),
82                MockCmd('crossystem', 0, CROSSYSTEM_RESULT))
83        self.assertEqual(host.get_platform(), 'reef')
84
85    def test_non_unibuild(self):
86        host = MockHost(
87                MockCmd('cat /etc/lsb-release', 0, NON_UNI_LSB_RELEASE_OUTPUT),
88                MockCmd('crossystem', 0, CROSSYSTEM_RESULT))
89        self.assertEqual(host.get_platform(), 'reef')
90
91    def test_cat_lsb_fails(self):
92        host = MockHost(
93                MockCmd('cat /etc/lsb-release', 1, ''),
94                MockCmd('crossystem', 0, CROSSYSTEM_RESULT))
95        self.assertEqual(host.get_platform(), 'reef')
96
97
98class DictFilteringTestCase(unittest.TestCase):
99    """Tests for dict filtering methods on CrosHost."""
100
101    def test_get_chameleon_arguments(self):
102        got = cros_host.CrosHost.get_chameleon_arguments({
103            'chameleon_host': 'host',
104            'spam': 'eggs',
105        })
106        self.assertEqual(got, {'chameleon_host': 'host'})
107
108    def test_get_pdtester_arguments(self):
109        got = cros_host.CrosHost.get_pdtester_arguments({
110            'pdtester_host': 'host',
111            'spam': 'eggs',
112        })
113        self.assertEqual(got, {'pdtester_host': 'host'})
114
115    def test_get_servo_arguments(self):
116        got = cros_host.CrosHost.get_servo_arguments({
117            servo_constants.SERVO_HOST_ATTR: 'host',
118            'spam': 'eggs',
119        })
120        self.assertEqual(got, {servo_constants.SERVO_HOST_ATTR: 'host'})
121
122
123class DictFilteringTestCase(unittest.TestCase):
124    """Test to verify servo_state was set-up as label in host_info_store"""
125
126    def create_host(self):
127        host = MockHost()
128        host.servo = None
129        host._servo_host = mock.Mock()
130        servo = mock.Mock()
131        servo.get_servo_type.return_value = None
132        host._servo_host.get_servo.return_value = servo
133        host._servo_host.get_servo_state.return_value = 'SOME_STATE'
134        host.host_info_store = host_info.InMemoryHostInfoStore()
135        return host
136
137    def test_do_not_update_label_when_servo_host_is_not_inited(self):
138        host = self.create_host()
139        host._servo_host = None
140
141        host.set_servo_state('some_status')
142        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'some_status')
143
144    def test_do_not_update_label_when_servo_state_is_None(self):
145        host = self.create_host()
146
147        host.set_servo_state(None)
148        host._servo_host.get_servo_state.assert_not_called()
149        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), '')
150
151    def test_repair_servo_set_servo_state_after_repair_when_repair_is_fail(self):
152        host = self.create_host()
153        host._servo_host.repair.side_effect = Exception('Something bad')
154
155        try:
156            host.repair_servo()
157            self.assertEqual("Exception is", 'expecting to raise')
158        except:
159            pass
160        host._servo_host.get_servo_state.assert_called()
161        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'SOME_STATE')
162
163    def test_repair_servo_set_servo_state_after_repair_when_repair_is_not_fail(self):
164        host = self.create_host()
165        try:
166            host.repair_servo()
167        except:
168            self.assertEqual("Exception is not", 'expected')
169            pass
170        host._servo_host.get_servo_state.assert_called()
171        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'SOME_STATE')
172
173    def test_set_servo_host_update_servo_state_when_host_exist(self):
174        host = self.create_host()
175        host._servo_host = mock.Mock()
176        servo = mock.Mock()
177        servo.get_servo_type.return_value = None
178        host._servo_host.get_servo.return_value = servo
179        host._servo_host.get_servo_state.return_value = 'SOME_STATE'
180        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), '')
181
182        try:
183            host.repair_servo()
184        except:
185            self.assertEqual("Exception is not", 'expected')
186            pass
187        host._servo_host.get_servo_state.assert_called()
188        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'SOME_STATE')
189
190    def test_set_servo_host_use_passed_servo_state_when_host_is_None(self):
191        host = self.create_host()
192
193        host.set_servo_host(None, 'passed_State')
194        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'passed_State')
195
196    def test_set_servo_host_use_servo_state_from_host_when_host_is_passed(self):
197        host = self.create_host()
198        servo_host = mock.Mock()
199        servo = mock.Mock()
200        servo.get_servo_type.return_value = None
201        servo_host.get_servo.return_value = servo
202        servo_host.get_servo_state.return_value = 'state_of_host'
203
204        host.set_servo_host(servo_host)
205        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'state_of_host')
206
207        servo_host.get_servo_state.return_value = 'state_of_host2'
208        host.set_servo_host(servo_host, 'passed_State')
209        self.assertEqual(host.host_info_store.get().get_label_value(SERVO_STATE_PREFIX), 'state_of_host2')
210
211class CrosHostTestCase(unittest.TestCase):
212    """Tests to verify CrosHost."""
213
214    class TestCrosHost(cros_host.CrosHost):
215        def __init__(self, *args, **kwargs):
216            self.hostname = 'hostname'
217            self.servo = mock.create_autospec(servo.Servo)
218
219    @mock.patch('autotest_lib.server.hosts.cros_host.dev_server')
220    def test_stage_build_to_usb(self, devserver_mock):
221        host = self.TestCrosHost()
222        image_server = mock.MagicMock()
223        devserver_mock.ImageServer.resolve.return_value = image_server
224        image_server.get_test_image_url.return_value = 'image_url'
225
226        host.stage_build_to_usb('board/version')
227
228        image_server.stage_artifacts.assert_called_with('board/version', ['test_image'])
229        host.servo.image_to_servo_usb.assert_called_with('image_url')
230
231        host.servo.get_power_state_controller.return_value.power_on.assert_called()
232
233    @mock.patch('autotest_lib.server.hosts.cros_host.dev_server')
234    def test_stage_build_to_usb_failure(self, devserver_mock):
235        host = self.TestCrosHost()
236        image_server = mock.MagicMock()
237        devserver_mock.ImageServer.resolve.return_value = image_server
238        image_server.get_test_image_url.return_value = 'image_url'
239        host.servo.image_to_servo_usb.side_effect = error.AutotestError('download')
240
241        with self.assertRaises(error.AutotestError):
242            host.stage_build_to_usb('board/version')
243
244        host.servo.get_power_state_controller.return_value.power_on.assert_called()
245
246
247if __name__ == "__main__":
248    unittest.main()
249