githubmirror/oppia

scripts/common_test.py
1 549 строк57 KB
Sean Lip
Fix #23201: Try to fix some e2e and acceptance tests. Add debug info. (#24011)
12 дек 2025, 01:39
12 дек 2025, 01:39f652db8
# Copyright 2019 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/common.py."""

from __future__ import annotations

import builtins
import contextlib
import errno
import getpass
import http.server
import io
import json
import os
import pathlib
import re
import shutil
import socketserver
import ssl
import stat
import subprocess
import sys
import tempfile
import time
from urllib import error as urlerror
from urllib import request as urlrequest

from core.tests import test_utils

import yaml
from typing import Generator, List, Literal, NoReturn, Tuple

from . import common


class CommonTests(test_utils.GenericTestBase):
    """Test the methods which handle common functionalities."""

    def setUp(self) -> None:
        super().setUp()
        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 tearDown(self) -> None:
        pathlib.Path.unlink(pathlib.Path('mock_app.yaml'), missing_ok=True)
        pathlib.Path.unlink(pathlib.Path('mock_app_dev.yaml'), missing_ok=True)
        super().tearDown()

    @contextlib.contextmanager
    def open_tcp_server_port(self) -> Generator[int, None, None]:
        """Context manager for starting and stoping an HTTP TCP server.

        Yields:
            int. The port number of the server.
        """
        handler = http.server.SimpleHTTPRequestHandler
        # NOTE: Binding to port 0 causes the OS to select a random free port
        # between 1024 to 65535.
        server = socketserver.TCPServer(('localhost', 0), handler)
        try:
            yield server.server_address[1]
        finally:
            server.server_close()

    def test_is_x64_architecture_in_x86(self) -> None:
        maxsize_swap = self.swap(sys, 'maxsize', 1)
        with maxsize_swap:
            self.assertFalse(common.is_x64_architecture())

    def test_is_x64_architecture_in_x64(self) -> None:
        maxsize_swap = self.swap(sys, 'maxsize', 2**32 + 1)
        with maxsize_swap:
            self.assertTrue(common.is_x64_architecture())

    def test_is_mac_os(self) -> None:
        with self.swap(common, 'OS_NAME', 'Darwin'):
            self.assertTrue(common.is_mac_os())
        with self.swap(common, 'OS_NAME', 'Linux'):
            self.assertFalse(common.is_mac_os())

    def test_is_linux_os(self) -> None:
        with self.swap(common, 'OS_NAME', 'Linux'):
            self.assertTrue(common.is_linux_os())
        with self.swap(common, 'OS_NAME', 'Windows'):
            self.assertFalse(common.is_linux_os())

    def test_is_windows_os(self) -> None:
        with self.swap(common, 'OS_NAME', 'Windows'):
            self.assertTrue(common.is_windows_os())
        with self.swap(common, 'OS_NAME', 'Linux'):
            self.assertFalse(common.is_windows_os())

    def test_run_cmd(self) -> None:
        self.assertEqual(
            common.run_cmd(('echo Test for common.py ').split(' ')),
            'Test for common.py',
        )

    def test_ensure_directory_exists_with_existing_dir(self) -> None:
        check_function_calls = {'makedirs_gets_called': False}

        def mock_makedirs(unused_dirpath: str) -> None:
            check_function_calls['makedirs_gets_called'] = True

        with self.swap(os, 'makedirs', mock_makedirs):
            common.ensure_directory_exists('assets')
        self.assertEqual(check_function_calls, {'makedirs_gets_called': False})

    def test_ensure_directory_exists_with_non_existing_dir(self) -> None:
        check_function_calls = {'makedirs_gets_called': False}

        def mock_makedirs(unused_dirpath: str) -> None:
            check_function_calls['makedirs_gets_called'] = True

        with self.swap(os, 'makedirs', mock_makedirs):
            common.ensure_directory_exists('test-dir')
        self.assertEqual(check_function_calls, {'makedirs_gets_called': True})

    def test_require_cwd_to_be_oppia_with_correct_cwd_and_unallowed_deploy_dir(
        self,
    ) -> None:
        common.require_cwd_to_be_oppia()

    def test_require_cwd_to_be_oppia_with_correct_cwd_and_allowed_deploy_dir(
        self,
    ) -> None:
        common.require_cwd_to_be_oppia(allow_deploy_dir=True)

    def test_require_cwd_to_be_oppia_with_wrong_cwd_and_unallowed_deploy_dir(
        self,
    ) -> None:
        def mock_getcwd() -> str:
            return 'invalid'

        getcwd_swap = self.swap(os, 'getcwd', mock_getcwd)
        with getcwd_swap, self.assertRaisesRegex(
            Exception, 'Please run this script from the oppia/ directory.'
        ):
            common.require_cwd_to_be_oppia()

    def test_require_cwd_to_be_oppia_with_wrong_cwd_and_allowed_deploy_dir(
        self,
    ) -> None:
        def mock_getcwd() -> str:
            return 'invalid'

        def mock_basename(unused_dirpath: str) -> str:
            return 'deploy-dir'

        def mock_isdir(unused_dirpath: str) -> Literal[True]:
            return True

        getcwd_swap = self.swap(os, 'getcwd', mock_getcwd)
        basename_swap = self.swap(os.path, 'basename', mock_basename)
        isdir_swap = self.swap(os.path, 'isdir', mock_isdir)
        with getcwd_swap, basename_swap, isdir_swap:
            common.require_cwd_to_be_oppia(allow_deploy_dir=True)

    def test_open_new_tab_in_browser_if_possible_with_user_manually_opening_url(
        self,
    ) -> None:
        try:
            check_function_calls = {
                'input_gets_called': 0,
                'check_call_gets_called': False,
            }
            expected_check_function_calls = {
                'input_gets_called': 1,
                'check_call_gets_called': False,
            }

            def mock_call(unused_cmd_tokens: List[str]) -> int:
                return 0

            def mock_check_call(unused_cmd_tokens: List[str]) -> None:
                check_function_calls['check_call_gets_called'] = True

            def mock_input() -> str:
                check_function_calls['input_gets_called'] += 1
                return 'n'

            call_swap = self.swap(subprocess, 'call', mock_call)
            check_call_swap = self.swap(
                subprocess, 'check_call', mock_check_call
            )
            input_swap = self.swap(builtins, 'input', mock_input)
            with call_swap, check_call_swap, input_swap:
                common.open_new_tab_in_browser_if_possible('test-url')
            self.assertEqual(
                check_function_calls, expected_check_function_calls
            )
        finally:
            common.USER_PREFERENCES['open_new_tab_in_browser'] = None

    def test_open_new_tab_in_browser_if_possible_with_url_opening_correctly(
        self,
    ) -> None:
        try:
            check_function_calls = {
                'input_gets_called': 0,
                'check_call_gets_called': False,
            }
            expected_check_function_calls = {
                'input_gets_called': 2,
                'check_call_gets_called': True,
            }

            def mock_call(unused_cmd_tokens: List[str]) -> int:
                return 0

            def mock_check_call(unused_cmd_tokens: List[str]) -> None:
                check_function_calls['check_call_gets_called'] = True

            def mock_input() -> str:
                check_function_calls['input_gets_called'] += 1
                if check_function_calls['input_gets_called'] == 2:
                    return '1'
                return 'y'

            call_swap = self.swap(subprocess, 'call', mock_call)
            check_call_swap = self.swap(
                subprocess, 'check_call', mock_check_call
            )
            input_swap = self.swap(builtins, 'input', mock_input)
            with call_swap, check_call_swap, input_swap:
                common.open_new_tab_in_browser_if_possible('test-url')
            self.assertEqual(
                check_function_calls, expected_check_function_calls
            )
        finally:
            common.USER_PREFERENCES['open_new_tab_in_browser'] = None

    def test_open_new_tab_in_browser_if_possible_with_url_not_opening_correctly(
        self,
    ) -> None:
        try:
            check_function_calls = {
                'input_gets_called': 0,
                'check_call_gets_called': False,
            }
            expected_check_function_calls = {
                'input_gets_called': 3,
                'check_call_gets_called': False,
            }

            def mock_call(unused_cmd_tokens: List[str]) -> int:
                return 1

            def mock_check_call(unused_cmd_tokens: List[str]) -> None:
                check_function_calls['check_call_gets_called'] = True

            def mock_input() -> str:
                check_function_calls['input_gets_called'] += 1
                if check_function_calls['input_gets_called'] == 2:
                    return '1'
                return 'y'

            call_swap = self.swap(subprocess, 'call', mock_call)
            check_call_swap = self.swap(
                subprocess, 'check_call', mock_check_call
            )
            input_swap = self.swap(builtins, 'input', mock_input)
            with call_swap, check_call_swap, input_swap:
                common.open_new_tab_in_browser_if_possible('test-url')
            self.assertEqual(
                check_function_calls, expected_check_function_calls
            )
        finally:
            common.USER_PREFERENCES['open_new_tab_in_browser'] = None

    def test_open_new_tab_in_browser_if_possible_no_new_tab(self) -> None:
        try:
            check_function_calls = {
                'input_gets_called': 0,
                'check_call_gets_called': False,
            }
            expected_check_function_calls = {
                'input_gets_called': 0,
                'check_call_gets_called': False,
            }

            def mock_call(unused_cmd_tokens: List[str]) -> int:
                return 0

            def mock_check_call(unused_cmd_tokens: List[str]) -> None:
                check_function_calls['check_call_gets_called'] = True

            def mock_input() -> str:
                check_function_calls['input_gets_called'] += 1
                if check_function_calls['input_gets_called'] == 2:
                    return '1'
                return 'no'

            call_swap = self.swap(subprocess, 'call', mock_call)
            check_call_swap = self.swap(
                subprocess, 'check_call', mock_check_call
            )
            input_swap = self.swap(builtins, 'input', mock_input)
            with call_swap, check_call_swap, input_swap:
                # Make it so the program asks the user to
                # Open the link in their browser.
                common.USER_PREFERENCES['open_new_tab_in_browser'] = 'no'
                common.open_new_tab_in_browser_if_possible('test-url')
            self.assertEqual(
                check_function_calls, expected_check_function_calls
            )
        finally:
            common.USER_PREFERENCES['open_new_tab_in_browser'] = None

    def test_get_remote_alias_with_correct_alias(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'remote1 url1\nremote2 url2'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.get_remote_alias(['url1']), 'remote1')

    def test_get_remote_alias_with_incorrect_alias(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'remote1 url1\nremote2 url2'

        check_output_swap = self.swap(
            subprocess, 'check_output', mock_check_output
        )
        with check_output_swap, self.assertRaisesRegex(
            Exception,
            'ERROR: There is no existing remote alias for the url3, url4 repo.',
        ):
            common.get_remote_alias(['url3', 'url4'])

    def test_verify_local_repo_is_clean_with_clean_repo(self) -> None:
        def mock_check_output(unused_cmd_tokens: List[str]) -> bytes:
            return b'nothing to commit, working directory clean'

        with self.swap(subprocess, 'check_output', mock_check_output):
            common.verify_local_repo_is_clean()

    def test_verify_local_repo_is_clean_with_unclean_repo(self) -> None:
        def mock_check_output(unused_cmd_tokens: List[str]) -> bytes:
            return b'invalid'

        check_output_swap = self.swap(
            subprocess, 'check_output', mock_check_output
        )
        with check_output_swap, self.assertRaisesRegex(
            Exception, 'ERROR: This script should be run from a clean branch.'
        ):
            common.verify_local_repo_is_clean()

    def test_get_current_branch_name(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch test'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.get_current_branch_name(), 'test')

    def test_update_branch_with_upstream(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch test'

        def mock_run_cmd(cmd: str) -> str:
            return cmd

        with self.swap(subprocess, 'check_output', mock_check_output):
            with self.swap(common, 'run_cmd', mock_run_cmd):
                common.update_branch_with_upstream()

    def test_get_current_release_version_number_with_non_hotfix_branch(
        self,
    ) -> None:
        self.assertEqual(
            common.get_current_release_version_number('release-1.2.3'), '1.2.3'
        )

    def test_get_current_release_version_number_with_hotfix_branch(
        self,
    ) -> None:
        self.assertEqual(
            common.get_current_release_version_number('release-1.2.3-hotfix-1'),
            '1.2.3',
        )

    def test_get_current_release_version_number_with_maintenance_branch(
        self,
    ) -> None:
        self.assertEqual(
            common.get_current_release_version_number(
                'release-maintenance-1.2.3'
            ),
            '1.2.3',
        )

    def test_get_current_release_version_number_with_invalid_branch(
        self,
    ) -> None:
        with self.assertRaisesRegex(
            Exception, 'Invalid branch name: invalid-branch.'
        ):
            common.get_current_release_version_number('invalid-branch')

    def test_is_current_branch_a_hotfix_branch_with_non_hotfix_branch(
        self,
    ) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch release-1.2.3'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_hotfix_branch(), False)

    def test_is_current_branch_a_hotfix_branch_with_hotfix_branch(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch release-1.2.3-hotfix-1'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_hotfix_branch(), True)

    def test_is_current_branch_a_release_branch_with_release_branch(
        self,
    ) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch release-1.2.3'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_release_branch(), True)

    def test_is_current_branch_a_release_branch_with_hotfix_branch(
        self,
    ) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch release-1.2.3-hotfix-1'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_release_branch(), True)

    def test_is_current_branch_a_release_branch_with_maintenance_branch(
        self,
    ) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch release-maintenance-1.2.3'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_release_branch(), True)

    def test_is_current_branch_a_release_branch_with_non_release_branch(
        self,
    ) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch test'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_release_branch(), False)

    def test_is_current_branch_a_test_branch_with_test_branch(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch test-common'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_test_branch(), True)

    def test_is_current_branch_a_test_branch_with_non_test_branch(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch invalid-test'

        with self.swap(subprocess, 'check_output', mock_check_output):
            self.assertEqual(common.is_current_branch_a_test_branch(), False)

    def test_verify_current_branch_name_with_correct_branch(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch test'

        with self.swap(subprocess, 'check_output', mock_check_output):
            common.verify_current_branch_name('test')

    def test_verify_current_branch_name_with_incorrect_branch(self) -> None:
        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'On branch invalid'

        check_output_swap = self.swap(
            subprocess, 'check_output', mock_check_output
        )
        with check_output_swap, self.assertRaisesRegex(
            Exception,
            'ERROR: This script can only be run from the "test" branch.',
        ):
            common.verify_current_branch_name('test')

    def test_is_port_in_use(self) -> None:
        with self.open_tcp_server_port() as port:
            self.assertTrue(common.is_port_in_use(port))
        self.assertFalse(common.is_port_in_use(port))

    def test_get_ports_in_use(self) -> None:
        with self.open_tcp_server_port() as port_in_use:
            # Test with one port in use and one not in use.
            ports = [port_in_use, port_in_use + 1]
            in_use = common.get_ports_in_use(ports)
            self.assertEqual(in_use, [port_in_use])

            # Test with no ports in use.
            ports_not_in_use = [port_in_use + 1, port_in_use + 2]
            in_use_empty = common.get_ports_in_use(ports_not_in_use)
            self.assertEqual(in_use_empty, [])

            # Test with all ports in use.
            with self.open_tcp_server_port() as port_in_use2:
                ports_all_in_use = [port_in_use, port_in_use2]
                in_use_all = common.get_ports_in_use(ports_all_in_use)
                self.assertEqual(set(in_use_all), set(ports_all_in_use))

    def test_wait_for_port_to_not_be_in_use_port_never_closes(self) -> None:
        def mock_sleep(unused_seconds: int) -> None:
            return

        def mock_is_port_in_use(unused_port_number: int) -> Literal[True]:
            return True

        sleep_swap = self.swap_with_checks(
            time, 'sleep', mock_sleep, expected_args=[(1,)] * 60
        )
        is_port_in_use_swap = self.swap(
            common, 'is_port_in_use', mock_is_port_in_use
        )

        with sleep_swap, is_port_in_use_swap:
            success = common.wait_for_port_to_not_be_in_use(9999)
        self.assertFalse(success)

    def test_wait_for_port_to_not_be_in_use_port_closes(self) -> None:
        def mock_sleep(unused_seconds: int) -> NoReturn:
            raise AssertionError('mock_sleep should not be called.')

        def mock_is_port_in_use(unused_port_number: int) -> Literal[False]:
            return False

        sleep_swap = self.swap(time, 'sleep', mock_sleep)
        is_port_in_use_swap = self.swap(
            common, 'is_port_in_use', mock_is_port_in_use
        )

        with sleep_swap, is_port_in_use_swap:
            success = common.wait_for_port_to_not_be_in_use(9999)
        self.assertTrue(success)

    def test_wait_for_port_to_be_in_use_port_never_opens(self) -> None:
        def mock_sleep(unused_seconds: int) -> None:
            return

        def mock_is_port_in_use(unused_port_number: int) -> Literal[False]:
            return False

        def mock_exit(unused_code: str) -> None:
            pass

        sleep_swap = self.swap_with_checks(
            time, 'sleep', mock_sleep, expected_args=[(1,)] * 60 * 5
        )
        is_port_in_use_swap = self.swap(
            common, 'is_port_in_use', mock_is_port_in_use
        )
        exit_swap = self.swap_with_checks(
            sys, 'exit', mock_exit, expected_args=[(1,)]
        )

        with sleep_swap, is_port_in_use_swap, exit_swap:
            common.wait_for_port_to_be_in_use(9999)

    def test_wait_for_port_to_be_in_use_port_opens(self) -> None:
        def mock_sleep(unused_seconds: int) -> NoReturn:
            raise AssertionError('mock_sleep should not be called.')

        def mock_is_port_in_use(unused_port_number: int) -> Literal[True]:
            return True

        def mock_exit(unused_code: str) -> NoReturn:
            raise AssertionError('mock_exit should not be called.')

        sleep_swap = self.swap(time, 'sleep', mock_sleep)
        is_port_in_use_swap = self.swap(
            common, 'is_port_in_use', mock_is_port_in_use
        )
        exit_swap = self.swap(sys, 'exit', mock_exit)

        with sleep_swap, is_port_in_use_swap, exit_swap:
            common.wait_for_port_to_be_in_use(9999)

    def test_wait_for_firebase_emulator_to_be_ready_success(self) -> None:
        class MockResponse:
            status = 200

        def mock_urlopen(
            unused_url: str, timeout: int  # pylint: disable=unused-argument
        ) -> MockResponse:
            return MockResponse()

        urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

        with urlopen_swap:
            # Should not raise an exception.
            common.wait_for_firebase_emulator_to_be_ready(9099)

    def test_wait_for_firebase_emulator_to_be_ready_retries_on_connection_error(
        self,
    ) -> None:
        attempt_count = 0

        class MockResponse:
            status = 200

        def mock_urlopen(
            unused_url: str, timeout: int  # pylint: disable=unused-argument
        ) -> MockResponse:
            nonlocal attempt_count
            attempt_count += 1
            if attempt_count < 3:
                raise urlerror.URLError('Connection refused')
            return MockResponse()

        def mock_sleep(unused_seconds: int) -> None:
            pass

        urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)
        sleep_swap = self.swap(time, 'sleep', mock_sleep)

        with urlopen_swap, sleep_swap:
            common.wait_for_firebase_emulator_to_be_ready(9099)

        self.assertEqual(attempt_count, 3)

    def test_wait_for_firebase_emulator_to_be_ready_timeout(self) -> None:
        def mock_urlopen(
            unused_url: str, timeout: int  # pylint: disable=unused-argument
        ) -> NoReturn:
            raise urlerror.URLError('Connection refused')

        def mock_sleep(unused_seconds: int) -> None:
            pass

        urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)
        sleep_swap = self.swap(time, 'sleep', mock_sleep)

        with urlopen_swap, sleep_swap:
            with self.assertRaisesRegex(
                Exception,
                'Firebase Auth emulator failed to become ready on port 9099',
            ):
                common.wait_for_firebase_emulator_to_be_ready(
                    9099, timeout_secs=5
                )

    def test_permissions_of_file(self) -> None:
        root_temp_dir = tempfile.mkdtemp()
        temp_dirpath = tempfile.mkdtemp(dir=root_temp_dir)
        temp_file = tempfile.NamedTemporaryFile(dir=temp_dirpath)
        # Here MyPy assumes that the 'name' attribute is read-only. In order to
        # silence the MyPy complaints `setattr` is used to set the attribute.
        setattr(temp_file, 'name', 'temp_file')
        temp_file_path = os.path.join(temp_dirpath, 'temp_file')
        with open(temp_file_path, 'w', encoding='utf-8') as f:
            f.write('content')

        common.recursive_chown(root_temp_dir, os.getuid(), -1)
        common.recursive_chmod(root_temp_dir, 0o744)

        for root, directories, filenames in os.walk(root_temp_dir):
            for directory in directories:
                self.assertEqual(
                    oct(
                        stat.S_IMODE(
                            os.stat(os.path.join(root, directory)).st_mode
                        )
                    ),
                    '0o744',
                )
                self.assertEqual(
                    os.stat(os.path.join(root, directory)).st_uid, os.getuid()
                )

            for filename in filenames:
                self.assertEqual(
                    oct(
                        stat.S_IMODE(
                            os.stat(os.path.join(root, filename)).st_mode
                        )
                    ),
                    '0o744',
                )
                self.assertEqual(
                    os.stat(os.path.join(root, filename)).st_uid, os.getuid()
                )

        temp_file.close()
        shutil.rmtree(root_temp_dir)

    def test_print_each_string_after_two_new_lines(self) -> None:
        @contextlib.contextmanager
        def _redirect_stdout(
            new_target: io.TextIOWrapper,
        ) -> Generator[io.TextIOWrapper, None, None]:
            """Redirect stdout to the new target.

            Args:
                new_target: TextIOWrapper. The new target to which stdout is
                    redirected.

            Yields:
                TextIOWrapper. The new target.
            """
            old_target = sys.stdout
            sys.stdout = new_target
            try:
                yield new_target
            finally:
                sys.stdout = old_target

        target_stdout = io.StringIO()
        with _redirect_stdout(target_stdout):
            common.print_each_string_after_two_new_lines(
                ['These', 'are', 'sample', 'strings.']
            )

        self.assertEqual(
            target_stdout.getvalue(), 'These\n\nare\n\nsample\n\nstrings.\n\n'
        )

    def test_install_npm_library(self) -> None:
        def _mock_subprocess_check_call(unused_command: str) -> None:
            """Mocks subprocess.check_call() to create a temporary file instead
            of the actual npm library.
            """
            temp_file = tempfile.NamedTemporaryFile()
            # Here MyPy assumes that the 'name' attribute is read-only.
            # In order to silence the MyPy complaints `setattr` is used to set
            # the attribute.
            setattr(temp_file, 'name', 'temp_file')
            with open('temp_file', 'w', encoding='utf-8') as f:
                f.write('content')

            self.assertTrue(os.path.exists('temp_file'))
            temp_file.close()
            if os.path.isfile('temp_file'):
                # Occasionally this temp file is not deleted.
                os.remove('temp_file')

        self.assertFalse(os.path.exists('temp_file'))

        with self.swap(subprocess, 'check_call', _mock_subprocess_check_call):
            common.install_npm_library('library_name', 'version', 'path')

        self.assertFalse(os.path.exists('temp_file'))

    def test_install_npm_library_path_exists(self) -> None:
        """Install an npm library that already exists."""

        def mock_exists(unused_file: str) -> bool:
            return True

        with self.swap(os.path, 'exists', mock_exists):
            common.install_npm_library(
                'moment', '2.29.4', common.OPPIA_TOOLS_DIR
            )

    def test_ask_user_to_confirm(self) -> None:
        def mock_input() -> str:
            return 'Y'

        with self.swap(builtins, 'input', mock_input):
            common.ask_user_to_confirm('Testing')

    def test_ask_user_to_confirm_n_then_y(self) -> None:
        check_function_calls = {
            'input_gets_called': 0,
        }

        def mock_input() -> str:
            check_function_calls['input_gets_called'] += 1
            if check_function_calls['input_gets_called'] == 1:
                return 'N'
            return 'Y'

        with self.swap(builtins, 'input', mock_input):
            common.ask_user_to_confirm('Testing')

    def test_get_personal_access_token_with_valid_token(self) -> None:
        def mock_getpass(prompt: str) -> str:  # pylint: disable=unused-argument
            return 'token'

        with self.swap(getpass, 'getpass', mock_getpass):
            self.assertEqual(common.get_personal_access_token(), 'token')

    def test_get_personal_access_token_with_token_as_none(self) -> None:
        def mock_getpass(  # pylint: disable=unused-argument
            prompt: str,
        ) -> None:
            return None

        getpass_swap = self.swap(getpass, 'getpass', mock_getpass)
        with getpass_swap, self.assertRaisesRegex(
            Exception,
            'No personal access token provided, please set up a personal '
            'access token at https://github.com/settings/tokens and re-run '
            'the script',
        ):
            common.get_personal_access_token()

    def test_inplace_replace_file(self) -> None:
        origin_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json'
        )

        backup_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json.bak'
        )
        shutil.copyfile(origin_filepath, backup_filepath)

        expected_lines = [
            '{\n',
            '    "RANDMON1" : "randomValue1",\n',
            '    "312RANDOM" : "ValueRanDom2",\n',
            '    "DEV_MODE": true,\n',
            '    "RAN213DOM" : "raNdoVaLue3"\n',
            '}\n',
        ]

        common.inplace_replace_file(
            origin_filepath,
            '"DEV_MODE": .*',
            '"DEV_MODE": true,',
            expected_number_of_replacements=1,
        )

        with open(origin_filepath, 'r', encoding='utf-8') as f:
            self.assertEqual(expected_lines, f.readlines())
        # Revert the file.
        shutil.move(backup_filepath, origin_filepath)

    def test_inplace_replace_file_with_expected_number_of_replacements_raises(
        self,
    ) -> None:
        origin_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json'
        )
        new_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json.new'
        )

        backup_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json.bak'
        )
        shutil.copyfile(origin_filepath, backup_filepath)

        with open(origin_filepath, 'r', encoding='utf-8') as f:
            origin_content = f.readlines()

        with self.assertRaisesRegex(
            ValueError, 'Wrong number of replacements. Expected 1. Performed 0.'
        ):
            common.inplace_replace_file(
                origin_filepath,
                '"DEV_MODEa": .*',
                '"DEV_MODE": true,',
                expected_number_of_replacements=1,
            )
        self.assertFalse(os.path.isfile(new_filepath))
        with open(origin_filepath, 'r', encoding='utf-8') as f:
            new_content = f.readlines()
        self.assertEqual(origin_content, new_content)
        # Revert the file.
        shutil.move(backup_filepath, origin_filepath)

    def test_inplace_replace_file_with_exception_raised(self) -> None:
        origin_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json'
        )
        new_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json.new'
        )

        backup_filepath = os.path.join(
            'core', 'tests', 'data', 'inplace_replace_test.json.bak'
        )
        shutil.copyfile(origin_filepath, backup_filepath)

        with open(origin_filepath, 'r', encoding='utf-8') as f:
            origin_content = f.readlines()

        def mock_compile(unused_arg: str) -> NoReturn:
            raise ValueError('Exception raised from compile()')

        compile_swap = self.swap_with_checks(re, 'compile', mock_compile)
        with self.assertRaisesRegex(
            ValueError, re.escape('Exception raised from compile()')
        ), compile_swap:
            common.inplace_replace_file(
                origin_filepath, '"DEV_MODE": .*', '"DEV_MODE": true,'
            )
        self.assertFalse(os.path.isfile(new_filepath))
        with open(origin_filepath, 'r', encoding='utf-8') as f:
            new_content = f.readlines()
        self.assertEqual(origin_content, new_content)
        # Revert the file.
        shutil.move(backup_filepath, origin_filepath)

    def test_create_readme(self) -> None:
        try:
            os.makedirs('readme_test_dir')
            common.create_readme('readme_test_dir', 'Testing readme.')
            with open('readme_test_dir/README.md', 'r', encoding='utf-8') as f:
                self.assertEqual(f.read(), 'Testing readme.')
        finally:
            if os.path.exists('readme_test_dir'):
                shutil.rmtree('readme_test_dir')

    def test_cd(self) -> None:
        def mock_chdir(unused_path: str) -> None:
            pass

        def mock_getcwd() -> str:
            return '/old/path'

        chdir_swap = self.swap_with_checks(
            os,
            'chdir',
            mock_chdir,
            expected_args=[
                ('/new/path',),
                ('/old/path',),
            ],
        )
        getcwd_swap = self.swap(os, 'getcwd', mock_getcwd)

        with chdir_swap, getcwd_swap:
            with common.CD('/new/path'):
                pass

    def test_swap_env_when_var_had_a_value(self) -> None:
        os.environ['ABC'] = 'Hard as Rocket Science'
        with common.swap_env('ABC', 'Easy as 123') as old_value:
            self.assertEqual(old_value, 'Hard as Rocket Science')
            self.assertEqual(os.environ['ABC'], 'Easy as 123')
        self.assertEqual(os.environ['ABC'], 'Hard as Rocket Science')

    def test_swap_env_when_var_did_not_exist(self) -> None:
        self.assertNotIn('DEF', os.environ)
        with common.swap_env('DEF', 'Easy as 123') as old_value:
            self.assertIsNone(old_value)
            self.assertEqual(os.environ['DEF'], 'Easy as 123')
        self.assertNotIn('DEF', os.environ)

    def test_write_stdout_safe_with_repeat_oserror_repeats_call_to_write(
        self,
    ) -> None:
        raised_once = False

        def write_raise_oserror(
            unused_fileno: int, bytes_to_write: bytes
        ) -> int:
            self.assertEqual(bytes_to_write, 'test'.encode('utf-8'))

            nonlocal raised_once
            if not raised_once:
                raised_once = True
                raise OSError(errno.EAGAIN, 'OS error that should be repeated')

            return 4

        write_swap = self.swap_with_checks(
            os,
            'write',
            write_raise_oserror,
            expected_args=(
                (sys.stdout.fileno(), b'test'),
                (sys.stdout.fileno(), b'test'),
            ),
        )
        with write_swap:
            # This test makes sure that when write fails (with errno.EAGAIN)
            # the call is repeated.
            common.write_stdout_safe('test')

        self.assertTrue(raised_once)

    def test_write_stdout_safe_with_oserror(self) -> None:
        write_swap = self.swap_to_always_raise(os, 'write', OSError('OS error'))
        with write_swap, self.assertRaisesRegex(OSError, 'OS error'):
            common.write_stdout_safe('test')

    def test_write_stdout_safe_with_unsupported_operation(self) -> None:
        mock_stdout = io.StringIO()

        write_swap = self.swap_to_always_raise(
            os, 'write', io.UnsupportedOperation('unsupported operation')
        )
        stdout_write_swap = self.swap(sys, 'stdout', mock_stdout)

        with write_swap, stdout_write_swap:
            common.write_stdout_safe('test')
        self.assertEqual(mock_stdout.getvalue(), 'test')

    def _assert_ssl_context_matches_default(
        self, context: ssl.SSLContext
    ) -> None:
        """Assert that an SSL context matches the default one.

        If we create two default SSL contexts, they will evaluate as unequal
        even though they are the same for our purposes. Therefore, this function
        checks that the provided context has the same important security
        properties as the default.

        Args:
            context: SSLContext. The context to compare.

        Raises:
            AssertionError. Raised if the contexts differ in any of their
                important attributes or behaviors.
        """
        default_context = ssl.create_default_context()
        for attribute in (
            'verify_flags',
            'verify_mode',
            'protocol',
            'hostname_checks_common_name',
            'options',
            'minimum_version',
            'maximum_version',
            'check_hostname',
        ):
            self.assertEqual(
                getattr(context, attribute), getattr(default_context, attribute)
            )
        for method in ('get_ca_certs', 'get_ciphers'):
            self.assertEqual(
                getattr(context, method)(), getattr(default_context, method)()
            )

    def test_chrome_bin_setup_with_google_chrome(self) -> None:
        isfile_swap = self.swap(
            os.path, 'isfile', lambda path: path == '/usr/bin/google-chrome'
        )
        with isfile_swap:
            common.setup_chrome_bin_env_variable()
        self.assertEqual(os.environ['CHROME_BIN'], '/usr/bin/google-chrome')

    def test_chrome_bin_setup_with_wsl_chrome_browser(self) -> None:
        isfile_swap = self.swap(
            os.path,
            'isfile',
            lambda path: path
            == (
                '/mnt/c/Program Files (x86)/Google/'
                'Chrome/Application/chrome.exe'
            ),
        )
        with isfile_swap:
            common.setup_chrome_bin_env_variable()
        self.assertEqual(
            os.environ['CHROME_BIN'],
            '/mnt/c/Program Files (x86)/Google/Chrome/Application/chrome.exe',
        )

    def test_chrome_bin_setup_with_error(self) -> None:
        print_arr = []

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

        isfile_swap = self.swap(os.path, 'isfile', lambda _: False)
        print_swap = self.swap(builtins, 'print', mock_print)

        with print_swap, isfile_swap, self.assertRaisesRegex(
            Exception, 'Chrome not found.'
        ):
            common.setup_chrome_bin_env_variable()
        self.assertIn('Chrome is not found, stopping...', print_arr)

    def test_modify_constants(self) -> None:
        mock_constants_path = 'mock_app_dev.yaml'
        mock_feconf_path = 'mock_app.yaml'
        constants_path_swap = self.swap(
            common, 'CONSTANTS_FILE_PATH', mock_constants_path
        )
        feconf_path_swap = self.swap(common, 'FECONF_PATH', mock_feconf_path)

        def mock_check_output(
            unused_cmd_tokens: List[str],
            encoding: str = 'utf-8',  # pylint: disable=unused-argument
        ) -> str:
            return 'test'

        check_output_swap = self.swap(
            subprocess, 'check_output', mock_check_output
        )

        constants_temp_file = tempfile.NamedTemporaryFile()
        # Here MyPy assumes that the 'name' attribute is read-only.
        # In order to silence the MyPy complaints `setattr` is used
        # to set the attribute.
        setattr(constants_temp_file, 'name', mock_constants_path)
        with open(mock_constants_path, 'w', encoding='utf-8') as tmp:
            tmp.write('export = {\n')
            tmp.write('  "DEV_MODE": true,\n')
            tmp.write('  "EMULATOR_MODE": false,\n')
            tmp.write('  "BRANCH_NAME": "",\n')
            tmp.write('  "SHORT_COMMIT_HASH": ""\n')
            tmp.write('};')

        feconf_temp_file = tempfile.NamedTemporaryFile()
        # Here MyPy assumes that the 'name' attribute is read-only.
        # In order to silence the MyPy complaints `setattr` is used
        # to set the attribute.
        setattr(feconf_temp_file, 'name', mock_feconf_path)
        with open(mock_feconf_path, 'w', encoding='utf-8') as tmp:
            tmp.write('ENABLE_MAINTENANCE_MODE = False')

        with constants_path_swap, feconf_path_swap, check_output_swap:
            common.modify_constants(prod_env=True, maintenance_mode=False)
            with open(
                mock_constants_path, 'r', encoding='utf-8'
            ) as constants_file:
                self.assertEqual(
                    constants_file.read(),
                    'export = {\n'
                    '  "DEV_MODE": false,\n'
                    '  "EMULATOR_MODE": true,\n'
                    '  "BRANCH_NAME": "test",\n'
                    '  "SHORT_COMMIT_HASH": "test"\n'
                    '};',
                )
            with open(mock_feconf_path, 'r', encoding='utf-8') as feconf_file:
                self.assertEqual(
                    feconf_file.read(), 'ENABLE_MAINTENANCE_MODE = False'
                )

            common.modify_constants(prod_env=False, maintenance_mode=True)
            with open(
                mock_constants_path, 'r', encoding='utf-8'
            ) as constants_file:
                self.assertEqual(
                    constants_file.read(),
                    'export = {\n'
                    '  "DEV_MODE": true,\n'
                    '  "EMULATOR_MODE": true,\n'
                    '  "BRANCH_NAME": "test",\n'
                    '  "SHORT_COMMIT_HASH": "test"\n'
                    '};',
                )
            with open(mock_feconf_path, 'r', encoding='utf-8') as feconf_file:
                self.assertEqual(
                    feconf_file.read(), 'ENABLE_MAINTENANCE_MODE = True'
                )

        constants_temp_file.close()
        feconf_temp_file.close()

        # Clean up spare files.
        os.remove(mock_constants_path)
        os.remove(mock_feconf_path)

    def test_set_constants_to_default(self) -> None:
        mock_constants_path = 'mock_app_dev.yaml'
        mock_feconf_path = 'mock_app.yaml'
        constants_path_swap = self.swap(
            common, 'CONSTANTS_FILE_PATH', mock_constants_path
        )
        feconf_path_swap = self.swap(common, 'FECONF_PATH', mock_feconf_path)

        constants_temp_file = tempfile.NamedTemporaryFile()
        # Here MyPy assumes that the 'name' attribute is read-only. In order to
        # silence the MyPy complaints `setattr` is used to set the attribute.
        setattr(constants_temp_file, 'name', mock_constants_path)
        with open(mock_constants_path, 'w', encoding='utf-8') as tmp:
            tmp.write('export = {\n')
            tmp.write('  "DEV_MODE": false,\n')
            tmp.write('  "EMULATOR_MODE": false,\n')
            tmp.write('  "BRANCH_NAME": "test",\n')
            tmp.write('  "SHORT_COMMIT_HASH": "test"\n')
            tmp.write('};')

        feconf_temp_file = tempfile.NamedTemporaryFile()
        # Here MyPy assumes that the 'name' attribute is read-only. In order to
        # silence the MyPy complaints `setattr` is used to set the attribute.
        setattr(feconf_temp_file, 'name', mock_feconf_path)
        with open(mock_feconf_path, 'w', encoding='utf-8') as tmp:
            tmp.write('ENABLE_MAINTENANCE_MODE = True')
        self.contextManager.__exit__(None, None, None)
        with constants_path_swap, feconf_path_swap:
            common.set_constants_to_default()
            with open(
                mock_constants_path, 'r', encoding='utf-8'
            ) as constants_file:
                self.assertEqual(
                    constants_file.read(),
                    'export = {\n'
                    '  "DEV_MODE": true,\n'
                    '  "EMULATOR_MODE": true,\n'
                    '  "BRANCH_NAME": "",\n'
                    '  "SHORT_COMMIT_HASH": ""\n'
                    '};',
                )
            with open(mock_feconf_path, 'r', encoding='utf-8') as feconf_file:
                self.assertEqual(
                    feconf_file.read(), 'ENABLE_MAINTENANCE_MODE = False'
                )
        constants_temp_file.close()
        feconf_temp_file.close()

        # Clean up spare files.
        os.remove(mock_constants_path)
        os.remove(mock_feconf_path)

    def test_is_oppia_server_already_running_when_ports_closed(self) -> None:
        with contextlib.ExitStack() as stack:
            stack.enter_context(
                self.swap_to_always_return(
                    common, 'is_port_in_use', value=False
                )
            )

            self.assertFalse(common.is_oppia_server_already_running())

    def test_is_oppia_server_already_running_when_a_port_is_open(self) -> None:
        with contextlib.ExitStack() as stack:
            stack.enter_context(
                self.swap_with_checks(
                    common,
                    'is_port_in_use',
                    lambda port: port == common.GAE_PORT_FOR_E2E_TESTING,
                )
            )

            self.assertTrue(common.is_oppia_server_already_running())

    def test_start_subprocess_for_result(self) -> None:
        process = subprocess.Popen(
            ['echo', 'test'], stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )

        def mock_popen(  # pylint: disable=unused-argument
            cmd_tokens: List[str], stdout: int, stderr: int
        ) -> subprocess.Popen[bytes]:
            return process

        popen_swap = self.swap(subprocess, 'Popen', mock_popen)

        with popen_swap:
            self.assertEqual(
                common.start_subprocess_for_result(['cmd']), (b'test\n', b'')
            )

    def test_workflow_permissions_set_to_read_all(self) -> None:
        workflows_dir = os.path.join(os.getcwd(), '.github', 'workflows')
        self.assertTrue(
            os.path.isdir(workflows_dir),
            f'{workflows_dir} directory not found.',
        )

        for filename in os.listdir(workflows_dir):
            if filename.endswith(('.yaml', '.yml')):
                filepath = os.path.join(workflows_dir, filename)
                with open(filepath, 'r', encoding='utf-8') as file:
                    try:
                        workflow_data = yaml.safe_load(file)
                        permissions = workflow_data.get('permissions')
                        self.assertEqual(
                            permissions,
                            'read-all',
                            f'Workflow file "{filename}" is missing a '
                            '"permissions: read-all" field.',
                        )
                    except yaml.YAMLError as e:
                        self.fail(f'Error parsing file "{filename}": {str(e)}')

    def test_write_hashes_json_file(self) -> None:
        """Test write_hashes_json_file writes provided hash dict correctly to
        JSON file.
        """
        with tempfile.TemporaryDirectory() as tmpdir:
            hashes_path = os.path.join(tmpdir, 'hashes.json')

            # Test writing a simple hash dict.
            hashes = {'path/file.js': '123456'}
            with self.swap(common, 'HASHES_JSON_FILEPATH', hashes_path):
                common.write_hashes_json_file(hashes)
            with open(hashes_path, 'r', encoding='utf-8') as hashes_file:
                self.assertEqual(
                    json.loads(hashes_file.read()),
                    {'path/file.js': '123456'},
                )

            # Test writing multiple hashes.
            hashes = {'file.js': '123456', 'file.min.js': '654321'}
            with self.swap(common, 'HASHES_JSON_FILEPATH', hashes_path):
                common.write_hashes_json_file(hashes)
            with open(hashes_path, 'r', encoding='utf-8') as hashes_file:
                self.assertEqual(
                    json.loads(hashes_file.read()),
                    {'file.min.js': '654321', 'file.js': '123456'},
                )

            # Test writing an empty dict (used by dev/test scripts).
            with self.swap(common, 'HASHES_JSON_FILEPATH', hashes_path):
                common.write_hashes_json_file({})
            with open(hashes_path, 'r', encoding='utf-8') as hashes_file:
                self.assertEqual(json.loads(hashes_file.read()), {})


