• 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__(
64                self.control_file,
65                (),
66                None,
67                'job_label',
68                'auser',
69                ['mach1', 'mach2'],
70                [
71                        {'hostname': 'mach1'},  # Incomplete machine_dict.
72                        {'hostname': 'mach2'},  # Incomplete machine_dict.
73                ],
74        )
75        self.god.unstub_all()
76
77
78class WarningManagerTest(unittest.TestCase):
79    def test_never_disabled(self):
80        manager = server_job.warning_manager()
81        self.assertEqual(manager.is_valid(10, "MSGTYPE"), True)
82
83
84    def test_only_enabled(self):
85        manager = server_job.warning_manager()
86        manager.enable_warnings("MSGTYPE", lambda: 10)
87        self.assertEqual(manager.is_valid(20, "MSGTYPE"), True)
88
89
90    def test_disabled_once(self):
91        manager = server_job.warning_manager()
92        manager.disable_warnings("MSGTYPE", lambda: 10)
93        self.assertEqual(manager.is_valid(5, "MSGTYPE"), True)
94        self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
95
96
97    def test_disable_and_enabled(self):
98        manager = server_job.warning_manager()
99        manager.disable_warnings("MSGTYPE", lambda: 10)
100        manager.enable_warnings("MSGTYPE", lambda: 20)
101        self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
102        self.assertEqual(manager.is_valid(25, "MSGTYPE"), True)
103
104
105    def test_disabled_changes_is_valid(self):
106        manager = server_job.warning_manager()
107        self.assertEqual(manager.is_valid(15, "MSGTYPE"), True)
108        manager.disable_warnings("MSGTYPE", lambda: 10)
109        self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
110
111
112    def test_multiple_disabled_calls(self):
113        manager = server_job.warning_manager()
114        manager.disable_warnings("MSGTYPE", lambda: 10)
115        manager.disable_warnings("MSGTYPE", lambda: 20)
116        manager.enable_warnings("MSGTYPE", lambda: 30)
117        self.assertEqual(manager.is_valid(15, "MSGTYPE"), False)
118        self.assertEqual(manager.is_valid(25, "MSGTYPE"), False)
119        self.assertEqual(manager.is_valid(35, "MSGTYPE"), True)
120
121
122    def test_multiple_types(self):
123        manager = server_job.warning_manager()
124        manager.disable_warnings("MSGTYPE1", lambda: 10)
125        manager.disable_warnings("MSGTYPE2", lambda: 20)
126        manager.enable_warnings("MSGTYPE2", lambda: 30)
127        self.assertEqual(manager.is_valid(15, "MSGTYPE1"), False)
128        self.assertEqual(manager.is_valid(15, "MSGTYPE2"), True)
129        self.assertEqual(manager.is_valid(25, "MSGTYPE1"), False)
130        self.assertEqual(manager.is_valid(25, "MSGTYPE2"), False)
131        self.assertEqual(manager.is_valid(35, "MSGTYPE1"), False)
132        self.assertEqual(manager.is_valid(35, "MSGTYPE2"), True)
133
134
135if __name__ == "__main__":
136    unittest.main()
137