# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt

"""Run performance comparisons for versions of coverage"""

from __future__ import annotations

import collections
import contextlib
import itertools
import json
import os
import random
import shutil
import statistics
import subprocess
import sys
import time
import traceback

from pathlib import Path

from dataclasses import dataclass
from io import TextIOWrapper
from types import TracebackType
from typing import Any, Dict, Iterable, Iterator, Mapping, Optional, Tuple, Type, cast

import requests
import tabulate

TweaksType = Optional[Iterable[Tuple[str, Any]]]
Env_VarsType = Optional[Dict[str, str]]


class ShellSession:
    """A logged shell session.

    The duration of the last command is available as .last_duration.
    """

    def __init__(self, output_filename: str):
        self.output_filename = output_filename
        self.last_duration: float = 0
        self.foutput: TextIOWrapper | None = None
        self.env_vars = {"PATH": os.getenv("PATH")}

    def __enter__(self) -> ShellSession:
        self.foutput = open(self.output_filename, "a", encoding="utf-8")
        print(f"Logging output to {os.path.abspath(self.output_filename)}")
        return self

    def __exit__(
        self,
        exc_type: Type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        if self.foutput is not None:
            self.foutput.close()

    @contextlib.contextmanager
    def set_env(self, *env_varss: dict[str, str] | None) -> Iterator[None]:
        """Set environment variables.

        All the arguments are dicts of name:value, or None.  All are applied
        to the environment variables.
        """
        old_env_vars = self.env_vars
        self.env_vars = dict(old_env_vars)
        for env_vars in env_varss:
            self.env_vars.update(env_vars or {})
        try:
            yield
        finally:
            self.env_vars = old_env_vars

    def print(self, *args: Any, **kwargs: Any) -> None:
        """Print a message to this shell's log."""
        print(*args, **kwargs, file=self.foutput)

    def print_banner(self, *args: Any, **kwargs: Any) -> None:
        """Print a distinguished banner to the log."""
        self.print("\n######> ", end="")
        self.print(*args, **kwargs)

    def run_command(self, cmd: str) -> str:
        """
        Run a command line (with a shell).

        Returns:
            str: the output of the command.

        """
        self.print(f"\n### ========================\n$ {cmd}")
        start = time.perf_counter()
        proc = subprocess.run(
            cmd,
            shell=True,
            check=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            env=cast(Mapping[str, str], self.env_vars),
        )
        output = proc.stdout.decode("utf-8")
        self.last_duration = time.perf_counter() - start
        self.print(output, end="")
        self.print(f"(was: {cmd})")
        self.print(f"(in {os.getcwd()}, duration: {self.last_duration:.3f}s)")

        if proc.returncode != 0:
            self.print(f"ERROR: command returned {proc.returncode}")
            raise Exception(
                f"Command failed ({proc.returncode}): {cmd!r}, output was:\n{output}"
            )

        return output.strip()


def rmrf(path: Path) -> None:
    """
    Remove a directory tree.  It's OK if it doesn't exist.
    """
    if path.exists():
        shutil.rmtree(path)


@contextlib.contextmanager
def change_dir(newdir: Path) -> Iterator[Path]:
    """
    Change to a new directory, and then change back.

    Will make the directory if needed.
    """
    old_dir = os.getcwd()
    newdir.mkdir(parents=True, exist_ok=True)
    os.chdir(newdir)
    try:
        yield newdir
    finally:
        os.chdir(old_dir)


@contextlib.contextmanager
def file_replace(file_name: Path, old_text: str, new_text: str) -> Iterator[None]:
    """
    Replace some text in `file_name`, and change it back.
    """
    file_text = ""
    if old_text:
        file_text = file_name.read_text()
        if old_text not in file_text:
            raise Exception("Old text {old_text!r} not found in {file_name}")
        updated_text = file_text.replace(old_text, new_text)
        file_name.write_text(updated_text)
    try:
        yield
    finally:
        if old_text:
            file_name.write_text(file_text)


def file_must_exist(file_name: str, kind: str = "file") -> Path:
    """
    Check that a file exists, for early validation of pip (etc) arguments.

    Raises an exception if it doesn't exist.  Returns the resolved path if it
    does exist so we can use relative paths and they'll still work once we've
    cd'd to the temporary workspace.
    """
    path = Path(file_name).expanduser().resolve()
    if not path.exists():
        kind = kind[0].upper() + kind[1:]
        raise RuntimeError(f"{kind} {file_name!r} doesn't exist")
    return path


def url_must_exist(url: str) -> bool:
    """
    Check that a URL exists, for early validation of pip (etc) arguments.

    Raises an exception if it doesn't exist.
    """
    resp = requests.head(url)
    resp.raise_for_status()
    return True


class ProjectToTest:
    """Information about a project to use as a test case."""

    # Where can we clone the project from?
    git_url: str = ""
    slug: str = ""
    env_vars: Env_VarsType = {}

    def __init__(self) -> None:
        url_must_exist(self.git_url)
        if not self.slug:
            if self.git_url:
                self.slug = self.git_url.split("/")[-1]

    def shell(self) -> ShellSession:
        return ShellSession(f"output_{self.slug}.log")

    def make_dir(self) -> None:
        self.dir = Path(f"work_{self.slug}")
        if self.dir.exists():
            rmrf(self.dir)

    def get_source(self, shell: ShellSession, retries: int = 5) -> None:
        """Get the source of the project."""
        for retry in range(retries):
            try:
                shell.run_command(f"git clone {self.git_url} {self.dir}")
                return
            except Exception as e:
                print(f"Retrying to clone {self.git_url} due to error:\n{e}")
                if retry == retries - 1:
                    raise e

    def prep_environment(self, env: Env) -> None:
        """Prepare the environment to run the test suite.

        This is not timed.
        """

    @contextlib.contextmanager
    def tweak_coverage_settings(self, settings: TweaksType) -> Iterator[None]:
        """Tweak the coverage settings.

        NOTE: This is not properly factored, and is only used by ToxProject now!!!
        """
        yield

    def pre_check(self, env: Env) -> None:
        pass

    def post_check(self, env: Env) -> None:
        pass

    def run_no_coverage(self, env: Env) -> float:
        """Run the test suite with no coverage measurement.

        Returns the duration of the run.
        """
        return 0.0

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        """Run the test suite with coverage measurement.

        Must install a particular version of coverage using `cov_ver.pip_args`.

        Returns the duration of the run.
        """
        return 0.0


class EmptyProject(ProjectToTest):
    """A dummy project for testing other parts of this code."""

    def __init__(self, slug: str = "empty", fake_durations: Iterable[float] = (1.23,)):
        self.slug = slug
        self.durations = iter(itertools.cycle(fake_durations))

    def get_source(self, shell: ShellSession, retries: int = 5) -> None:
        pass

    def run_no_coverage(self, env: Env) -> float:
        """Run the test suite with coverage measurement."""
        return next(self.durations)

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        """Run the test suite with coverage measurement."""
        return next(self.durations)


class ToxProject(ProjectToTest):
    """A project using tox to run the test suite."""

    env_vars: Env_VarsType = {
        **(ProjectToTest.env_vars or {}),
        # Allow some environment variables into the tox execution.
        "TOX_OVERRIDE": "testenv.pass_env+=COVERAGE_DEBUG,COVERAGE_CORE,COVERAGE_FORCE_CONFIG",
        "COVERAGE_DEBUG": "config,sys",
    }

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install tox")
        self.run_tox(env, env.pyver.toxenv, "--notest")

    def run_tox(self, env: Env, toxenv: str, toxargs: str = "") -> float:
        """Run a tox command. Return the duration."""
        env.shell.run_command(f"{env.python} -m tox -v -e {toxenv} {toxargs}")
        return env.shell.last_duration

    def run_no_coverage(self, env: Env) -> float:
        return self.run_tox(env, env.pyver.toxenv, "--skip-pkg-install")

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        self.run_tox(env, env.pyver.toxenv, "--notest")
        env.shell.run_command(
            f".tox/{env.pyver.toxenv}/bin/python -m pip install {cov_ver.pip_args}"
        )
        with self.tweak_coverage_settings(cov_ver.tweaks):
            self.pre_check(env)  # NOTE: Not properly factored, and only used from here.
            duration = self.run_tox(env, env.pyver.toxenv, "--skip-pkg-install")
            self.post_check(
                env
            )  # NOTE: Not properly factored, and only used from here.
        return duration


class ProjectPytestHtml(ToxProject):
    """pytest-dev/pytest-html"""

    git_url = "https://github.com/pytest-dev/pytest-html"

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        raise Exception("This doesn't work because options changed to tweaks")
        covenv = env.pyver.toxenv + "-cov"  # type: ignore[unreachable]
        self.run_tox(env, covenv, "--notest")
        env.shell.run_command(
            f".tox/{covenv}/bin/python -m pip install {cov_ver.pip_args}"
        )
        if cov_ver.tweaks:
            replace = ("# reference: https", f"[run]\n{cov_ver.tweaks}\n#")
        else:
            replace = ("", "")
        with file_replace(Path(".coveragerc"), *replace):
            env.shell.run_command("cat .coveragerc")
            env.shell.run_command(f".tox/{covenv}/bin/python -m coverage debug sys")
            return self.run_tox(env, covenv, "--skip-pkg-install")


class ProjectDateutil(ToxProject):
    """dateutil/dateutil"""

    git_url = "https://github.com/dateutil/dateutil"

    def prep_environment(self, env: Env) -> None:
        super().prep_environment(env)
        env.shell.run_command(f"{env.python} updatezinfo.py")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command("echo No option to run without coverage")
        return 0


class ProjectAttrs(ToxProject):
    """python-attrs/attrs"""

    git_url = "https://github.com/python-attrs/attrs"

    @contextlib.contextmanager
    def tweak_coverage_settings(self, tweaks: TweaksType) -> Iterator[None]:
        return tweak_toml_coverage_settings("pyproject.toml", tweaks)

    def pre_check(self, env: Env) -> None:
        env.shell.run_command("cat pyproject.toml")

    def post_check(self, env: Env) -> None:
        env.shell.run_command("ls -al")


class ProjectDjangoAuthToolkit(ToxProject):
    """jazzband/django-oauth-toolkit"""

    git_url = "https://github.com/jazzband/django-oauth-toolkit"

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command("echo No option to run without coverage")
        return 0


class ProjectDjango(ToxProject):
    """django/django"""

    # brew install libmemcached
    # pip install -e .
    # coverage run tests/runtests.py --settings=test_sqlite
    # coverage report --format=total --precision=6
    # 32.848540


class ProjectMashumaro(ProjectToTest):
    git_url = "https://github.com/Fatal1ty/mashumaro"

    def __init__(self, more_pytest_args: str = ""):
        super().__init__()
        self.more_pytest_args = more_pytest_args

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install .")
        env.shell.run_command(f"{env.python} -m pip install -r requirements-dev.txt")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest {self.more_pytest_args}")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(
            f"{env.python} -m pytest --cov=mashumaro --cov=tests {self.more_pytest_args}"
        )
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectMashumaroBranch(ProjectMashumaro):
    def __init__(self, more_pytest_args: str = ""):
        super().__init__(more_pytest_args="--cov-branch " + more_pytest_args)
        self.slug = "mashbranch"


