Home | History | Annotate | Download | only in bin 
      1 #!/usr/bin/python
      2 #pylint: disable-msg=C0111
      3 import logging, os, shutil, sys, StringIO
      4 import common
      5 
      6 from autotest_lib.client.bin import job, setup_job
      7 from autotest_lib.client.bin import utils
      8 from autotest_lib.client.common_lib import base_job
      9 from autotest_lib.client.common_lib import logging_manager, logging_config
     10 from autotest_lib.client.common_lib import base_job_unittest
     11 from autotest_lib.client.common_lib.test_utils import mock, unittest
     12 
     13 
     14 class 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 
     33 class 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 
     48 class 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 
     59 class 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 
     88 class dummy(object):
     89     """A simple placeholder for attributes"""
     90     pass
     91 
     92 
     93 class 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 
    102 class 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 
    248 if __name__ == "__main__":
    249     unittest.main()
    250