class UrlRetrieveTests(CommonTests):
    """Test the methods related to retrieving URLs."""

    def setUp(self) -> None:
        super().setUp()

        self.curl_is_called = False

        successful_mock_curl_process = subprocess.Popen(
            ['echo', 'test'], stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )

        def mock_successful_curl_popen(  # pylint: disable=unused-argument
            cmd_tokens: List[str], stdout: int, stderr: int, encoding: str
        ) -> subprocess.Popen[bytes]:
            self.assertEqual(cmd_tokens[0], 'curl')
            self.curl_is_called = True
            return successful_mock_curl_process

        class MockErrorProcess:
            def __init__(self) -> None:
                self.returncode = 1

            def communicate(self) -> Tuple[str, str]:
                """Return required method."""
                return '', 'Failure starting curl'

            def __enter__(self) -> None:
                pass

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

        def mock_failing_curl_popen(  # pylint: disable=unused-argument
            cmd_tokens: List[str], stdout: int, stderr: int, encoding: str
        ) -> MockErrorProcess:
            self.assertEqual(cmd_tokens[0], 'curl')
            return MockErrorProcess()

        self.swap_curl_success = self.swap(
            subprocess, 'Popen', mock_successful_curl_popen
        )
        self.swap_curl_failure = self.swap(
            subprocess, 'Popen', mock_failing_curl_popen
        )

    def test_url_open(self) -> None:
        # Use a URL that Oppia's CI environment is expected to have access to.
        github_api_url = (
            'https://api.github.com/repos/oppia/oppia/releases/latest'
        )
        response = common.url_open(github_api_url)
        self.assertEqual(response.getcode(), 200)
        self.assertEqual(response.url, github_api_url)

    def test_url_retrieve_tries_curl_at_outset(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            output_path = os.path.join(tempdir, 'output')

            function_call_records = {
                'urlopen': False,
            }

            def mock_urlopen(
                url: str, context: ssl.SSLContext
            ) -> io.BufferedIOBase:
                self.assertEqual(url, 'https://example.com')
                self._assert_ssl_context_matches_default(context)
                function_call_records['urlopen'] = True
                return io.BytesIO(b'content')

            urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)
            with urlopen_swap, self.swap_curl_success:
                common.url_retrieve('https://example.com', output_path)
            self.assertTrue(self.curl_is_called)
            self.assertFalse(function_call_records['urlopen'])

    def test_url_retrieve_with_successful_https_works(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            output_path = os.path.join(tempdir, 'buffer')
            attempts = []

            def mock_urlopen(
                url: str, context: ssl.SSLContext
            ) -> io.BufferedIOBase:
                attempts.append(url)
                self.assertLessEqual(len(attempts), 1)
                self.assertEqual(url, 'https://example.com')
                self._assert_ssl_context_matches_default(context)
                return io.BytesIO(b'content')

            urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

            with urlopen_swap, self.swap_curl_failure:
                common.url_retrieve('https://example.com', output_path)
            with open(output_path, 'rb', encoding=None) as buffer:
                self.assertEqual(buffer.read(), b'content')

    def test_url_retrieve_with_successful_https_works_on_retry(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            output_path = os.path.join(tempdir, 'output')
            attempts = []

            def mock_urlopen(
                url: str, context: ssl.SSLContext
            ) -> io.BufferedIOBase:
                attempts.append(url)
                self.assertLessEqual(len(attempts), 2)
                self.assertEqual(url, 'https://example.com')
                self._assert_ssl_context_matches_default(context)
                if len(attempts) == 1:
                    raise ssl.SSLError()
                return io.BytesIO(b'content')

            urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

            with urlopen_swap, self.swap_curl_failure:
                common.url_retrieve('https://example.com', output_path)
            with open(output_path, 'rb', encoding=None) as buffer:
                self.assertEqual(buffer.read(), b'content')

    def test_url_retrieve_runs_out_of_attempts(self) -> None:
        attempts = []

        def mock_open(
            _path: str, _options: str, encoding: str = 'utf-8'
        ) -> NoReturn:
            raise AssertionError('open() should not be called')

        def mock_urlopen(
            url: str, context: ssl.SSLContext
        ) -> io.BufferedIOBase:
            attempts.append(url)
            self.assertLessEqual(len(attempts), 2)
            self.assertEqual(url, 'https://example.com')
            self._assert_ssl_context_matches_default(context)
            raise ssl.SSLError('test_error')

        open_swap = self.swap(builtins, 'open', mock_open)
        urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

        with open_swap, urlopen_swap, self.swap_curl_failure:
            with self.assertRaisesRegex(ssl.SSLError, 'test_error'):
                common.url_retrieve('https://example.com', 'test_path')

    def test_url_retrieve_https_check_fails(self) -> None:
        def mock_open(
            _path: str, _options: str, encoding: str = 'utf-8'
        ) -> NoReturn:
            raise AssertionError('open() should not be called')

        def mock_urlopen(
            url: str, context: ssl.SSLContext
        ) -> NoReturn:  # pylint: disable=unused-argument
            raise AssertionError('urlopen() should not be called')

        open_swap = self.swap(builtins, 'open', mock_open)
        urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

        with open_swap, urlopen_swap, self.swap_curl_failure:
            with self.assertRaisesRegex(
                Exception, 'The URL http://example.com should use HTTPS.'
            ):
                common.url_retrieve('http://example.com', 'test_path')

    def test_url_retrieve_with_successful_http_works(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            output_path = os.path.join(tempdir, 'output')
            attempts = []

            def mock_urlopen(
                url: str, context: ssl.SSLContext
            ) -> io.BufferedIOBase:
                attempts.append(url)
                self.assertLessEqual(len(attempts), 1)
                self.assertEqual(url, 'https://example.com')
                self._assert_ssl_context_matches_default(context)
                return io.BytesIO(b'content')

            urlopen_swap = self.swap(urlrequest, 'urlopen', mock_urlopen)

            with urlopen_swap, self.swap_curl_failure:
                common.url_retrieve(
                    'https://example.com', output_path, enforce_https=False
                )
            with open(output_path, 'rb', encoding=None) as buffer:
                self.assertEqual(buffer.read(), b'content')