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 TestBaseAutotest(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
36        # stubs
37        self.god.stub_function(utils, "get_server_dir")
38        self.god.stub_function(utils, "run")
39        self.god.stub_function(utils, "get")
40        self.god.stub_function(utils, "read_keyval")
41        self.god.stub_function(utils, "write_keyval")
42        self.god.stub_function(utils, "system")
43        self.god.stub_function(tempfile, "mkstemp")
44        self.god.stub_function(tempfile, "mktemp")
45        self.god.stub_function(os, "getcwd")
46        self.god.stub_function(os, "system")
47        self.god.stub_function(os, "chdir")
48        self.god.stub_function(os, "makedirs")
49        self.god.stub_function(os, "remove")
50        self.god.stub_function(os, "fdopen")
51        self.god.stub_function(os.path, "exists")
52        self.god.stub_function(autotest, "open")
53        self.god.stub_function(autotest.global_config.global_config,
54                               "get_config_value")
55        self.god.stub_function(logging, "exception")
56        self.god.stub_class(autotest, "_Run")
57        self.god.stub_class(autotest, "log_collector")
58
59
60    def tearDown(self):
61        self.god.unstub_all()
62
63
64    def construct(self):
65        # setup
66        self.serverdir = "serverdir"
67
68        # record
69        utils.get_server_dir.expect_call().and_return(self.serverdir)
70
71        # create the autotest object
72        self.base_autotest = autotest.BaseAutotest(self.host)
73        self.base_autotest.job = self.host.job
74        self.god.stub_function(self.base_autotest, "_install_using_send_file")
75
76        # stub out abspath
77        self.god.stub_function(os.path, "abspath")
78
79        # check
80        self.god.check_playback()
81
82
83    def record_install_prologue(self):
84        self.construct()
85
86        # setup
87        self.god.stub_class(packages, "PackageManager")
88        self.base_autotest.got = False
89        location = os.path.join(self.serverdir, '../client')
90        location = os.path.abspath.expect_call(location).and_return(location)
91
92        # record
93        os.getcwd.expect_call().and_return('cwd')
94        os.chdir.expect_call(os.path.join(self.serverdir, '../client'))
95        utils.system.expect_call('tools/make_clean', ignore_status=True)
96        os.chdir.expect_call('cwd')
97        utils.get.expect_call(os.path.join(self.serverdir,
98            '../client')).and_return('source_material')
99
100        self.host.wait_up.expect_call(timeout=30)
101        self.host.setup.expect_call()
102        self.host.get_autodir.expect_call().and_return("autodir")
103        self.host.set_autodir.expect_call("autodir")
104        self.host.run.expect_call('mkdir -p autodir')
105        self.host.run.expect_call('rm -rf autodir/results/*',
106                                  ignore_status=True)
107
108
109    def test_constructor(self):
110        self.construct()
111
112        # we should check the calls
113        self.god.check_playback()
114
115
116    def test_full_client_install(self):
117        self.record_install_prologue()
118
119        self.host.run.expect_call('rm -f "autodir/packages.checksum"')
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',
125                                        delete_dest=True)
126
127        # run and check
128        self.base_autotest.install_full_client()
129        self.god.check_playback()
130
131
132    def test_autoserv_install(self):
133        self.record_install_prologue()
134
135        c = autotest.global_config.global_config
136        c.get_config_value.expect_call('PACKAGES',
137            'fetch_location', type=list, default=[]).and_return([])
138
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,
143                                                                'autodir')
144        # run and check
145        self.base_autotest.install()
146        self.god.check_playback()
147
148
149    def test_packaging_install(self):
150        self.record_install_prologue()
151
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)
164
165        # run and check
166        self.base_autotest.install()
167        self.god.check_playback()
168
169
170    def test_run(self):
171        self.construct()
172
173        # setup
174        control = "control"
175
176        # stub out install
177        self.god.stub_function(self.base_autotest, "install")
178
179        # record
180        self.base_autotest.install.expect_call(self.host, use_packaging=True)
181        self.host.wait_up.expect_call(timeout=30)
182        os.path.abspath.expect_call('.').and_return('.')
183        run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)
184        tag = None
185        run_obj.manual_control_file = os.path.join('autodir',
186                                                   'control.%s' % tag)
187        run_obj.remote_control_file = os.path.join('autodir',
188                                                   'control.%s.autoserv' % tag)
189        run_obj.tag = tag
190        run_obj.autodir = 'autodir'
191        run_obj.verify_machine.expect_call()
192        run_obj.background = False
193        debug = os.path.join('.', 'debug')
194        os.makedirs.expect_call(debug)
195        delete_file_list = [run_obj.remote_control_file,
196                            run_obj.remote_control_file + '.state',
197                            run_obj.manual_control_file,
198                            run_obj.manual_control_file + '.state']
199        cmd = ';'.join('rm -f ' + control for control in delete_file_list)
200        self.host.run.expect_call(cmd, ignore_status=True)
201
202        utils.get.expect_call(control, local_copy=True).and_return("temp")
203
204        c = autotest.global_config.global_config
205        c.get_config_value.expect_call("PACKAGES",
206            'fetch_location', type=list, default=[]).and_return(['repo'])
207
208        cfile = self.god.create_mock_class(file, "file")
209        cfile_orig = "original control file"
210        cfile_new = "args = []\njob.add_repository(['repo'])\n"
211        cfile_new += cfile_orig
212
213        autotest.open.expect_call("temp").and_return(cfile)
214        cfile.read.expect_call().and_return(cfile_orig)
215        autotest.open.expect_call("temp", 'w').and_return(cfile)
216        cfile.write.expect_call(cfile_new)
217
218        self.host.job.preprocess_client_state.expect_call().and_return(
219            '/job/tmp/file1')
220        self.host.send_file.expect_call(
221            "/job/tmp/file1", "autodir/control.None.autoserv.init.state")
222        os.remove.expect_call("/job/tmp/file1")
223
224        self.host.send_file.expect_call("temp", run_obj.remote_control_file)
225        os.path.abspath.expect_call('temp').and_return('control_file')
226        os.path.abspath.expect_call('control').and_return('control')
227        os.remove.expect_call("temp")
228
229        run_obj.execute_control.expect_call(timeout=30,
230                                            client_disconnect_timeout=240)
231
232        # run and check output
233        self.base_autotest.run(control, timeout=30)
234        self.god.check_playback()
235
236
237    def _stub_get_client_autodir_paths(self):
238        def mock_get_client_autodir_paths(cls, host):
239            return ['/some/path', '/another/path']
240        self.god.stub_with(autotest.Autotest, 'get_client_autodir_paths',
241                           classmethod(mock_get_client_autodir_paths))
242
243
244    def _expect_failed_run(self, command):
245        (self.host.run.expect_call(command)
246         .and_raises(error.AutoservRunError('dummy', object())))
247
248
249    def test_get_installed_autodir(self):
250        self._stub_get_client_autodir_paths()
251        self.host.get_autodir.expect_call().and_return(None)
252        self._expect_failed_run('test -x /some/path/bin/autotest')
253        self.host.run.expect_call('test -x /another/path/bin/autotest')
254        self.host.run.expect_call('test -w /another/path')
255
256        autodir = autotest.Autotest.get_installed_autodir(self.host)
257        self.assertEquals(autodir, '/another/path')
258
259
260    def test_get_install_dir(self):
261        self._stub_get_client_autodir_paths()
262        self.host.get_autodir.expect_call().and_return(None)
263        self._expect_failed_run('test -x /some/path/bin/autotest')
264        self._expect_failed_run('test -x /another/path/bin/autotest')
265        self._expect_failed_run('mkdir -p /some/path')
266        self.host.run.expect_call('mkdir -p /another/path')
267        self.host.run.expect_call('test -w /another/path')
268
269        install_dir = autotest.Autotest.get_install_dir(self.host)
270        self.assertEquals(install_dir, '/another/path')
271
272
273    def test_client_logger_process_line_log_copy_collection_failure(self):
274        collector = autotest.log_collector.expect_new(self.host, '', '')
275        logger = autotest.client_logger(self.host, '', '')
276        collector.collect_client_job_results.expect_call().and_raises(
277                Exception('log copy failure'))
278        logging.exception.expect_call(mock.is_string_comparator())
279        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')
280
281
282    def test_client_logger_process_line_log_copy_fifo_failure(self):
283        collector = autotest.log_collector.expect_new(self.host, '', '')
284        logger = autotest.client_logger(self.host, '', '')
285        collector.collect_client_job_results.expect_call()
286        self.host.run.expect_call('echo A > /autotest/fifo2').and_raises(
287                Exception('fifo failure'))
288        logging.exception.expect_call(mock.is_string_comparator())
289        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')
290
291
292    def test_client_logger_process_line_package_install_fifo_failure(self):
293        collector = autotest.log_collector.expect_new(self.host, '', '')
294        logger = autotest.client_logger(self.host, '', '')
295        self.god.stub_function(logger, '_send_tarball')
296
297        c = autotest.global_config.global_config
298        c.get_config_value.expect_call('PACKAGES',
299                                       'serve_packages_from_autoserv',
300                                       type=bool).and_return(True)
301        c.get_config_value.expect_call('PACKAGES',
302                                       'serve_packages_from_autoserv',
303                                       type=bool).and_return(True)
304        logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')
305
306        self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(
307                Exception('fifo failure'))
308        logging.exception.expect_call(mock.is_string_comparator())
309        logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
310                             '/autotest/dest/:/autotest/fifo3')
311
312
313if __name__ == "__main__":
314    unittest.main()
315