scripts/servers.py - develop - githubmirror/oppia

scripts/servers.py
895 строк33 KB
Meeth
Fix part of #25877 Remove webpack build architecture (#26078)
10 июл 2026, 12:38
10 июл 2026, 12:389089f71
# Copyright 2014 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.

"""Utility functions for managing server processes required by Oppia."""

from __future__ import annotations

import contextlib
import json
import logging
import os
import re
import shutil
import signal
import subprocess
import sys
import threading

from core import feconf
from scripts import common

import psutil
from typing import (
    Any,
    Callable,
    ContextManager,
    Dict,
    Iterator,
    List,
    Optional,
    Sequence,
    Union,
)


# Here we use type Any because the argument 'popen_kwargs' can accept an
# arbitrary number of keyword arguments with different types of values.
@contextlib.contextmanager
def managed_process(
    command_args: Sequence[Union[int, str]],
    human_readable_name: str = 'Process',
    shell: bool = False,
    timeout_secs: int = 60,
    raise_on_nonzero_exit: bool = True,
    **popen_kwargs: Any,
) -> Iterator[psutil.Process]:
    """Context manager for starting and stopping a process gracefully.

    Args:
        command_args: list(int|str). A sequence of program arguments, where the
            program to execute is the first item. Ints are allowed in order to
            accomodate e.g. port numbers.
        human_readable_name: str. The human-readable name of the process. Used
            by the function's logging logic to improve readability.
        shell: bool. Whether the command should be run inside of its own shell.
            WARNING: Executing shell commands that incorporate unsanitized input
            from an untrusted source makes a program vulnerable to
            [shell injection](https://w.wiki/_Ac2), a serious security flaw
            which can result in arbitrary command execution. For this reason,
            the use of `shell=True` is **strongly discouraged** in cases where
            the command string is constructed from external input.
        timeout_secs: int. The time allotted for the managed process and its
            descendants to terminate themselves. After the timeout, any
            remaining processes will be killed abruptly.
        raise_on_nonzero_exit: bool. If True, raise an Exception when the
            managed process has a nonzero exit code. If False, no Exception is
            raised, and it is the caller's responsibility to handle the error.
        **popen_kwargs: dict(str: *). Same kwargs as `subprocess.Popen`.

    Yields:
        psutil.Process. The process managed by the context manager.

    Raises:
        Exception. The process exited unexpectedly (only raised if
            raise_on_nonzero_exit is True).
    """
    get_proc_info: Callable[[psutil.Process], str] = lambda p: (
        '%s(name="%s", pid=%d)' % (human_readable_name, p.name(), p.pid)
        if p.is_running()
        else '%s(pid=%d)' % (human_readable_name, p.pid)
    )

    stripped_args = (('%s' % arg).strip() for arg in command_args)
    non_empty_args = (s for s in stripped_args if s)

    command = ' '.join(non_empty_args) if shell else list(non_empty_args)
    human_readable_command = command if shell else ' '.join(command)
    msg = 'Starting new %s: %s' % (human_readable_name, human_readable_command)
    print(msg)
    popen_proc = psutil.Popen(command, shell=shell, **popen_kwargs)

    try:
        yield popen_proc
    finally:
        proc_name = get_proc_info(popen_proc)
        print('Stopping %s...' % proc_name)
        procs_still_alive = [popen_proc]

        try:
            if popen_proc.is_running():
                # Children must be terminated before the parent, otherwise they
                # may become zombie processes.
                procs_still_alive = popen_proc.children(recursive=True) + [
                    popen_proc
                ]

            procs_to_kill = []
            for proc in procs_still_alive:
                if proc.is_running():
                    logging.info('Terminating %s...' % get_proc_info(proc))
                    proc.terminate()
                    procs_to_kill.append(proc)
                else:
                    logging.info('%s has already ended.' % get_proc_info(proc))

            procs_gone, procs_still_alive = psutil.wait_procs(
                procs_to_kill, timeout=timeout_secs
            )
            for proc in procs_still_alive:
                logging.warning('Forced to kill %s!' % get_proc_info(proc))
                proc.kill()
            for proc in procs_gone:
                logging.info('%s has already ended.' % get_proc_info(proc))
        except Exception:
            # NOTE: Raising an exception while exiting a context manager is bad
            # practice, so we log and suppress exceptions instead.
            logging.exception(
                'Failed to stop %s gracefully!' % get_proc_info(popen_proc)
            )

        exit_code = popen_proc.returncode
        # Note that negative values indicate termination by a signal: SIGTERM,
        # SIGINT, etc. Also, exit code 143 indicates that the process received
        # a SIGTERM from the OS, and it succeeded in gracefully terminating.
        if (
            exit_code is not None
            and exit_code > 0
            and exit_code != 143
            and raise_on_nonzero_exit
        ):
            raise Exception(
                'Process %s exited unexpectedly with exit code %s'
                % (proc_name, exit_code)
            )


