• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python
2
3import os
4import tempfile
5import unittest
6
7import common
8from autotest_lib.server import server_job
9from autotest_lib.client.common_lib import base_job_unittest
10from autotest_lib.client.common_lib.test_utils import mock
11
12
13class 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
34class 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
68class 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
125if __name__ == "__main__":
126    unittest.main()
127