1#!/usr/bin/python
2#pylint: disable-msg=C0111
3import logging
4import os
5import shutil
6import StringIO
7import sys
8import unittest
9
10import common
11from autotest_lib.client.bin import job, setup_job
12from autotest_lib.client.bin import utils
13from autotest_lib.client.common_lib import base_job
14from autotest_lib.client.common_lib import logging_manager, logging_config
15from autotest_lib.client.common_lib import base_job_unittest
16from autotest_lib.client.common_lib.test_utils import mock
17
18
19class setup_job_test_case(unittest.TestCase):
20    """Generic job TestCase class that defines a standard job setUp and
21    tearDown, with some standard stubs."""
22
23    job_class = setup_job.setup_job
24
25    def setUp(self):
26        self.god = mock.mock_god()
27        self.god.stub_with(setup_job.setup_job, '_get_environ_autodir',
28                           classmethod(lambda cls: '/adir'))
29        self.job = self.job_class.__new__(self.job_class)
30        self.job._job_directory = base_job_unittest.stub_job_directory
31        self.job.args = []
32
33
34    def tearDown(self):
35        self.god.unstub_all()
36
37
38class test_find_base_directories(
39        base_job_unittest.test_find_base_directories.generic_tests,
40        setup_job_test_case):
41
42    def test_autodir_equals_clientdir(self):
43        autodir, clientdir, _ = self.job._find_base_directories()
44        self.assertEqual(autodir, '/adir')
45        self.assertEqual(clientdir, '/adir')
46
47
48    def test_serverdir_is_none(self):
49        _, _, serverdir = self.job._find_base_directories()
50        self.assertEqual(serverdir, None)
51
52
53class abstract_test_init(base_job_unittest.test_init.generic_tests):
54    """Generic client job mixin used when defining variations on the
55    job.__init__ generic tests."""
56    PUBLIC_ATTRIBUTES = (
57        base_job_unittest.test_init.generic_tests.PUBLIC_ATTRIBUTES
58        - set(['bootloader', 'control', 'drop_caches',
59               'drop_caches_between_iterations', 'harness', 'hosts', 'logging',
60               'machines', 'num_tests_failed', 'num_tests_run', 'profilers',
61               'sysinfo', 'user',  'warning_loggers', 'warning_manager']))
62
63
64class test_init_minimal_options(abstract_test_init, setup_job_test_case):
65    def call_init(self):
66        # TODO(jadmanski): refactor more of the __init__ code to not need to
67        # stub out countless random APIs
68        self.god.stub_function_to_return(setup_job.os, 'mkdir', None)
69        self.god.stub_function_to_return(setup_job.os.path, 'exists', True)
70        self.god.stub_function_to_return(self.job, '_load_state', None)
71        self.god.stub_function_to_return(setup_job.logging_manager,
72                                         'configure_logging', None)
73        class manager:
74            def start_logging(self):
75                return None
76        self.god.stub_function_to_return(setup_job.logging_manager,
77                                         'get_logging_manager', manager())
78
79        class options:
80            tag = ''
81            verbose = False
82            cont = False
83            harness = 'stub'
84            hostname = None
85            user = None
86            log = False
87            tap_report = None
88            output_dir = False
89
90        self.job.__init__(options)
91
92
93class dummy(object):
94    """A simple placeholder for attributes"""
95    pass
96
97
98class first_line_comparator(mock.argument_comparator):
99    def __init__(self, first_line):
100        self.first_line = first_line
101
102
103    def is_satisfied_by(self, parameter):
104        return self.first_line == parameter.splitlines()[0]
105
106
107class test_setup_job(unittest.TestCase):
108    def setUp(self):
109        # make god
110        self.god = mock.mock_god()
111
112        # need to set some environ variables
113        self.autodir = "autodir"
114        os.environ['AUTODIR'] = self.autodir
115
116        # set up some variables
117        self.jobtag = "jobtag"
118
119        # get rid of stdout and logging
120        sys.stdout = StringIO.StringIO()
121        logging_manager.configure_logging(logging_config.TestingConfig())
122        logging.disable(logging.CRITICAL)
123        def dummy_configure_logging(*args, **kwargs):
124            pass
125        self.god.stub_with(logging_manager, 'configure_logging',
126                           dummy_configure_logging)
127        real_get_logging_manager = logging_manager.get_logging_manager
128        def get_logging_manager_no_fds(manage_stdout_and_stderr=False,
129                                       redirect_fds=False):
130            return real_get_logging_manager(manage_stdout_and_stderr, False)
131        self.god.stub_with(logging_manager, 'get_logging_manager',
132                           get_logging_manager_no_fds)
133
134        # stub out some stuff
135        self.god.stub_function(os.path, 'exists')
136        self.god.stub_function(os.path, 'isdir')
137        self.god.stub_function(os, 'makedirs')
138        self.god.stub_function(os, 'mkdir')
139        self.god.stub_function(os, 'remove')
140        self.god.stub_function(shutil, 'rmtree')
141        self.god.stub_function(shutil, 'copyfile')
142        self.god.stub_function(setup_job, 'open')
143        self.god.stub_function(utils, 'system')
144
145        self.god.stub_class_method(job.base_client_job,
146                                   '_cleanup_debugdir_files')
147        self.god.stub_class_method(job.base_client_job, '_cleanup_results_dir')
148
149        self.god.stub_with(base_job.job_directory, '_ensure_valid',
150                           lambda *_: None)
151
152
153    def tearDown(self):
154        sys.stdout = sys.__stdout__
155        self.god.unstub_all()
156
157
158    def _setup_pre_record_init(self):
159        resultdir = os.path.join(self.autodir, 'results', self.jobtag)
160        tmpdir = os.path.join(self.autodir, 'tmp')
161        job.base_client_job._cleanup_debugdir_files.expect_call()
162        job.base_client_job._cleanup_results_dir.expect_call()
163
164        return resultdir
165
166    def construct_job(self):
167        # will construct class instance using __new__
168        self.job = setup_job.setup_job.__new__(setup_job.setup_job)
169
170        resultdir = self._setup_pre_record_init()
171
172        # finish constructor
173        options = dummy()
174        options.tag = self.jobtag
175        options.log = False
176        options.verbose = False
177        options.hostname = 'localhost'
178        options.user = 'my_user'
179        options.tap_report = None
180        options.output_dir = False
181        self.job.__init__(options)
182
183        # check
184        self.god.check_playback()
185
186
187    def get_partition_mock(self, devname):
188        """
189        Create a mock of a partition object and return it.
190        """
191        class mock(object):
192            device = devname
193            get_mountpoint = self.god.create_mock_function('get_mountpoint')
194        return mock
195
196
197    def test_constructor_first_run(self):
198        self.construct_job()
199
200
201    def test_constructor_continuation(self):
202        self.construct_job()
203
204
205    def test_setup_dirs_raise(self):
206        self.construct_job()
207
208        # setup
209        results_dir = 'foo'
210        tmp_dir = 'bar'
211
212        # record
213        os.path.exists.expect_call(tmp_dir).and_return(True)
214        os.path.isdir.expect_call(tmp_dir).and_return(False)
215
216        # test
217        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)
218        self.god.check_playback()
219
220
221    def test_setup_dirs(self):
222        self.construct_job()
223
224        # setup
225        results_dir1 = os.path.join(self.job.resultdir, 'build')
226        results_dir2 = os.path.join(self.job.resultdir, 'build.2')
227        results_dir3 = os.path.join(self.job.resultdir, 'build.3')
228        tmp_dir = 'bar'
229
230        # record
231        os.path.exists.expect_call(tmp_dir).and_return(False)
232        os.mkdir.expect_call(tmp_dir)
233        os.path.isdir.expect_call(tmp_dir).and_return(True)
234        os.path.exists.expect_call(results_dir1).and_return(True)
235        os.path.exists.expect_call(results_dir2).and_return(True)
236        os.path.exists.expect_call(results_dir3).and_return(False)
237        os.path.exists.expect_call(results_dir3).and_return(False)
238        os.mkdir.expect_call(results_dir3)
239
240        # test
241        self.assertEqual(self.job.setup_dirs(None, tmp_dir),
242                         (results_dir3, tmp_dir))
243        self.god.check_playback()
244
245
246if __name__ == "__main__":
247    unittest.main()
248