@contextlib.contextmanager
def managed_dev_appserver(
    app_yaml_path: str,
    env: Optional[Dict[str, str]] = None,
    log_level: str = 'info',
    host: str = '0.0.0.0',
    port: int = 8080,
    admin_host: str = '0.0.0.0',
    admin_port: int = feconf.GAE_ADMIN_SERVER_PORT,
    enable_host_checking: bool = True,
    automatic_restart: bool = False,
    skip_sdk_update_check: bool = False,
) -> Iterator[psutil.Process]:
    """Returns a context manager to start up and shut down a GAE dev appserver.

    Args:
        app_yaml_path: str. Path to the app.yaml file which defines the
            structure of the server.
        env: dict(str: str) or None. Defines the environment variables for the
            new process.
        log_level: str. The lowest log level generated by the application code
            and the development server. Expected values are: debug, info,
            warning, error, critical.
        host: str. The host name to which the app server should bind.
        port: int. The lowest port to which application modules should bind.
        admin_host: str. The host name to which the admin server should bind.
        admin_port: int. The port to which the admin server should bind.
        enable_host_checking: bool. Whether to enforce HTTP Host checking for
            application modules, API server, and admin server. Host checking
            protects against DNS rebinding attacks, so only disable after
            understanding the security implications.
        automatic_restart: bool. Whether to restart instances automatically when
            files relevant to their module are changed.
        skip_sdk_update_check: bool. Whether to skip checking for SDK updates.
            If false, uses .appcfg_nag to decide.

    Yields:
        psutil.Process. The dev_appserver process.
    """
    dev_appserver_args: List[Union[str, int]] = [
        common.CURRENT_PYTHON_BIN,
        common.DEV_APPSERVER_PATH,
        '--host',
        host,
        '--port',
        port,
        '--admin_host',
        admin_host,
        '--admin_port',
        admin_port,
        '--enable_host_checking',
        'true' if enable_host_checking else 'false',
        '--automatic_restart',
        'true' if automatic_restart else 'false',
        '--skip_sdk_update_check',
        'true' if skip_sdk_update_check else 'false',
        '--log_level',
        log_level,
        '--dev_appserver_log_level',
        log_level,
        app_yaml_path,
    ]
    with contextlib.ExitStack() as stack:
        # OK to use shell=True here because we are not passing anything that
        # came from an untrusted user, only other callers of the script,
        # so there's no risk of shell-injection attacks.
        proc = stack.enter_context(
            managed_process(
                dev_appserver_args,
                human_readable_name='GAE Development Server',
                shell=True,
                env=env,
            )
        )
        common.wait_for_port_to_be_in_use(port)
        yield proc


@contextlib.contextmanager
def managed_firebase_auth_emulator(
    recover_users: bool = False,
) -> Iterator[psutil.Process]:
    """Returns a context manager to manage the Firebase auth emulator.

    Args:
        recover_users: bool. Whether to recover users created by the previous
            instance of the Firebase auth emulator.

    Yields:
        psutil.Process. The Firebase emulator process.
    """
    emulator_args = [
        common.FIREBASE_PATH,
        'emulators:start',
        '--only',
        'auth',
        '--project',
        common.DEV_PROJECT_ID,
        '--config',
        feconf.FIREBASE_EMULATOR_CONFIG_PATH,
    ]

    emulator_args.extend(
        ['--import', common.FIREBASE_EMULATOR_CACHE_DIR, '--export-on-exit']
        if recover_users
        else ['--export-on-exit', common.FIREBASE_EMULATOR_CACHE_DIR]
    )

    # OK to use shell=True here because we are passing string literals and
    # constants, so there is no risk of a shell-injection attack.
    proc_context = managed_process(
        emulator_args, human_readable_name='Firebase Emulator', shell=True
    )
    with proc_context as proc:
        # Verify that the emulator is actually responding to HTTP requests, not
        # just that the port is open. This prevents race conditions where the
        # emulator binds the port but crashes during initialization.
        common.wait_for_firebase_emulator_to_be_ready(
            feconf.FIREBASE_EMULATOR_PORT
        )
        yield proc