class ProjectOperator(ProjectToTest):
    git_url = "https://github.com/nedbat/operator"

    def __init__(self, more_pytest_args: str = ""):
        super().__init__()
        self.more_pytest_args = more_pytest_args

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install tox")
        Path("/tmp/operator_tmp").mkdir(exist_ok=True)
        env.shell.run_command(f"{env.python} -m tox -e unit --notest")
        env.shell.run_command(f"{env.python} -m tox -e unitnocov --notest")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(
            f"TMPDIR=/tmp/operator_tmp {env.python} -m tox -e unitnocov --skip-pkg-install"
            + f" -- {self.more_pytest_args}"
        )
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(
            f"TMPDIR=/tmp/operator_tmp {env.python} -m tox -e unit --skip-pkg-install"
            + f" -- {self.more_pytest_args}"
        )
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectPygments(ToxProject):
    git_url = "https://github.com/pygments/pygments"

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        self.run_tox(env, env.pyver.toxenv, "--notest")
        env.shell.run_command(
            f".tox/{env.pyver.toxenv}/bin/python -m pip install {cov_ver.pip_args}"
        )
        with self.tweak_coverage_settings(cov_ver.tweaks):
            self.pre_check(env)  # NOTE: Not properly factored, and only used here.
            duration = self.run_tox(
                env, env.pyver.toxenv, "--skip-pkg-install -- --cov"
            )
            self.post_check(env)  # NOTE: Not properly factored, and only used here.
        return duration


