• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python2
2# Copyright 2015 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5# pylint: disable-msg=C0111
6
7import os, unittest
8import mox
9import common
10import shutil
11import tempfile
12import types
13from autotest_lib.client.common_lib import control_data
14from autotest_lib.server.cros.dynamic_suite import control_file_getter
15from autotest_lib.server.cros.dynamic_suite import suite as suite_module
16from autotest_lib.server.hosts import host_info
17from autotest_lib.site_utils import test_runner_utils
18
19
20class StartsWithList(mox.Comparator):
21    def __init__(self, start_of_list):
22        """Mox comparator which returns True if the argument
23        to the mocked function is a list that begins with the elements
24        in start_of_list.
25        """
26        self._lhs = start_of_list
27
28    def equals(self, rhs):
29        if len(rhs)<len(self._lhs):
30            return False
31        for (x, y) in zip(self._lhs, rhs):
32            if x != y:
33                return False
34        return True
35
36
37class ContainsSublist(mox.Comparator):
38    def __init__(self, sublist):
39        """Mox comparator which returns True if the argument
40        to the mocked function is a list that contains sublist
41        as a sub-list.
42        """
43        self._sublist = sublist
44
45    def equals(self, rhs):
46        n = len(self._sublist)
47        if len(rhs)<n:
48            return False
49        return any((self._sublist == rhs[i:i+n])
50                   for i in xrange(len(rhs) - n + 1))
51
52class DummyJob(object):
53    def __init__(self, id=1):
54        self.id = id
55
56class TestRunnerUnittests(mox.MoxTestBase):
57
58    def setUp(self):
59        mox.MoxTestBase.setUp(self)
60
61
62    def test_fetch_local_suite(self):
63        # Deferred until fetch_local_suite knows about non-local builds.
64        pass
65
66
67    def _results_directory_from_results_list(self, results_list):
68        """Generate a temp directory filled with provided test results.
69
70        @param results_list: List of results, each result is a tuple of strings
71                             (test_name, test_status_message).
72        @returns: Absolute path to the results directory.
73        """
74        global_dir = tempfile.mkdtemp()
75        for index, (test_name, test_status_message) in enumerate(results_list):
76            dir_name = '-'.join(['results',
77                                 "%02.f" % (index + 1),
78                                 test_name])
79            local_dir = os.path.join(global_dir, dir_name)
80            os.mkdir(local_dir)
81            os.mkdir('%s/debug' % local_dir)
82            with open("%s/status.log" % local_dir, mode='w+') as status:
83                status.write(test_status_message)
84                status.flush()
85        return global_dir
86
87
88    def test_handle_local_result_for_good_test(self):
89        getter = self.mox.CreateMock(control_file_getter.DevServerGetter)
90        getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([])
91        job = DummyJob()
92        test = self.mox.CreateMock(control_data.ControlData)
93        test.job_retries = 5
94        self.mox.StubOutWithMock(test_runner_utils.LocalSuite,
95                                 '_retry_local_result')
96        self.mox.ReplayAll()
97        suite = test_runner_utils.LocalSuite([], "tag", [], None, getter,
98                                             job_retry=True)
99        suite._retry_handler = suite_module.RetryHandler({job.id: test})
100
101        #No calls, should not be retried
102        directory = self._results_directory_from_results_list([
103            ("dummy_Good", "GOOD: nonexistent test completed successfully")])
104        new_id = suite.handle_local_result(
105            job.id, directory,
106            lambda log_entry, log_in_subdir=False: None)
107        self.assertIsNone(new_id)
108        shutil.rmtree(directory)
109
110
111    def test_handle_local_result_for_bad_test(self):
112        getter = self.mox.CreateMock(control_file_getter.DevServerGetter)
113        getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([])
114        job = DummyJob()
115        test = self.mox.CreateMock(control_data.ControlData)
116        test.job_retries = 5
117        self.mox.StubOutWithMock(test_runner_utils.LocalSuite,
118                                 '_retry_local_result')
119        test_runner_utils.LocalSuite._retry_local_result(
120            job.id, mox.IgnoreArg()).AndReturn(42)
121        self.mox.ReplayAll()
122        suite = test_runner_utils.LocalSuite([], "tag", [], None, getter,
123                                             job_retry=True)
124        suite._retry_handler = suite_module.RetryHandler({job.id: test})
125
126        directory = self._results_directory_from_results_list([
127            ("dummy_Bad", "FAIL")])
128        new_id = suite.handle_local_result(
129            job.id, directory,
130            lambda log_entry, log_in_subdir=False: None)
131        self.assertIsNotNone(new_id)
132        shutil.rmtree(directory)
133
134
135    def test_generate_report_status_code_success_with_retries(self):
136        global_dir = self._results_directory_from_results_list([
137            ("dummy_Flaky", "FAIL"),
138            ("dummy_Flaky", "GOOD: nonexistent test completed successfully")])
139        status_code = test_runner_utils.generate_report(
140            global_dir, just_status_code=True)
141        self.assertEquals(status_code, 0)
142        shutil.rmtree(global_dir)
143
144
145    def test_generate_report_status_code_failure_with_retries(self):
146        global_dir = self._results_directory_from_results_list([
147            ("dummy_Good", "GOOD: nonexistent test completed successfully"),
148            ("dummy_Bad", "FAIL"),
149            ("dummy_Bad", "FAIL")])
150        status_code = test_runner_utils.generate_report(
151            global_dir, just_status_code=True)
152        self.assertNotEquals(status_code, 0)
153        shutil.rmtree(global_dir)
154
155
156    def test_get_predicate_for_test_arg(self):
157        # Assert the type signature of get_predicate_for_test(...)
158        # Because control.test_utils_wrapper calls this function,
159        # it is imperative for backwards compatilbility that
160        # the return type of the tested function does not change.
161        tests = ['dummy_test', 'e:name_expression', 'f:expression',
162                 'suite:suitename']
163        for test in tests:
164            pred, desc = test_runner_utils.get_predicate_for_test_arg(test)
165            self.assertTrue(isinstance(pred, types.FunctionType))
166            self.assertTrue(isinstance(desc, str))
167
168    def test_perform_local_run(self):
169        afe = test_runner_utils.setup_local_afe()
170        autotest_path = 'ottotest_path'
171        suite_name = 'sweet_name'
172        test_arg = 'suite:' + suite_name
173        remote = 'remoat'
174        build = 'bild'
175        board = 'bored'
176        fast_mode = False
177        suite_control_files = ['c1', 'c2', 'c3', 'c4']
178        results_dir = '/tmp/test_that_results_fake'
179        id_digits = 1
180        ssh_verbosity = 2
181        ssh_options = '-F /dev/null -i /dev/null'
182        args = 'matey'
183        ignore_deps = False
184        retry = True
185
186        # Fake suite objects that will be returned by fetch_local_suite
187        class fake_suite(object):
188            def __init__(self, suite_control_files, hosts):
189                self._suite_control_files = suite_control_files
190                self._hosts = hosts
191                self._jobs = []
192                self._jobs_to_tests = {}
193                self.retry_hack = True
194
195            def schedule(self, *args, **kwargs):
196                for control_file in self._suite_control_files:
197                    job_id = afe.create_job(control_file, hosts=self._hosts)
198                    self._jobs.append(job_id)
199                    self._jobs_to_tests[job_id] = control_file
200
201            def handle_local_result(self, job_id, results_dir, logger,
202                                    **kwargs):
203                if results_dir == "success_directory":
204                    return None
205                retries = True
206                if 'retries' in kwargs:
207                    retries = kwargs['retries']
208                if retries and self.retry_hack:
209                    self.retry_hack = False
210                else:
211                    return None
212                control_file = self._jobs_to_tests.get(job_id)
213                job_id = afe.create_job(control_file, hosts=self._hosts)
214                self._jobs.append(job_id)
215                self._jobs_to_tests[job_id] = control_file
216                return job_id
217
218            @property
219            def jobs(self):
220                return self._jobs
221
222            def test_name_from_job(self, id):
223                return ""
224
225        # Mock out scheduling of suite and running of jobs.
226        self.mox.StubOutWithMock(test_runner_utils, 'fetch_local_suite')
227        test_runner_utils.fetch_local_suite(autotest_path, mox.IgnoreArg(),
228                afe, test_arg=test_arg, remote=remote, build=build,
229                board=board, results_directory=results_dir,
230                no_experimental=False,
231                ignore_deps=ignore_deps,
232                job_retry=retry
233                ).AndReturn(fake_suite(suite_control_files, [remote]))
234        self.mox.StubOutWithMock(test_runner_utils, 'run_job')
235        self.mox.StubOutWithMock(test_runner_utils, 'run_provisioning_job')
236        self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels')
237
238        test_runner_utils._auto_detect_labels(afe, remote).AndReturn([])
239        # Test perform_local_run. Enforce that run_provisioning_job,
240        # run_job and _auto_detect_labels are called correctly.
241        test_runner_utils.run_provisioning_job(
242                'cros-version:' + build,
243                remote,
244                mox.IsA(host_info.HostInfo),
245                autotest_path,
246                results_dir,
247                fast_mode,
248                ssh_verbosity,
249                ssh_options,
250                False,
251                False,
252        )
253
254        for control_file in suite_control_files:
255            test_runner_utils.run_job(
256                    mox.ContainsAttributeValue('control_file', control_file),
257                    remote,
258                    mox.IsA(host_info.HostInfo),
259                    autotest_path,
260                    results_dir,
261                    fast_mode,
262                    id_digits,
263                    ssh_verbosity,
264                    ssh_options,
265                    mox.StrContains(args),
266                    False,
267                    False,
268            ).AndReturn((0, '/fake/dir'))
269        self.mox.ReplayAll()
270        test_runner_utils.perform_local_run(
271                afe, autotest_path, ['suite:'+suite_name], remote, fast_mode,
272                build=build, board=board, ignore_deps=False,
273                ssh_verbosity=ssh_verbosity, ssh_options=ssh_options,
274                args=args, results_directory=results_dir, job_retry=retry)
275
276
277if __name__ == '__main__':
278    unittest.main()
279