@contextlib.contextmanager
def managed_elasticsearch_dev_server() -> Iterator[psutil.Process]:
    """Returns a context manager for ElasticSearch server for running tests
    in development mode and running a local dev server. This is only required
    in a development environment.

    Yields:
        psutil.Process. The ElasticSearch server process.
    """
    # Clear previous data stored in the local cluster.
    if os.path.exists(common.ES_PATH_DATA_DIR):
        shutil.rmtree(common.ES_PATH_DATA_DIR)

    es_args = [
        '%s/bin/elasticsearch' % common.ES_PATH,
        # -q is the quiet flag.
        '-q',
        '-E',
        # Disable security for the local ElasticSearch server.
        'xpack.security.enabled=false',
        # Disable the disk threshold checks. These checks can cause issues on
        # machines with low disk space.
        '-E',
        'cluster.routing.allocation.disk.threshold_enabled=false',
    ]
    # Override the default path to ElasticSearch config files.
    es_env = {
        'ES_PATH_CONF': common.ES_PATH_CONFIG_DIR,
        # Set the minimum heap size to 100 MB and maximum to 500 MB.
        'ES_JAVA_OPTS': '-Xms100m -Xmx500m',
    }
    # OK to use shell=True here because we are passing string literals and
    # constants, so there is no risk of a shell-injection attack.
    proc_context = managed_process(
        es_args,
        human_readable_name='ElasticSearch Server',
        env=es_env,
        shell=True,
    )
    with proc_context as proc:
        common.wait_for_port_to_be_in_use(feconf.ES_LOCALHOST_PORT)
        yield proc


@contextlib.contextmanager
def managed_cloud_datastore_emulator(
    clear_datastore: bool = False,
) -> Iterator[psutil.Process]:
    """Returns a context manager for the Cloud Datastore emulator.

    Args:
        clear_datastore: bool. Whether to delete the datastore's config and data
            before starting the emulator.

    Yields:
        psutil.Process. The emulator process.
    """
    emulator_hostport = '%s:%d' % (
        feconf.CLOUD_DATASTORE_EMULATOR_HOST,
        feconf.CLOUD_DATASTORE_EMULATOR_PORT,
    )
    emulator_args = [
        common.GCLOUD_PATH,
        'beta',
        'emulators',
        'datastore',
        'start',
        '--project',
        common.DEV_PROJECT_ID,
        '--data-dir',
        common.CLOUD_DATASTORE_EMULATOR_DATA_DIR,
        '--host-port',
        emulator_hostport,
        '--consistency=1.0',
        '--quiet',
    ]

    if clear_datastore:
        emulator_args.append('--no-store-on-disk')

    with contextlib.ExitStack() as stack:
        data_dir_exists = os.path.exists(
            common.CLOUD_DATASTORE_EMULATOR_DATA_DIR
        )
        if clear_datastore and data_dir_exists:
            # Replace it with an empty directory.
            shutil.rmtree(common.CLOUD_DATASTORE_EMULATOR_DATA_DIR)
            os.makedirs(common.CLOUD_DATASTORE_EMULATOR_DATA_DIR)
        elif not data_dir_exists:
            os.makedirs(common.CLOUD_DATASTORE_EMULATOR_DATA_DIR)

        # OK to use shell=True here because we are passing string literals and
        # constants, so there is no risk of a shell-injection attack.
        proc = stack.enter_context(
            managed_process(
                emulator_args,
                human_readable_name='Cloud Datastore Emulator',
                shell=True,
            )
        )

        common.wait_for_port_to_be_in_use(feconf.CLOUD_DATASTORE_EMULATOR_PORT)

        # Environment variables required to communicate with the emulator.
        stack.enter_context(
            common.swap_env('DATASTORE_DATASET', common.DEV_PROJECT_ID)
        )
        stack.enter_context(
            common.swap_env('DATASTORE_EMULATOR_HOST', emulator_hostport)
        )
        stack.enter_context(
            common.swap_env(
                'DATASTORE_EMULATOR_HOST_PATH',
                '%s/datastore' % emulator_hostport,
            )
        )
        stack.enter_context(
            common.swap_env('DATASTORE_HOST', 'http://%s' % emulator_hostport)
        )
        stack.enter_context(
            common.swap_env('DATASTORE_PROJECT_ID', common.DEV_PROJECT_ID)
        )
        stack.enter_context(
            common.swap_env('DATASTORE_USE_PROJECT_ID_AS_APP_ID', 'true')
        )
        stack.enter_context(
            common.swap_env('GOOGLE_CLOUD_PROJECT', common.DEV_PROJECT_ID)
        )

        yield proc


