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