Home | History | Annotate | Download | only in server
      1 #!/usr/bin/python
      2 
      3 import os
      4 import tempfile
      5 import unittest
      6 
      7 import common
      8 from autotest_lib.server import server_job
      9 from autotest_lib.client.common_lib import base_job_unittest
     10 from autotest_lib.client.common_lib.test_utils import mock
     11 
     12 
     13 class test_find_base_directories(
     14         unittest.TestCase,
     15         base_job_unittest.test_find_base_directories.generic_tests):
     16 
     17     def setUp(self):
     18         self.job = server_job.server_job.__new__(server_job.server_job)
     19 
     20 
     21     def test_relative_path_layout(self):
     22         existing_file = server_job.__file__
     23         server_job.__file__ = '/rootdir/atest/server/server_job.py'
     24         try:
     25             autodir, clientdir, serverdir = (
     26                 self.job._find_base_directories())
     27             self.assertEqual(autodir, '/rootdir/atest')
     28             self.assertEqual(clientdir, '/rootdir/atest/client')
     29             self.assertEqual(serverdir, '/rootdir/atest/server')
     30         finally:
     31             server_job.__file__ = existing_file
     32 
     33 
     34 class test_init(base_job_unittest.test_init.generic_tests, unittest.TestCase):
     35     OPTIONAL_ATTRIBUTES = (
     36         base_job_unittest.test_init.generic_tests.OPTIONAL_ATTRIBUTES
     37         - set(['serverdir', 'num_tests_run', 'num_tests_failed',
     38                'warning_manager', 'warning_loggers', 'in_lab']))
     39 
     40     def setUp(self):
     41         self.god = mock.mock_god()
     42         self.job = server_job.server_job.__new__(
     43             server_job.server_job)
     44         self.job._job_directory = base_job_unittest.stub_job_directory
     45         _, self.control_file = tempfile.mkstemp()
     46 
     47     def tearDown(self):
     48         """Cleanup the test control file."""
     49         os.remove(self.control_file)
     50 
     51     def call_init(self):
     52         # TODO(jadmanski): refactor more of the __init__ code to not need to
     53         # stub out countless random APIs
     54         self.god.stub_with(server_job.os, 'mkdir', lambda p: None)
     55         class manager:
     56             pass
     57         self.god.stub_with(server_job.logging_manager, 'get_logging_manager',
     58                            lambda *a,**k: manager())
     59         class sysi:
     60             log_per_reboot_data = lambda self: None
     61         self.god.stub_with(server_job.sysinfo, 'sysinfo', lambda r: sysi())
     62 
     63         self.job.__init__(self.control_file, (), None, 'job_label',
     64                           'auser', ['mach1', 'mach2'])
     65         self.god.unstub_all()
     66 
     67 
     68 class WarningManagerTest(unittest.TestCase):
     69     def test_never_disabled(self):
     70         manager = server_job.warning_manager()
     71         self.assertEqual(manager.is_valid(10, "MSGTYPE"), True)
     72 
     73 
     74     def test_only_enabled(self):
     75         manager = server_job.warning_manager()
     76         manager.enable_warnings("MSGTYPE", lambda: 10)
     77         self.assertEqual(manager.is_valid(20, "MSGTYPE"), True)
     78 
     79 
     80     def test_disabled_once(self):
     81         manager = server_job.warning_manager()
     82         manager.disable_warnings("MSGTYPE", lambda: 10)
     83         self.assertEqual(manager.is_valid(5, "MSGTYPE"), True)
     84         self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
     85 
     86 
     87     def test_disable_and_enabled(self):
     88         manager = server_job.warning_manager()
     89         manager.disable_warnings("MSGTYPE", lambda: 10)
     90         manager.enable_warnings("MSGTYPE", lambda: 20)
     91         self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
     92         self.assertEqual(manager.is_valid(25, "MSGTYPE"), True)
     93 
     94 
     95     def test_disabled_changes_is_valid(self):
     96         manager = server_job.warning_manager()
     97         self.assertEqual(manager.is_valid(15, "MSGTYPE"), True)
     98         manager.disable_warnings("MSGTYPE", lambda: 10)
     99         self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
    100 
    101 
    102     def test_multiple_disabled_calls(self):
    103         manager = server_job.warning_manager()
    104         manager.disable_warnings("MSGTYPE", lambda: 10)
    105         manager.disable_warnings("MSGTYPE", lambda: 20)
    106         manager.enable_warnings("MSGTYPE", lambda: 30)
    107         self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
    108         self.assertEqual(manager.is_valid(25, "MSGTYPE"), False)
    109         self.assertEqual(manager.is_valid(35, "MSGTYPE"), True)
    110 
    111 
    112     def test_multiple_types(self):
    113         manager = server_job.warning_manager()
    114         manager.disable_warnings("MSGTYPE1", lambda: 10)
    115         manager.disable_warnings("MSGTYPE2", lambda: 20)
    116         manager.enable_warnings("MSGTYPE2", lambda: 30)
    117         self.assertEqual(manager.is_valid(15, "MSGTYPE1"), False)
    118         self.assertEqual(manager.is_valid(15, "MSGTYPE2"), True)
    119         self.assertEqual(manager.is_valid(25, "MSGTYPE1"), False)
    120         self.assertEqual(manager.is_valid(25, "MSGTYPE2"), False)
    121         self.assertEqual(manager.is_valid(35, "MSGTYPE1"), False)
    122         self.assertEqual(manager.is_valid(35, "MSGTYPE2"), True)
    123 
    124 
    125 if __name__ == "__main__":
    126     unittest.main()
    127