@contextlib.contextmanager
def managed_redis_server() -> Iterator[psutil.Process]:
    """Run the redis server within a context manager that ends it gracefully."""

    # Check if a redis dump file currently exists. This file contains residual
    # data from a previous run of the redis server. If it exists, removes the
    # dump file so that the redis server starts with a clean slate.
    if os.path.exists(common.REDIS_DUMP_PATH):
        os.remove(common.REDIS_DUMP_PATH)

    # OK to use shell=True here because we are passing string literals and
    # constants, so there is no risk of a shell-injection attack.
    proc_context = managed_process(
        [common.REDIS_SERVER_PATH, common.REDIS_CONF_PATH],
        human_readable_name='Redis Server',
        shell=True,
    )
    with proc_context as proc:
        common.wait_for_port_to_be_in_use(feconf.REDISPORT)
        try:
            yield proc
        finally:
            subprocess.check_call([common.REDIS_CLI_PATH, 'shutdown', 'nosave'])


def create_managed_web_browser(port: int) -> ContextManager[psutil.Process]:
    """Returns a ContextManager for a web browser targeting the given port on
    localhost. If a web browser cannot be opened on the current system by Oppia,
    then raises an exception.

    Args:
        port: int. The port number to open in the web browser.

    Returns:
        ContextManager. The ContextManager to a web browser window if the
        current operating system can be identified and a web browser can be
        launched automatically.

    Raises:
        Exception. Unable to launch the web browser (this happens when the
            Operating System cannot be identified).
    """
    url = 'http://localhost:%s/' % port
    human_readable_name = 'Web Browser'
    if common.is_linux_os():
        return managed_process(
            ['xdg-open', url], human_readable_name=human_readable_name
        )
    elif common.is_mac_os():
        return managed_process(
            ['open', url], human_readable_name=human_readable_name
        )
    else:
        raise Exception(
            'Unable to identify the Operating System and therefore, unable to '
            'launch the web browser.'
        )


@contextlib.contextmanager
def managed_ng_build(
    *,
    use_prod_env: bool = False,
    watch_mode: bool = False,
    source_maps: bool = False,
) -> Iterator[psutil.Process]:
    """Returns context manager to start/stop the ng compiler gracefully.

    Args:
        use_prod_env: bool. Whether to compile for use in production.
        watch_mode: bool. Run the compiler in watch mode, which rebuilds on file
            change.
        source_maps: bool. Whether to generate source maps for the compiled
            files. This is useful for debugging the compiled code.

    Yields:
        psutil.Process. The ng compiler process.

    Raises:
        OSError. First build never completed.
    """
    compiler_args = [common.NG_BIN_PATH, 'build']
    if use_prod_env:
        compiler_args.append('--prod')
    if watch_mode:
        compiler_args.append('--watch')
    if source_maps:
        compiler_args.append('--source-map')

    with contextlib.ExitStack() as exit_stack:
        # OK to use shell=True here because we are passing string literals and
        # constants, so there is no risk of a shell-injection attack.
        proc = exit_stack.enter_context(
            managed_process(
                compiler_args,
                human_readable_name='Angular Compiler',
                shell=True,
                # Capture compiler's output to detect when builds have completed.
                stdout=subprocess.PIPE,
            )
        )

        read_line_func: Callable[[], Optional[bytes]] = (
            lambda: proc.stdout.readline() or None
        )
        if watch_mode:
            for line in iter(read_line_func, None):
                common.write_stdout_safe(line)
                # Message printed when a compilation has succeeded. We break
                # after the first one to ensure the site is ready to be visited.
                if b'Build at: ' in line:
                    break
            else:
                # If none of the lines contained the string 'Built at',
                # raise an error because a build hasn't finished successfully.
                raise IOError('First build never completed')

        def print_proc_output() -> None:
            """Prints the proc's output until it is exhausted."""
            for line in iter(read_line_func, None):
                common.write_stdout_safe(line)

        # Start a thread to print the rest of the compiler's output to stdout.
        printer_thread = threading.Thread(target=print_proc_output)
        printer_thread.start()
        exit_stack.callback(printer_thread.join)

        yield proc


