run-tests

#!/usr/bin/env python3
# Don't run tests from the root repo dir.
# We want to ensure we're importing from the installed
# binary package not from the CWD.

import argparse
import os
from contextlib import contextmanager
from subprocess import check_call

_dname = os.path.dirname

REPO_ROOT = _dname(_dname(_dname(os.path.abspath(__file__))))


@contextmanager
def cd(path):
    """Change directory while inside context manager."""
    cwd = os.getcwd()
    try:
        os.chdir(path)
        yield
    finally:
        os.chdir(cwd)


def run(command, allow_repo_root_on_path=False):
    env = os.environ.copy()
    if not allow_repo_root_on_path:
        env['TESTS_REMOVE_REPO_ROOT_FROM_PATH'] = 'true'
    return check_call(command, shell=True, env=env)


def process_args(args):
    runner = args.test_runner
    test_args = ""
    if args.ignore:
        test_args += " ".join(f"--ignore {ignore}" for ignore in args.ignore)
        test_args += " "

    if args.with_cov:
        test_args += (
            # Even though botocore and s3transfer reside in the awscli package,
            # they are often imported via their top-level import alias in the
            # codebase. Explicitly including them in --cov args ensures we
            # include usage of these aliases for coverage purposes.
            f"--cov=awscli --cov=botocore --cov=s3transfer "
            f"--cov-config={os.path.join(REPO_ROOT, '.coveragerc')} "
            f"--cov-report xml "
        )
    if args.junit_xml_path:
        test_args += f'--junit-xml={args.junit_xml_path} '
    if args.markers:
        test_args += f'-m {args.markers} '
    if os.getenv('PYTEST_XDIST_AUTO_NUM_WORKERS'):
        # If the user set PYTEST_XDIST_AUTO_NUM_WORKERS,
        # use that to limit max processes
        max_processes = int(os.environ['PYTEST_XDIST_AUTO_NUM_WORKERS'])
    elif os.getenv('GITHUB_ACTIONS', 'false') == 'true':
        # Limiting parallelism in GitHub, to avoid the
        # 'Error: The operation was canceled.' issues in 2026
        max_processes=3
    else:
        max_processes=4
    test_args += f'--numprocesses=auto --dist=loadfile --maxprocesses={max_processes} '
    dirs = " ".join(args.test_dirs)

    return runner, test_args, dirs


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "test_dirs",
        default=["unit/", "functional/"],
        nargs="*",
        help="One or more directories containing tests.",
    )
    parser.add_argument(
        "-r",
        "--test-runner",
        default="pytest",
        help="Test runner to execute tests. Defaults to pytest.",
    )
    parser.add_argument(
        "-c",
        "--with-cov",
        default=False,
        action="store_true",
        help="Run default test-runner with code coverage enabled.",
    )
    parser.add_argument(
        "--allow-repo-root-on-path",
        default=False,
        action="store_true",
        help=(
            "Do not remove the repository root from the Python path when "
            "running tests. This allows you to run the tests against the "
            "current repository without have to install the package as a "
            "distribution."
        ),
    )
    parser.add_argument(
        "--ignore",
        nargs='+',
        default=[],
        help=("Ignore a test subdirectory. Can be specified multiple times."),
    )
    parser.add_argument(
        "--tests-path",
        default=None,
        type=os.path.abspath,
        help=("Optional path to an alternate test directory to use."),
    )
    parser.add_argument(
        "--junit-xml-path",
        default=None,
        type=os.path.abspath,
        help="Optional path to output a JUnit XML result file to.",
    )
    parser.add_argument(
        "-m",
        "--markers",
        default=None,
        help="Run all tests decorated with the specified marker expression.",
    )

    raw_args = parser.parse_args()
    test_runner, test_args, test_dirs = process_args(raw_args)

    tests_path = raw_args.tests_path
    if tests_path is None:
        tests_path = os.path.join(REPO_ROOT, "tests")

    cmd = f"{test_runner} {test_args}{test_dirs}"
    print(f"CDing to {tests_path}")
    print(f"Running {cmd}...")
    with cd(tests_path):
        run(cmd, allow_repo_root_on_path=raw_args.allow_repo_root_on_path)