1#!/usr/bin/python
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 subprocess
11import types
12from autotest_lib.server import utils
13from autotest_lib.server.cros.dynamic_suite import constants
14from autotest_lib.site_utils import test_runner_utils
15
16
17class StartsWithList(mox.Comparator):
18    def __init__(self, start_of_list):
19        """Mox comparator which returns True if the argument
20        to the mocked function is a list that begins with the elements
21        in start_of_list.
22        """
23        self._lhs = start_of_list
24
25    def equals(self, rhs):
26        if len(rhs)<len(self._lhs):
27            return False
28        for (x, y) in zip(self._lhs, rhs):
29            if x != y:
30                return False
31        return True
32
33
34class ContainsSublist(mox.Comparator):
35    def __init__(self, sublist):
36        """Mox comparator which returns True if the argument
37        to the mocked function is a list that contains sublist
38        as a sub-list.
39        """
40        self._sublist = sublist
41
42    def equals(self, rhs):
43        n = len(self._sublist)
44        if len(rhs)<n:
45            return False
46        return any((self._sublist == rhs[i:i+n])
47                   for i in xrange(len(rhs) - n + 1))
48
49
50class TestRunnerUnittests(unittest.TestCase):
51
52    def test_fetch_local_suite(self):
53        # Deferred until fetch_local_suite knows about non-local builds.
54        pass
55
56    def test_get_predicate_for_test_arg(self):
57        # Assert the type signature of get_predicate_for_test(...)
58        # Because control.test_utils_wrapper calls this function,
59        # it is imperative for backwards compatilbility that
60        # the return type of the tested function does not change.
61        tests = ['dummy_test', 'e:name_expression', 'f:expression',
62                 'suite:suitename']
63        for test in tests:
64            pred, desc = test_runner_utils.get_predicate_for_test_arg(test)
65            self.assertTrue(isinstance(pred, types.FunctionType))
66            self.assertTrue(isinstance(desc, str))
67
68    def test_run_job(self):
69        class Object():
70            pass
71
72        autotest_path = 'htap_tsetotua'
73        autoserv_command = os.path.join(autotest_path, 'server', 'autoserv')
74        remote = 'etomer'
75        results_dir = '/tmp/fakeresults'
76        fast_mode = False
77        job1_results_dir = '/tmp/fakeresults/results-1-gilbert'
78        job2_results_dir = '/tmp/fakeresults/results-2-sullivan'
79        args = 'matey'
80        expected_args_sublist = ['--args', args]
81        experimental_keyval = {constants.JOB_EXPERIMENTAL_KEY: False}
82        self.mox = mox.Mox()
83
84        # Create some dummy job objects.
85        job1 = Object()
86        job2 = Object()
87        setattr(job1, 'control_type', 'cLiEnT')
88        setattr(job1, 'control_file', 'c1')
89        setattr(job1, 'id', 1)
90        setattr(job1, 'name', 'gilbert')
91        setattr(job1, 'keyvals', experimental_keyval)
92
93        setattr(job2, 'control_type', 'Server')
94        setattr(job2, 'control_file', 'c2')
95        setattr(job2, 'id', 2)
96        setattr(job2, 'name', 'sullivan')
97        setattr(job2, 'keyvals', experimental_keyval)
98
99        id_digits = 1
100
101        # Stub out subprocess.Popen and wait calls.
102        # Make them expect correct arguments.
103        def fake_readline():
104            return b''
105        mock_process_1 = self.mox.CreateMock(subprocess.Popen)
106        mock_process_2 = self.mox.CreateMock(subprocess.Popen)
107        fake_stdout = self.mox.CreateMock(file)
108        fake_returncode = 0
109        mock_process_1.stdout = fake_stdout
110        mock_process_1.returncode = fake_returncode
111        mock_process_2.stdout = fake_stdout
112        mock_process_2.returncode = fake_returncode
113
114        self.mox.StubOutWithMock(os, 'makedirs')
115        self.mox.StubOutWithMock(utils, 'write_keyval')
116        self.mox.StubOutWithMock(subprocess, 'Popen')
117
118        os.makedirs(job1_results_dir)
119        utils.write_keyval(job1_results_dir, experimental_keyval)
120        arglist_1 = [autoserv_command, '-p', '-r', job1_results_dir,
121                     '-m', remote, '--no_console_prefix', '-l', 'gilbert',
122                     '-c']
123        subprocess.Popen(mox.And(StartsWithList(arglist_1),
124                                 ContainsSublist(expected_args_sublist)),
125                         stdout=subprocess.PIPE,
126                         stderr=subprocess.STDOUT
127                        ).AndReturn(mock_process_1)
128        mock_process_1.stdout.readline().AndReturn(b'')
129        mock_process_1.wait().AndReturn(0)
130
131        os.makedirs(job2_results_dir)
132        utils.write_keyval(job2_results_dir, experimental_keyval)
133        arglist_2 = [autoserv_command, '-p', '-r', job2_results_dir,
134                     '-m', remote,  '--no_console_prefix', '-l', 'sullivan',
135                     '-s']
136        subprocess.Popen(mox.And(StartsWithList(arglist_2),
137                                 ContainsSublist(expected_args_sublist)),
138                         stdout=subprocess.PIPE,
139                         stderr=subprocess.STDOUT
140                        ).AndReturn(mock_process_2)
141        mock_process_2.stdout.readline().AndReturn(b'')
142        mock_process_2.wait().AndReturn(0)
143
144        # Test run_job.
145        self.mox.ReplayAll()
146        code, job_res = test_runner_utils.run_job(
147                job1, remote, autotest_path,results_dir, fast_mode, id_digits,
148                0, None, args)
149        self.assertEqual(job_res, job1_results_dir)
150        self.assertEqual(code, 0)
151        code, job_res = test_runner_utils.run_job(
152                job2, remote, autotest_path, results_dir, fast_mode, id_digits,
153                0, None, args)
154
155        self.assertEqual(job_res, job2_results_dir)
156        self.assertEqual(code, 0)
157        self.mox.UnsetStubs()
158        self.mox.VerifyAll()
159        self.mox.ResetAll()
160
161    def test_perform_local_run(self):
162        afe = test_runner_utils.setup_local_afe()
163        autotest_path = 'ottotest_path'
164        suite_name = 'sweet_name'
165        test_arg = 'suite:' + suite_name
166        remote = 'remoat'
167        build = 'bild'
168        board = 'bored'
169        fast_mode = False
170        suite_control_files = ['c1', 'c2', 'c3', 'c4']
171        results_dir = '/tmp/test_that_results_fake'
172        id_digits = 1
173        ssh_verbosity = 2
174        ssh_options = '-F /dev/null -i /dev/null'
175        args = 'matey'
176        ignore_deps = False
177
178        # Fake suite objects that will be returned by fetch_local_suite
179        class fake_suite(object):
180            def __init__(self, suite_control_files, hosts):
181                self._suite_control_files = suite_control_files
182                self._hosts = hosts
183
184            def schedule(self, *args, **kwargs):
185                for control_file in self._suite_control_files:
186                    afe.create_job(control_file, hosts=self._hosts)
187
188        # Mock out scheduling of suite and running of jobs.
189        self.mox = mox.Mox()
190
191        self.mox.StubOutWithMock(test_runner_utils, 'fetch_local_suite')
192        test_runner_utils.fetch_local_suite(autotest_path, mox.IgnoreArg(),
193                afe, test_arg=test_arg, remote=remote, build=build,
194                board=board, results_directory=results_dir,
195                no_experimental=False,
196                ignore_deps=ignore_deps
197                ).AndReturn(fake_suite(suite_control_files, [remote]))
198        self.mox.StubOutWithMock(test_runner_utils, 'run_job')
199        self.mox.StubOutWithMock(test_runner_utils, 'run_provisioning_job')
200        self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels')
201
202        test_runner_utils._auto_detect_labels(afe, remote)
203        # Test perform_local_run. Enforce that run_provisioning_job,
204        # run_job and _auto_detect_labels are called correctly.
205        test_runner_utils.run_provisioning_job(
206                'cros-version:' + build, remote, autotest_path,
207                 results_dir, fast_mode,
208                 ssh_verbosity, ssh_options,
209                 False, False)
210
211        for control_file in suite_control_files:
212            test_runner_utils.run_job(
213                    mox.ContainsAttributeValue('control_file', control_file),
214                    remote, autotest_path, results_dir, fast_mode,id_digits,
215                    ssh_verbosity, ssh_options,args, False,
216                    False, {}).AndReturn((0, '/fake/dir'))
217        self.mox.ReplayAll()
218        test_runner_utils.perform_local_run(
219                afe, autotest_path, ['suite:'+suite_name], remote, fast_mode,
220                build=build, board=board, ignore_deps=False,
221                ssh_verbosity=ssh_verbosity, ssh_options=ssh_options,
222                args=args, results_directory=results_dir)
223        self.mox.UnsetStubs()
224        self.mox.VerifyAll()
225
226
227if __name__ == '__main__':
228    unittest.main()
229