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