scripts/run_backend_tests_test.py - develop - githubmirror/oppia

scripts/run_backend_tests_test.py
1 419 строк51 KB
Simran kaur
Fix #24136: Fix average test case time calculation in backend test reports (#25079)
20 мар 2026, 12:05
20 мар 2026, 12:05d159b69
# coding: utf-8
#
# Copyright 2022 The Oppia Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unit tests for scripts/run_backend_tests.py."""

from __future__ import annotations

import builtins
import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest.mock

from core.tests import test_utils
from scripts import (
    common,
    concurrent_task_utils,
    git_changes_utils,
    install_third_party_libs,
    run_backend_tests,
    servers,
)

from typing import Callable, Final, List, Set, Tuple

TEST_RUNNER_PATH: Final = os.path.join(
    os.getcwd(), 'core', 'tests', 'gae_suite.py'
)
SHARDS_SPEC_PATH: Final = os.path.join(
    os.getcwd(), 'scripts', 'backend_test_shards.json'
)
SHARDS_WIKI_LINK: Final = (
    'https://github.com/oppia/oppia/wiki/Writing-backend-tests#common-errors'
)
_LOAD_TESTS_DIR: Final = os.path.join(
    os.getcwd(), 'core', 'tests', 'load_tests'
)


def test_function(_: str) -> Callable[[], None]:
    def task_func() -> None:
        pass

    return task_func


class MockCompiler:
    def wait(self) -> None:  # pylint: disable=missing-docstring
        pass


class MockCompilerContextManager:
    def __init__(self) -> None:
        pass

    def __enter__(self) -> MockCompiler:
        return MockCompiler()

    def __exit__(self, *unused_args: str) -> None:
        pass


class MockProcessOutput:
    returncode = 0
    stdout = ''


