1#!/usr/bin/python
2
3import os
4import unittest
5
6import common
7from autotest_lib.client.common_lib.test_utils import mock
8from autotest_lib.client.common_lib import autotemp
9from autotest_lib.client.bin import local_host
10
11
12class test_local_host_class(unittest.TestCase):
13    def setUp(self):
14        self.god = mock.mock_god()
15        self.god.stub_function(local_host.utils, 'run')
16
17        self.tmpdir = autotemp.tempdir(unique_id='localhost_unittest')
18
19
20    def tearDown(self):
21        self.god.unstub_all()
22        self.tmpdir.clean()
23
24
25    def test_init(self):
26        self.god.stub_function(local_host.platform, 'node')
27        local_host.platform.node.expect_call().and_return('foo')
28
29        # run the actual test
30        host = local_host.LocalHost()
31        self.assertEqual(host.hostname, 'foo')
32        self.god.check_playback()
33
34        host = local_host.LocalHost(hostname='bar')
35        self.assertEqual(host.hostname, 'bar')
36        self.god.check_playback()
37
38
39    def test_wait_up(self):
40        # just test that wait_up always works
41        host = local_host.LocalHost()
42        host.wait_up(1)
43        self.god.check_playback()
44
45
46    def _setup_run(self, result):
47        host = local_host.LocalHost()
48
49        (local_host.utils.run.expect_call(result.command, timeout=123,
50                ignore_status=True, stdout_tee=local_host.utils.TEE_TO_LOGS,
51                stderr_tee=local_host.utils.TEE_TO_LOGS, stdin=None,
52                ignore_timeout=False, args=())
53                .and_return(result))
54
55        return host
56
57
58    def test_run_success(self):
59        result = local_host.utils.CmdResult(command='yes', stdout='y',
60                stderr='', exit_status=0, duration=1)
61
62        host = self._setup_run(result)
63
64        self.assertEqual(host.run('yes', timeout=123, ignore_status=True,
65                stdout_tee=local_host.utils.TEE_TO_LOGS,
66                stderr_tee=local_host.utils.TEE_TO_LOGS, stdin=None), result)
67        self.god.check_playback()
68
69
70    def test_run_failure_raised(self):
71        result = local_host.utils.CmdResult(command='yes', stdout='',
72                stderr='err', exit_status=1, duration=1)
73
74        host = self._setup_run(result)
75
76        self.assertRaises(local_host.error.AutotestHostRunError, host.run,
77                          'yes', timeout=123)
78        self.god.check_playback()
79
80
81    def test_run_failure_ignored(self):
82        result = local_host.utils.CmdResult(command='yes', stdout='',
83                stderr='err', exit_status=1, duration=1)
84
85        host = self._setup_run(result)
86
87        self.assertEqual(host.run('yes', timeout=123, ignore_status=True),
88                         result)
89        self.god.check_playback()
90
91
92    def test_list_files_glob(self):
93        host = local_host.LocalHost()
94
95        files = (os.path.join(self.tmpdir.name, 'file1'),
96                 os.path.join(self.tmpdir.name, 'file2'))
97
98        # create some files in tmpdir
99        open(files[0], 'w').close()
100        open(files[1], 'w').close()
101
102        self.assertItemsEqual(
103                files,
104                host.list_files_glob(os.path.join(self.tmpdir.name, '*')))
105
106
107    def test_symlink_closure_does_not_add_existent_file(self):
108        host = local_host.LocalHost()
109
110        # create a file and a symlink to it
111        fname = os.path.join(self.tmpdir.name, 'file')
112        sname = os.path.join(self.tmpdir.name, 'sym')
113        open(fname, 'w').close()
114        os.symlink(fname, sname)
115
116        # test that when the symlinks point to already know files
117        # nothing is added
118        self.assertItemsEqual(
119                [fname, sname],
120                host.symlink_closure([fname, sname]))
121
122
123    def test_symlink_closure_adds_missing_files(self):
124        host = local_host.LocalHost()
125
126        # create a file and a symlink to it
127        fname = os.path.join(self.tmpdir.name, 'file')
128        sname = os.path.join(self.tmpdir.name, 'sym')
129        open(fname, 'w').close()
130        os.symlink(fname, sname)
131
132        # test that when the symlinks point to unknown files they are added
133        self.assertItemsEqual(
134                [fname, sname],
135                host.symlink_closure([sname]))
136
137
138    def test_get_file(self):
139        """Tests get_file() copying a regular file."""
140        host = local_host.LocalHost()
141
142        source_file = os.path.join(self.tmpdir.name, 'file')
143        open(os.path.join(source_file), 'w').close()
144
145        dest_file = os.path.join(self.tmpdir.name, 'dest')
146
147        host.get_file(source_file, dest_file)
148        self.assertTrue(os.path.isfile(dest_file))
149
150
151    def test_get_directory_into_new_directory(self):
152        """Tests get_file() copying a directory into a new dir."""
153        host = local_host.LocalHost()
154
155        source_dir = os.path.join(self.tmpdir.name, 'dir')
156        os.mkdir(source_dir)
157        open(os.path.join(source_dir, 'file'), 'w').close()
158
159        dest_dir = os.path.join(self.tmpdir.name, 'dest')
160
161        host.get_file(source_dir, dest_dir)
162
163        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
164
165
166    def test_get_directory_into_existing_directory(self):
167        """Tests get_file() copying a directory into an existing dir."""
168        host = local_host.LocalHost()
169
170        source_dir = os.path.join(self.tmpdir.name, 'dir')
171        os.mkdir(source_dir)
172        open(os.path.join(source_dir, 'file'), 'w').close()
173
174        dest_dir = os.path.join(self.tmpdir.name, 'dest')
175        os.mkdir(dest_dir)
176
177        host.get_file(source_dir, dest_dir)
178
179        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
180
181
182    def test_get_directory_delete_dest(self):
183        """Tests get_file() replacing a dir."""
184        host = local_host.LocalHost()
185
186        source_dir = os.path.join(self.tmpdir.name, 'dir')
187        os.mkdir(source_dir)
188        open(os.path.join(source_dir, 'file'), 'w').close()
189
190        dest_dir = os.path.join(self.tmpdir.name, 'dest')
191        os.mkdir(dest_dir)
192        os.mkdir(os.path.join(dest_dir, 'dir'))
193        open(os.path.join(dest_dir, 'dir', 'file2'), 'w').close()
194
195        host.get_file(source_dir, dest_dir, delete_dest=True)
196
197        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'dir', 'file')))
198        self.assertFalse(os.path.isfile(os.path.join(dest_dir, 'dir', 'file2')))
199
200
201    def test_get_directory_contents_into_new_directory(self):
202        """Tests get_file() copying dir contents to a new dir."""
203        host = local_host.LocalHost()
204
205        source_dir = os.path.join(self.tmpdir.name, 'dir')
206        os.mkdir(source_dir)
207        open(os.path.join(source_dir, 'file'), 'w').close()
208
209        dest_dir = os.path.join(self.tmpdir.name, 'dest')
210
211        # End the source with '/' to copy the contents only.
212        host.get_file(os.path.join(source_dir, ''), dest_dir)
213
214        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
215
216
217    def test_get_directory_contents_into_existing_directory(self):
218        """Tests get_file() copying dir contents into an existing dir."""
219        host = local_host.LocalHost()
220
221        source_dir = os.path.join(self.tmpdir.name, 'dir')
222        os.mkdir(source_dir)
223        open(os.path.join(source_dir, 'file'), 'w').close()
224
225        dest_dir = os.path.join(self.tmpdir.name, 'dest')
226        os.mkdir(dest_dir)
227
228        # End the source with '/' to copy the contents only.
229        host.get_file(os.path.join(source_dir, ''), dest_dir)
230
231        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
232
233
234    def test_get_directory_contents_delete_dest(self):
235        """Tests get_file() replacing contents of 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        open(os.path.join(dest_dir, 'file2'), 'w').close()
245
246        # End the source with '/' to copy the contents only.
247        host.get_file(os.path.join(source_dir, ''), dest_dir, delete_dest=True)
248
249        self.assertTrue(os.path.isfile(os.path.join(dest_dir, 'file')))
250        self.assertFalse(os.path.isfile(os.path.join(dest_dir, 'file2')))
251
252
253if __name__ == "__main__":
254    unittest.main()
255