xref: /aosp_15_r20/external/autotest/server/server_job_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3

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, version=None: 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()