#!/usr/bin/python
|
|
import os
|
import tempfile
|
import unittest
|
|
import common
|
from autotest_lib.server import server_job
|
from autotest_lib.client.common_lib import base_job_unittest
|
from autotest_lib.client.common_lib.test_utils import mock
|
|
|
class test_find_base_directories(
|
unittest.TestCase,
|
base_job_unittest.test_find_base_directories.generic_tests):
|
|
def setUp(self):
|
self.job = server_job.server_job.__new__(server_job.server_job)
|
|
|
def test_relative_path_layout(self):
|
existing_file = server_job.__file__
|
server_job.__file__ = '/rootdir/atest/server/server_job.py'
|
try:
|
autodir, clientdir, serverdir = (
|
self.job._find_base_directories())
|
self.assertEqual(autodir, '/rootdir/atest')
|
self.assertEqual(clientdir, '/rootdir/atest/client')
|
self.assertEqual(serverdir, '/rootdir/atest/server')
|
finally:
|
server_job.__file__ = existing_file
|
|
|
class test_init(base_job_unittest.test_init.generic_tests, unittest.TestCase):
|
OPTIONAL_ATTRIBUTES = (
|
base_job_unittest.test_init.generic_tests.OPTIONAL_ATTRIBUTES
|
- set(['serverdir', 'num_tests_run', 'num_tests_failed',
|
'warning_manager', 'warning_loggers', 'in_lab']))
|
|
def setUp(self):
|
self.god = mock.mock_god()
|
self.job = server_job.server_job.__new__(
|
server_job.server_job)
|
self.job._job_directory = base_job_unittest.stub_job_directory
|
_, self.control_file = tempfile.mkstemp()
|
|
def tearDown(self):
|
"""Cleanup the test control file."""
|
os.remove(self.control_file)
|
|
def call_init(self):
|
# TODO(jadmanski): refactor more of the __init__ code to not need to
|
# stub out countless random APIs
|
self.god.stub_with(server_job.os, 'mkdir', lambda p: None)
|
class manager:
|
pass
|
self.god.stub_with(server_job.logging_manager, 'get_logging_manager',
|
lambda *a,**k: manager())
|
class sysi:
|
log_per_reboot_data = lambda self: None
|
self.god.stub_with(server_job.sysinfo, 'sysinfo', lambda r: sysi())
|
|
self.job.__init__(
|
self.control_file,
|
(),
|
None,
|
'job_label',
|
'auser',
|
['mach1', 'mach2'],
|
[
|
{'hostname': 'mach1'}, # Incomplete machine_dict.
|
{'hostname': 'mach2'}, # Incomplete machine_dict.
|
],
|
)
|
self.god.unstub_all()
|
|
|
class WarningManagerTest(unittest.TestCase):
|
def test_never_disabled(self):
|
manager = server_job.warning_manager()
|
self.assertEqual(manager.is_valid(10, "MSGTYPE"), True)
|
|
|
def test_only_enabled(self):
|
manager = server_job.warning_manager()
|
manager.enable_warnings("MSGTYPE", lambda: 10)
|
self.assertEqual(manager.is_valid(20, "MSGTYPE"), True)
|
|
|
def test_disabled_once(self):
|
manager = server_job.warning_manager()
|
manager.disable_warnings("MSGTYPE", lambda: 10)
|
self.assertEqual(manager.is_valid(5, "MSGTYPE"), True)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
|
|
|
def test_disable_and_enabled(self):
|
manager = server_job.warning_manager()
|
manager.disable_warnings("MSGTYPE", lambda: 10)
|
manager.enable_warnings("MSGTYPE", lambda: 20)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
|
self.assertEqual(manager.is_valid(25, "MSGTYPE"), True)
|
|
|
def test_disabled_changes_is_valid(self):
|
manager = server_job.warning_manager()
|
self.assertEqual(manager.is_valid(15, "MSGTYPE"), True)
|
manager.disable_warnings("MSGTYPE", lambda: 10)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
|
|
|
def test_multiple_disabled_calls(self):
|
manager = server_job.warning_manager()
|
manager.disable_warnings("MSGTYPE", lambda: 10)
|
manager.disable_warnings("MSGTYPE", lambda: 20)
|
manager.enable_warnings("MSGTYPE", lambda: 30)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
|
self.assertEqual(manager.is_valid(25, "MSGTYPE"), False)
|
self.assertEqual(manager.is_valid(35, "MSGTYPE"), True)
|
|
|
def test_multiple_types(self):
|
manager = server_job.warning_manager()
|
manager.disable_warnings("MSGTYPE1", lambda: 10)
|
manager.disable_warnings("MSGTYPE2", lambda: 20)
|
manager.enable_warnings("MSGTYPE2", lambda: 30)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE1"), False)
|
self.assertEqual(manager.is_valid(15, "MSGTYPE2"), True)
|
self.assertEqual(manager.is_valid(25, "MSGTYPE1"), False)
|
self.assertEqual(manager.is_valid(25, "MSGTYPE2"), False)
|
self.assertEqual(manager.is_valid(35, "MSGTYPE1"), False)
|
self.assertEqual(manager.is_valid(35, "MSGTYPE2"), True)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