class ProjectRich(ToxProject):
    git_url = "https://github.com/Textualize/rich"

    def prep_environment(self, env: Env) -> None:
        raise Exception("Doesn't work due to poetry install error.")


class ProjectTornado(ToxProject):
    git_url = "https://github.com/tornadoweb/tornado"

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m tornado.test")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(f"{env.python} -m coverage run -m tornado.test")
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectDulwich(ToxProject):
    git_url = "https://github.com/jelmer/dulwich"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install -r requirements.txt")
        env.shell.run_command(f"{env.python} -m pip install .")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m unittest tests.test_suite")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(
            f"{env.python} -m coverage run -m unittest tests.test_suite"
        )
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectBlack(ToxProject):
    git_url = "https://github.com/psf/black"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install -r test_requirements.txt")
        env.shell.run_command(f"{env.python} -m pip install -e .[d]")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(
            f"{env.python} -m pytest tests --run-optional no_jupyter --no-cov --numprocesses 1"
        )
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(
            f"{env.python} -m pytest tests --run-optional no_jupyter --cov --numprocesses 1"
        )
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectMpmath(ProjectToTest):
    git_url = "https://github.com/mpmath/mpmath"
    select = "-k 'not (torture or extra or functions2 or calculus or cli or elliptic or quad)'"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install .[develop]")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest {self.select} --no-cov")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(f"{env.python} -m pytest {self.select} --cov=mpmath")
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectMypy(ToxProject):
    git_url = "https://github.com/python/mypy"

    SLOW_TESTS = " or ".join([
        "PythonCmdline",
        "PEP561Suite",
        "PythonEvaluation",
        "testdaemon",
        "StubgenCmdLine",
        "StubgenPythonSuite",
        "TestRun",
        "TestRunMultiFile",
        "TestExternal",
        "TestCommandLine",
        "ErrorStreamSuite",
    ])

    FAST = f"-k 'not ({SLOW_TESTS})'"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install -r test-requirements.txt")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest {self.FAST} --no-cov")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        pforce = Path("force.ini")
        pforce.write_text("[run]\nbranch=false\n")
        with env.shell.set_env({"COVERAGE_FORCE_CONFIG": str(pforce.resolve())}):
            env.shell.run_command(f"{env.python} -m pytest {self.FAST} --cov")
            duration = env.shell.last_duration
            report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectHtml5lib(ToxProject):
    git_url = "https://github.com/html5lib/html5lib-python"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install -r requirements-test.txt")
        env.shell.run_command(f"{env.python} -m pip install .")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(f"{env.python} -m coverage run -m pytest")
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectSphinx(ToxProject):
    git_url = "https://github.com/sphinx-doc/sphinx"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install .[test]")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(f"{env.python} -m coverage run -m pytest")
        duration = env.shell.last_duration
        env.shell.run_command(f"{env.python} -m coverage combine")
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


