diff --git a/assets/conx.png b/assets/conx.png new file mode 100644 index 0000000..3736844 Binary files /dev/null and b/assets/conx.png differ diff --git a/assets/conx.svg b/assets/conx.svg new file mode 100644 index 0000000..f0b2430 --- /dev/null +++ b/assets/conx.svg @@ -0,0 +1,134 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/condax/__init__.py b/condax/__init__.py index a96c2d4..012e76d 100644 --- a/condax/__init__.py +++ b/condax/__init__.py @@ -2,8 +2,8 @@ if sys.version_info >= (3, 8): - import importlib.metadata as metadata + import importlib.metadata as _metadata else: - import importlib_metadata as metadata + import importlib_metadata as _metadata -__version__ = metadata.version(__package__) +__version__ = _metadata.version(__package__) diff --git a/condax/cli/__init__.py b/condax/cli/__init__.py index e1c83de..aade6f7 100644 --- a/condax/cli/__init__.py +++ b/condax/cli/__init__.py @@ -1,4 +1,5 @@ import click +from click_aliases import ClickAliasedGroup import condax.config as config from condax import __version__ @@ -11,7 +12,8 @@ Conda environment location is {config.DEFAULT_PREFIX_DIR}\n Links to apps are placed in {config.DEFAULT_BIN_DIR} - """ + """, + cls=ClickAliasedGroup, ) @click.version_option( __version__, diff --git a/condax/cli/__main__.py b/condax/cli/__main__.py index 764d769..63c7c42 100644 --- a/condax/cli/__main__.py +++ b/condax/cli/__main__.py @@ -1,10 +1,9 @@ import logging import sys from urllib.error import HTTPError -from condax import config from condax.exceptions import CondaxError from .install import install -from .remove import remove, uninstall +from .remove import remove from .update import update from .list import run_list from .ensure_path import ensure_path @@ -19,7 +18,6 @@ def main(): for subcommand in ( install, remove, - uninstall, update, run_list, ensure_path, @@ -34,10 +32,6 @@ def main(): logger = logging.getLogger(__package__) try: - try: - config.set_via_file(config.DEFAULT_CONFIG) - except config.MissingConfigFileError: - pass cli() except CondaxError as e: if e.exit_code: diff --git a/condax/cli/install.py b/condax/cli/install.py index 2fa4a54..0498ae4 100644 --- a/condax/cli/install.py +++ b/condax/cli/install.py @@ -1,11 +1,8 @@ import logging -from typing import List +from typing import Iterable, List -import click - -import condax.config as config -import condax.core as core -from condax import __version__ +from condax import __version__, consts, core +from condax.condax import Condax from . import cli, options @@ -15,7 +12,7 @@ Install a package with condax. This will install a package into a new conda environment and link the executable - provided by it to `{config.DEFAULT_BIN_DIR}`. + provided by it to `{consts.DEFAULT_PATHS.bin_dir}`. """ ) @options.channels @@ -25,10 +22,13 @@ def install( packages: List[str], is_forcing: bool, - log_level: int, + channels: Iterable[str], + condax: Condax, **_, ): for pkg in packages: - core.install_package( - pkg, is_forcing=is_forcing, conda_stdout=log_level <= logging.INFO + condax.install_package( + pkg, + is_forcing=is_forcing, + channels=channels, ) diff --git a/condax/cli/options.py b/condax/cli/options.py index 4220b00..6b08053 100644 --- a/condax/cli/options.py +++ b/condax/cli/options.py @@ -1,22 +1,28 @@ import logging +import subprocess import rainbowlog +import yaml from statistics import median -from typing import Callable, Sequence +from typing import Any, Callable, Mapping, Optional, Sequence from pathlib import Path from functools import wraps -from condax import config +from condax import consts +from condax.condax import Condax +from condax.conda import Conda import click +from condax.utils import FullPath + def common(f: Callable) -> Callable: """ This decorator adds common options to the CLI. """ options: Sequence[Callable] = ( - config_file, - log_level, + condax, + setup_logging, click.help_option("-h", "--help"), ) @@ -28,12 +34,27 @@ def common(f: Callable) -> Callable: packages = click.argument("packages", nargs=-1, required=True) -config_file = click.option( + +def _config_file_callback(_, __, config_file: Path) -> Mapping[str, Any]: + try: + with (config_file or consts.DEFAULT_PATHS.conf_file).open() as cf: + config = yaml.safe_load(cf) or {} + except FileNotFoundError: + config = {} + + if not isinstance(config, dict): + raise click.BadParameter( + f"Config file {config_file} must contain a dict as its root." + ) + + return config + + +config = click.option( "--config", - "config_file", type=click.Path(exists=True, path_type=Path), - help=f"Custom path to a condax config file in YAML. Default: {config.DEFAULT_CONFIG}", - callback=lambda _, __, f: (f and config.set_via_file(f)) or f, + help=f"Custom path to a condax config file in YAML. Default: {consts.DEFAULT_PATHS.conf_file}", + callback=_config_file_callback, ) channels = click.option( @@ -41,9 +62,7 @@ def common(f: Callable) -> Callable: "-c", "channels", multiple=True, - help=f"""Use the channels specified to install. If not specified condax will - default to using {config.DEFAULT_CHANNELS}, or 'channels' in the config file.""", - callback=lambda _, __, c: (c and config.set_via_value(channels=c)) or c, + help="Use the channels specified in addition to those in the configuration files of condax, conda, and/or mamba.", ) envname = click.option( @@ -80,11 +99,67 @@ def common(f: Callable) -> Callable: help="Decrease verbosity level.", ) +bin_dir = click.option( + "-b", + "--bin-dir", + type=click.Path(exists=True, path_type=Path), + help=f"Custom path to the condax bin directory. Default: {consts.DEFAULT_PATHS.bin_dir}", +) + + +def conda(f: Callable) -> Callable: + """ + This click option decorator adds the --config option to the CLI. + It constructs a `Conda` object and passes it to the decorated function as `conda`. + It reads the config file and passes it as a dict to the decorated function as `config`. + """ + + @config + @wraps(f) + def construct_conda_hook(config: Mapping[str, Any], **kwargs): + return f( + conda=Conda(config.get("channels", [])), + config=config, + **kwargs, + ) + + return construct_conda_hook + + +def condax(f: Callable) -> Callable: + """ + This click option decorator adds the --bin-dir option as well as all those added by `options.conda` to the CLI. + It then constructs a `Condax` object and passes it to the decorated function as `condax`. + """ + + @conda + @bin_dir + @wraps(f) + def construct_condax_hook( + conda: Conda, config: Mapping[str, Any], bin_dir: Optional[Path], **kwargs + ): + return f( + condax=Condax( + conda, + bin_dir + or config.get("bin_dir", None) + or config.get("target_destination", None) # Compatibility <=0.0.5 + or consts.DEFAULT_PATHS.bin_dir, + FullPath( + config.get("prefix_dir", None) + or config.get("prefix_path", None) # Compatibility <=0.0.5 + or consts.DEFAULT_PATHS.prefix_dir + ), + ), + **kwargs, + ) + + return construct_condax_hook + -def log_level(f: Callable) -> Callable: +def setup_logging(f: Callable) -> Callable: """ This click option decorator adds -v and -q options to the CLI, then sets up logging with the specified level. - It passes the level to the decorated function as `log_level`. """ @verbose @@ -101,6 +176,6 @@ def setup_logging_hook(verbose: int, quiet: int, **kwargs): ) ) logger.setLevel(level) - return f(log_level=level, **kwargs) + return f(**kwargs) return setup_logging_hook diff --git a/condax/cli/remove.py b/condax/cli/remove.py index 3e51d29..98dfa89 100644 --- a/condax/cli/remove.py +++ b/condax/cli/remove.py @@ -1,9 +1,5 @@ -import logging from typing import List -import click - -import condax.core as core -from condax import __version__ +from condax.condax import Condax from . import cli, options @@ -14,21 +10,11 @@ This will remove a package installed with condax and destroy the underlying conda environment. - """ + """, + aliases=["uninstall"], ) @options.common @options.packages -def remove(packages: List[str], log_level: int, **_): +def remove(packages: List[str], condax: Condax, **_): for pkg in packages: - core.remove_package(pkg, conda_stdout=log_level <= logging.INFO) - - -@cli.command( - help=""" - Alias for condax remove. - """ -) -@options.common -@options.packages -def uninstall(packages: List[str], **_): - remove(packages) + condax.remove_package(pkg) diff --git a/condax/cli/repair.py b/condax/cli/repair.py index a9f8cbd..eed4b10 100644 --- a/condax/cli/repair.py +++ b/condax/cli/repair.py @@ -27,5 +27,5 @@ def repair(is_migrating, **_): if is_migrating: migrate.from_old_version() - conda.setup_micromamba() + conda.install_micromamba() core.fix_links() diff --git a/condax/cli/update.py b/condax/cli/update.py index 026f8cb..60754d8 100644 --- a/condax/cli/update.py +++ b/condax/cli/update.py @@ -1,11 +1,9 @@ -import logging -import sys from typing import List import click import condax.core as core -from condax import __version__ +from condax.condax import Condax from . import cli, options @@ -24,22 +22,29 @@ "--update-specs", is_flag=True, help="Update based on provided specifications." ) @options.common +@options.is_forcing +@options.channels @click.argument("packages", required=False, nargs=-1) -@click.pass_context def update( - ctx: click.Context, - all: bool, + condax: Condax, packages: List[str], + all: bool, update_specs: bool, - log_level: int, + is_forcing: bool, + channels: List[str], **_ ): + + if not (all or packages): + raise click.BadArgumentUsage( + "No packages specified. To update all packages use --all." + ) + + if all and packages: + raise click.BadArgumentUsage("Cannot specify packages and --all.") + if all: - core.update_all_packages(update_specs) - elif packages: - for pkg in packages: - core.update_package( - pkg, update_specs, conda_stdout=log_level <= logging.INFO - ) + condax.update_all_packages(channels, is_forcing) else: - ctx.fail("No packages specified.") + for pkg in packages: + condax.update_package(pkg, channels, update_specs, is_forcing) diff --git a/condax/conda.py b/condax/conda.py index 162baf3..3d7c9c6 100644 --- a/condax/conda.py +++ b/condax/conda.py @@ -1,3 +1,4 @@ +from functools import partial import io import json import logging @@ -15,71 +16,13 @@ from condax.config import C from condax.exceptions import CondaxError -from condax.utils import to_path +from condax.utils import FullPath import condax.utils as utils logger = logging.getLogger(__name__) -def _ensure(execs: Iterable[str], installer: Callable[[], Path]) -> Path: - for exe in execs: - exe_path = shutil.which(exe) - if exe_path is not None: - return to_path(exe_path) - - logger.info("No existing conda installation found. Installing the standalone") - return installer() - - -def ensure_conda() -> Path: - return _ensure(("conda", "mamba"), setup_conda) - - -def ensure_micromamba() -> Path: - return _ensure(("micromamba",), setup_micromamba) - - -def setup_conda() -> Path: - url = utils.get_conda_url() - resp = requests.get(url, allow_redirects=True) - resp.raise_for_status() - utils.mkdir(C.bin_dir()) - exe_name = "conda.exe" if os.name == "nt" else "conda" - target_filename = C.bin_dir() / exe_name - with open(target_filename, "wb") as fo: - fo.write(resp.content) - st = os.stat(target_filename) - os.chmod(target_filename, st.st_mode | stat.S_IXUSR) - return target_filename - - -def setup_micromamba() -> Path: - utils.mkdir(C.bin_dir()) - exe_name = "micromamba.exe" if os.name == "nt" else "micromamba" - umamba_exe = C.bin_dir() / exe_name - _download_extract_micromamba(umamba_exe) - return umamba_exe - - -def _download_extract_micromamba(umamba_dst: Path) -> None: - url = utils.get_micromamba_url() - print(f"Downloading micromamba from {url}") - response = requests.get(url, allow_redirects=True) - response.raise_for_status() - - utils.mkdir(umamba_dst.parent) - tarfile_obj = io.BytesIO(response.content) - with tarfile.open(fileobj=tarfile_obj) as tar, open(umamba_dst, "wb") as f: - p = "Library/bin/micromamba.exe" if os.name == "nt" else "bin/micromamba" - extracted = tar.extractfile(p) - if extracted: - shutil.copyfileobj(extracted, f) - - st = os.stat(umamba_dst) - os.chmod(umamba_dst, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - - ## Need to activate if using micromamba as drop-in replacement # def _activate_umamba(umamba_path: Path) -> None: # print("Activating micromamba") @@ -89,33 +32,6 @@ def _download_extract_micromamba(umamba_dst: Path) -> None: # ) -def create_conda_environment(spec: str, stdout: bool) -> None: - """Create an environment by installing a package. - - NOTE: `spec` may contain version specificaitons. - """ - conda_exe = ensure_conda() - prefix = conda_env_prefix(spec) - - channels = C.channels() - channels_args = [x for c in channels for x in ["--channel", c]] - - _subprocess_run( - [ - conda_exe, - "create", - "--prefix", - prefix, - "--override-channels", - *channels_args, - "--quiet", - "--yes", - shlex.quote(spec), - ], - suppress_stdout=not stdout, - ) - - def inject_to_conda_env(specs: Iterable[str], env_name: str, stdout: bool) -> None: """Add packages onto existing `env_name`. @@ -163,66 +79,6 @@ def uninject_from_conda_env( ) -def remove_conda_env(package: str, stdout: bool) -> None: - """Remove a conda environment.""" - conda_exe = ensure_conda() - - _subprocess_run( - [conda_exe, "remove", "--prefix", conda_env_prefix(package), "--all", "--yes"], - suppress_stdout=not stdout, - ) - - -def update_conda_env(spec: str, update_specs: bool, stdout: bool) -> None: - """Update packages in an environment. - - NOTE: More controls of package updates might be needed. - """ - _, match_spec = utils.split_match_specs(spec) - conda_exe = ensure_conda() - prefix = conda_env_prefix(spec) - channels_args = [x for c in C.channels() for x in ["--channel", c]] - update_specs_args = ["--update-specs"] if update_specs else [] - # NOTE: `conda update` does not support version specification. - # It suggets to use `conda install` instead. - args: Iterable[str] - if conda_exe.name == "conda" and match_spec: - subcmd = "install" - args = (shlex.quote(spec),) - elif match_spec: - subcmd = "update" - args = (*update_specs_args, shlex.quote(spec)) - else: - ## FIXME: this update process is inflexible - subcmd = "update" - args = (*update_specs_args, "--all") - - command: List[Union[Path, str]] = [ - conda_exe, - subcmd, - "--prefix", - prefix, - "--override-channels", - "--quiet", - "--yes", - *channels_args, - *args, - ] - - _subprocess_run(command, suppress_stdout=not stdout) - - -def has_conda_env(package: str) -> bool: - # TODO: check some properties of a conda environment - p = conda_env_prefix(package) - return p.exists() and p.is_dir() - - -def conda_env_prefix(spec: str) -> Path: - package, _ = utils.split_match_specs(spec) - return C.prefix_dir() / package - - def get_package_info(package: str, specific_name=None) -> Tuple[str, str, str]: env_prefix = conda_env_prefix(package) package_name = package if specific_name is None else specific_name @@ -245,46 +101,6 @@ def get_package_info(package: str, specific_name=None) -> Tuple[str, str, str]: return ("", "", "") -class DeterminePkgFilesError(CondaxError): - def __init__(self, package: str): - super().__init__(40, f"Could not determine package files: {package}.") - - -def determine_executables_from_env( - package: str, injected_package: Optional[str] = None -) -> List[Path]: - def is_good(p: Union[str, Path]) -> bool: - p = to_path(p) - return p.parent.name in ("bin", "sbin", "scripts", "Scripts") - - env_prefix = conda_env_prefix(package) - target_name = injected_package if injected_package else package - - conda_meta_dir = env_prefix / "conda-meta" - for file_name in conda_meta_dir.glob(f"{target_name}*.json"): - with file_name.open() as fo: - package_info = json.load(fo) - if package_info["name"] == target_name: - potential_executables: Set[str] = { - fn - for fn in package_info["files"] - if (fn.startswith("bin/") and is_good(fn)) - or (fn.startswith("sbin/") and is_good(fn)) - # They are Windows style path - or (fn.lower().startswith("scripts") and is_good(fn)) - or (fn.lower().startswith("library") and is_good(fn)) - } - break - else: - raise DeterminePkgFilesError(target_name) - - return sorted( - env_prefix / fn - for fn in potential_executables - if utils.is_executable(env_prefix / fn) - ) - - def _get_conda_package_dirs() -> List[Path]: """ Get the conda's global package directories. @@ -297,7 +113,7 @@ def _get_conda_package_dirs() -> List[Path]: return [] d = json.loads(res.stdout.decode()) - return [to_path(p) for p in d["pkgs_dirs"]] + return [FullPath(p) for p in d["pkgs_dirs"]] def _get_dependencies(package: str, pkg_dir: Path) -> List[str]: diff --git a/condax/conda/__init__.py b/condax/conda/__init__.py new file mode 100644 index 0000000..1df59fb --- /dev/null +++ b/condax/conda/__init__.py @@ -0,0 +1,3 @@ +from .conda import Conda + +__all__ = ["Conda"] diff --git a/condax/conda/conda.py b/condax/conda/conda.py new file mode 100644 index 0000000..1c02141 --- /dev/null +++ b/condax/conda/conda.py @@ -0,0 +1,148 @@ +import itertools +from pathlib import Path +import shlex +import subprocess +import logging +import sys +from typing import IO, Iterable, Optional +from halo import Halo + +from condax import consts, utils +from condax.conda.exceptions import CondaCommandError +from .installers import ensure_conda + + +logger = logging.getLogger(__name__) + + +class Conda: + def __init__(self, channels: Iterable[str] = ()) -> None: + """This class is a wrapper for conda's CLI. + + Args: + channels: Additional channels to use. + """ + self.channels = tuple(channels) + self.exe = ensure_conda(consts.DEFAULT_PATHS.bin_dir) + + def remove_env(self, env: Path) -> None: + """Remove a conda environment. + + Args: + env: The path to the environment to remove. + """ + self._run( + f"env remove --prefix {env} --yes", + stdout_level=logging.DEBUG, + stderr_level=logging.INFO, + ) + + def create_env( + self, + prefix: Path, + spec: str, + extra_channels: Iterable[str] = (), + ) -> None: + """Create an environment by installing a package. + + NOTE: `spec` may contain version specificaitons. + + Args: + prefix: The path to the environment to create. + spec: Package spec to install. e.g. "python=3.6", "python>=3.6", "python", etc. + extra_channels: Additional channels to search for packages in. + """ + channels = " ".join( + f"--channel {c}" for c in itertools.chain(extra_channels, self.channels) + ) + cmd = f"create --prefix {prefix} {channels} --quiet --yes {shlex.quote(spec)}" + if logger.getEffectiveLevel() <= logging.INFO: + with Halo( + text=f"Creating environment for {spec}", + spinner="dots", + stream=sys.stderr, + ): + self._run(cmd) + else: + self._run(cmd) + + def update_env( + self, + prefix: Path, + spec: str, + update_specs: bool, + extra_channels: Iterable[str] = (), + ) -> None: + """Update packages in an environment.""" + version_info = utils.version_info(spec) + # NOTE: `conda update` does not support version specification. + # It suggets to use `conda install` instead. + ## FIXME: this update process is inflexible + subcmd = "install" if self.exe.name == "conda" and version_info else "update" + + channels = " ".join( + f"--channel {c}" for c in itertools.chain(extra_channels, self.channels) + ) + cmd = f"{subcmd} --prefix {prefix} {channels} --quiet --yes {'--update-specs' if update_specs else ''} {shlex.quote(spec) if version_info else '--all'}" + if logger.getEffectiveLevel() <= logging.INFO: + with Halo( + text=f"Updating environment for {spec}", + spinner="dots", + stream=sys.stderr, + ): + self._run(cmd) + else: + self._run(cmd) + + def _run( + self, + command: str, + stdout_level: int = logging.DEBUG, + stderr_level: int = logging.ERROR, + ) -> subprocess.CompletedProcess: + """Run a conda command. + + Args: + command: The command to run excluding the conda executable. + """ + cmd = f"{self.exe} {command}" + logger.debug(f"Running: {cmd}") + cmd_list = shlex.split(cmd) + + p = subprocess.Popen( + cmd_list, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + + stdout_done, stderr_done = False, False + while not stdout_done or not stderr_done: + stdout_done = self._log_stream(p.stdout, stdout_level) + stderr_done = self._log_stream(p.stderr, stderr_level) + + p.wait() + + if p.returncode != 0: + raise CondaCommandError(cmd, p) + + return subprocess.CompletedProcess( + cmd_list, + p.returncode, + p.stdout.read() if p.stdout else None, + p.stderr.read() if p.stderr else None, + ) + + def _log_stream(self, stream: Optional[IO[str]], log_level: int) -> bool: + """Log one line of process ouput. + + Args: + stream: The stream to read from. + log_level: The log level to use. + + Returns: + True if the stream is depleted. False otherwise. + """ + if stream is None: + return True + line = stream.readline() + if line: + logger.log(log_level, f"\r{line.rstrip()}") + return not line diff --git a/condax/conda/env_info.py b/condax/conda/env_info.py new file mode 100644 index 0000000..bb3fb36 --- /dev/null +++ b/condax/conda/env_info.py @@ -0,0 +1,63 @@ +import json +from pathlib import Path +from typing import Union, Set + +from condax.utils import FullPath +from condax import utils +from .exceptions import NoPackageMetadataError + + +def is_env(path: Path) -> bool: + return (path / "conda-meta/history").is_file() + + +def find_exes(prefix: Path, package: str) -> Set[Path]: + """Find executables in environment `prefix` provided py a given `package`. + + Args: + prefix: The environment to search in. + package: The package whose executables to search for. + + Returns: + A list of executables in `prefix` provided by `package`. + + Raises: + NoPackageMetadataError: If the package files could not be determined. + """ + + def is_exe(p: Union[str, Path]) -> bool: + return FullPath(p).parent.name in ("bin", "sbin", "scripts", "Scripts") + + conda_meta_dir = prefix / "conda-meta" + for file_name in conda_meta_dir.glob(f"{package}*.json"): + with file_name.open() as fo: + package_info = json.load(fo) + if package_info["name"] == package: + potential_executables: Set[str] = { + fn + for fn in package_info["files"] + if (fn.startswith("bin/") and is_exe(fn)) + or (fn.startswith("sbin/") and is_exe(fn)) + # They are Windows style path + or (fn.lower().startswith("scripts") and is_exe(fn)) + or (fn.lower().startswith("library") and is_exe(fn)) + } + break + else: + raise NoPackageMetadataError(package) + + return { + prefix / fn for fn in potential_executables if utils.is_executable(prefix / fn) + } + + +def find_envs(directory: Path) -> Set[Path]: + """Find all environments in `directory`. + + Args: + directory: The directory to search in. + + Returns: + A list of environment prefixes in `directory`. + """ + return {prefix for prefix in directory.iterdir() if is_env(prefix)} diff --git a/condax/conda/exceptions.py b/condax/conda/exceptions.py new file mode 100644 index 0000000..289c497 --- /dev/null +++ b/condax/conda/exceptions.py @@ -0,0 +1,14 @@ +from subprocess import Popen +from condax.exceptions import CondaxError + + +class NoPackageMetadataError(CondaxError): + def __init__(self, package: str): + super().__init__(201, f"Could not determine package files: {package}.") + + +class CondaCommandError(CondaxError): + def __init__(self, command: str, p: Popen[str]): + super().__init__( + 202, f"Conda command `{command}` failed with exit code {p.returncode}." + ) diff --git a/condax/conda/installers.py b/condax/conda/installers.py new file mode 100644 index 0000000..f5aeca5 --- /dev/null +++ b/condax/conda/installers.py @@ -0,0 +1,77 @@ +import io +import shutil +import logging +import tarfile +import requests +import os +import stat +from functools import partial +from pathlib import Path +from typing import Callable, Iterable + +from condax.utils import FullPath +from condax import utils, consts + +logger = logging.getLogger(__name__) + + +DEFAULT_CONDA_BINS_DIR = consts.DEFAULT_PATHS.data_dir / "bins" + + +def _ensure(execs: Iterable[str], installer: Callable[[], Path]) -> Path: + path = os.pathsep.join((os.environ.get("PATH", ""), str(DEFAULT_CONDA_BINS_DIR))) + for exe in execs: + exe_path = shutil.which(exe, path=path) + if exe_path is not None: + return FullPath(exe_path) + + logger.info("No existing conda installation found. Installing the standalone") + return installer() + + +def ensure_conda(bin_dir: Path = DEFAULT_CONDA_BINS_DIR) -> Path: + return _ensure(("conda", "mamba"), partial(install_conda, bin_dir)) + + +def ensure_micromamba(bin_dir: Path = DEFAULT_CONDA_BINS_DIR) -> Path: + return _ensure(("micromamba",), partial(install_micromamba, bin_dir)) + + +def install_conda(bin_dir: Path = DEFAULT_CONDA_BINS_DIR) -> Path: + url = utils.get_conda_url() + resp = requests.get(url, allow_redirects=True) + resp.raise_for_status() + utils.mkdir(bin_dir) + exe_name = "conda.exe" if os.name == "nt" else "conda" + target_filename = bin_dir / exe_name + with open(target_filename, "wb") as fo: + fo.write(resp.content) + st = os.stat(target_filename) + os.chmod(target_filename, st.st_mode | stat.S_IXUSR) + return target_filename + + +def install_micromamba(bin_dir: Path = DEFAULT_CONDA_BINS_DIR) -> Path: + utils.mkdir(bin_dir) + exe_name = "micromamba.exe" if os.name == "nt" else "micromamba" + umamba_exe = bin_dir / exe_name + _download_extract_micromamba(umamba_exe) + return umamba_exe + + +def _download_extract_micromamba(umamba_dst: Path) -> None: + url = utils.get_micromamba_url() + print(f"Downloading micromamba from {url}") + response = requests.get(url, allow_redirects=True) + response.raise_for_status() + + utils.mkdir(umamba_dst.parent) + tarfile_obj = io.BytesIO(response.content) + with tarfile.open(fileobj=tarfile_obj) as tar, open(umamba_dst, "wb") as f: + p = "Library/bin/micromamba.exe" if os.name == "nt" else "bin/micromamba" + extracted = tar.extractfile(p) + if extracted: + shutil.copyfileobj(extracted, f) + + st = os.stat(umamba_dst) + os.chmod(umamba_dst, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) diff --git a/condax/condax/__init__.py b/condax/condax/__init__.py new file mode 100644 index 0000000..c46c37b --- /dev/null +++ b/condax/condax/__init__.py @@ -0,0 +1,3 @@ +from .condax import Condax + +__all__ = ["Condax"] diff --git a/condax/condax/condax.py b/condax/condax/condax.py new file mode 100644 index 0000000..c6302e3 --- /dev/null +++ b/condax/condax/condax.py @@ -0,0 +1,128 @@ +from pathlib import Path +from typing import Iterable, Set +import logging + +from condax import utils +from condax.conda import Conda, env_info, exceptions as conda_exceptions +from .exceptions import PackageInstalledError, NotAnEnvError, PackageNotInstalled + +from . import links +from .metadata import metadata + +logger = logging.getLogger(__name__) + + +class Condax: + def __init__(self, conda: Conda, bin_dir: Path, prefix_dir: Path) -> None: + """ + Args: + conda: A conda object to use for executing conda commands. + bin_dir: The directory to make executables available in. + prefix_dir: The directory where to create new conda environments. + """ + self.conda = conda + self.bin_dir = bin_dir + self.prefix_dir = prefix_dir + + def install_package( + self, + spec: str, + channels: Iterable[str], + is_forcing: bool = False, + ): + """Create a new conda environment with the package provided by `spec` and make all its executables available in `self.bin_dir`. + + Args: + spec: The package to install. Can have version constraints. + channels: Additional channels to search for packages in. + is_forcing: If True, install even if the package is already installed. + """ + package = utils.package_name(spec) + env = self.prefix_dir / package + + if env_info.is_env(env): + if is_forcing: + logger.warning(f"Overwriting environment for {package}") + self.conda.remove_env(env) + else: + raise PackageInstalledError(package, env) + elif env.exists() and (not env.is_dir() or tuple(env.iterdir())): + raise NotAnEnvError(env, "Cannot install to this location") + + self.conda.create_env(env, spec, channels) + executables = env_info.find_exes(env, package) + utils.mkdir(self.bin_dir) + links.create_links(env, executables, self.bin_dir, is_forcing=is_forcing) + metadata.create(env, package, executables) + logger.info(f"`{package}` has been installed by condax") + + def remove_package(self, package: str): + env = self.prefix_dir / package + if not env_info.is_env(env): + logger.warning(f"{package} is not installed with condax") + return + + apps_to_unlink = metadata.load(env).apps + links.remove_links(env, apps_to_unlink, self.bin_dir) + self.conda.remove_env(env) + logger.info(f"`{package}` has been removed from condax") + + def update_all_packages( + self, channels: Iterable[str] = (), is_forcing: bool = False + ): + for env in env_info.find_envs(self.prefix_dir): + self.update_package(env.name, channels, is_forcing=is_forcing) + + def update_package( + self, + spec: str, + channels: Iterable[str] = (), + update_specs: bool = False, + is_forcing: bool = False, + ) -> None: + pkg_name = utils.package_name(spec) + env = self.prefix_dir / pkg_name + meta = metadata.load(env) + + if not env_info.is_env(env): + raise PackageNotInstalled(pkg_name) + + try: + exes_before = self._find_all_exes(env, meta) + self.conda.update_env(env, spec, update_specs, channels) + exes_after = self._find_all_exes(env, meta) + + to_create = exes_after - exes_before + to_remove = exes_before - exes_after + + links.create_links(env, to_create, self.bin_dir, is_forcing) + links.remove_links(env, (exe.name for exe in to_remove), self.bin_dir) + + logger.info(f"{pkg_name} updated successfully") + + except conda_exceptions.CondaCommandError as e: + logger.error(str(e)) + logger.error(f"Failed to update `{env}`") + logger.warning(f"Recreating the environment...") + + self.remove_package(pkg_name) + self.install_package(spec, channels, is_forcing=is_forcing) + for pkg in meta.injected_packages: + self.inject_package(pkg.name, env, is_forcing=is_forcing) + + # Update metadata file + metadata.create(env) + for pkg in meta.injected_packages: + metadata.inject(env, (pkg.name,), pkg.include_apps) + + def _find_all_exes(self, env: Path, meta: metadata.CondaxMetaData) -> Set[Path]: + """Get exes of main and injected packages in env directory (not in self.bin_dir)""" + return { + utils.FullPath(exe) + for exe in env_info.find_exes(env, meta.main_package.name).union( + *( + env_info.find_exes(env, injected.name) + for injected in meta.injected_packages + ) + ) + } diff --git a/condax/condax/exceptions.py b/condax/condax/exceptions.py new file mode 100644 index 0000000..2aaeacc --- /dev/null +++ b/condax/condax/exceptions.py @@ -0,0 +1,23 @@ +from pathlib import Path +from condax.exceptions import CondaxError + + +class PackageInstalledError(CondaxError): + def __init__(self, package: str, location: Path): + super().__init__( + 101, + f"Package `{package}` is already installed at {location / package}. Use `--force` to overwrite.", + ) + + +class NotAnEnvError(CondaxError): + def __init__(self, location: Path, msg: str = ""): + super().__init__( + 102, + f"{location} exists, is not empty, and is not a conda environment. {msg}", + ) + + +class PackageNotInstalled(CondaxError): + def __init__(self, package: str): + super().__init__(103, f"Package `{package}` is not installed with condax") diff --git a/condax/condax/links.py b/condax/condax/links.py new file mode 100644 index 0000000..a34e44d --- /dev/null +++ b/condax/condax/links.py @@ -0,0 +1,126 @@ +import logging +import os +from pathlib import Path +import shutil +from typing import Iterable, List + +from condax.conda import installers +from condax import utils, wrapper + +logger = logging.getLogger(__name__) + + +def create_links( + env: Path, + executables_to_link: Iterable[Path], + location: Path, + is_forcing: bool = False, +): + """Create links to the executables in `executables_to_link` in `location`. + + Args: + env: The conda environment to link executables from. + executables_to_link: The executables to link. + location: The location to put the links in. + is_forcing: If True, overwrite existing links. + """ + linked = ( + exe.name + for exe in sorted(executables_to_link) + if create_link(env, exe, location, is_forcing=is_forcing) + ) + if executables_to_link: + logger.info("\n - ".join(("Created the following entrypoint links:", *linked))) + + +def create_link(env: Path, exe: Path, location: Path, is_forcing: bool = False) -> bool: + """Create a link to the executable in `exe` in `location`. + + Args: + env: The conda environment to link executables from. + exe: The executable to link. + location: The location to put the link in. + is_forcing: If True, overwrite existing links. + + Returns: + bool: True if a link was created, False otherwise. + """ + micromamba_exe = installers.ensure_micromamba() + if os.name == "nt": + script_lines = [ + "@rem Entrypoint created by condax\n", + f'@call "{micromamba_exe}" run --prefix "{env}" "{exe}" %*\n', + ] + else: + script_lines = [ + "#!/usr/bin/env bash\n", + "\n", + "# Entrypoint created by condax\n", + f'{utils.quote(micromamba_exe)} run --prefix {utils.quote(env)} {utils.quote(exe)} "$@"\n', + ] + if utils.to_bool(os.environ.get("CONDAX_HIDE_EXITCODE", False)): + # Let scripts to return exit code 0 constantly + script_lines.append("exit 0\n") + + script_path = location / _get_wrapper_name(exe.name) + if script_path.exists() and not is_forcing: + answer = input(f"{exe.name} already exists in {location}. Overwrite? (y/N) ") + if answer.strip().lower() not in ("y", "yes"): + logger.warning(f"Skipped creating entrypoint: {exe.name}") + return False + + if script_path.exists(): + logger.warning(f"Overwriting entrypoint: {exe.name}") + utils.unlink(script_path) + with open(script_path, "w") as fo: + fo.writelines(script_lines) + shutil.copystat(exe, script_path) + return True + + +def remove_links(env: Path, executables_to_unlink: Iterable[str], location: Path): + """Remove links in `location` which point to to executables in `env` whose names match those in `executables_to_unlink`. + + Args: + env: The conda environment which the links must point to to be removed. + location: The location the links are in. + executables_to_unlink: The names of the executables to unlink. + """ + unlinked: List[str] = [] + executables_to_unlink = tuple(executables_to_unlink) + for executable_name in executables_to_unlink: + link_path = location / _get_wrapper_name(executable_name) + if os.name == "nt": + # FIXME: this is hand-waving for now + utils.unlink(link_path) + else: + wrapper_env = wrapper.read_prefix(link_path) + + if wrapper_env is None: + utils.unlink(link_path) + unlinked.append(f"{executable_name} \t (failed to get env)") + continue + + if wrapper_env.samefile(env): + logger.warning( + f"Keeping {executable_name} as it runs in environment `{wrapper_env}`, not `{env}`." + ) + continue + + link_path.unlink() + + unlinked.append(executable_name) + + if executables_to_unlink: + logger.info( + "\n - ".join(("Removed the following entrypoint links:", *unlinked)) + ) + + +def _get_wrapper_name(name: str) -> str: + """Get the file name of the entrypoint script for the executable with the given name. + + On Windows, the file name is the executable name with a .bat extension. + On Unix, the file name is unchanged. + """ + return f"{Path(name).stem}.bat" if os.name == "nt" else name diff --git a/condax/condax/metadata/__init__.py b/condax/condax/metadata/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/condax/condax/metadata/exceptions.py b/condax/condax/metadata/exceptions.py new file mode 100644 index 0000000..809f7b6 --- /dev/null +++ b/condax/condax/metadata/exceptions.py @@ -0,0 +1,14 @@ +from pathlib import Path +from condax.exceptions import CondaxError + + +class BadMetadataError(CondaxError): + def __init__(self, metadata_path: Path, msg: str): + super().__init__( + 301, f"Error loading condax metadata at {metadata_path}: {msg}" + ) + + +class NoMetadataError(CondaxError): + def __init__(self, prefix: Path): + super().__init__(302, f"Failed to recreate condax_metadata.json in {prefix}") diff --git a/condax/condax/metadata/metadata.py b/condax/condax/metadata/metadata.py new file mode 100644 index 0000000..cd577b4 --- /dev/null +++ b/condax/condax/metadata/metadata.py @@ -0,0 +1,156 @@ +import json +from pathlib import Path +from typing import Any, Dict, Iterable, Optional, Set +import logging + +from condax.conda import env_info + +from .package import MainPackage, InjectedPackage +from .exceptions import BadMetadataError, NoMetadataError +from .serializable import Serializable + +logger = logging.getLogger(__name__) + + +class CondaxMetaData(Serializable): + """ + Handle metadata information written in `condax_metadata.json` + placed in each environment. + """ + + metadata_file = "condax_metadata.json" + + def __init__( + self, + main_package: MainPackage, + injected_packages: Iterable[InjectedPackage] = (), + ): + self._main_package = main_package + self._injected_packages = {pkg.name: pkg for pkg in injected_packages} + + def inject(self, package: InjectedPackage): + self._injected_packages[package.name] = package + + def uninject(self, name: str): + self._injected_packages.pop(name, None) + + @property + def apps(self) -> Set[str]: + """All the executable apps in the condax environment, including injected ones.""" + return self._main_package._apps.union( + *(pkg._apps for pkg in self._injected_packages.values()) + ) + + @property + def main_package(self) -> MainPackage: + return self._main_package + + @property + def injected_packages(self) -> Set[InjectedPackage]: + return set(self._injected_packages.values()) + + def serialize(self) -> Dict[str, Any]: + return { + "main_package": self._main_package.serialize(), + "injected_packages": [ + pkg.serialize() for pkg in self._injected_packages.values() + ], + } + + @classmethod + def deserialize(cls, serialized: Dict[str, Any]): + assert isinstance(serialized, dict) + assert isinstance(serialized["main_package"], dict) + assert isinstance(serialized["injected_packages"], list) + serialized.update( + main_package=MainPackage.deserialize(serialized["main_package"]), + injected_packages=[ + InjectedPackage.deserialize(pkg) + for pkg in serialized["injected_packages"] + ], + ) + return cls(**serialized) + + def save(self) -> None: + metadata_path = self._main_package.prefix / self.metadata_file + with metadata_path.open("w") as f: + json.dump(self.serialize(), f, indent=4) + + +def create( + prefix: Path, + package: Optional[str] = None, + executables: Optional[Iterable[Path]] = None, +): + """ + Create the metadata file. + + Args: + prefix: The conda environment to create the metadata file for. + package: The package to add to the metadata. By default it is the name of the environment's directory. + executables: The executables to add to the metadata. If not provided, they are searched for in conda's metadata. + """ + package = package or prefix.name + apps = (p.name for p in (executables or env_info.find_exes(prefix, package))) + main = MainPackage(package, prefix, apps) + meta = CondaxMetaData(main) + meta.save() + + +def inject(prefix: Path, packages_to_inject: Iterable[str], include_apps: bool = False): + """ + Inject the given packages into the condax_metadata.json file for the environment at `prefix`. + + Args: + prefix: The path to the environment. + packages_to_inject: The names of the packages to inject. + include_apps: Whether to make links to the executables of the injected packages. + """ + meta = load(prefix) + for pkg in packages_to_inject: + apps = (p.name for p in env_info.find_exes(prefix, pkg)) + pkg_to_inject = InjectedPackage(pkg, apps, include_apps=include_apps) + meta.inject(pkg_to_inject) + meta.save() + + +def load(prefix: Path) -> CondaxMetaData: + """Load the metadata object for the given environment. + + If the metadata doesn't exist, it is created. + + Args: + prefix (Path): The path to the environment. + + Returns: + CondaxMetaData: The metadata object for the environment. + """ + meta = _load(prefix) + # For backward compatibility: metadata can be absent + if meta is None: + logger.info(f"Recreating condax_metadata.json in {prefix}...") + create(prefix) + meta = _load(prefix) + if meta is None: + raise NoMetadataError(prefix) + return meta + + +def _load(prefix: Path) -> Optional[CondaxMetaData]: + """Does the heavy lifting for loading the metadata. + + `load` is the exposed wrapper that tries to create it if it doesn't exist. + """ + p = prefix / CondaxMetaData.metadata_file + if not p.exists(): + return None + + with open(p) as f: + d = json.load(f) + + try: + return CondaxMetaData.deserialize(d) + except AssertionError as e: + raise BadMetadataError(p, f"A value is of the wrong type. {e}") from e + except KeyError as e: + raise BadMetadataError(p, f"Key {e} is missing.") from e diff --git a/condax/condax/metadata/package.py b/condax/condax/metadata/package.py new file mode 100644 index 0000000..48fdce5 --- /dev/null +++ b/condax/condax/metadata/package.py @@ -0,0 +1,61 @@ +from pathlib import Path +from typing import Any, Dict, Iterable, Set + +from condax.utils import FullPath +from .serializable import Serializable + + +class _PackageBase(Serializable): + def __init__(self, name: str, apps: Iterable[str], include_apps: bool): + self.name = name + self._apps = set(apps) + self.include_apps = include_apps + + @property + def apps(self) -> Set[str]: + """The executable apps provided by the package.""" + return self._apps + + def __lt__(self, other): + return self.name < other.name + + def serialize(self) -> Dict[str, Any]: + return { + "name": self.name, + "apps": list(self._apps), + "include_apps": self.include_apps, + } + + @classmethod + def deserialize(cls, serialized: Dict[str, Any]): + assert isinstance(serialized, dict) + assert isinstance(serialized["name"], str) + assert isinstance(serialized["apps"], list) + assert all(isinstance(app, str) for app in serialized["apps"]) + assert isinstance(serialized["include_apps"], bool) + serialized.update(apps=set(serialized["apps"])) + return cls(**serialized) + + +class MainPackage(_PackageBase): + def __init__( + self, name: str, prefix: Path, apps: Iterable[str], include_apps: bool = True + ): + super().__init__(name, apps, include_apps) + self.prefix = prefix + + def serialize(self) -> Dict[str, Any]: + return { + **super().serialize(), + "prefix": str(self.prefix), + } + + @classmethod + def deserialize(cls, serialized: Dict[str, Any]): + assert isinstance(serialized["prefix"], str) + serialized.update(prefix=FullPath(serialized["prefix"])) + return super().deserialize(serialized) + + +class InjectedPackage(_PackageBase): + pass diff --git a/condax/condax/metadata/serializable.py b/condax/condax/metadata/serializable.py new file mode 100644 index 0000000..6b1e517 --- /dev/null +++ b/condax/condax/metadata/serializable.py @@ -0,0 +1,16 @@ +from abc import ABC, abstractmethod +from typing import Any, Dict, Type, TypeVar + + +S = TypeVar("S", bound="Serializable") + + +class Serializable(ABC): + @classmethod + @abstractmethod + def deserialize(cls: Type[S], serialized: Dict[str, Any]) -> S: + raise NotImplementedError() + + @abstractmethod + def serialize(self) -> Dict[str, Any]: + raise NotImplementedError() diff --git a/condax/config.py b/condax/config.py index 51cd125..83422d8 100644 --- a/condax/config.py +++ b/condax/config.py @@ -4,7 +4,7 @@ from typing import Any, Dict, List, Optional, Union from condax.exceptions import CondaxError -from condax.utils import to_path +from condax.utils import FullPath import condax.condarc as condarc import yaml @@ -17,7 +17,7 @@ _localappdata_dir, "condax", "condax", _config_filename ) _default_config = _default_config_windows if os.name == "nt" else _default_config_unix -DEFAULT_CONFIG = to_path(os.environ.get("CONDAX_CONFIG", _default_config)) +DEFAULT_CONFIG = FullPath(os.environ.get("CONDAX_CONFIG", _default_config)) _xdg_data_home = os.environ.get("XDG_DATA_HOME", "~/.local/share") _default_prefix_dir_unix = os.path.join(_xdg_data_home, "condax", "envs") @@ -25,22 +25,22 @@ _default_prefix_dir = ( _default_prefix_dir_win if os.name == "nt" else _default_prefix_dir_unix ) -DEFAULT_PREFIX_DIR = to_path(os.environ.get("CONDAX_PREFIX_DIR", _default_prefix_dir)) +DEFAULT_PREFIX_DIR = FullPath(os.environ.get("CONDAX_PREFIX_DIR", _default_prefix_dir)) -DEFAULT_BIN_DIR = to_path(os.environ.get("CONDAX_BIN_DIR", "~/.local/bin")) +DEFAULT_BIN_DIR = FullPath(os.environ.get("CONDAX_BIN_DIR", "~/.local/bin")) _channels_in_condarc = condarc.load_channels() DEFAULT_CHANNELS = ( os.environ.get("CONDAX_CHANNELS", " ".join(_channels_in_condarc)).strip().split() ) -CONDA_ENVIRONMENT_FILE = to_path("~/.conda/environments.txt") +CONDA_ENVIRONMENT_FILE = FullPath("~/.conda/environments.txt") conda_path = shutil.which("conda") MAMBA_ROOT_PREFIX = ( - to_path(conda_path).parent.parent + FullPath(conda_path).parent.parent if conda_path is not None - else to_path(os.environ.get("MAMBA_ROOT_PREFIX", "~/micromamba")) + else FullPath(os.environ.get("MAMBA_ROOT_PREFIX", "~/micromamba")) ) @@ -94,7 +94,7 @@ def set_via_file(config_file: Union[str, Path]): Raises: BadConfigFileError: If the config file is not valid. """ - config_file = to_path(config_file) + config_file = FullPath(config_file) try: with config_file.open() as f: config = yaml.safe_load(f) @@ -107,20 +107,20 @@ def set_via_file(config_file: Union[str, Path]): # For compatibility with condax 0.0.5 if "prefix_path" in config: - prefix_dir = to_path(config["prefix_path"]) + prefix_dir = FullPath(config["prefix_path"]) C._set("prefix_dir", prefix_dir) # For compatibility with condax 0.0.5 if "target_destination" in config: - bin_dir = to_path(config["target_destination"]) + bin_dir = FullPath(config["target_destination"]) C._set("bin_dir", bin_dir) if "prefix_dir" in config: - prefix_dir = to_path(config["prefix_dir"]) + prefix_dir = FullPath(config["prefix_dir"]) C._set("prefix_dir", prefix_dir) if "bin_dir" in config: - bin_dir = to_path(config["bin_dir"]) + bin_dir = FullPath(config["bin_dir"]) C._set("bin_dir", bin_dir) if "channels" in config: @@ -137,10 +137,10 @@ def set_via_value( Set a part of values in the object C by passing values directly. """ if prefix_dir: - C._set("prefix_dir", to_path(prefix_dir)) + C._set("prefix_dir", FullPath(prefix_dir)) if bin_dir: - C._set("bin_dir", to_path(bin_dir)) + C._set("bin_dir", FullPath(bin_dir)) if channels: C._set("channels", channels + C.channels()) diff --git a/condax/consts.py b/condax/consts.py new file mode 100644 index 0000000..91d474d --- /dev/null +++ b/condax/consts.py @@ -0,0 +1,54 @@ +import os +from dataclasses import dataclass +from pathlib import Path + + +from condax.utils import FullPath + + +IS_WIN = os.name == "nt" +IS_UNIX = not IS_WIN + + +@dataclass +class Paths: + conf_dir: Path + bin_dir: Path + data_dir: Path + conf_file_name: str = "config.yaml" + envs_dir_name: str = "envs" + + @property + def conf_file(self) -> Path: + return self.conf_dir / self.conf_file_name + + @property + def prefix_dir(self) -> Path: + return self.data_dir / self.envs_dir_name + + +class _WindowsPaths(Paths): + def __init__(self): + conf_dir = data_dir = ( + FullPath(os.environ.get("LOCALAPPDATA", "~/AppData/Local")) + / "condax/condax" + ) + super().__init__( + conf_dir=conf_dir, + bin_dir=conf_dir / "bin", + data_dir=data_dir, + ) + + +class _UnixPaths(Paths): + def __init__(self): + super().__init__( + conf_dir=FullPath(os.environ.get("XDG_CONFIG_HOME", "~/.config")) + / "condax", + bin_dir=FullPath("~/.local/bin"), + data_dir=FullPath(os.environ.get("XDG_DATA_HOME", "~/.local/share")) + / "condax", + ) + + +DEFAULT_PATHS: Paths = _UnixPaths() if IS_UNIX else _WindowsPaths() diff --git a/condax/core.py b/condax/core.py index a110e04..e052caf 100644 --- a/condax/core.py +++ b/condax/core.py @@ -10,123 +10,17 @@ import condax.conda as conda from condax.exceptions import CondaxError -import condax.metadata as metadata +import condax.condax.metadata as metadata import condax.wrapper as wrapper import condax.utils as utils import condax.config as config from condax.config import C +from condax.conda import env_info logger = logging.getLogger(__name__) -def create_link(package: str, exe: Path, is_forcing: bool = False) -> bool: - micromamba_exe = conda.ensure_micromamba() - executable_name = exe.name - # FIXME: Enforcing conda (not mamba) for `conda run` for now - prefix = conda.conda_env_prefix(package) - if os.name == "nt": - script_lines = [ - "@rem Entrypoint created by condax\n", - f"@call {utils.quote(micromamba_exe)} run --prefix {utils.quote(prefix)} {utils.quote(exe)} %*\n", - ] - else: - script_lines = [ - "#!/usr/bin/env bash\n", - "\n", - "# Entrypoint created by condax\n", - f'{utils.quote(micromamba_exe)} run --prefix {utils.quote(prefix)} {utils.quote(exe)} "$@"\n', - ] - if utils.to_bool(os.environ.get("CONDAX_HIDE_EXITCODE", False)): - # Let scripts to return exit code 0 constantly - script_lines.append("exit 0\n") - - script_path = _get_wrapper_path(executable_name) - if script_path.exists() and not is_forcing: - user_input = input(f"{executable_name} already exists. Overwrite? (y/N) ") - if user_input.strip().lower() not in ("y", "yes"): - logger.warning(f"Skipped creating entrypoint: {executable_name}") - return False - - if script_path.exists(): - logger.warning(f"Overwriting entrypoint: {executable_name}") - utils.unlink(script_path) - with open(script_path, "w") as fo: - fo.writelines(script_lines) - shutil.copystat(exe, script_path) - return True - - -def create_links( - package: str, executables_to_link: Iterable[Path], is_forcing: bool = False -): - linked = ( - exe.name - for exe in sorted(executables_to_link) - if create_link(package, exe, is_forcing=is_forcing) - ) - if executables_to_link: - logger.info("\n - ".join(("Created the following entrypoint links:", *linked))) - - -def remove_links(package: str, app_names_to_unlink: Iterable[str]): - unlinked: List[str] = [] - if os.name == "nt": - # FIXME: this is hand-waving for now - for executable_name in app_names_to_unlink: - link_path = _get_wrapper_path(executable_name) - utils.unlink(link_path) - else: - for executable_name in app_names_to_unlink: - link_path = _get_wrapper_path(executable_name) - wrapper_env = wrapper.read_env_name(link_path) - if wrapper_env is None: - utils.unlink(link_path) - unlinked.append(f"{executable_name} \t (failed to get env)") - elif wrapper_env == package: - link_path.unlink() - unlinked.append(executable_name) - else: - logger.warning( - f"Keeping {executable_name} as it runs in environment `{wrapper_env}`, not `{package}`." - ) - - if app_names_to_unlink: - logger.info( - "\n - ".join(("Removed the following entrypoint links:", *unlinked)) - ) - - -class PackageInstalledError(CondaxError): - def __init__(self, package: str): - super().__init__( - 20, - f"Package `{package}` is already installed. Use `--force` to force install.", - ) - - -def install_package( - spec: str, - is_forcing: bool = False, - conda_stdout: bool = False, -): - package, _ = utils.split_match_specs(spec) - - if conda.has_conda_env(package): - if is_forcing: - logger.warning(f"Overwriting environment for {package}") - conda.remove_conda_env(package, conda_stdout) - else: - raise PackageInstalledError(package) - - conda.create_conda_environment(spec, conda_stdout) - executables_to_link = conda.determine_executables_from_env(package) - utils.mkdir(C.bin_dir()) - create_links(package, executables_to_link, is_forcing=is_forcing) - _create_metadata(package) - logger.info(f"`{package}` has been installed by condax") - - def inject_package_to( env_name: str, injected_specs: List[str], @@ -190,33 +84,6 @@ def uninject_package_from( logger.info(f"`{pkgs_str}` has been uninjected from `{env_name}`") -class PackageNotInstalled(CondaxError): - def __init__(self, package: str, error: bool = True): - super().__init__( - 21 if error else 0, - f"Package `{package}` is not installed with condax", - ) - - -def exit_if_not_installed(package: str, error: bool = True): - prefix = conda.conda_env_prefix(package) - if not prefix.exists(): - raise PackageNotInstalled(package, error) - - -def remove_package(package: str, conda_stdout: bool = False): - exit_if_not_installed(package, error=False) - apps_to_unlink = _get_apps(package) - remove_links(package, apps_to_unlink) - conda.remove_conda_env(package, conda_stdout) - logger.info(f"`{package}` has been removed from condax") - - -def update_all_packages(update_specs: bool = False, is_forcing: bool = False): - for package in _get_all_envs(): - update_package(package, update_specs=update_specs, is_forcing=is_forcing) - - def list_all_packages(short=False, include_injected=False) -> None: if short: _list_all_packages_short(include_injected) @@ -308,112 +175,6 @@ def _print_condax_dirs() -> None: ) -def update_package( - spec: str, - update_specs: bool = False, - is_forcing: bool = False, - conda_stdout: bool = False, -) -> None: - - env, _ = utils.split_match_specs(spec) - exit_if_not_installed(env) - try: - main_apps_before_update = set(conda.determine_executables_from_env(env)) - injected_apps_before_update = { - injected: set(conda.determine_executables_from_env(env, injected)) - for injected in _get_injected_packages(env) - } - conda.update_conda_env(spec, update_specs, conda_stdout) - main_apps_after_update = set(conda.determine_executables_from_env(env)) - injected_apps_after_update = { - injected: set(conda.determine_executables_from_env(env, injected)) - for injected in _get_injected_packages(env) - } - - if ( - main_apps_before_update == main_apps_after_update - and injected_apps_before_update == injected_apps_after_update - ): - logger.info(f"No updates found: {env}") - - to_create = main_apps_after_update - main_apps_before_update - to_delete = main_apps_before_update - main_apps_after_update - to_delete_apps = [path.name for path in to_delete] - - # Update links of main apps - create_links(env, to_create, is_forcing) - remove_links(env, to_delete_apps) - - # Update links of injected apps - for pkg in _get_injected_packages(env): - to_delete = ( - injected_apps_before_update[pkg] - injected_apps_after_update[pkg] - ) - to_delete_apps = [p.name for p in to_delete] - remove_links(env, to_delete_apps) - - to_create = ( - injected_apps_after_update[pkg] - injected_apps_before_update[pkg] - ) - create_links(env, to_create, is_forcing) - - logger.info(f"{env} update successfully") - - except subprocess.CalledProcessError: - logger.error(f"Failed to update `{env}`") - logger.warning(f"Recreating the environment...") - - remove_package(env, conda_stdout) - install_package(env, is_forcing=is_forcing, conda_stdout=conda_stdout) - - # Update metadata file - _create_metadata(env) - for pkg in _get_injected_packages(env): - _inject_to_metadata(env, pkg) - - -def _create_metadata(package: str): - """ - Create metadata file - """ - apps = [p.name for p in conda.determine_executables_from_env(package)] - main = metadata.MainPackage(package, apps) - meta = metadata.CondaxMetaData(main) - meta.save() - - -class NoMetadataError(CondaxError): - def __init__(self, env: str): - super().__init__(22, f"Failed to recreate condax_metadata.json in {env}") - - -def _load_metadata(env: str) -> metadata.CondaxMetaData: - meta = metadata.load(env) - # For backward compatibility: metadata can be absent - if meta is None: - logger.info(f"Recreating condax_metadata.json in {env}...") - _create_metadata(env) - meta = metadata.load(env) - if meta is None: - raise NoMetadataError(env) - return meta - - -def _inject_to_metadata( - env: str, packages_to_inject: Iterable[str], include_apps: bool = False -): - """ - Inject the package into the condax_metadata.json file for the env. - """ - meta = _load_metadata(env) - for pkg in packages_to_inject: - apps = [p.name for p in conda.determine_executables_from_env(env, pkg)] - pkg_to_inject = metadata.InjectedPackage(pkg, apps, include_apps=include_apps) - meta.uninject(pkg) # overwrites if necessary - meta.inject(pkg_to_inject) - meta.save() - - def _uninject_from_metadata(env: str, packages_to_uninject: Iterable[str]): """ Uninject the package from the condax_metadata.json file for the env. @@ -430,9 +191,7 @@ def _get_all_envs() -> List[str]: """ utils.mkdir(C.prefix_dir()) return sorted( - pkg_dir.name - for pkg_dir in C.prefix_dir().iterdir() - if utils.is_env_dir(pkg_dir) + pkg_dir.name for pkg_dir in C.prefix_dir().iterdir() if env_info.is_env(pkg_dir) ) @@ -486,13 +245,6 @@ def _get_apps(env_name: str) -> List[str]: ] -def _get_wrapper_path(cmd_name: str) -> Path: - p = C.bin_dir() / cmd_name - if os.name == "nt": - p = p.parent / (p.stem + ".bat") - return p - - def export_all_environments(out_dir: str, conda_stdout: bool = False) -> None: """Export all environments to a directory. @@ -599,7 +351,7 @@ def _prune_links(): if not wrapper.is_wrapper(link): continue - target_env = wrapper.read_env_name(link) + target_env = wrapper.read_prefix(link) if target_env is None: logging.info(f"Failed to read env name from {link}") continue diff --git a/condax/metadata.py b/condax/metadata.py deleted file mode 100644 index 002b0a8..0000000 --- a/condax/metadata.py +++ /dev/null @@ -1,78 +0,0 @@ -import json -from pathlib import Path -from typing import List, Optional - -from condax.config import C - - -class _PackageBase(object): - def __init__(self, name: str, apps: List[str], include_apps: bool): - self.name = name - self.apps = apps - self.include_apps = include_apps - - -class MainPackage(_PackageBase): - def __init__(self, name: str, apps: List[str], include_apps: bool = True): - self.name = name - self.apps = apps - self.include_apps = True - - -class InjectedPackage(_PackageBase): - pass - - -class CondaxMetaData(object): - """ - Handle metadata information written in `condax_metadata.json` - placed in each environment. - """ - - metadata_file = "condax_metadata.json" - - @classmethod - def get_path(cls, package: str) -> Path: - p = C.prefix_dir() / package / cls.metadata_file - return p - - def __init__(self, main: MainPackage, injected: List[InjectedPackage] = []): - self.main_package = main - self.injected_packages = injected - - def inject(self, package: InjectedPackage): - if self.injected_packages is None: - self.injected_packages = [] - already_injected = [p.name for p in self.injected_packages] - if package.name in already_injected: - return - self.injected_packages.append(package) - - def uninject(self, name: str): - self.injected_packages = [p for p in self.injected_packages if p.name != name] - - def to_json(self) -> str: - return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4) - - def save(self) -> None: - p = CondaxMetaData.get_path(self.main_package.name) - with open(p, "w") as fo: - fo.write(self.to_json()) - - -def load(package: str) -> Optional[CondaxMetaData]: - p = CondaxMetaData.get_path(package) - if not p.exists(): - return None - - with open(p) as f: - d = json.load(f) - if not d: - raise ValueError(f"Failed to read the metadata from {p}") - return _from_dict(d) - - -def _from_dict(d: dict) -> CondaxMetaData: - main = MainPackage(**d["main_package"]) - injected = [InjectedPackage(**p) for p in d["injected_packages"]] - return CondaxMetaData(main, injected) diff --git a/condax/paths.py b/condax/paths.py index 47f15a1..3ee1051 100644 --- a/condax/paths.py +++ b/condax/paths.py @@ -1,5 +1,4 @@ import logging -import sys from pathlib import Path from typing import Union diff --git a/condax/utils.py b/condax/utils.py index de285de..2613f42 100644 --- a/condax/utils.py +++ b/condax/utils.py @@ -1,14 +1,16 @@ +import logging import os from pathlib import Path import platform -from typing import List, Tuple, Union +import shlex +from typing import Iterable, TextIO, Tuple, Union import re import urllib.parse from condax.exceptions import CondaxError -pat = re.compile(r"<=|>=|==|!=|<|>|=") +pat = re.compile(r"(?=<=|>=|==|!=|<|>|=|$)") def split_match_specs(package_with_specs: str) -> Tuple[str, str]: @@ -36,26 +38,36 @@ def split_match_specs(package_with_specs: str) -> Tuple[str, str]: >>> split_match_specs("numpy") ("numpy", "") """ - name, *_ = pat.split(package_with_specs) - # replace with str.removeprefix() once Python>=3.9 is assured - match_specs = package_with_specs[len(name) :] + name, match_specs = pat.split(package_with_specs, 1) return name.strip(), match_specs.strip() -def to_path(path: Union[str, Path]) -> Path: +def package_name(package_with_specs: str) -> str: """ - Convert a string to a pathlib.Path object. + Get the package name from a match specification. """ - return Path(path).expanduser().resolve() + return split_match_specs(package_with_specs)[0] -def mkdir(path: Union[Path, str]) -> None: +def version_info(package_with_specs: str) -> str: + """ + Get the version info from a match specification. + """ + return split_match_specs(package_with_specs)[1] + + +class FullPath(Path): + def __new__(cls, *args, **kwargs): + return super().__new__(Path, Path(*args, **kwargs).expanduser().resolve()) + + +def mkdir(path: Path) -> None: """mkdir -p path""" - to_path(path).mkdir(exist_ok=True, parents=True) + path.mkdir(exist_ok=True, parents=True) def quote(path: Union[Path, str]) -> str: - return f'"{str(path)}"' + return shlex.quote(str(path)) def is_executable(path: Path) -> bool: @@ -180,9 +192,3 @@ def to_bool(value: Union[str, bool]) -> bool: pass return False - - -def is_env_dir(path: Union[Path, str]) -> bool: - """Check if a path is a conda environment directory.""" - p = to_path(path) - return (p / "conda-meta" / "history").exists() diff --git a/condax/wrapper.py b/condax/wrapper.py index 7677990..511b466 100644 --- a/condax/wrapper.py +++ b/condax/wrapper.py @@ -6,16 +6,16 @@ from pathlib import Path from typing import Optional, List, Union -from condax.utils import to_path +from condax.utils import FullPath -def read_env_name(script_path: Union[str, Path]) -> Optional[str]: +def read_prefix(script_path: Union[str, Path]) -> Optional[Path]: """ Read a condax bash script. Returns the environment name within which conda run is executed. """ - path = to_path(script_path) + path = FullPath(script_path) script_name = path.name if not path.exists(): logging.warning(f"File missing: `{path}`.") @@ -45,14 +45,14 @@ def read_env_name(script_path: Union[str, Path]) -> Optional[str]: logging.warning(msg) return None - return namespace.prefix.name + return namespace.prefix def is_wrapper(exec_path: Union[str, Path]) -> bool: """ Check if a file is a condax wrapper script. """ - path = to_path(exec_path) + path = FullPath(exec_path) if not path.exists(): return False @@ -106,7 +106,7 @@ def _parse_line(cls, line: str) -> Optional[argparse.Namespace]: return None first_word = words[0] - cmd = to_path(first_word).stem + cmd = FullPath(first_word).stem if cmd not in ("conda", "mamba", "micromamba"): return None diff --git a/docs/config.md b/docs/config.md index 66d69d1..05e5fe9 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1,8 +1,8 @@ Condax generally requires very little configuration. -Condax will read configuration settings from a `~/.config/condax/config.yaml` file. +Condax will read configuration settings from a `~/.config/condax/config.yaml` file. This path can be overridden by the `--config` command line argument. -This is the default state for this file. +This is the expected format for the configuration file. All settings are optional. ```yaml prefix_dir: "~/.local/share/condax/envs" diff --git a/poetry.lock b/poetry.lock index a4f4ea8..53d4327 100644 --- a/poetry.lock +++ b/poetry.lock @@ -85,6 +85,20 @@ python-versions = ">=3.7" colorama = {version = "*", markers = "platform_system == \"Windows\""} importlib-metadata = {version = "*", markers = "python_version < \"3.8\""} +[[package]] +name = "click-aliases" +version = "1.0.1" +description = "Enable aliases for Click" +category = "main" +optional = false +python-versions = "*" + +[package.dependencies] +click = "*" + +[package.extras] +dev = ["wheel", "coveralls", "pytest-cov", "pytest", "tox-travis", "flake8-import-order", "flake8"] + [[package]] name = "colorama" version = "0.4.5" @@ -148,6 +162,24 @@ python-dateutil = ">=2.6.0" requests = ">=2.18" uritemplate = ">=3.0.0" +[[package]] +name = "halo" +version = "0.0.31" +description = "Beautiful terminal spinners in Python" +category = "main" +optional = false +python-versions = ">=3.4" + +[package.dependencies] +colorama = ">=0.3.9" +log-symbols = ">=0.0.14" +six = ">=1.12.0" +spinners = ">=0.0.24" +termcolor = ">=1.1.0" + +[package.extras] +ipython = ["ipywidgets (==7.1.0)", "IPython (==5.7.0)"] + [[package]] name = "idna" version = "3.3" @@ -189,6 +221,17 @@ category = "dev" optional = false python-versions = "*" +[[package]] +name = "log-symbols" +version = "0.0.14" +description = "Colored symbols for various log levels for Python" +category = "main" +optional = false +python-versions = "*" + +[package.dependencies] +colorama = ">=0.3.9" + [[package]] name = "mypy" version = "0.971" @@ -418,10 +461,26 @@ python-versions = ">=3.5" name = "six" version = "1.16.0" description = "Python 2 and 3 compatibility utilities" -category = "dev" +category = "main" optional = false python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*" +[[package]] +name = "spinners" +version = "0.0.24" +description = "Spinners for terminals" +category = "main" +optional = false +python-versions = "*" + +[[package]] +name = "termcolor" +version = "1.1.0" +description = "ANSII Color formatting for output in terminal." +category = "main" +optional = false +python-versions = "*" + [[package]] name = "tomli" version = "2.0.1" @@ -537,7 +596,7 @@ testing = ["pytest (>=6)", "pytest-checkdocs (>=2.4)", "pytest-flake8", "pytest- [metadata] lock-version = "1.1" python-versions = "^3.7" -content-hash = "7c447397b203af9883341e58d8d23c76ffd15dcdc122db06edb885cdd264d2dd" +content-hash = "b30c76dcea8398059c52429c6e7b3cf5579fb6cfef5ca1d6e0e008a8562e3a10" [metadata.files] atomicwrites = [ @@ -650,6 +709,10 @@ click = [ {file = "click-8.1.3-py3-none-any.whl", hash = "sha256:bb4d8133cb15a609f44e8213d9b391b0809795062913b383c62be0ee95b1db48"}, {file = "click-8.1.3.tar.gz", hash = "sha256:7682dc8afb30297001674575ea00d1814d808d6a36af415a82bd481d37ba7b8e"}, ] +click-aliases = [ + {file = "click-aliases-1.0.1.tar.gz", hash = "sha256:f48012077e0788eb02f4f8ee458fef3601873fec6c998e9ea8b4554394e705a3"}, + {file = "click_aliases-1.0.1-py2.py3-none-any.whl", hash = "sha256:229ecab12a97d1d5ce3f1fd7ce16da0e4333a24ebe3b34d8b7a6d0a1d2cfab90"}, +] colorama = [ {file = "colorama-0.4.5-py2.py3-none-any.whl", hash = "sha256:854bf444933e37f5824ae7bfc1e98d5bce2ebe4160d46b5edf346a89358e99da"}, {file = "colorama-0.4.5.tar.gz", hash = "sha256:e6c6b4334fc50988a639d9b98aa429a0b57da6e17b9a44f0451f930b6967b7a4"}, @@ -729,6 +792,10 @@ cryptography = [ {file = "github3.py-3.2.0-py2.py3-none-any.whl", hash = "sha256:a9016e40609c6f5cb9954dd188d08257dafd09c4da8c0e830a033fca00054b0d"}, {file = "github3.py-3.2.0.tar.gz", hash = "sha256:09b72be1497d346b0968cde8360a0d6af79dc206d0149a63cd3ec86c65c377cc"}, ] +halo = [ + {file = "halo-0.0.31-py2-none-any.whl", hash = "sha256:5350488fb7d2aa7c31a1344120cee67a872901ce8858f60da7946cef96c208ab"}, + {file = "halo-0.0.31.tar.gz", hash = "sha256:7b67a3521ee91d53b7152d4ee3452811e1d2a6321975137762eb3d70063cc9d6"}, +] idna = [ {file = "idna-3.3-py3-none-any.whl", hash = "sha256:84d9dd047ffa80596e0f246e2eab0b391788b0503584e8945f2368256d2735ff"}, {file = "idna-3.3.tar.gz", hash = "sha256:9d643ff0a55b762d5cdb124b8eaa99c66322e2157b69160bc32796e824360e6d"}, @@ -744,6 +811,10 @@ iniconfig = [ lazyasd = [ {file = "lazyasd-0.1.4.tar.gz", hash = "sha256:a3196f05cff27f952ad05767e5735fd564b4ea4e89b23f5ea1887229c3db145b"}, ] +log-symbols = [ + {file = "log_symbols-0.0.14-py3-none-any.whl", hash = "sha256:4952106ff8b605ab7d5081dd2c7e6ca7374584eff7086f499c06edd1ce56dcca"}, + {file = "log_symbols-0.0.14.tar.gz", hash = "sha256:cf0bbc6fe1a8e53f0d174a716bc625c4f87043cc21eb55dd8a740cfe22680556"}, +] mypy = [ {file = "mypy-0.971-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:f2899a3cbd394da157194f913a931edfd4be5f274a88041c9dc2d9cdcb1c315c"}, {file = "mypy-0.971-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:98e02d56ebe93981c41211c05adb630d1d26c14195d04d95e49cd97dbc046dc5"}, @@ -894,6 +965,13 @@ six = [ {file = "six-1.16.0-py2.py3-none-any.whl", hash = "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"}, {file = "six-1.16.0.tar.gz", hash = "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926"}, ] +spinners = [ + {file = "spinners-0.0.24-py3-none-any.whl", hash = "sha256:2fa30d0b72c9650ad12bbe031c9943b8d441e41b4f5602b0ec977a19f3290e98"}, + {file = "spinners-0.0.24.tar.gz", hash = "sha256:1eb6aeb4781d72ab42ed8a01dcf20f3002bf50740d7154d12fb8c9769bf9e27f"}, +] +termcolor = [ + {file = "termcolor-1.1.0.tar.gz", hash = "sha256:1d6d69ce66211143803fbc56652b41d73b4a400a2891d7bf7a1cdf4c02de613b"}, +] tomli = [ {file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"}, {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"}, diff --git a/pyproject.toml b/pyproject.toml index aad14da..f3efb3b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,8 @@ userpath = "^1.8.0" PyYAML = "^6.0" importlib-metadata = "^4.12.0" rainbowlog = "^2.0.1" +halo = "^0.0.31" +click-aliases = "^1.0.1" [tool.poetry.dev-dependencies] pytest = "^7.1.2" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b05447a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1 @@ +pytest_plugins = ["tests.fixtures"] diff --git a/tests/fixtures/__init__.py b/tests/fixtures/__init__.py new file mode 100644 index 0000000..6129b37 --- /dev/null +++ b/tests/fixtures/__init__.py @@ -0,0 +1,4 @@ +from .conda import conda +from .envs import env_read_only, empty_env + +__all__ = ["conda", "env_read_only", "empty_env"] diff --git a/tests/fixtures/conda.py b/tests/fixtures/conda.py new file mode 100644 index 0000000..4529034 --- /dev/null +++ b/tests/fixtures/conda.py @@ -0,0 +1,8 @@ +import pytest + +from condax.conda.conda import Conda + + +@pytest.fixture(scope="session") +def conda() -> Conda: + return Conda(channels=("conda-forge",)) diff --git a/tests/fixtures/envs.py b/tests/fixtures/envs.py new file mode 100644 index 0000000..38de4d6 --- /dev/null +++ b/tests/fixtures/envs.py @@ -0,0 +1,31 @@ +from pathlib import Path +import shlex +import subprocess +import tempfile +from typing import Generator +import pytest + +from condax.conda.conda import Conda + + +@pytest.fixture(scope="session") +def env_read_only(conda: Conda) -> Generator[Path, None, None]: + """For efficiency, this env can be reused by all tests which won't modify it. + + This env is guaranteed to contain pip=22.2.2 and some version of python, which it depends on.""" + with tempfile.TemporaryDirectory() as tmp_path: + prefix = Path(tmp_path) / "env_read_only" + conda.create_env(prefix, "pip=22.2.2") + yield prefix + + +@pytest.fixture(scope="session") +def empty_env(conda: Conda) -> Generator[Path, None, None]: + """For efficiency, this env can be reused by all tests which won't modify it. + This env is guaranteed to contain no packages.""" + with tempfile.TemporaryDirectory() as tmp_path: + prefix = Path(tmp_path) / "empty_env" + subprocess.run( + shlex.split(f"{conda.exe} create --prefix {prefix} --yes --quiet") + ) + yield prefix diff --git a/tests/test_conda/test_env_info.py b/tests/test_conda/test_env_info.py new file mode 100644 index 0000000..749f14e --- /dev/null +++ b/tests/test_conda/test_env_info.py @@ -0,0 +1,50 @@ +from pathlib import Path +import pytest + +from condax.conda.env_info import find_envs, is_env, find_exes +from condax.conda.exceptions import NoPackageMetadataError + + +def test_is_env(env_read_only: Path): + assert is_env(env_read_only) + + +def test_is_env_empty_env(empty_env: Path): + assert is_env(empty_env) + + +def test_is_env_empty_dir(tmp_path: Path): + assert not is_env(tmp_path) + + +def test_is_env_file(tmp_path: Path): + (tmp_path / "foo.txt").touch() + assert not is_env(tmp_path) + + +def test_is_env_not_exists(tmp_path: Path): + assert not is_env(tmp_path / "foo/bar/biz/") + + +def test_find_exes(env_read_only: Path): + exes = {exe.name for exe in find_exes(env_read_only, "pip")} + assert "pip" in exes + assert "pip3" in exes + assert "python" not in exes + + with pytest.raises(NoPackageMetadataError): + assert not find_exes(env_read_only, "foo") + + +def test_find_exes_empty_env(empty_env: Path): + with pytest.raises(NoPackageMetadataError): + find_exes(empty_env, "pip") + + +def test_find_envs(env_read_only: Path, empty_env: Path): + for env in (env_read_only, empty_env): + assert env in find_envs(env.parent) + + +def test_find_envs_empty_dir(tmp_path: Path): + assert find_envs(tmp_path) == set() diff --git a/tests/test_condax.py b/tests/test_condax.py index 081d386..6dad8cb 100644 --- a/tests/test_condax.py +++ b/tests/test_condax.py @@ -8,12 +8,12 @@ def test_pipx_install_roundtrip(): """ from condax.core import install_package, remove_package import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge", "default"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) @@ -53,12 +53,12 @@ def test_install_specific_version(): """ from condax.core import install_package, remove_package import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge", "default"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) @@ -102,12 +102,12 @@ def test_inject_then_uninject(): """ from condax.core import install_package, inject_package_to, uninject_package_from import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge", "default"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) @@ -194,13 +194,13 @@ def test_inject_with_include_apps(): remove_package, ) import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath # prep prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge", "default"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) diff --git a/tests/test_condax_more.py b/tests/test_condax_more.py index 1f034c6..3725545 100644 --- a/tests/test_condax_more.py +++ b/tests/test_condax_more.py @@ -15,18 +15,18 @@ def test_export_import(): import_environments, ) import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath # prep prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) export_dir_fp = tempfile.TemporaryDirectory() - export_dir = to_path(export_dir_fp.name) + export_dir = FullPath(export_dir_fp.name) gh = "gh" injected_rg_name = "ripgrep" diff --git a/tests/test_condax_repair.py b/tests/test_condax_repair.py index e6a058e..92f2d3f 100644 --- a/tests/test_condax_repair.py +++ b/tests/test_condax_repair.py @@ -8,13 +8,13 @@ def test_fix_links(): """ from condax.core import install_package, inject_package_to, fix_links import condax.config as config - from condax.utils import to_path + from condax.utils import FullPath # prep prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) @@ -106,14 +106,14 @@ def test_fix_links_without_metadata(): fix_links, ) import condax.config as config - import condax.metadata as metadata - from condax.utils import to_path + import condax.condax.metadata as metadata + from condax.utils import FullPath # prep prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) diff --git a/tests/test_condax_update.py b/tests/test_condax_update.py index a31f9f9..1720cdd 100644 --- a/tests/test_condax_update.py +++ b/tests/test_condax_update.py @@ -9,14 +9,15 @@ def test_condax_update_main_apps(): update_package, ) import condax.config as config - from condax.utils import to_path, is_env_dir - import condax.metadata as metadata + from condax.utils import FullPath + import condax.condax.metadata as metadata + from condax.conda.env_info import is_env # prep prefix_fp = tempfile.TemporaryDirectory() - prefix_dir = to_path(prefix_fp.name) + prefix_dir = FullPath(prefix_fp.name) bin_fp = tempfile.TemporaryDirectory() - bin_dir = to_path(bin_fp.name) + bin_dir = FullPath(bin_fp.name) channels = ["conda-forge", "bioconda"] config.set_via_value(prefix_dir=prefix_dir, bin_dir=bin_dir, channels=channels) @@ -46,13 +47,13 @@ def test_condax_update_main_apps(): exe_main = bin_dir / "gff3-to-ddbj" # Before installation there should be nothing - assert not is_env_dir(env_dir) + assert not is_env(env_dir) assert all(not app.exists() for app in apps_before_update) install_package(main_spec_before_update) # After installtion there should be an environment and apps - assert is_env_dir(env_dir) + assert is_env(env_dir) assert all(app.exists() and app.is_file() for app in apps_before_update) # gff3-to-ddbj --version was not implemented as of 0.1.1 @@ -62,7 +63,7 @@ def test_condax_update_main_apps(): update_package(main_spec_after_update, update_specs=True) # After update there should be an environment and update apps - assert is_env_dir(env_dir) + assert is_env(env_dir) assert all(app.exists() and app.is_file() for app in apps_after_update) to_be_removed = apps_before_update - apps_after_update diff --git a/tests/test_metadata.py b/tests/test_metadata.py index 69e9119..a6cda3b 100644 --- a/tests/test_metadata.py +++ b/tests/test_metadata.py @@ -1,5 +1,6 @@ import textwrap -from condax.metadata import MainPackage, InjectedPackage, CondaxMetaData +from condax.condax.metadata.metadata import CondaxMetaData +from condax.condax.metadata.package import MainPackage, InjectedPackage def test_metadata_to_json():