3 __author__
= "raphtee@google.com (Travis Miller)"
5 import unittest
, os
, tempfile
, logging
8 from autotest_lib
.server
import autotest
, utils
, hosts
, server_job
, profilers
9 from autotest_lib
.client
.bin
import sysinfo
10 from autotest_lib
.client
.common_lib
import utils
as client_utils
, packages
11 from autotest_lib
.client
.common_lib
import error
12 from autotest_lib
.client
.common_lib
.test_utils
import mock
15 class TestBaseAutotest(unittest
.TestCase
):
18 self
.god
= mock
.mock_god()
20 # create mock host object
21 self
.host
= self
.god
.create_mock_class(hosts
.RemoteHost
, "host")
22 self
.host
.hostname
= "hostname"
23 self
.host
.job
= self
.god
.create_mock_class(server_job
.server_job
,
25 self
.host
.job
.run_test_cleanup
= True
26 self
.host
.job
.last_boot_tag
= 'Autotest'
27 self
.host
.job
.sysinfo
= self
.god
.create_mock_class(
28 sysinfo
.sysinfo
, "sysinfo")
29 self
.host
.job
.profilers
= self
.god
.create_mock_class(
30 profilers
.profilers
, "profilers")
31 self
.host
.job
.profilers
.add_log
= {}
32 self
.host
.job
.tmpdir
= "/job/tmp"
33 self
.host
.job
.default_profile_only
= False
34 self
.host
.job
.args
= []
35 self
.host
.job
.record
= lambda *args
: None
38 self
.god
.stub_function(utils
, "get_server_dir")
39 self
.god
.stub_function(utils
, "run")
40 self
.god
.stub_function(utils
, "get")
41 self
.god
.stub_function(utils
, "read_keyval")
42 self
.god
.stub_function(utils
, "write_keyval")
43 self
.god
.stub_function(utils
, "system")
44 self
.god
.stub_function(tempfile
, "mkstemp")
45 self
.god
.stub_function(tempfile
, "mktemp")
46 self
.god
.stub_function(os
, "getcwd")
47 self
.god
.stub_function(os
, "system")
48 self
.god
.stub_function(os
, "chdir")
49 self
.god
.stub_function(os
, "makedirs")
50 self
.god
.stub_function(os
, "remove")
51 self
.god
.stub_function(os
, "fdopen")
52 self
.god
.stub_function(os
.path
, "exists")
53 self
.god
.stub_function(autotest
, "open")
54 self
.god
.stub_function(autotest
.global_config
.global_config
,
56 self
.god
.stub_function(logging
, "exception")
57 self
.god
.stub_class(autotest
, "_Run")
58 self
.god
.stub_class(autotest
, "log_collector")
67 self
.serverdir
= "serverdir"
70 utils
.get_server_dir
.expect_call().and_return(self
.serverdir
)
72 # create the autotest object
73 self
.base_autotest
= autotest
.BaseAutotest(self
.host
)
74 self
.base_autotest
.job
= self
.host
.job
75 self
.god
.stub_function(self
.base_autotest
, "_install_using_send_file")
78 self
.god
.stub_function(os
.path
, "abspath")
81 self
.god
.check_playback()
84 def record_install_prologue(self
):
88 self
.god
.stub_class(packages
, "PackageManager")
89 self
.base_autotest
.got
= False
90 location
= os
.path
.join(self
.serverdir
, '../client')
91 location
= os
.path
.abspath
.expect_call(location
).and_return(location
)
94 os
.getcwd
.expect_call().and_return('cwd')
95 os
.chdir
.expect_call(os
.path
.join(self
.serverdir
, '../client'))
96 utils
.system
.expect_call('tools/make_clean', ignore_status
=True)
97 os
.chdir
.expect_call('cwd')
98 utils
.get
.expect_call(os
.path
.join(self
.serverdir
,
99 '../client')).and_return('source_material')
101 self
.host
.wait_up
.expect_call(timeout
=30)
102 self
.host
.setup
.expect_call()
103 self
.host
.get_autodir
.expect_call().and_return("autodir")
104 self
.host
.set_autodir
.expect_call("autodir")
105 self
.host
.run
.expect_call('mkdir -p autodir')
106 self
.host
.run
.expect_call('rm -rf autodir/results/*',
110 def test_constructor(self
):
113 # we should check the calls
114 self
.god
.check_playback()
117 def test_full_client_install(self
):
118 self
.record_install_prologue()
120 c
= autotest
.global_config
.global_config
121 c
.get_config_value
.expect_call('PACKAGES',
122 'serve_packages_from_autoserv',
123 type=bool).and_return(False)
124 self
.host
.send_file
.expect_call('source_material', 'autodir',
128 self
.base_autotest
.install_full_client()
129 self
.god
.check_playback()
132 def test_autoserv_install(self
):
133 self
.record_install_prologue()
135 c
= autotest
.global_config
.global_config
136 c
.get_config_value
.expect_call('PACKAGES',
137 'fetch_location', type=list, default
=[]).and_return([])
139 c
.get_config_value
.expect_call('PACKAGES',
140 'serve_packages_from_autoserv',
141 type=bool).and_return(True)
142 self
.base_autotest
._install
_using
_send
_file
.expect_call(self
.host
,
145 self
.base_autotest
.install()
146 self
.god
.check_playback()
149 def test_packaging_install(self
):
150 self
.record_install_prologue()
152 c
= autotest
.global_config
.global_config
153 c
.get_config_value
.expect_call('PACKAGES',
154 'fetch_location', type=list, default
=[]).and_return(['repo'])
155 pkgmgr
= packages
.PackageManager
.expect_new('autodir',
156 repo_urls
=['repo'], hostname
='hostname', do_locking
=False,
157 run_function
=self
.host
.run
, run_function_dargs
=dict(timeout
=600))
158 pkg_dir
= os
.path
.join('autodir', 'packages')
159 cmd
= ('cd autodir && ls | grep -v "^packages$"'
160 ' | xargs rm -rf && rm -rf .[^.]*')
161 self
.host
.run
.expect_call(cmd
)
162 pkgmgr
.install_pkg
.expect_call('autotest', 'client', pkg_dir
,
163 'autodir', preserve_install_dir
=True)
166 self
.base_autotest
.install()
167 self
.god
.check_playback()
177 self
.god
.stub_function(self
.base_autotest
, "install")
178 self
.god
.stub_class(packages
, "PackageManager")
181 self
.base_autotest
.install
.expect_call(self
.host
)
182 self
.host
.wait_up
.expect_call(timeout
=30)
183 os
.path
.abspath
.expect_call('.').and_return('.')
184 run_obj
= autotest
._Run
.expect_new(self
.host
, '.', None, False, False)
186 run_obj
.manual_control_file
= os
.path
.join('autodir',
188 run_obj
.remote_control_file
= os
.path
.join('autodir',
189 'control.%s.autoserv' % tag
)
191 run_obj
.autodir
= 'autodir'
192 run_obj
.verify_machine
.expect_call()
193 run_obj
.verify_machine
.expect_call()
194 run_obj
.background
= False
195 debug
= os
.path
.join('.', 'debug')
196 os
.makedirs
.expect_call(debug
)
197 delete_file_list
= [run_obj
.remote_control_file
,
198 run_obj
.remote_control_file
+ '.state',
199 run_obj
.manual_control_file
,
200 run_obj
.manual_control_file
+ '.state']
201 cmd
= ';'.join('rm -f ' + control
for control
in delete_file_list
)
202 self
.host
.run
.expect_call(cmd
, ignore_status
=True)
204 utils
.get
.expect_call(control
).and_return("temp")
206 c
= autotest
.global_config
.global_config
207 c
.get_config_value
.expect_call("PACKAGES",
208 'fetch_location', type=list).and_return(['repo'])
209 pkgmgr
= packages
.PackageManager
.expect_new('autotest',
213 cfile
= self
.god
.create_mock_class(file, "file")
214 cfile_orig
= "original control file"
215 cfile_new
= "args = []\njob.add_repository(['repo'])\n"
216 cfile_new
+= cfile_orig
218 autotest
.open.expect_call("temp").and_return(cfile
)
219 cfile
.read
.expect_call().and_return(cfile_orig
)
220 autotest
.open.expect_call("temp", 'w').and_return(cfile
)
221 cfile
.write
.expect_call(cfile_new
)
223 self
.host
.job
.preprocess_client_state
.expect_call().and_return(
225 self
.host
.send_file
.expect_call(
226 "/job/tmp/file1", "autodir/control.None.autoserv.init.state")
227 os
.remove
.expect_call("/job/tmp/file1")
229 self
.host
.send_file
.expect_call("temp", run_obj
.remote_control_file
)
230 os
.path
.abspath
.expect_call('temp').and_return('control_file')
231 os
.path
.abspath
.expect_call('control').and_return('control')
232 os
.remove
.expect_call("temp")
234 run_obj
.execute_control
.expect_call(timeout
=30,
235 client_disconnect_timeout
=1800)
237 # run and check output
238 self
.base_autotest
.run(control
, timeout
=30)
239 self
.god
.check_playback()
242 def _stub_get_client_autodir_paths(self
):
243 def mock_get_client_autodir_paths(cls
, host
):
244 return ['/some/path', '/another/path']
245 self
.god
.stub_with(autotest
.Autotest
, 'get_client_autodir_paths',
246 classmethod(mock_get_client_autodir_paths
))
249 def _expect_failed_run(self
, command
):
250 (self
.host
.run
.expect_call(command
)
251 .and_raises(error
.AutoservRunError('dummy', object())))
254 def test_get_installed_autodir(self
):
255 self
._stub
_get
_client
_autodir
_paths
()
256 self
.host
.get_autodir
.expect_call().and_return(None)
257 self
._expect
_failed
_run
('test -x /some/path/bin/autotest')
258 self
.host
.run
.expect_call('test -x /another/path/bin/autotest')
259 self
.host
.run
.expect_call('test -w /another/path')
261 autodir
= autotest
.Autotest
.get_installed_autodir(self
.host
)
262 self
.assertEquals(autodir
, '/another/path')
265 def test_get_install_dir(self
):
266 self
._stub
_get
_client
_autodir
_paths
()
267 self
.host
.get_autodir
.expect_call().and_return(None)
268 self
._expect
_failed
_run
('test -x /some/path/bin/autotest')
269 self
._expect
_failed
_run
('test -x /another/path/bin/autotest')
270 self
._expect
_failed
_run
('mkdir -p /some/path')
271 self
.host
.run
.expect_call('mkdir -p /another/path')
272 self
.host
.run
.expect_call('test -w /another/path')
274 install_dir
= autotest
.Autotest
.get_install_dir(self
.host
)
275 self
.assertEquals(install_dir
, '/another/path')
278 def test_client_logger_process_line_log_copy_collection_failure(self
):
279 collector
= autotest
.log_collector
.expect_new(self
.host
, '', '')
280 logger
= autotest
.client_logger(self
.host
, '', '')
281 collector
.collect_client_job_results
.expect_call().and_raises(
282 Exception('log copy failure'))
283 logging
.exception
.expect_call(mock
.is_string_comparator())
284 logger
._process
_line
('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')
287 def test_client_logger_process_line_log_copy_fifo_failure(self
):
288 collector
= autotest
.log_collector
.expect_new(self
.host
, '', '')
289 logger
= autotest
.client_logger(self
.host
, '', '')
290 collector
.collect_client_job_results
.expect_call()
291 self
.host
.run
.expect_call('echo A > /autotest/fifo2').and_raises(
292 Exception('fifo failure'))
293 logging
.exception
.expect_call(mock
.is_string_comparator())
294 logger
._process
_line
('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')
297 def test_client_logger_process_line_package_install_fifo_failure(self
):
298 collector
= autotest
.log_collector
.expect_new(self
.host
, '', '')
299 logger
= autotest
.client_logger(self
.host
, '', '')
300 self
.god
.stub_function(logger
, '_send_tarball')
302 c
= autotest
.global_config
.global_config
303 c
.get_config_value
.expect_call('PACKAGES',
304 'serve_packages_from_autoserv',
305 type=bool).and_return(True)
306 logger
._send
_tarball
.expect_call('pkgname.tar.bz2', '/autotest/dest/')
308 self
.host
.run
.expect_call('echo B > /autotest/fifo3').and_raises(
309 Exception('fifo failure'))
310 logging
.exception
.expect_call(mock
.is_string_comparator())
311 logger
._process
_line
('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
312 '/autotest/dest/:/autotest/fifo3')
315 class test_autotest_mixin(unittest
.TestCase
):
317 # a dummy Autotest and job class for use in the mixin
318 class stub_autotest(object):
321 def get_state(self
, var
, default
):
322 return self
.state_dict
.get(var
, default
)
326 def run(control_file
, host
=None):
327 self
.control_file
= control_file
330 self
.mixin
= autotest
.AutotestHostMixin()
331 self
.mixin
._Autotest
= stub_autotest
332 self
.job
= self
.mixin
._Autotest
.job
335 def test_passes(self
):
336 self
.job
.state_dict
['test_result'] = True
337 self
.assertEqual(True, self
.mixin
.run_test('sleeptest', seconds
=1))
338 self
.assert_("job.run_test('sleeptest', seconds=1)\n"
339 in self
.control_file
)
340 self
.assertEqual(self
.mixin
, self
.host
)
343 def test_fails_clean(self
):
344 self
.job
.state_dict
['test_result'] = False
345 self
.assertEqual(False, self
.mixin
.run_test('sleeptest', seconds
='2'))
346 self
.assert_("job.run_test('sleeptest', seconds='2')\n"
347 in self
.control_file
)
348 self
.assertEqual(self
.mixin
, self
.host
)
351 def test_fails_with_exception(self
):
352 self
.assertEqual(False, self
.mixin
.run_test('sleeptest'))
353 self
.assert_("job.run_test('sleeptest')\n" in self
.control_file
)
354 self
.assertEqual(self
.mixin
, self
.host
)
357 if __name__
== "__main__":