class ProjectUrllib3(ProjectToTest):
    git_url = "https://github.com/urllib3/urllib3"

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install -r dev-requirements.txt")
        env.shell.run_command(f"{env.python} -m pip install .")

    def run_no_coverage(self, env: Env) -> float:
        env.shell.run_command(f"{env.python} -m pytest")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        env.shell.run_command(f"{env.python} -m coverage run -m pytest")
        duration = env.shell.last_duration
        report = env.shell.run_command(f"{env.python} -m coverage report --precision=6")
        print("Results:", report.splitlines()[-1])
        return duration


def tweak_toml_coverage_settings(toml_file: str, tweaks: TweaksType) -> Iterator[None]:
    if tweaks:
        toml_inserts = []
        for name, value in tweaks:
            if isinstance(value, bool):
                toml_inserts.append(f"{name} = {str(value).lower()}")
            elif isinstance(value, str):
                toml_inserts.append(f"{name} = '{value}'")
            else:
                raise Exception(f"Can't tweak toml setting: {name} = {value!r}")
        header = "[tool.coverage.run]\n"
        insert = header + "\n".join(toml_inserts) + "\n"
    else:
        header = insert = ""
    return file_replace(Path(toml_file), header, insert)  # type: ignore


class AdHocProject(ProjectToTest):
    """A standalone program to run locally."""

    def __init__(
        self, python_file: str, cur_dir: str | None = None, pip_args: str = ""
    ):
        super().__init__()
        self.python_file = Path(python_file)
        if not self.python_file.exists():
            raise ValueError(f"Couldn't find {self.python_file} to run ad-hoc.")
        self.cur_dir = Path(cur_dir or self.python_file.parent)
        if not self.cur_dir.exists():
            raise ValueError(f"Couldn't find {self.cur_dir} to run in.")
        self.pip_args = pip_args
        self.slug = self.python_file.name

    def get_source(self, shell: ShellSession, retries: int = 5) -> None:
        pass

    def prep_environment(self, env: Env) -> None:
        env.shell.run_command(f"{env.python} -m pip install {self.pip_args}")

    def run_no_coverage(self, env: Env) -> float:
        with change_dir(self.cur_dir):
            env.shell.run_command(f"{env.python} {self.python_file}")
        return env.shell.last_duration

    def run_with_coverage(self, env: Env, cov_ver: Coverage) -> float:
        env.shell.run_command(f"{env.python} -m pip install {cov_ver.pip_args}")
        with change_dir(self.cur_dir):
            env.shell.run_command(f"{env.python} -m coverage run {self.python_file}")
        return env.shell.last_duration


