Source code for edgetest.lib

"""Default virtual environment hook."""

import logging
import platform
from pathlib import Path

import click
import pluggy
from uv import find_uv_bin

from edgetest.utils import _run_command

LOG = logging.getLogger(__name__)

hookimpl = pluggy.HookimplMarker("edgetest")




@hookimpl(trylast=True)
def path_to_python(basedir: str, envname: str) -> str:
    """Return the path to the python executable."""
    if platform.system() == "Windows":
        return str(Path(basedir) / envname / "Scripts" / "python.exe")
    else:
        return str(Path(basedir) / envname / "bin" / "python")





@hookimpl(trylast=True)
def create_environment(basedir: str, envname: str, conf: dict):
    """Create the virtual environment for testing.

    Creates an environment using ``uv``.

    Parameters
    ----------
    basedir : str
        The base directory location for the environment.
    envname : str
        The name of the virtual environment.
    conf : dict
        Ignored.

    Raises
    ------
    RuntimeError
        Error raised if the environment cannot be created.
    """
    try:
        uv_ = find_uv_bin()
        callargs_ = [uv_, "venv", str(Path(basedir, envname))]
        if (py_version := conf.get("python_version")) is not None:
            callargs_.append(f"--python={py_version}")
        if Path(basedir, envname).is_dir():
            callargs_.append("--clear")
        _run_command(*callargs_)
    except Exception as err:
        raise RuntimeError(f"Unable to create {envname} in {basedir}") from err





@hookimpl(trylast=True)
def run_update(basedir: str, envname: str, upgrade: list, conf: dict):
    """Update packages from upgrade list.

    Parameters
    ----------
    basedir : str
        The base directory location for the environment.
    envname : str
        The name of the virtual environment.
    upgrade : list
        The list of packages to upgrade
    conf : dict
        The configuration dictionary. The configuration is parsed for ``exclude_newer``,
        which maps to the parameter of the same name in ``uv``.

    Raises
    ------
    RuntimeError
        Error raised if the packages cannot be updated.
    """
    python_path = path_to_python(basedir, envname)
    uv_ = find_uv_bin()
    callargs_ = [
        uv_,
        "pip",
        "install",
        f"--python={python_path}",
        *upgrade,
        "--upgrade",
    ]
    ctx = click.get_current_context()
    if (cooldown_ := conf.get("exclude_newer")) is not None and not ctx.params[
        "ignore_cooldown"
    ]:
        callargs_.append(f"--exclude-newer={cooldown_}")
    try:
        _run_command(*callargs_)
    except Exception as err:
        raise RuntimeError(f"Unable to pip upgrade: {upgrade}") from err





@hookimpl(trylast=True)
def run_install_lower(basedir: str, envname: str, lower: list[str], conf: dict):
    """Install lower bounds of packages provided.

    Parameters
    ----------
    basedir : str
        The base directory location for the environment.
    envname : str
        Environment to install into.
    lower : list[str]
        Lower bounds of packages to install.
    conf : dict
        The configuration dictionary for the environment. This is useful if you
        want to add configuration arguments for additional dependencies that can
        only be installed through the environment manager (e.g. Conda).
    """
    python_path = path_to_python(basedir, envname)
    try:
        uv_ = find_uv_bin()
        _run_command(uv_, "pip", "install", f"--python={python_path}", *lower)
    except Exception as err:
        raise RuntimeError(f"Unable to pip install: {lower}") from err





@hookimpl(tryfirst=True)
def post_run_hook(testers: list, conf: dict):
    """Refresh ``uv.lock`` based on the test output."""
    ctx = click.get_current_context()
    if not ctx.params["export"]:
        LOG.info(
            "Skipping ``uv lock --upgrade`` as the requirements have not been updated."
        )
    elif not all(tester.status for tester in testers):
        LOG.info("Skipping ``uv.lock`` refresh as not every environment passed.")
    elif (Path(ctx.params["config"]).parent / "uv.lock").is_file():
        # uv.lock exists already and the whole matrix is green
        try:
            uv_ = find_uv_bin()
            _run_command(uv_, "lock", "--upgrade")
        except RuntimeError:
            LOG.info("Unable to update the ``uv.lock`` file.")
    else:
        LOG.info(
            "Skipping ``uv.lock`` refresh as we couldn't find an existing ``uv.lock`` file."
        )