class RunBackendTestsTests(test_utils.GenericTestBase):
    """Test the methods for run_backend_tests script."""

    def setUp(self) -> None:
        super().setUp()
        self.semaphore = threading.Semaphore(1)
        self.print_arr: list[str] = []

        def mock_print(msg: str) -> None:
            self.print_arr.append(msg)

        self.print_swap = self.swap(builtins, 'print', mock_print)

        def mock_install_third_party_libs() -> None:
            pass

        # We need to create a swap for install_third_party_libs because
        # run_backend_tests.py script installs third party libraries whenever
        # it is imported.
        self.swap_install_third_party_libs = self.swap(
            install_third_party_libs, 'main', mock_install_third_party_libs
        )

        test_target_flag = '--test_target=random_test'
        self.coverage_exc_list = [
            sys.executable,
            '-m',
            'coverage',
            'run',
            '--branch',
            TEST_RUNNER_PATH,
            test_target_flag,
        ]
        self.coverage_combine_cmd = [
            sys.executable,
            '-m',
            'coverage',
            'combine',
        ]
        self.coverage_check_cmd = [
            sys.executable,
            '-m',
            'coverage',
            'report',
            '--omit="%s*","third_party/*","/usr/share/*"'
            % common.OPPIA_TOOLS_DIR,
            '--show-missing',
            '--skip-covered',
        ]
        self.call_count = 0

        self.terminal_logs: List[str] = []

        def mock_log(msg: str) -> None:
            self.terminal_logs.append(msg)

        self.swap_logs = self.swap(concurrent_task_utils, 'log', mock_log)

        def mock_context_manager(**_: str) -> MockCompilerContextManager:
            return MockCompilerContextManager()

        self.swap_redis_server = self.swap(
            servers, 'managed_redis_server', mock_context_manager
        )
        self.swap_cloud_datastore_emulator = self.swap(
            servers, 'managed_cloud_datastore_emulator', mock_context_manager
        )
        self.swap_execute_task = self.swap(
            concurrent_task_utils, 'execute_tasks', lambda *unused_args: None
        )

    def test_run_shell_command_successfully(self) -> None:
        class MockProcess:
            returncode = 0

            def communicate(  # pylint: disable=missing-docstring
                self,
            ) -> tuple[bytes, bytes]:
                return (b'LOG_INFO_TEST: This is task output.\n', b'')

        def mock_popen(  # pylint: disable=unused-argument
            cmd_tokens: list[str], **unsued_kwargs: str
        ) -> MockProcess:
            return MockProcess()

        swap_popen = self.swap_with_checks(
            subprocess,
            'Popen',
            mock_popen,
            expected_args=((self.coverage_exc_list,),),
        )

        expected_result = 'LOG_INFO_TEST: This is task output.\n'
        with swap_popen, self.swap_logs:
            returned_result = run_backend_tests.run_shell_cmd(
                self.coverage_exc_list
            )

        self.assertIn('INFO: This is task output.', self.terminal_logs)
        self.assertEqual(expected_result, returned_result)

    def test_run_shell_command_failure_throws_error(self) -> None:
        class MockProcess:
            returncode = 1

            def communicate(  # pylint: disable=missing-docstring
                self,
            ) -> Tuple[bytes, bytes]:
                return (b'', b'Error XYZ occured.')

        def mock_popen(  # pylint: disable=unused-argument
            cmd_tokens: List[str], **unsued_kwargs: str
        ) -> MockProcess:
            return MockProcess()

        swap_popen = self.swap_with_checks(
            subprocess,
            'Popen',
            mock_popen,
            expected_args=((self.coverage_exc_list,),),
        )
        with swap_popen, self.swap_logs:
            with self.assertRaisesRegex(
                Exception, 'Error 1\nError XYZ occured.'
            ):
                run_backend_tests.run_shell_cmd(self.coverage_exc_list)

    def test_duplicate_test_files_in_shards_throws_error(self) -> None:

        with open(SHARDS_SPEC_PATH, 'r', encoding='utf-8') as shards_file:
            shards_spec = json.load(shards_file)

        shards_spec['1'].append(shards_spec['1'][0])
        swap_shard_modules = self.swap(
            json, 'loads', lambda *unused_args, **unused_kwargs: shards_spec
        )

        with swap_shard_modules:
            returned_error_msg = run_backend_tests.check_shards_match_tests()

        self.assertEqual(
            '%s duplicated in %s' % (shards_spec['1'][0], SHARDS_SPEC_PATH),
            returned_error_msg,
        )

    def test_module_in_shards_not_found_throws_error(self) -> None:

        with open(SHARDS_SPEC_PATH, 'r', encoding='utf-8') as shards_file:
            shards_spec = json.load(shards_file)

        shards_spec['1'].append('scripts.new_script_test')
        swap_shard_modules = self.swap(
            json, 'loads', lambda *unused_args, **unused_kwargs: shards_spec
        )

        with swap_shard_modules:
            returned_error_msg = run_backend_tests.check_shards_match_tests()

        self.assertEqual(
            'Modules %s are in the backend test shards but missing from the '
            'filesystem. See %s.'
            % ({'scripts.new_script_test'}, SHARDS_WIKI_LINK),
            returned_error_msg,
        )

    def test_module_not_in_shards_throws_error(self) -> None:

        test_modules = run_backend_tests.get_all_test_targets_from_path()
        test_modules.append('scripts.new_script_test')

        swap_test_modules = self.swap(
            run_backend_tests,
            'get_all_test_targets_from_path',
            lambda *unused_args, **unused_kwargs: test_modules,
        )

        with swap_test_modules:
            returned_error_msg = run_backend_tests.check_shards_match_tests()

        self.assertEqual(
            'Modules %s are present on the filesystem but are not listed in '
            'the backend test shards. See %s.'
            % ({'scripts.new_script_test'}, SHARDS_WIKI_LINK),
            returned_error_msg,
        )

    def test_tests_in_load_tests_dir_are_not_included_when_flag_is_passed(
        self,
    ) -> None:
        test_modules = run_backend_tests.get_all_test_targets_from_path(
            include_load_tests=False
        )
        self.assertNotIn(
            os.path.join(_LOAD_TESTS_DIR, 'new_test.py'), test_modules
        )

    def test_subprocess_error_while_execution_throws_error(self) -> None:

        test_cmd = 'python -m scripts.run_backend_tests'
        task1 = concurrent_task_utils.create_task(
            test_function('unused_arg'),
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task1.exception = subprocess.CalledProcessError(
            returncode=1, cmd=test_cmd
        )
        task1.finished = True

        tasks = [task1]
        task_to_taskspec = {}
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            'scripts.new_script.py', False
        )

        expected_error_msg = (
            'Command \'%s\' returned non-zero exit status 1.' % test_cmd
        )
        with self.assertRaisesRegex(
            subprocess.CalledProcessError, expected_error_msg
        ):
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

    def test_empty_test_files_show_no_tests_were_run(self) -> None:

        task1 = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task1.exception = Exception('No tests were run.')
        task1.finished = True

        tasks = [task1]
        task_to_taskspec = {}
        test_target = 'scripts.new_script.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test_target, False
        )

        with self.print_swap:
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

        self.assertIn(
            'ERROR     %s: No tests found.' % test_target, self.print_arr
        )

    def test_failed_test_suite_throws_error(self) -> None:

        task1 = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task1.exception = Exception(
            'Test suite failed: 6 tests run, 0 errors, 2 failures'
        )
        task1.finished = True

        tasks = [task1]
        task_to_taskspec = {}
        test_target = 'scripts.new_script.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test_target, False
        )

        with self.print_swap:
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

        self.assertIn(
            'FAILED    %s: %s errors, %s failures' % (test_target, 0, 2),
            self.print_arr,
        )

    def test_tests_failed_due_to_internal_error(self) -> None:

        task1 = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task1.exception = Exception('Some internal error.')
        task1.finished = True

        tasks = [task1]
        task_to_taskspec = {}
        test_target = 'scripts.new_script.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test_target, False
        )

        with self.print_swap, self.assertRaisesRegex(
            Exception, 'Some internal error.'
        ):
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

        self.assertIn(
            '    WARNING: FAILED TO RUN %s' % test_target, self.print_arr
        )
        self.assertIn(
            '    This is most likely due to an import error.', self.print_arr
        )

    def test_unfinished_tests_are_cancelled(self) -> None:

        task = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task.finished = False
        task_output = ['Ran 9 tests in 1.244s', '98']
        task_result = concurrent_task_utils.TaskResult(
            'task1', False, task_output, task_output
        )
        task.task_results.append(task_result)

        tasks = [task]
        task_to_taskspec = {}
        test_target = 'scripts.new_script.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test_target, False
        )

        with self.print_swap:
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

        self.assertIn('CANCELED  %s' % test_target, self.print_arr)

    def test_successfull_test_run_message_is_printed_correctly(self) -> None:

        task1 = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task1.finished = True
        task1_output = ['Ran 9 tests in 1.234s', '100']
        task1_result = concurrent_task_utils.TaskResult(
            'task1', False, task1_output, task1_output
        )
        task1.task_results.append(task1_result)

        task2 = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task2.finished = True
        task2_output = ['Ran 9 tests in 2.542s', '100']
        task2_result = concurrent_task_utils.TaskResult(
            'task2', False, task2_output, task2_output
        )
        task2.task_results.append(task2_result)

        tasks = [task1, task2]
        task_to_taskspec = {}
        test1_target = 'scripts.new_script_one_test.py'
        task2_target = 'scripts.new_script_two_test.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test1_target, True
        )
        task_to_taskspec[tasks[1]] = run_backend_tests.TestingTaskSpec(
            task2_target, True
        )

        with self.print_swap:
            _, _, _, time_report = run_backend_tests.check_test_results(
                tasks, task_to_taskspec
            )

        # Check first values match exactly.
        self.assertEqual(
            time_report['scripts.new_script_one_test.py'][0], 1.234
        )
        self.assertEqual(
            time_report['scripts.new_script_two_test.py'][0], 2.542
        )

        # Check second values (averages) match with tolerance.
        self.assertAlmostEqual(
            time_report['scripts.new_script_one_test.py'][1],
            1.234 / 9,
            places=5,
        )
        self.assertAlmostEqual(
            time_report['scripts.new_script_two_test.py'][1],
            2.542 / 9,
            places=5,
        )

        self.assertIn(
            'SUCCESS   %s: 9 tests (1.2 secs)' % test1_target, self.print_arr
        )
        self.assertIn(
            'SUCCESS   %s: 9 tests (2.5 secs)' % task2_target, self.print_arr
        )

    def test_successful_test_run_with_generate_time_report_flag(self) -> None:
        expected_time_report = {
            'scripts.new_script_one_test.py': [1.234, 1.234 / 9],
            'scripts.new_script_two_test.py': [2.542, 2.542 / 9],
        }
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (
                100,
                0,
                0,
                expected_time_report,
            ),
        )
        swap_check_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('Coverage report', 100.00),
        )
        time_report_temp_file = tempfile.NamedTemporaryFile('w+')
        time_report_path = time_report_temp_file.name
        swap_time_report_path = self.swap(
            run_backend_tests, 'TIME_REPORT_PATH', time_report_path
        )

        with self.swap_execute_task, swap_check_coverage:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                with swap_time_report_path, self.swap_redis_server:
                    with self.print_swap:
                        run_backend_tests.main(args=['--generate_time_report'])
        loaded_time_report = json.loads(time_report_temp_file.read())
        self.assertEqual(loaded_time_report, expected_time_report)
        time_report_temp_file.close()

    def test_test_failed_due_to_error_in_parsing_coverage_report(self) -> None:

        task = concurrent_task_utils.create_task(
            test_function,
            False,
            self.semaphore,
            name='test',
            errors_to_retry_on=['Error -11'],
        )
        task.finished = True
        task_output = ['XYZ', '100']
        task_result = concurrent_task_utils.TaskResult(
            'task1', False, task_output, task_output
        )
        task.task_results = [task_result]

        tasks = [task]
        task_to_taskspec = {}
        test_target = 'scripts.random_script.py'
        task_to_taskspec[tasks[0]] = run_backend_tests.TestingTaskSpec(
            test_target, True
        )

        with self.print_swap:
            run_backend_tests.check_test_results(tasks, task_to_taskspec)

        self.assertIn(
            'An unexpected error occurred. Task output:\nXYZ', self.print_arr
        )

    def test_invalid_directory_in_sys_path_throws_error(self) -> None:

        def mock_path_exists(dirname: str) -> bool:
            for directory in common.DIRS_TO_ADD_TO_SYS_PATH:
                if os.path.dirname(directory) == dirname:
                    return False
            return True

        swap_path_exists = self.swap(os.path, 'exists', mock_path_exists)

        with swap_path_exists, self.assertRaisesRegex(
            Exception,
            'Directory %s does not exist.' % common.DIRS_TO_ADD_TO_SYS_PATH[0],
        ):
            run_backend_tests.main(args=[])

    def test_invalid_delimiter_in_test_path_argument_throws_error(self) -> None:

        with self.assertRaisesRegex(
            Exception, r'The delimiter in test_path should be a slash \(/\)'
        ):
            run_backend_tests.main(
                args=['--test_path', 'scripts.run_backend_tests']
            )

    def test_invalid_delimiter_in_test_targets_argument_throws_error(
        self,
    ) -> None:

        with self.swap_redis_server, self.swap_cloud_datastore_emulator:
            with self.assertRaisesRegex(
                Exception,
                r'The delimiter in each test_target should be a dot \(\.\)',
            ):
                run_backend_tests.main(
                    args=['--test_targets', 'scripts/run_backend_tests']
                )

    def test_invalid_test_targets_message_is_displayed_correctly(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )
        swapcheck_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('', 100.00),
        )
        with self.swap_execute_task, swapcheck_coverage, self.swap_redis_server:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                with self.print_swap:
                    run_backend_tests.main(
                        args=['--test_targets', 'scripts.run_backend_tests.py']
                    )

        self.assertIn(
            'WARNING : each test_target should point to the test file.',
            self.print_arr,
        )
        self.assertIn(
            'Redirecting to its corresponding test file...', self.print_arr
        )

    def test_error_in_matching_shards_with_tests_throws_error(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )
        swapcheck_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('', 100.00),
        )

        error_msg = 'Some error in matching shards with tests.'

        def mockcheck_shards_match_tests(**unused_kwargs: str) -> str:
            return error_msg

        swapcheck_shards_match_tests = self.swap_with_checks(
            run_backend_tests,
            'check_shards_match_tests',
            mockcheck_shards_match_tests,
            expected_kwargs=[{'include_load_tests': True}],
        )
        with self.swap_execute_task, swapcheck_coverage, self.swap_redis_server:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                with self.print_swap, swapcheck_shards_match_tests:
                    with self.assertRaisesRegex(Exception, error_msg):
                        run_backend_tests.main(args=['--test_shard', '1'])

    def test_no_tests_run_raises_error(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (0, 0, 0, {}),
        )
        swapcheck_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('', 100.00),
        )

        with swapcheck_coverage, self.swap_cloud_datastore_emulator:
            with self.swap_redis_server, swap_check_results:
                with self.swap_execute_task, self.assertRaisesRegex(
                    Exception, 'WARNING: No tests were run.'
                ):
                    run_backend_tests.main(
                        args=[
                            '--test_targets',
                            'scripts.run_backend_tests_test',
                        ]
                    )

    def test_incomplete_overall_backend_coverage_throws_error(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )
        swapcheck_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('Coverage report', 98.00),
        )

        with swapcheck_coverage, self.swap_redis_server, self.print_swap:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                with self.swap_execute_task:
                    with self.assertRaisesRegex(
                        Exception, 'Backend test coverage is not 100%'
                    ):
                        run_backend_tests.main(
                            args=['--generate_coverage_report']
                        )

        self.assertIn('Coverage report', self.print_arr)

    def test_failure_in_test_execution_throws_error(self) -> None:

        def mock_execute_tasks(*_: str) -> None:
            raise Exception('XYZ error occured.')

        self.swap_execute_task = self.swap(
            concurrent_task_utils, 'execute_tasks', mock_execute_tasks
        )
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )

        with self.swap_execute_task, self.swap_redis_server, swap_check_results:
            with self.swap_cloud_datastore_emulator, self.assertRaisesRegex(
                Exception, 'Task execution failed.'
            ):
                run_backend_tests.main(args=[])

    def test_errors_in_test_suite_throw_error(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 2, 0, {}),
        )

        with self.swap_execute_task, self.swap_redis_server, swap_check_results:
            with self.swap_cloud_datastore_emulator, self.print_swap:
                with self.assertRaisesRegex(Exception, '2 errors, 0 failures'):
                    run_backend_tests.main(args=['--test_shard', '1'])

        self.assertIn('(2 ERRORS, 0 FAILURES)', self.print_arr)

    def test_individual_test_in_test_file_is_run_successfully(self) -> None:

        executed_tasks = []
        test_target = (
            'scripts.new_test_file_test.NewTestFileTests.test_for_something'
        )

        def mock_execute(
            tasks: List[concurrent_task_utils.TaskThread], *_: str
        ) -> None:
            for task in tasks:
                executed_tasks.append(task)

        swap_execute_task = self.swap(
            concurrent_task_utils, 'execute_tasks', mock_execute
        )
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )
        swap_check_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('Coverage report', 100.00),
        )

        args = [
            '--skip_install',
            '--test_targets',
            test_target,
            '--generate_coverage_report',
        ]
        with self.print_swap:
            with swap_check_coverage, self.swap_redis_server, swap_execute_task:
                with self.swap_cloud_datastore_emulator, swap_check_results:
                    run_backend_tests.main(args=args)

        self.assertEqual(len(executed_tasks), 1)
        self.assertEqual(executed_tasks[0].name, test_target)
        self.assertIn('All tests passed.', self.print_arr)
        self.assertIn('Done!', self.print_arr)

    def test_multiple_tests_in_test_targets_argument_is_run_successfully(
        self,
    ) -> None:

        executed_tasks = []
        test_targets = 'scripts.test_file_test,scripts.another_test_file_test'

        def mock_execute(
            tasks: List[concurrent_task_utils.TaskThread], *_: str
        ) -> None:
            for task in tasks:
                executed_tasks.append(task)

        swap_execute_task = self.swap(
            concurrent_task_utils, 'execute_tasks', mock_execute
        )
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )

        args = ['--skip_install', '--test_targets', test_targets]
        with self.print_swap, self.swap_redis_server, swap_execute_task:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                run_backend_tests.main(args=args)

        self.assertEqual(len(executed_tasks), 2)
        self.assertEqual(
            executed_tasks[0].name, test_targets.split(',', maxsplit=1)[0]
        )
        self.assertEqual(executed_tasks[1].name, test_targets.split(',')[1])
        self.assertIn('All tests passed.', self.print_arr)
        self.assertIn('Done!', self.print_arr)

    def test_successful_test_run_with_run_on_changed_files_in_branch_flag(
        self,
    ) -> None:

        def mock_get_changed_python_test_files() -> Set[str]:
            return {'test.file1_test', 'test.file2_test', 'test.file3_test'}

        executed_tasks = []

        def mock_execute(
            tasks: List[concurrent_task_utils.TaskThread], *_: str
        ) -> None:
            for task in tasks:
                executed_tasks.append(task)

        swap_execute_task = self.swap(
            concurrent_task_utils, 'execute_tasks', mock_execute
        )
        get_changed_python_test_files_swap = self.swap(
            git_changes_utils,
            'get_changed_python_test_files',
            mock_get_changed_python_test_files,
        )
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )

        with swap_execute_task, self.print_swap:
            with self.swap_cloud_datastore_emulator, swap_check_results:
                with self.swap_redis_server, get_changed_python_test_files_swap:
                    run_backend_tests.main(
                        args=['--run_on_changed_files_in_branch']
                    )

        self.assertEqual(len(executed_tasks), 3)
        self.assertIn('All tests passed.', self.print_arr)
        self.assertIn('Done!', self.print_arr)

    def test_backend_tests_with_run_on_changed_files_in_branch_no_remote(
        self,
    ) -> None:

        def mock_get_remote_name() -> str:
            return ''

        get_remote_name_swap = self.swap(
            git_changes_utils,
            'get_local_git_repository_remote_name',
            mock_get_remote_name,
        )

        with get_remote_name_swap, self.swap_redis_server:
            with self.swap_cloud_datastore_emulator, self.assertRaisesRegex(
                SystemExit, 'Error: No remote repository found.'
            ):
                run_backend_tests.main(
                    args=['--run_on_changed_files_in_branch']
                )

    def test_all_test_pass_successfully_with_full_coverage(self) -> None:
        swap_check_results = self.swap(
            run_backend_tests,
            'check_test_results',
            lambda *unused_args, **unused_kwargs: (100, 0, 0, {}),
        )
        swap_check_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('Coverage report', 100.00),
        )

        with self.swap_execute_task, swap_check_results:
            with swap_check_coverage, self.swap_redis_server, self.print_swap:
                with self.swap_cloud_datastore_emulator:
                    run_backend_tests.main(args=['--generate_coverage_report'])

        self.assertIn('Coverage report', self.print_arr)
        self.assertIn('All tests passed.', self.print_arr)
        self.assertIn('Done!', self.print_arr)

    def test_failure_to_combine_coverage_report_throws_error(self) -> None:
        failed_process_output = MockProcessOutput()
        failed_process_output.returncode = 1

        def mock_subprocess_run(cmd: List[str], **_: str) -> MockProcessOutput:
            if cmd == self.coverage_combine_cmd:
                return failed_process_output
            elif cmd == self.coverage_check_cmd:
                return MockProcessOutput()
            else:
                raise Exception(
                    'Invalid command passed to subprocess.run() method'
                )

        swap_subprocess_run = self.swap(subprocess, 'run', mock_subprocess_run)
        error_msg = (
            'Failed to combine coverage because subprocess failed.'
            '\n%s' % failed_process_output
        )
        with swap_subprocess_run, self.assertRaisesRegex(
            RuntimeError, error_msg
        ):
            run_backend_tests.check_coverage(True)

    def test_failure_to_calculate_coverage_report_throws_error(self) -> None:
        failed_process_output = MockProcessOutput()
        failed_process_output.returncode = 1

        def mock_subprocess_run(cmd: List[str], **_: str) -> MockProcessOutput:
            if cmd == self.coverage_combine_cmd:
                return MockProcessOutput()
            elif cmd == self.coverage_check_cmd:
                return failed_process_output
            else:
                raise Exception(
                    'Invalid command passed to subprocess.run() method'
                )

        swap_subprocess_run = self.swap(subprocess, 'run', mock_subprocess_run)
        error_msg = (
            'Failed to calculate coverage because subprocess failed. '
            '%s' % failed_process_output
        )
        with swap_subprocess_run, self.assertRaisesRegex(
            RuntimeError, error_msg
        ):
            run_backend_tests.check_coverage(True)

    def test_coverage_is_calculated_correctly_for_specific_files(self) -> None:
        include_files = (
            'scripts/run_backend_tests.py',
            'core/domain/exp_domain.py',
        )
        self.coverage_check_cmd.append('--include=%s' % ','.join(include_files))
        coverage_report_output = 'TOTAL       283     36    112     10    86% '
        process = MockProcessOutput()
        process.stdout = coverage_report_output

        def mock_subprocess_run(cmd: List[str], **_: str) -> MockProcessOutput:
            if cmd == self.coverage_combine_cmd:
                return MockProcessOutput()
            elif cmd == self.coverage_check_cmd:
                return process
            else:
                raise Exception(
                    'Invalid command passed to subprocess.run() method'
                )

        swap_subprocess_run = self.swap(subprocess, 'run', mock_subprocess_run)
        with swap_subprocess_run:
            returned_output, coverage = run_backend_tests.check_coverage(
                True, include=include_files
            )

        self.assertEqual(returned_output, coverage_report_output)
        self.assertEqual(coverage, 86)

    def test_coverage_is_calculated_correctly_for_a_single_file(self) -> None:
        data_file = '.coverage.hostname.12345.987654321'
        coverage_report_output = (
            'Name                                                       '
            '                    Stmts   Miss Branch BrPart  Cover   Missing\n'
            'core/constants.py                         '
            '                         '
            '               37      8      6      1    70%   108-112, 119-123\n'
            'TOTAL                                                         '
            '                 53906  17509  16917   1191    62%\n'
        )
        process = MockProcessOutput()
        process.stdout = coverage_report_output

        def mock_subprocess_run(cmd: List[str], **_: str) -> MockProcessOutput:
            if cmd == self.coverage_combine_cmd:
                return MockProcessOutput()
            elif cmd == self.coverage_check_cmd:
                return process
            else:
                raise Exception(
                    'Invalid command passed to subprocess.run() method'
                )

        swap_subprocess_run = self.swap(subprocess, 'run', mock_subprocess_run)
        with swap_subprocess_run:
            returned_output, coverage = run_backend_tests.check_coverage(
                False, data_file=data_file
            )

        self.assertEqual(coverage_report_output, returned_output)
        self.assertEqual(coverage, 62.0)

    def test_no_data_to_report_returns_full_coverage(self) -> None:
        coverage_report_output = 'No data to report.'
        process = MockProcessOutput()
        process.stdout = coverage_report_output

        def mock_subprocess_run(cmd: List[str], **_: str) -> MockProcessOutput:
            if cmd == self.coverage_combine_cmd:
                return MockProcessOutput()
            elif cmd == self.coverage_check_cmd:
                return process
            else:
                raise Exception(
                    'Invalid command passed to subprocess.run() method'
                )

        swap_subprocess_run = self.swap(subprocess, 'run', mock_subprocess_run)
        with swap_subprocess_run:
            returned_output, coverage = run_backend_tests.check_coverage(True)

        self.assertEqual(returned_output, coverage_report_output)
        self.assertEqual(coverage, 100)

    def test_failure_to_run_test_tasks_throws_error(self) -> None:

        def mock_run_shell_cmd(*_: str, **__: str) -> None:
            raise Exception('XYZ error.')

        swap_run_shell_cmd = self.swap(
            run_backend_tests, 'run_shell_cmd', mock_run_shell_cmd
        )
        swap_hostname = self.swap(socket, 'gethostname', lambda: 'IamEzio')
        swap_getpid = self.swap(os, 'getpid', lambda: 12345)

        task = run_backend_tests.TestingTaskSpec(
            'scripts.run_backend_tests_test', False
        )
        with swap_run_shell_cmd, swap_hostname, swap_getpid:
            with self.assertRaisesRegex(Exception, 'XYZ error.'):
                task.run()

    def test_tasks_run_again_if_race_condition_occurs(self) -> None:

        def mock_run_shell_cmd(*_: str, **__: str) -> str:
            if self.call_count == 1:
                return 'Task result'
            self.call_count = 1
            raise Exception('ev_epollex_linux.cc')

        swap_run_shell_cmd = self.swap(
            run_backend_tests, 'run_shell_cmd', mock_run_shell_cmd
        )
        swapcheck_coverage = self.swap(
            run_backend_tests,
            'check_coverage',
            lambda *unused_args, **unused_kwargs: ('Coverage report', 100.00),
        )

        task = run_backend_tests.TestingTaskSpec(
            'scripts.run_backend_tests_test', True
        )
        with swap_run_shell_cmd, swapcheck_coverage:
            results = task.run()

        self.assertIn('Task result', results[0].messages)

    def test_coverage_is_not_calculated_when_flag_is_not_passed(self) -> None:

        def mock_run_shell_cmd(*_: str, **__: str) -> str:
            if self.call_count == 1:
                return 'Task result'
            self.call_count = 1
            raise Exception('ev_epollex_linux.cc')

        swap_run_shell_cmd = self.swap(
            run_backend_tests, 'run_shell_cmd', mock_run_shell_cmd
        )
        swap_hostname = self.swap(socket, 'gethostname', lambda: 'IamEzio')
        swap_getpid = self.swap(os, 'getpid', lambda: 12345)

        task = run_backend_tests.TestingTaskSpec('scripts.random_test', False)
        with swap_run_shell_cmd, swap_hostname, swap_getpid:
            results = task.run()

        self.assertIn('Task result', results[0].messages)
        self.assertEqual(len(results[0].messages), 1)

    def test_skip_install_with_third_party_installation_skipped(self) -> None:
        with unittest.mock.patch(
            'scripts.install_third_party_libs.main'
        ) as mock_third_party_install:
            swap_check_results = self.swap(
                run_backend_tests,
                'check_test_results',
                lambda *_: (100, 0, 0, {}),
            )
            swap_coverage = self.swap(
                run_backend_tests, 'check_coverage', lambda *_: ('', 100.0)
            )
            with self.swap_redis_server:
                with self.swap_cloud_datastore_emulator, swap_check_results:
                    with swap_coverage, self.swap_execute_task:
                        run_backend_tests.main(args=['--skip_install'])
            mock_third_party_install.assert_not_called()

    def test_third_party_install_with_skip_flag_not_set(self) -> None:
        with unittest.mock.patch(
            'scripts.install_third_party_libs.main'
        ) as mock_third_party_install:
            swap_check_results = self.swap(
                run_backend_tests,
                'check_test_results',
                lambda *_: (100, 0, 0, {}),
            )
            swap_coverage = self.swap(
                run_backend_tests, 'check_coverage', lambda *_: ('', 100.0)
            )
            with self.swap_redis_server:
                with self.swap_cloud_datastore_emulator, swap_check_results:
                    with swap_coverage, self.swap_execute_task:
                        run_backend_tests.main(args=[])
            mock_third_party_install.assert_called_once()

    def test_convert_args_to_pytest_with_verbose_flag(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--verbose', '--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('-v', pytest_args)
        self.assertNotIn('-q', pytest_args)

    def test_convert_args_to_pytest_without_verbose_flag(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('-q', pytest_args)
        self.assertNotIn('-v', pytest_args)

    def test_convert_args_to_pytest_with_coverage_report(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--generate_coverage_report', '--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('--cov=.', pytest_args)
        self.assertIn('--cov-report=term-missing', pytest_args)
        self.assertIn('--cov-fail-under=100', pytest_args)

    def test_convert_args_to_pytest_with_coverage_ignore(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            [
                '--generate_coverage_report',
                '--ignore_coverage',
                '--skip_install',
            ]
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('--cov=.', pytest_args)
        self.assertIn('--cov-report=term-missing', pytest_args)
        self.assertNotIn('--cov-fail-under=100', pytest_args)

    def test_convert_args_to_pytest_with_different_test_targets(self) -> None:
        """Test conversion of various --test_targets formats."""
        test_data: List[Tuple[str, str]] = [
            # Test module.
            ('core.utils_test', 'core/utils_test.py'),
            # Test class.
            (
                'core.utils_test.SingletonMetaTests',
                'core/utils_test.py::SingletonMetaTests',
            ),
            # Test method.
            (
                'core.utils_test.SingletonMetaTests.test_singleton',
                'core/utils_test.py::SingletonMetaTests::test_singleton',
            ),
            # Target without _test suffix.
            ('core.utils', 'core/utils_test.py'),
        ]

        for test_target, expected_path in test_data:
            parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
                [f'--test_targets={test_target}', '--skip_install']
            )
            pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
            self.assertIn(
                expected_path,
                pytest_args,
                msg=f'\n\nTest target: {test_target}',
            )

    def test_convert_args_to_pytest_with_multiple_test_targets(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            [
                '--test_targets=core.utils_test,core.feconf_test',
                '--skip_install',
            ]
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('core/utils_test.py', pytest_args)
        self.assertIn('core/feconf_test.py', pytest_args)

    def test_convert_args_to_pytest_with_test_path(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--test_path=core/controllers', '--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('core/controllers', pytest_args)

    def test_convert_args_to_pytest_with_test_shard(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--test_shard=1', '--skip_install']
        )

        # Mock the shard contents.
        mock_shard_tests = [
            'core.controllers.base_test',
            'core.controllers.editor_test',
            'scripts.linters.run_lint_checks_test',
        ]

        def mock_check_shards_match_tests(
            include_load_tests: bool = True,  # pylint: disable=unused-argument
        ) -> str:
            return ''

        def mock_get_all_test_targets_from_shard(
            shard_name: str,  # pylint: disable=unused-argument
        ) -> List[str]:
            return mock_shard_tests

        swap_check_shards = self.swap(
            run_backend_tests,
            'check_shards_match_tests',
            mock_check_shards_match_tests,
        )
        swap_get_shard_tests = self.swap(
            run_backend_tests,
            'get_all_test_targets_from_shard',
            mock_get_all_test_targets_from_shard,
        )

        with swap_check_shards, swap_get_shard_tests:
            pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)

        # Verify that only the shard tests are included.
        test_file_args = [arg for arg in pytest_args if arg.endswith('.py')]
        self.assertEqual(len(test_file_args), 3)
        self.assertEqual(
            set(test_file_args),
            {
                'core/controllers/base_test.py',
                'core/controllers/editor_test.py',
                'scripts/linters/run_lint_checks_test.py',
            },
        )

    def test_convert_args_to_pytest_with_exclude_load_tests(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--exclude_load_tests', '--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('--ignore=core/tests/load_tests', pytest_args)
        self.assertIn('.', pytest_args)

    def test_convert_args_to_pytest_default_runs_all_tests(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--skip_install']
        )
        pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)
        self.assertIn('.', pytest_args)

    def test_convert_args_to_pytest_with_test_shard_validation_error(
        self,
    ) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--test_shard=1', '--skip_install']
        )

        def mock_check_shards_match_tests(
            include_load_tests: bool = True,  # pylint: disable=unused-argument
        ) -> str:
            return 'Shard validation error'

        swap_check_shards = self.swap(
            run_backend_tests,
            'check_shards_match_tests',
            mock_check_shards_match_tests,
        )

        with swap_check_shards:
            with self.assertRaisesRegex(Exception, 'Shard validation error'):
                run_backend_tests.convert_args_to_pytest(parsed_args)

    def test_convert_args_to_pytest_with_run_on_changed_files(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--run_on_changed_files_in_branch', '--skip_install']
        )

        def mock_get_changed_files() -> Set[str]:
            return {'core.utils_test', 'core.feconf_test'}

        swap_changed_files = self.swap(
            git_changes_utils,
            'get_changed_python_test_files',
            mock_get_changed_files,
        )

        with swap_changed_files:
            pytest_args = run_backend_tests.convert_args_to_pytest(parsed_args)

        # Verify exactly the changed files are included.
        self.assertIn('core/utils_test.py', pytest_args)
        self.assertIn('core/feconf_test.py', pytest_args)
        # Verify no other test files are included.
        test_file_args = [arg for arg in pytest_args if arg.endswith('.py')]
        self.assertEqual(len(test_file_args), 2)
        self.assertEqual(
            set(test_file_args), {'core/utils_test.py', 'core/feconf_test.py'}
        )

    def test_run_tests_with_pytest_success(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--test_targets=core.utils_test', '--skip_install']
        )

        with unittest.mock.patch('pytest.main', return_value=0) as mock_pytest:
            with self.print_swap:
                exit_code = run_backend_tests.run_tests_with_pytest(parsed_args)

        self.assertEqual(exit_code, 0)
        mock_pytest.assert_called_once()
        # Check that pytest was called with correct arguments.
        call_args = mock_pytest.call_args[0][0]
        self.assertIn('core/utils_test.py', call_args)

    def test_run_tests_with_pytest_failure(self) -> None:
        parsed_args = run_backend_tests._PARSER.parse_args(  # pylint: disable=protected-access
            ['--test_targets=core.utils_test', '--skip_install']
        )

        with unittest.mock.patch('pytest.main', return_value=1) as mock_pytest:
            with self.print_swap:
                exit_code = run_backend_tests.run_tests_with_pytest(parsed_args)

        self.assertEqual(exit_code, 1)
        mock_pytest.assert_called_once()

    def test_main_with_use_pytest_flag_success(self) -> None:
        with unittest.mock.patch('pytest.main', return_value=0):
            with self.swap_redis_server:
                with self.swap_cloud_datastore_emulator:
                    with self.print_swap:
                        # Should not raise an exception.
                        run_backend_tests.main(
                            args=[
                                '--use_pytest',
                                '--test_targets=core.utils_test',
                                '--skip_install',
                            ]
                        )
        self.assertIn('Done!', self.print_arr)

    def test_main_with_use_pytest_flag_failure(self) -> None:
        with unittest.mock.patch('pytest.main', return_value=1):
            with self.swap_redis_server:
                with self.swap_cloud_datastore_emulator:
                    with self.print_swap:
                        with self.assertRaisesRegex(
                            Exception, 'Tests failed with exit code 1'
                        ):
                            run_backend_tests.main(
                                args=[
                                    '--use_pytest',
                                    '--test_targets=core.utils_test',
                                    '--skip_install',
                                ]
                            )

    def test_main_with_use_pytest_uses_correct_emulator_context(self) -> None:
        emulator_context_entered = {'cloud_datastore': False, 'redis': False}

        def mock_cloud_datastore_emulator(
            **_: str,
        ) -> MockCompilerContextManager:
            emulator_context_entered['cloud_datastore'] = True
            return MockCompilerContextManager()

        def mock_redis_server(**_: str) -> MockCompilerContextManager:
            emulator_context_entered['redis'] = True
            return MockCompilerContextManager()

        swap_redis = self.swap(
            servers, 'managed_redis_server', mock_redis_server
        )
        swap_datastore = self.swap(
            servers,
            'managed_cloud_datastore_emulator',
            mock_cloud_datastore_emulator,
        )

        with unittest.mock.patch('pytest.main', return_value=0):
            with swap_redis, swap_datastore, self.print_swap:
                run_backend_tests.main(
                    args=[
                        '--use_pytest',
                        '--test_targets=core.utils_test',
                        '--skip_install',
                    ]
                )

        # Verify emulator contexts were entered.
        self.assertTrue(emulator_context_entered['cloud_datastore'])
        self.assertTrue(emulator_context_entered['redis'])