def get_chromedriver_version() -> str:
    """Returns the version of Chromedriver compatible with the installed version
    of Chrome.
    """
    # Although there are spaces between Google and Chrome in the path, we
    # don't need to escape them for Popen (as opposed to on the terminal, in
    # which case we would need to escape them for the command to run).
    chrome_command = (
        '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome'
        if common.is_mac_os()
        else 'google-chrome'
    )
    try:
        output = subprocess.check_output([chrome_command, '--version'])
    except OSError as e:
        # For the error message on macOS, we need to add the backslashes in.
        # This is because it is likely that a user will try to run the
        # command on their terminal and, as mentioned above, the macOS
        # chrome version command has spaces in the path which need to be
        # escaped for successful terminal use.
        raise Exception(
            'Failed to execute "%s --version" command. This is used to '
            'determine the chromedriver version to use. Please set the '
            'chromedriver version manually using the '
            '--chrome_driver_version flag. To determine the '
            'chromedriver version to be used, please follow the '
            'instructions mentioned in the following URL:\n'
            'https://chromedriver.chromium.org/downloads/version-selection'
            % chrome_command.replace(' ', r'\ ')
        ) from e

    installed_version_bytes = b''.join(re.findall(rb'[0-9.]', output))
    installed_version_parts = installed_version_bytes.decode('utf-8').split('.')
    # For Chrome versions 115 and above, compatible Chromedriver and Chrome
    # versions have the same version numbers. For earlier versions, we use a
    # Google API to find the compatible Chromedriver version. See
    # https://chromedriver.chromium.org/downloads/version-selection for details.
    if int(installed_version_parts[0]) >= 115:
        chromedriver_version: str = '.'.join(installed_version_parts)
    else:
        response = common.url_open(
            'https://chromedriver.storage.googleapis.com/LATEST_RELEASE_%s'
            % ('.'.join(installed_version_parts[:-1]))
        )
        chromedriver_version = response.read().decode('utf-8')

    return chromedriver_version


@contextlib.contextmanager
def managed_portserver() -> Iterator[psutil.Process]:
    """Returns context manager to start/stop the portserver gracefully.

    The portserver listens at PORTSERVER_SOCKET_FILEPATH and allocates free
    ports to clients. This prevents race conditions when two clients request
    ports in quick succession. The local Google App Engine server that we use to
    serve the development version of Oppia uses python_portpicker, which is
    compatible with the portserver this function starts, to request ports.

    By "compatible" we mean that python_portpicker requests a port by sending a
    request consisting of the PID of the requesting process and expects a
    response consisting of the allocated port number. This is the interface
    provided by this portserver.

    Yields:
        psutil.Popen. The Popen subprocess object.
    """
    # Check if a socket file exists. This file can exist when previous instance
    # of the portserver did not close properly. We need to remove as otherwise
    # the portserver will fail to start.
    if os.path.exists(common.PORTSERVER_SOCKET_FILEPATH):
        os.remove(common.PORTSERVER_SOCKET_FILEPATH)

    portserver_args = [
        'python',
        '-m',
        'scripts.run_portserver',
        '--portserver_unix_socket_address',
        common.PORTSERVER_SOCKET_FILEPATH,
    ]
    # OK to use shell=True here because we are passing string literals and
    # constants, so there is no risk of a shell-injection attack.
    proc_context = managed_process(
        portserver_args, human_readable_name='Portserver', shell=True
    )

    with proc_context as proc:
        try:
            yield proc
        finally:
            # Before exiting the proc_context, try to end the process with
            # SIGINT. The portserver is configured to shut down cleanly upon
            # receiving this signal.
            try:
                proc.send_signal(signal.SIGINT)
            except OSError:
                # Raises when the process has already shutdown, in which case we
                # can just return immediately.
                return  # pylint: disable=lost-exception
            else:
                # Otherwise, give the portserver 10 seconds to shut down after
                # sending CTRL-C (SIGINT).
                try:
                    proc.wait(timeout=10)
                except psutil.TimeoutExpired:
                    # If the server fails to shut down, allow proc_context to
                    # end it by calling terminate() and/or kill().
                    logging.error(
                        'Portserver failed to shut down after 10 seconds.'
                    )


