• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python
2#pylint: disable-msg=C0111
3__author__ = "raphtee@google.com (Travis Miller)"
4
5import unittest, os, tempfile, logging
6
7import common
8from autotest_lib.server import autotest, utils, hosts, server_job, profilers
9from autotest_lib.client.bin import sysinfo
10from autotest_lib.client.common_lib import packages
11from autotest_lib.client.common_lib import error
12from autotest_lib.client.common_lib.test_utils import mock
13
14
15class TestAutotest(unittest.TestCase):
16    def setUp(self):
17        # create god
18        self.god = mock.mock_god()
19
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,
24                                                   "job")
25        self.host.job.run_test_cleanup = True
26        self.host.job.sysinfo = self.god.create_mock_class(
27            sysinfo.sysinfo, "sysinfo")
28        self.host.job.profilers = self.god.create_mock_class(
29            profilers.profilers, "profilers")
30        self.host.job.profilers.add_log = {}
31        self.host.job.tmpdir = "/job/tmp"
32        self.host.job.default_profile_only = False
33        self.host.job.args = []
34        self.host.job.record = lambda *args: None
35        self.host.verify_job_repo_url = lambda *args: None
36
37        # stubs
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,
55                               "get_config_value")
56        self.god.stub_function(logging, "exception")
57        self.god.stub_class(autotest, "_Run")
58        self.god.stub_class(autotest, "log_collector")
59
60
61    def tearDown(self):
62        self.god.unstub_all()
63
64
65    def construct(self):
66        # setup
67        self.serverdir = "serverdir"
68
69        # record
70        utils.get_server_dir.expect_call().and_return(self.serverdir)
71
72        # create the autotest object
73        self.autotest = autotest.Autotest(self.host)
74        self.autotest.job = self.host.job
75        self.god.stub_function(self.autotest, "_install_using_send_file")
76
77        # stub out abspath
78        self.god.stub_function(os.path, "abspath")
79
80        # check
81        self.god.check_playback()
82
83
84    def record_install_prologue(self):
85        self.construct()
86
87        # setup
88        self.god.stub_class(packages, "PackageManager")
89        self.autotest.got = False
90        location = os.path.join(self.serverdir, '../client')
91        location = os.path.abspath.expect_call(location).and_return(location)
92
93        # record
94        utils.get.expect_call(os.path.join(self.serverdir,
95            '../client')).and_return('source_material')
96
97        self.host.wait_up.expect_call(timeout=30)
98        self.host.setup.expect_call()
99        self.host.get_autodir.expect_call().and_return("autodir")
100        self.host.set_autodir.expect_call("autodir")
101        self.host.run.expect_call('mkdir -p autodir')
102        self.host.run.expect_call('rm -rf autodir/results/*',
103                                  ignore_status=True)
104
105
106    def test_constructor(self):
107        self.construct()
108
109        # we should check the calls
110        self.god.check_playback()
111
112
113    def test_full_client_install(self):
114        self.record_install_prologue()
115
116        self.host.run.expect_call('rm -f "autodir/packages.checksum"')
117        c = autotest.global_config.global_config
118        c.get_config_value.expect_call('PACKAGES',
119                                       'serve_packages_from_autoserv',
120                                       type=bool).and_return(False)
121        self.host.send_file.expect_call('source_material', 'autodir',
122                                        delete_dest=True)
123        self.god.stub_function(autotest.Autotest, "_send_shadow_config")
124        autotest.Autotest._send_shadow_config.expect_call()
125
126        # run and check
127        self.autotest.install_full_client()
128        self.god.check_playback()
129
130
131    def test_autoserv_install(self):
132        self.record_install_prologue()
133
134        c = autotest.global_config.global_config
135        c.get_config_value.expect_call('PACKAGES',
136            'fetch_location', type=list, default=[]).and_return([])
137
138        os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
139        c.get_config_value.expect_call('PACKAGES',
140                                       'serve_packages_from_autoserv',
141                                       type=bool).and_return(True)
142        self.autotest._install_using_send_file.expect_call(self.host,
143                                                           'autodir')
144        self.god.stub_function(autotest.Autotest, "_send_shadow_config")
145        autotest.Autotest._send_shadow_config.expect_call()
146        # run and check
147        self.autotest.install()
148        self.god.check_playback()
149
150
151    def test_packaging_install(self):
152        self.record_install_prologue()
153
154        c = autotest.global_config.global_config
155        c.get_config_value.expect_call('PACKAGES',
156            'fetch_location', type=list, default=[]).and_return(['repo'])
157        os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
158        pkgmgr = packages.PackageManager.expect_new('autodir',
159            repo_urls=['repo'], hostname='hostname', do_locking=False,
160            run_function=self.host.run, run_function_dargs=dict(timeout=600))
161        pkg_dir = os.path.join('autodir', 'packages')
162        cmd = ('cd autodir && ls | grep -v "^packages$" | '
163               'grep -v "^result_tools$" | '
164               'xargs rm -rf && rm -rf .[!.]*')
165        self.host.run.expect_call(cmd)
166        pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,
167                                       'autodir', preserve_install_dir=True)
168
169        # run and check
170        self.autotest.install()
171        self.god.check_playback()
172
173
174    def test_run(self):
175        self.construct()
176
177        # setup
178        control = "control"
179
180        # stub out install
181        self.god.stub_function(self.autotest, "install")
182
183        # record
184        self.autotest.install.expect_call(self.host, use_packaging=True)
185        self.host.wait_up.expect_call(timeout=30)
186        os.path.abspath.expect_call('.').and_return('.')
187        run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)
188        tag = None
189        run_obj.manual_control_file = os.path.join('autodir',
190                                                   'control.%s' % tag)
191        run_obj.remote_control_file = os.path.join('autodir',
192                                                   'control.%s.autoserv' % tag)
193        run_obj.tag = tag
194        run_obj.autodir = 'autodir'
195        run_obj.verify_machine.expect_call()
196        run_obj.background = False
197        debug = os.path.join('.', 'debug')
198        os.makedirs.expect_call(debug)
199        delete_file_list = [run_obj.remote_control_file,
200                            run_obj.remote_control_file + '.state',
201                            run_obj.manual_control_file,
202                            run_obj.manual_control_file + '.state']
203        cmd = ';'.join('rm -f ' + control for control in delete_file_list)
204        self.host.run.expect_call(cmd, ignore_status=True)
205
206        utils.get.expect_call(control, local_copy=True).and_return("temp")
207
208        c = autotest.global_config.global_config
209        c.get_config_value.expect_call("PACKAGES",
210            'fetch_location', type=list, default=[]).and_return(['repo'])
211
212        cfile = self.god.create_mock_class(file, "file")
213        cfile_orig = "original control file"
214        cfile_new = "args = []\njob.add_repository(['repo'])\n"
215        cfile_new += cfile_orig
216
217        os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
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)
222
223        self.host.job.preprocess_client_state.expect_call().and_return(
224            '/job/tmp/file1')
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")
228
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")
233
234        run_obj.execute_control.expect_call(timeout=30,
235                                            client_disconnect_timeout=240)
236
237        # run and check output
238        self.autotest.run(control, timeout=30)
239        self.god.check_playback()
240
241
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))
247
248
249    def _expect_failed_run(self, command):
250        (self.host.run.expect_call(command)
251         .and_raises(error.AutoservRunError('dummy', object())))
252
253
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')
260
261        autodir = autotest.Autotest.get_installed_autodir(self.host)
262        self.assertEquals(autodir, '/another/path')
263
264
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')
273
274        install_dir = autotest.Autotest.get_install_dir(self.host)
275        self.assertEquals(install_dir, '/another/path')
276
277
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')
285
286
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')
295
296
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')
301
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        c.get_config_value.expect_call('PACKAGES',
307                                       'serve_packages_from_autoserv',
308                                       type=bool).and_return(True)
309        logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')
310
311        self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(
312                Exception('fifo failure'))
313        logging.exception.expect_call(mock.is_string_comparator())
314        logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
315                             '/autotest/dest/:/autotest/fifo3')
316
317
318if __name__ == "__main__":
319    unittest.main()
320