class SlipcoverBenchmark(AdHocProject):
    """
    For running code from the Slipcover benchmarks.

    Clone https://github.com/plasma-umass/slipcover to /src/slipcover

    """

    def __init__(self, python_file: str):
        super().__init__(
            python_file=f"/src/slipcover/benchmarks/{python_file}",
            cur_dir="/src/slipcover",
            pip_args="six pyperf",
        )


class PyVersion:
    """A version of Python to use."""

    # The command to run this Python
    command: str
    # Short word for messages, directories, etc
    slug: str
    # The tox environment to run this Python
    toxenv: str


class Python(PyVersion):
    """A version of CPython to use."""

    def __init__(self, major: int, minor: int):
        self.command = self.slug = f"python{major}.{minor}"
        self.toxenv = f"py{major}{minor}"


class PyPy(PyVersion):
    """A version of PyPy to use."""

    def __init__(self, major: int, minor: int):
        self.command = self.slug = f"pypy{major}.{minor}"
        self.toxenv = f"pypy{major}{minor}"


class AdHocPython(PyVersion):
    """A custom build of Python to use."""

    def __init__(self, path: str, slug: str):
        self.command = f"{path}/bin/python3"
        file_must_exist(self.command, "python command")
        self.slug = slug
        self.toxenv = ""


@dataclass
class Coverage:
    """A version of coverage.py to use, maybe None."""

    # Short word for messages, directories, etc
    slug: str
    # Arguments for "pip install ..."
    pip_args: str | None = None
    # Tweaks to the .coveragerc file
    tweaks: TweaksType = None
    # Environment variables to set
    env_vars: Env_VarsType = None


class NoCoverage(Coverage):
    """Run without coverage at all."""

    def __init__(self, slug: str = "nocov"):
        super().__init__(slug=slug, pip_args=None)


class CoveragePR(Coverage):
    """A version of coverage.py from a pull request."""

    def __init__(
        self, number: int, tweaks: TweaksType = None, env_vars: Env_VarsType = None
    ):
        url = f"https://github.com/nedbat/coveragepy.git@refs/pull/{number}/merge"
        url_must_exist(url)
        super().__init__(
            slug=f"#{number}",
            pip_args=f"git+{url}",
            tweaks=tweaks,
            env_vars=env_vars,
        )


class CoverageCommit(Coverage):
    """A version of coverage.py from a specific commit."""

    def __init__(
        self, sha: str, tweaks: TweaksType = None, env_vars: Env_VarsType = None
    ):
        url = f"https://github.com/nedbat/coveragepy.git@{sha}"
        url_must_exist(url)
        super().__init__(
            slug=sha,
            pip_args=f"git+{url}",
            tweaks=tweaks,
            env_vars=env_vars,
        )


