xref: /aosp_15_r20/external/autotest/client/bin/local_host_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7) 
1#!/usr/bin/python3
2
3import os
4import six
5import unittest
6from unittest import mock
7
8import common
9from autotest_lib.client.common_lib import autotemp
10from autotest_lib.client.common_lib import error
11from autotest_lib.client.bin import local_host
12
13
14class test_local_host_class(unittest.TestCase):
15    def setUp(self):
16        self.tmpdir = autotemp.tempdir(unique_id='localhost_unittest')
17        self.addCleanup(self.tmpdir.clean)
18
19
20    @mock.patch('autotest_lib.client.bin.local_host.platform.node')
21    def test_init_default_hostname(self, mock_node):
22        mock_node.return_value = 'foo'
23        host = local_host.LocalHost()
24        self.assertEqual(host.hostname, 'foo')
25
26
27    @mock.patch('autotest_lib.client.bin.local_host.platform.node')
28    def test_init_with_hostname(self, mock_node):
29        mock_node.return_value = 'foo'
30        host = local_host.LocalHost(hostname='bar')
31        self.assertEqual(host.hostname, 'bar')
32
33
34    def test_wait_up(self):
35        # just test that wait_up always works
36        host = local_host.LocalHost()
37        host.wait_up(1)
38
39
40    @mock.patch('autotest_lib.client.bin.local_host.utils.run')
41    def test_run_success(self, mock_run):
42        result = local_host.utils.CmdResult(
43                command='yes',
44                stdout='y',
45                stderr='',
46                exit_status=0,
47                duration=1,
48        )
49        mock_run.return_value = result
50
51        host = local_host.LocalHost()
52        got = host.run(
53                'yes',
54                timeout=123,
55                ignore_status=True,
56                stdout_tee=local_host.utils.TEE_TO_LOGS,
57                stderr_tee=local_host.utils.TEE_TO_LOGS,
58                stdin=None,
59        )
60
61        self.assertEqual(got, result)
62        mock_run.assert_called_once_with(
63                result.command,
64                timeout=123,
65                ignore_status=True,
66                stdout_tee=local_host.utils.TEE_TO_LOGS,
67                stderr_tee=local_host.utils.TEE_TO_LOGS,
68                stdin=None,
69                ignore_timeout=False,
70                args=(),
71        )
72
73
74    @mock.patch('autotest_lib.client.bin.local_host.utils.run')
75    def test_run_cmd_failure_raised(self, mock_run):
76        mock_result = mock.MagicMock()
77        mock_run.side_effect = error.CmdError('yes', mock_result)
78
79        host = local_host.LocalHost()
80        with self.assertRaises(error.AutotestHostRunCmdError) as exc_cm:
81            host.run('yes', timeout=123)
82
83        self.assertEqual(exc_cm.exception.result_obj, mock_result)
84        mock_run.assert_called_once_with(
85                'yes',
86                timeout=123,
87                ignore_status=False,
88                stdout_tee=local_host.utils.TEE_TO_LOGS,
89                stderr_tee=local_host.utils.TEE_TO_LOGS,
90                stdin=None,
91                ignore_timeout=False,
92                args=(),
93        )
94
95
96    @mock.patch('autotest_lib.client.bin.local_host.utils.run')
97    def test_run_cmd_timeout_raised(self, mock_run):
98        mock_result = mock.MagicMock()
99        mock_run.side_effect = error.CmdTimeoutError('yes', mock_result)
100
101        host = local_host.LocalHost()
102        with self.assertRaises(error.AutotestHostRunTimeoutError) as exc_cm:
103            host.run('yes', timeout=123)
104
105        self.assertEqual(exc_cm.exception.result_obj, mock_result)
106        mock_run.assert_called_once_with(
107                'yes',
108                timeout=123,
109                ignore_status=False,
110                stdout_tee=local_host.utils.TEE_TO_LOGS,
111                stderr_tee=local_host.utils.TEE_TO_LOGS,
112                stdin=None,
113                ignore_timeout=False,
114                args=(),
115        )
116
117
118    @mock.patch('autotest_lib.client.bin.local_host.utils.run')
119    def test_run_failure_ignored(self, mock_run):
120        result = local_host.utils.CmdResult(
121                command='yes',
122                stdout='',
123                stderr='err',
124                exit_status=1,
125                duration=1,
126        )
127        mock_run.return_value = result
128
129        host = local_host.LocalHost()
130        got = host.run('yes', timeout=123, ignore_status=True)
131
132        self.assertEqual(got, result)
133        mock_run.assert_called_once_with(
134                result.command,
135                timeout=123,
136                ignore_status=True,
137                stdout_tee=local_host.utils.TEE_TO_LOGS,
138                stderr_tee=local_host.utils.TEE_TO_LOGS,
139                stdin=None,
140                ignore_timeout=False,
141                args=(),
142        )
143
144
145    def test_list_files_glob(self):
146        host = local_host.LocalHost()
147
148        files = (os.path.join(self.tmpdir.name, 'file1'),
149                 os.path.join(self.tmpdir.name, 'file2'))
150
151        # create some files in tmpdir
152        open(files[0], 'w').close()
153        open(files[1], 'w').close()
154        if six.PY2:
155            self.assertItemsEqual(
156                    files,
157                    host.list_files_glob(os.path.join(self.tmpdir.name, '*')))
158        else:
159            self.assertCountEqual(
160                    files,
161                    host.list_files_glob(os.path.join(self.tmpdir.name, '*')))
162
163
164    def test_symlink_closure_does_not_add_existent_file(self):
165        host = local_host.LocalHost()
166
167        # create a file and a symlink to it
168        fname = os.path.join(self.tmpdir.name, 'file')
169        sname = os.path.join(self.tmpdir.name, 'sym')
170        open(fname, 'w').close()
171        os.symlink(fname, sname)
172
173        # test that when the symlinks point to already know files
174        # nothing is added
175        if six.PY2:
176            self.assertItemsEqual([fname, sname],
177                                  host.symlink_closure([fname, sname]))
178        else:
179            self.assertCountEqual([fname, sname],
180                                  host.symlink_closure([fname, sname]))
181
182
183    def test_symlink_closure_adds_missing_files(self):
184        host = local_host.LocalHost()
185
186        # create a file and a symlink to it
187        fname = os.path.join(self.tmpdir.name, 'file')
188        sname = os.path.join(self.tmpdir.name, 'sym')
189        open(fname, 'w').close()
190        os.symlink(fname, sname)
191
192        # test that when the symlinks point to unknown files they are added
193        if six.PY2:
194            self.assertItemsEqual([fname, sname],
195                                  host.symlink_closure([sname]))
196        else:
197            self.assertCountEqual([fname, sname],
198                                  host.symlink_closure([sname]))
199
200
201    def test_get_file(self):
202        """Tests get_file() copying a regular file."""
203        host = local_host.LocalHost()
204
205        source_file = os.path.join(self.tmpdir.name, 'file')
206        open(os.path.join(source_file), 'w').close()
207
208        dest_file = os.path.join(self.tmpdir.name, 'dest')
209
210        host.get_file(source_file, dest_file)
211        self.assertTrue(os.path.isfile(dest_file))
212
213
214    def test_get_directory_into_new_directory(self):
215        """Tests get_file() copying a directory into a new dir."""
216        host = local_host.LocalHost()
217
218        source_dir = os.path.join(self.tmpdir.name, 'dir')
219        os.mkdir(source_dir)
220        open(os.path.join(source_dir, 'file'), 'w').close()
221
222        dest_dir = os.path.join(self.tmpdir.name, 'dest')
223
224        host.get_file(source_dir, dest_dir)
225
226        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
227
228
229    def test_get_directory_into_existing_directory(self):
230        """Tests get_file() copying a directory into an existing dir."""
231        host = local_host.LocalHost()
232
233        source_dir = os.path.join(self.tmpdir.name, 'dir')
234        os.mkdir(source_dir)
235        open(os.path.join(source_dir, 'file'), 'w').close()
236
237        dest_dir = os.path.join(self.tmpdir.name, 'dest')
238        os.mkdir(dest_dir)
239
240        host.get_file(source_dir, dest_dir)
241
242        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
243
244
245    def test_get_directory_delete_dest(self):
246        """Tests get_file() replacing a dir."""
247        host = local_host.LocalHost()
248
249        source_dir = os.path.join(self.tmpdir.name, 'dir')
250        os.mkdir(source_dir)
251        open(os.path.join(source_dir, 'file'), 'w').close()
252
253        dest_dir = os.path.join(self.tmpdir.name, 'dest')
254        os.mkdir(dest_dir)
255        os.mkdir(os.path.join(dest_dir, 'dir'))
256        open(os.path.join(dest_dir, 'dir', 'file2'), 'w').close()
257
258        host.get_file(source_dir, dest_dir, delete_dest=True)
259
260        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
261        self.assertFalse(os.path.isfile(os.path.join(dest_dir, 'dir', 'file2')))
262
263
264    def test_get_directory_contents_into_new_directory(self):
265        """Tests get_file() copying dir contents to a new dir."""
266        host = local_host.LocalHost()
267
268        source_dir = os.path.join(self.tmpdir.name, 'dir')
269        os.mkdir(source_dir)
270        open(os.path.join(source_dir, 'file'), 'w').close()
271
272        dest_dir = os.path.join(self.tmpdir.name, 'dest')
273
274        # End the source with '/' to copy the contents only.
275        host.get_file(os.path.join(source_dir, ''), dest_dir)
276
277        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
278
279
280    def test_get_directory_contents_into_existing_directory(self):
281        """Tests get_file() copying dir contents into an existing dir."""
282        host = local_host.LocalHost()
283
284        source_dir = os.path.join(self.tmpdir.name, 'dir')
285        os.mkdir(source_dir)
286        open(os.path.join(source_dir, 'file'), 'w').close()
287
288        dest_dir = os.path.join(self.tmpdir.name, 'dest')
289        os.mkdir(dest_dir)
290
291        # End the source with '/' to copy the contents only.
292        host.get_file(os.path.join(source_dir, ''), dest_dir)
293
294        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
295
296
297    def test_get_directory_contents_delete_dest(self):
298        """Tests get_file() replacing contents of a dir."""
299        host = local_host.LocalHost()
300
301        source_dir = os.path.join(self.tmpdir.name, 'dir')
302        os.mkdir(source_dir)
303        open(os.path.join(source_dir, 'file'), 'w').close()
304
305        dest_dir = os.path.join(self.tmpdir.name, 'dest')
306        os.mkdir(dest_dir)
307        open(os.path.join(dest_dir, 'file2'), 'w').close()
308
309        # End the source with '/' to copy the contents only.
310        host.get_file(os.path.join(source_dir, ''), dest_dir, delete_dest=True)
311
312        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
313        self.assertFalse(os.path.isfile(os.path.join(dest_dir, 'file2')))
314
315
316if __name__ == "__main__":
317    unittest.main()
318