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