server/server_job_unittest.py

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