class CoverageSource(Coverage):
    """The coverage.py in a working tree."""

    def __init__(
        self,
        directory_name: str = "..",
        slug: str = "source",
        tweaks: TweaksType = None,
        env_vars: Env_VarsType = None,
    ):
        directory = file_must_exist(directory_name, "coverage directory")
        super().__init__(
            slug=slug,
            pip_args=str(directory),
            tweaks=tweaks,
            env_vars=env_vars,
        )


@dataclass
class Env:
    """An environment to run a test suite in."""

    pyver: PyVersion
    python: Path
    shell: ShellSession


ResultKey = Tuple[str, str, str]

DIMENSION_NAMES = ["proj", "pyver", "cov"]


class Experiment:
    """A particular time experiment to run."""

    def __init__(
        self,
        py_versions: list[PyVersion],
        cov_versions: list[Coverage],
        projects: list[ProjectToTest],
        results_file: str = "results.json",
        load: bool = False,
        cwd: str = "",
    ):
        self.py_versions = py_versions
        self.cov_versions = cov_versions
        self.projects = projects
        self.results_file = Path(cwd) / Path(results_file)
        self.result_data: dict[ResultKey, list[float]] = (
            self.load_results() if load else {}
        )
        self.summary_data: dict[ResultKey, float] = {}

    def save_results(self) -> None:
        """Save current results to the JSON file."""
        with self.results_file.open("w") as f:
            json.dump({" ".join(k): v for k, v in self.result_data.items()}, f)

    def load_results(self) -> dict[ResultKey, list[float]]:
        """Load results from the JSON file if it exists."""
        if self.results_file.exists():
            with self.results_file.open("r") as f:
                data: dict[str, list[float]] = json.load(f)
            return {
                (k.split()[0], k.split()[1], k.split()[2]): v for k, v in data.items()
            }
        return {}

    def run(self, num_runs: int = 3) -> None:
        total_runs = (
            len(self.projects)
            * len(self.py_versions)
            * len(self.cov_versions)
            * num_runs
        )
        total_run_nums = iter(itertools.count(start=1))

        all_runs = []

        for proj in self.projects:
            with proj.shell() as shell:
                print(f"Prepping project {proj.slug}")
                shell.print_banner(f"Prepping project {proj.slug}")
                proj.make_dir()
                proj.get_source(shell)

                for pyver in self.py_versions:
                    print(f"Making venv for {proj.slug} {pyver.slug}")
                    venv_dir = f"venv_{proj.slug}_{pyver.slug}"
                    shell.run_command(f"{pyver.command} -m venv {venv_dir}")
                    python = Path.cwd() / f"{venv_dir}/bin/python"
                    shell.run_command(f"{python} -V")
                    shell.run_command(f"{python} -m pip install -U pip")
                    env = Env(pyver, python, shell)

                    with change_dir(proj.dir):
                        print(f"Prepping for {proj.slug} {pyver.slug}")
                        proj.prep_environment(env)
                        for cov_ver in self.cov_versions:
                            all_runs.append((proj, pyver, cov_ver, env))

        all_runs *= num_runs
        random.shuffle(all_runs)

        run_data: dict[ResultKey, list[float]] = collections.defaultdict(list)
        run_data.update(self.result_data)

        for proj, pyver, cov_ver, env in all_runs:
            result_key = (proj.slug, pyver.slug, cov_ver.slug)
            total_run_num = next(total_run_nums)
            if (
                result_key in self.result_data
                and len(self.result_data[result_key]) >= num_runs
            ):
                print(f"Skipping {result_key} as results already exist.")
                continue

            with env.shell:
                banner = (
                    "Running tests: "
                    + f"proj={proj.slug}, py={pyver.slug}, cov={cov_ver.slug}, "
                    + f"{total_run_num} of {total_runs}"
                )
                print(banner)
                env.shell.print_banner(banner)
                with change_dir(proj.dir):
                    with env.shell.set_env(proj.env_vars, cov_ver.env_vars):
                        try:
                            if cov_ver.pip_args is None:
                                dur = proj.run_no_coverage(env)
                            else:
                                dur = proj.run_with_coverage(env, cov_ver)
                        except Exception as exc:
                            print(f"!!! {exc = }")
                            traceback.print_exc(file=env.shell.foutput)
                            dur = float("NaN")
            print(f"Tests took {dur:.3f}s")
            if result_key not in self.result_data:
                self.result_data[result_key] = []
            self.result_data[result_key].append(dur)
            run_data[result_key].append(dur)
            self.save_results()

        # Summarize and collect the data.
        print("# Results")
        for proj in self.projects:
            for pyver in self.py_versions:
                for cov_ver in self.cov_versions:
                    result_key = (proj.slug, pyver.slug, cov_ver.slug)
                    data = run_data[result_key]
                    med = statistics.median(data)
                    self.summary_data[result_key] = med
                    stdev = statistics.stdev(data) if len(data) > 1 else 0.0
                    summary = (
                        f"Median for {proj.slug}, {pyver.slug}, {cov_ver.slug}: "
                        + f"{med:.3f}s, "
                        + f"stdev={stdev:.3f}"
                    )
                    if 1:
                        data_sum = ", ".join(f"{d:.3f}" for d in data)
                        summary += f", data={data_sum}"
                    print(summary)

    def show_results(
        self,
        rows: list[str],
        column: str,
        ratios: Iterable[tuple[str, str, str]] = (),
    ) -> None:
        dimensions = {
            "cov": [cov_ver.slug for cov_ver in self.cov_versions],
            "pyver": [pyver.slug for pyver in self.py_versions],
            "proj": [proj.slug for proj in self.projects],
        }

        table_axes = [dimensions[rowname] for rowname in rows]
        data_order = [*rows, column]
        remap = [data_order.index(datum) for datum in DIMENSION_NAMES]

        header = []
        header.extend(rows)
        header.extend(dimensions[column])
        header.extend(slug for slug, _, _ in ratios)

        aligns = ["left"] * len(rows) + ["right"] * (len(header) - len(rows))
        data = []

        for tup in itertools.product(*table_axes):
            row: list[str] = []
            row.extend(tup)
            col_data = {}
            for col in dimensions[column]:
                key = (*tup, col)
                key = tuple(key[i] for i in remap)
                key = cast(ResultKey, key)
                result_time = self.summary_data[key]
                row.append(f"{result_time:.1f}s")
                col_data[col] = result_time
            for _, num, denom in ratios:
                ratio = col_data[num] / col_data[denom]
                row.append(f"{ratio * 100:.0f}%")
            data.append(row)

        print()
        print(tabulate.tabulate(data, headers=header, colalign=aligns, tablefmt="pipe"))