@contextlib.contextmanager
def managed_webdriverio_server(
    suite_name: str = 'full',
    dev_mode: bool = True,
    debug_mode: bool = False,
    sharding_instances: int = 1,
    chrome_version: Optional[str] = None,
    mobile: bool = False,
    stdout: int = subprocess.PIPE,
) -> Iterator[psutil.Process]:
    """Returns context manager to start/stop the WebdriverIO server gracefully.

    Args:
        suite_name: str. The suite name whose tests should be run. If the value
            is `full`, all tests will run.
        dev_mode: bool. Whether the test is running on dev_mode.
        debug_mode: bool. Whether to run the webdriverio tests in debugging
            mode. Read the following instructions to learn how to run e2e
            tests in debugging mode:
            https://webdriver.io/docs/debugging/#the-debug-command.
        sharding_instances: int. How many sharding instances to be running.
        chrome_version: str|None. The version of Google Chrome to run the tests
            on. If None, then the currently-installed version of Google Chrome
            is used instead.
        stdout: int. This parameter specifies the executed program's standard
            output file handle.
        mobile: bool. Whether to run the webdriverio tests in mobile mode.

    Yields:
        psutil.Process. The webdriverio process.

    Raises:
        ValueError. Number of sharding instances are less than 0.
    """
    if sharding_instances <= 0:
        raise ValueError('Sharding instance should be larger than 0')

    if chrome_version is None:
        chrome_version = get_chromedriver_version()

    if mobile:
        os.environ['MOBILE'] = 'true'
    else:
        os.environ['MOBILE'] = 'false'

    webdriverio_args = [
        common.NPX_BIN_PATH,
        # This flag ensures tests fail if the `waitFor()` calls time out.
        '--unhandled-rejections=strict',
        common.NODEMODULES_WDIO_BIN_PATH,
        common.WEBDRIVERIO_CONFIG_FILE_PATH,
        '--suite',
        suite_name,
        chrome_version,
        '--params.devMode=%s' % dev_mode,
    ]

    # Capabilities in wdio.conf.js are added as an array of object,
    # so in order to set the value of maxmium instances of chrome
    # in wdio.conf.js, we need to provide the index of the capability
    # at which chrome is present, i.e. 0.
    if sharding_instances > 1:
        webdriverio_args.extend(
            [
                '--capabilities[0].maxInstances=%d' % sharding_instances,
            ]
        )

    if debug_mode:
        webdriverio_args.insert(0, 'DEBUG=true')

    # OK to use shell=True here because we are passing string literals and
    # constants, so there is no risk of a shell-injection attack.
    managed_webdriverio_proc = managed_process(
        webdriverio_args,
        human_readable_name='WebdriverIO Server',
        shell=True,
        raise_on_nonzero_exit=False,
        stdout=stdout,
    )

    try:
        with managed_webdriverio_proc as proc:
            yield proc
    finally:
        del os.environ['MOBILE']


