1 #!/usr/bin/python 2 3 import mock 4 import os 5 import unittest 6 7 import common 8 from autotest_lib.client.common_lib import autotemp 9 from autotest_lib.client.common_lib import error 10 from autotest_lib.client.bin import local_host 11 12 13 class 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 305 if __name__ == "__main__": 306 unittest.main() 307