PERF_DIR = Path("/tmp/covperf")


def run_experiment(
    py_versions: list[PyVersion],
    cov_versions: list[Coverage],
    projects: list[ProjectToTest],
    rows: list[str],
    column: str,
    ratios: Iterable[tuple[str, str, str]] = (),
    num_runs: int = int(sys.argv[1]),
    load: bool = False,
) -> None:
    """
    Run a benchmarking experiment and print a table of results.

    Arguments:

        py_versions: The Python versions to test.
        cov_versions: The coverage versions to test.
        projects: The projects to run.
        rows: A list of strings chosen from `"pyver"`, `"cov"`, and `"proj"`.
        column: The remaining dimension not used in `rows`.
        ratios: A list of triples: (title, slug1, slug2).
        num_runs: The number of times to run each matrix element.

    """
    slugs = [v.slug for v in py_versions + cov_versions + projects]
    if len(set(slugs)) != len(slugs):
        raise Exception(f"Slugs must be unique: {slugs}")
    if any(" " in slug for slug in slugs):
        raise Exception(f"No spaces in slugs please: {slugs}")
    ratio_slugs = [rslug for ratio in ratios for rslug in ratio[1:]]
    if any(rslug not in slugs for rslug in ratio_slugs):
        raise Exception(f"Ratio slug doesn't match a slug: {ratio_slugs}, {slugs}")
    if set(rows + [column]) != set(DIMENSION_NAMES):
        raise Exception(
            f"All of these must be in rows or column: {', '.join(DIMENSION_NAMES)}"
        )

    print(f"Removing and re-making {PERF_DIR}")
    rmrf(PERF_DIR)

    cwd = str(Path.cwd())
    with change_dir(PERF_DIR):
        exp = Experiment(
            py_versions=py_versions,
            cov_versions=cov_versions,
            projects=projects,
            load=load,
            cwd=cwd,
        )
        exp.run(num_runs=int(num_runs))
        exp.show_results(rows=rows, column=column, ratios=ratios)