@contextlib.contextmanager
def managed_acceptance_tests_server(
    suite_name: str,
    headless: bool = False,
    mobile: bool = False,
    prod_env: bool = False,
    update_snapshots: bool = False,
    stdout: int = subprocess.PIPE,
) -> Iterator[psutil.Process]:
    """Returns context manager to start/stop the acceptance tests
    server gracefully. If the suite_name is not in the list of the
    acceptance tests suite names, then raises an exception.

    Args:
        suite_name: str. The suite name whose tests should be run.
        headless: bool. Whether to run the acceptance tests in headless mode.
        mobile: bool. Whether to run the acceptance tests in mobile mode.
        prod_env: bool. Whether to run the acceptance tests in production mode.
        update_snapshots: bool. Whether to update Playwright screenshot
            baselines instead of comparing against them. Ignored for
            Puppeteer suites.
        stdout: int. The file descriptor where the standard output of the
            subprocess is sent.

    Yields:
        psutil.Process. The jest testing process.

    Raises:
        Exception. The suite_name is not in the list of the acceptance tests
            suite names.
    """
    available_suites = {}
    with open(
        common.ACCEPTANCE_TEST_CONFIG_FILE_PATH, 'r', encoding='utf-8'
    ) as f:
        filedata = json.load(f)
        for suites in filedata.values():
            for suite in suites:
                available_suites[suite['name']] = {
                    'module': suite['module'],
                    'framework': suite['framework'],
                }

    if suite_name not in available_suites:
        raise Exception('Invalid suite name: %s' % suite_name)

    suite_config = available_suites[suite_name]
    module = suite_config['module']
    framework = suite_config['framework']

    # TODO(#24715): Remove the puppeteer framework once the
    # migration from Puppeteer to Playwright is complete.
    if framework == 'puppeteer':
        os.environ['HEADLESS'] = 'true' if headless else 'false'
        os.environ['MOBILE'] = 'true' if mobile else 'false'
        os.environ['SPEC_NAME'] = suite_name
        os.environ['PROD_ENV'] = 'true' if prod_env else 'false'

        nodemodules_jest_bin_path = os.path.join(
            common.NODE_MODULES_PATH, '.bin', 'jest'
        )

        acceptance_tests_args = [
            nodemodules_jest_bin_path,
            '%s' % os.path.join(module),
            '--config=./core/tests/puppeteer-acceptance-tests/jest.config.js',
        ]

        # OK to use shell=True here because we are passing string literals,
        # and verifying that the passed suite-name are within the list of
        # the suites we have, so there is no risk of a shell-injection attack.
        managed_acceptance_tests_proc = managed_process(
            acceptance_tests_args,
            human_readable_name='Acceptance Tests Server',
            shell=True,
            raise_on_nonzero_exit=False,
            stdout=stdout,
        )

        with managed_acceptance_tests_proc as proc:
            yield proc

    elif framework == 'playwright':
        playwright_dir = os.path.join(
            common.CURR_DIR, 'core', 'tests', 'playwright-acceptance-tests'
        )
        # Use the local Playwright binary directly instead of npx
        # to avoid version resolution or download attempts.
        playwright_bin = os.path.join(
            playwright_dir, 'node_modules', '.bin', 'playwright'
        )
        playwright_env = {
            **os.environ,
            # Prepend Node 20 bin so npx uses the correct runtime.
            'PATH': os.pathsep.join(
                [
                    os.path.join(common.PLAYWRIGHT_NODE_PATH, 'bin'),
                    os.environ['PATH'],
                ]
            ),
            'NODE': os.path.join(common.PLAYWRIGHT_NODE_PATH, 'bin', 'node'),
            'HEADLESS': 'true' if headless else 'false',
            'MOBILE': 'true' if mobile else 'false',
            'PROD_ENV': 'true' if prod_env else 'false',
            'SPEC_NAME': suite_name,
            'UPDATE_SNAPSHOTS': 'true' if update_snapshots else 'false',
        }
        playwright_args = [
            playwright_bin,
            'test',
            module,
            '--config=./core/tests/playwright-acceptance-tests/'
            'playwright.config.ts',
        ]
        if headless is False:
            playwright_args.append('--headed')
        if update_snapshots:
            playwright_args.append('--update-snapshots')

        managed_proc = managed_process(
            playwright_args,
            human_readable_name='Acceptance Tests Server',
            shell=False,
            raise_on_nonzero_exit=False,
            stdout=stdout,
            env=playwright_env,
        )
        with managed_proc as proc:
            yield proc

    else:
        raise Exception(
            'Suite "%s" has invalid framework: "%s". '
            'Must be "puppeteer" or "playwright".' % (suite_name, framework)
        )


def run_ng_compilation() -> None:
    """Runs angular compilation."""
    max_tries = 2
    ng_bundles_dir_name = 'dist/oppia-angular'
    for _ in range(max_tries):
        try:
            with managed_ng_build() as proc:
                proc.wait()
        except subprocess.CalledProcessError as error:
            print(error.output)
            sys.exit(error.returncode)
        if os.path.isdir(ng_bundles_dir_name):
            break
    if not os.path.isdir(ng_bundles_dir_name):
        print('Failed to complete ng build compilation, exiting...')
        sys.exit(1)