• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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            output_dir = False
88
89        self.job.__init__(options)
90
91
92class dummy(object):
93    """A simple placeholder for attributes"""
94    pass
95
96
97class first_line_comparator(mock.argument_comparator):
98    def __init__(self, first_line):
99        self.first_line = first_line
100
101
102    def is_satisfied_by(self, parameter):
103        return self.first_line == parameter.splitlines()[0]
104
105
106class test_setup_job(unittest.TestCase):
107    def setUp(self):
108        # make god
109        self.god = mock.mock_god()
110
111        # need to set some environ variables
112        self.autodir = "autodir"
113        os.environ['AUTODIR'] = self.autodir
114
115        # set up some variables
116        self.jobtag = "jobtag"
117
118        # get rid of stdout and logging
119        sys.stdout = StringIO.StringIO()
120        logging_manager.configure_logging(logging_config.TestingConfig())
121        logging.disable(logging.CRITICAL)
122        def dummy_configure_logging(*args, **kwargs):
123            pass
124        self.god.stub_with(logging_manager, 'configure_logging',
125                           dummy_configure_logging)
126        real_get_logging_manager = logging_manager.get_logging_manager
127        def get_logging_manager_no_fds(manage_stdout_and_stderr=False,
128                                       redirect_fds=False):
129            return real_get_logging_manager(manage_stdout_and_stderr, False)
130        self.god.stub_with(logging_manager, 'get_logging_manager',
131                           get_logging_manager_no_fds)
132
133        # stub out some stuff
134        self.god.stub_function(os.path, 'exists')
135        self.god.stub_function(os.path, 'isdir')
136        self.god.stub_function(os, 'makedirs')
137        self.god.stub_function(os, 'mkdir')
138        self.god.stub_function(os, 'remove')
139        self.god.stub_function(shutil, 'rmtree')
140        self.god.stub_function(shutil, 'copyfile')
141        self.god.stub_function(setup_job, 'open')
142        self.god.stub_function(utils, 'system')
143
144        self.god.stub_class_method(job.base_client_job,
145                                   '_cleanup_debugdir_files')
146        self.god.stub_class_method(job.base_client_job, '_cleanup_results_dir')
147
148        self.god.stub_with(base_job.job_directory, '_ensure_valid',
149                           lambda *_: None)
150
151
152    def tearDown(self):
153        sys.stdout = sys.__stdout__
154        self.god.unstub_all()
155
156
157    def _setup_pre_record_init(self):
158        resultdir = os.path.join(self.autodir, 'results', self.jobtag)
159        tmpdir = os.path.join(self.autodir, 'tmp')
160        job.base_client_job._cleanup_debugdir_files.expect_call()
161        job.base_client_job._cleanup_results_dir.expect_call()
162
163        return resultdir
164
165    def construct_job(self):
166        # will construct class instance using __new__
167        self.job = setup_job.setup_job.__new__(setup_job.setup_job)
168
169        resultdir = self._setup_pre_record_init()
170
171        # finish constructor
172        options = dummy()
173        options.tag = self.jobtag
174        options.log = False
175        options.verbose = False
176        options.hostname = 'localhost'
177        options.user = 'my_user'
178        options.output_dir = False
179        self.job.__init__(options)
180
181        # check
182        self.god.check_playback()
183
184
185    def get_partition_mock(self, devname):
186        """
187        Create a mock of a partition object and return it.
188        """
189        class mock(object):
190            device = devname
191            get_mountpoint = self.god.create_mock_function('get_mountpoint')
192        return mock
193
194
195    def test_constructor_first_run(self):
196        self.construct_job()
197
198
199    def test_constructor_continuation(self):
200        self.construct_job()
201
202
203    def test_setup_dirs_raise(self):
204        self.construct_job()
205
206        # setup
207        results_dir = 'foo'
208        tmp_dir = 'bar'
209
210        # record
211        os.path.exists.expect_call(tmp_dir).and_return(True)
212        os.path.isdir.expect_call(tmp_dir).and_return(False)
213
214        # test
215        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)
216        self.god.check_playback()
217
218
219    def test_setup_dirs(self):
220        self.construct_job()
221
222        # setup
223        results_dir1 = os.path.join(self.job.resultdir, 'build')
224        results_dir2 = os.path.join(self.job.resultdir, 'build.2')
225        results_dir3 = os.path.join(self.job.resultdir, 'build.3')
226        tmp_dir = 'bar'
227
228        # record
229        os.path.exists.expect_call(tmp_dir).and_return(False)
230        os.mkdir.expect_call(tmp_dir)
231        os.path.isdir.expect_call(tmp_dir).and_return(True)
232        os.path.exists.expect_call(results_dir1).and_return(True)
233        os.path.exists.expect_call(results_dir2).and_return(True)
234        os.path.exists.expect_call(results_dir3).and_return(False)
235        os.path.exists.expect_call(results_dir3).and_return(False)
236        os.mkdir.expect_call(results_dir3)
237
238        # test
239        self.assertEqual(self.job.setup_dirs(None, tmp_dir),
240                         (results_dir3, tmp_dir))
241        self.god.check_playback()
242
243
244if __name__ == "__main__":
245    unittest.main()
246