runner.py

from __future__ import annotations

import logging
import os
import re
from abc import ABC, abstractmethod
from hashlib import sha256
from typing import TYPE_CHECKING, Any, Iterable, List

from tox.config.types import Command, EnvList

from .api import ToxEnv, ToxEnvCreateArgs
from .package import Package, PackageToxEnv, PathPackage
from .util import add_change_dir_conf

if TYPE_CHECKING:
    from tox.journal import EnvJournal


class RunToxEnv(ToxEnv, ABC):
    def __init__(self, create_args: ToxEnvCreateArgs) -> None:
        self.package_env: PackageToxEnv | None = None
        self._packages: list[Package] = []
        super().__init__(create_args)
        self._package_envs: list[PackageToxEnv | Exception] | None = None

    def register_config(self) -> None:
        def ensure_one_line(value: str) -> str:
            return re.sub(r"\s+", " ", value.replace("\r", "").replace("\n", " "))

        self.conf.add_config(
            keys=["description"],
            of_type=str,
            default="",
            desc="description attached to the tox environment",
            post_process=ensure_one_line,
        )
        self.conf.add_config(
            "depends",
            of_type=EnvList,
            desc="tox environments that this environment depends on (must be run after those)",
            default=EnvList([]),
        )
        super().register_config()
        self.conf.add_config(
            keys=["commands_pre"],
            of_type=List[Command],
            default=[],
            desc="the commands to be called before testing",
        )
        self.conf.add_config(
            keys=["commands"],
            of_type=List[Command],
            default=[],
            desc="the commands to be called for testing",
        )
        self.conf.add_config(
            keys=["commands_post"],
            of_type=List[Command],
            default=[],
            desc="the commands to be called after testing",
        )
        add_change_dir_conf(self.conf, self.core)
        self.conf.add_config(
            keys=["args_are_paths"],
            of_type=bool,
            default=True,
            desc="if True rewrite relative posargs paths from cwd to change_dir",
        )
        self.conf.add_config(
            keys=["ignore_errors"],
            of_type=bool,
            default=False,
            desc="when executing the commands keep going even if a sub-command exits with non-zero exit code",
        )
        self.conf.add_config(
            keys=["ignore_outcome"],
            of_type=bool,
            default=False,
            desc="if set to true a failing result of this testenv will not make tox fail (instead just warn)",
        )

    def _teardown(self) -> None:
        super()._teardown()
        self._call_pkg_envs("teardown_env", self.conf)

    def interrupt(self) -> None:
        super().interrupt()
        self._call_pkg_envs("interrupt")

    def get_package_env_types(self) -> tuple[str, str] | None:
        if self._register_package_conf():
            has_external_pkg = self.conf["package"] == "external"
            self.core.add_config(
                keys=["package_env", "isolated_build_env"],
                of_type=str,
                default=self._default_package_env,
                desc="tox environment used to package",
            )
            self.conf.add_config(
                keys=["package_env"],
                of_type=str,
                default=f'{self.core["package_env"]}{"_external" if has_external_pkg else ""}',
                desc="tox environment used to package",
            )
            is_external = self.conf["package"] == "external"
            self.conf.add_constant(
                keys=["package_tox_env_type"],
                desc="tox package type used to generate the package",
                value=self._external_pkg_tox_env_type if is_external else self._package_tox_env_type,
            )
            return self.conf["package_env"], self.conf["package_tox_env_type"]
        return None

    def _call_pkg_envs(self, method_name: str, *args: Any) -> None:
        for package_env in self.package_envs:
            with package_env.display_context(suspend=self._has_display_suspended):
                getattr(package_env, method_name)(*args)

    def _clean(self, transitive: bool = False) -> None:  # noqa: FBT001, FBT002
        super()._clean(transitive)
        if transitive:
            self._call_pkg_envs("_clean")

    @property
    def _default_package_env(self) -> str:
        return ".pkg"

    @property
    @abstractmethod
    def _package_tox_env_type(self) -> str:
        raise NotImplementedError

    @property
    @abstractmethod
    def _external_pkg_tox_env_type(self) -> str:
        raise NotImplementedError

    def _setup_with_env(self) -> None:
        if self.package_env is not None:
            skip_pkg_install: bool = getattr(self.options, "skip_pkg_install", False)
            if skip_pkg_install is True:
                logging.warning("skip building and installing the package")
            else:
                self._setup_pkg()

    def _register_package_conf(self) -> bool:
        """If this returns True package_env and package_tox_env_type configurations must be defined."""
        self.core.add_config(
            keys=["no_package", "skipsdist"],
            of_type=bool,
            default=False,
            desc="is there any packaging involved in this project",
        )
        core_no_package: bool = self.core["no_package"]
        if core_no_package is True:
            return False
        self.conf.add_config(
            keys="skip_install",
            of_type=bool,
            default=False,
            desc="skip installation",
        )
        skip_install: bool = self.conf["skip_install"]
        return not skip_install

    def _setup_pkg(self) -> None:
        self._packages = self._build_packages()
        if not self.options.package_only:
            self._install(self._packages, RunToxEnv.__name__, "package")
        self._handle_journal_package(self.journal, self._packages)

    @staticmethod
    def _handle_journal_package(journal: EnvJournal, packages: list[Package]) -> None:
        if not journal:
            return
        installed_meta = []
        for package in packages:
            if isinstance(package, PathPackage):
                pkg = package.path
                of_type = "file" if pkg.is_file() else ("dir" if pkg.is_dir() else "N/A")
                meta = {"basename": pkg.name, "type": of_type}
                if of_type == "file":
                    meta["sha256"] = sha256(pkg.read_bytes()).hexdigest()
            else:
                raise NotImplementedError
            installed_meta.append(meta)
        if installed_meta:
            journal["installpkg"] = installed_meta[0] if len(installed_meta) == 1 else installed_meta

    @property
    def environment_variables(self) -> dict[str, str]:
        environment_variables = super().environment_variables
        if self.package_env is not None and self._packages:
            # if package(s) have been built insert them as environment variable
            environment_variables["TOX_PACKAGE"] = os.pathsep.join(str(i) for i in self._packages)
        return environment_variables

    @abstractmethod
    def _build_packages(self) -> list[Package]:
        """:returns: a list of packages installed in the environment"""
        raise NotImplementedError

    @property
    def package_envs(self) -> Iterable[PackageToxEnv]:
        if self.package_env is not None:
            yield self.package_env
            yield from self.package_env.child_pkg_envs(self.conf)

    def mark_active(self) -> None:
        for pkg_env in self.package_envs:
            pkg_env.mark_active_run_env(self)