diff --git a/.config/dictionary.txt b/.config/dictionary.txt index b1fcbbf..1e874f0 100644 --- a/.config/dictionary.txt +++ b/.config/dictionary.txt @@ -1,9 +1,11 @@ bindep bindir bthornto +caplog fileh levelname levelno +netcommon pip4a reqs uninstallation diff --git a/src/pip4a/app.py b/src/pip4a/app.py deleted file mode 100644 index a80ed1f..0000000 --- a/src/pip4a/app.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Application internals.""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import TYPE_CHECKING - - -if TYPE_CHECKING: - from argparse import Namespace - - -@dataclass -class App: - """Application internals.""" - - args: Namespace - collection_name: str - collection_dependencies: dict[str, str] diff --git a/src/pip4a/arg_parser.py b/src/pip4a/arg_parser.py index 7955ca0..df2cb20 100644 --- a/src/pip4a/arg_parser.py +++ b/src/pip4a/arg_parser.py @@ -37,6 +37,7 @@ def parse() -> argparse.Namespace: description="valid subcommands", help="additional help", dest="subcommand", + required=True, ) parent_parser = argparse.ArgumentParser(add_help=False) @@ -45,6 +46,7 @@ def parse() -> argparse.Namespace: "-v", "--verbose", action="store_true", + default=False, help="Increase output verbosity.", ) parent_parser.add_argument( diff --git a/src/pip4a/base.py b/src/pip4a/base.py deleted file mode 100644 index d25a296..0000000 --- a/src/pip4a/base.py +++ /dev/null @@ -1,144 +0,0 @@ -"""A base class for pip4a.""" - -from __future__ import annotations - -import json -import logging -import os -import subprocess -import sys - -from pathlib import Path -from typing import TYPE_CHECKING - - -if TYPE_CHECKING: - from .app import App - -logger = logging.getLogger(__name__) - - -class Base: - """A base class for pip4a.""" - - def __init__(self: Base, app: App) -> None: - """Initialize the installer. - - Arguments: - app: The app instance - """ - self.app: App = app - self.bindir: Path - self.interpreter: Path - self.site_pkg_path: Path - self.python_path: Path - - def _set_interpreter( # noqa: PLR0912 - self: Base, - create: bool = False, # noqa: FBT001, FBT002 - ) -> None: - """Set the interpreter.""" - venv = None - if self.app.args.venv: - venv = Path(self.app.args.venv).resolve() - else: - venv_str = os.environ.get("VIRTUAL_ENV") - if venv_str: - venv = Path(venv_str).resolve() - if venv: - if not venv.exists(): - if create: - if not venv.is_relative_to(Path.cwd()): - err = "Virtual environment must relative to cwd to create it." - logger.critical(err) - msg = f"Creating virtual environment: {venv}" - logger.info(msg) - command = f"python -m venv {venv}" - msg = f"Running command: {command}" - logger.debug(msg) - try: - subprocess.run( - command, - check=True, - shell=True, # noqa: S602 - capture_output=self.app.args.verbose, - ) - except subprocess.CalledProcessError as exc: - err = f"Failed to create virtual environment: {exc}" - logger.critical(err) - else: - err = f"Cannot find virtual environment: {venv}." - logger.critical(err) - msg = f"Virtual environment: {venv}" - logger.info(msg) - interpreter = venv / "bin" / "python" - if not interpreter.exists(): - err = f"Cannot find interpreter: {interpreter}." - logger.critical(err) - - msg = f"Using specified interpreter: {interpreter}" - logger.info(msg) - self.interpreter = interpreter - else: - self.interpreter = Path(sys.executable) - msg = f"Using current interpreter: {self.interpreter}" - logger.info(msg) - - command = "python -c 'import sys; print(sys.executable)'" - msg = f"Running command: {command}" - logger.debug(msg) - try: - proc = subprocess.run( - command, - check=True, - capture_output=True, - shell=True, # noqa: S602 - text=True, - ) - except subprocess.CalledProcessError as exc: - err = f"Failed to find interpreter in use: {exc}" - logger.critical(err) - - self.python_path = Path(proc.stdout.strip()) - - def _set_bindir(self: Base) -> None: - """Set the bindir.""" - self.bindir = self.interpreter.parent - if not self.bindir.exists(): - err = f"Cannot find bindir: {self.bindir}." - logger.critical(err) - - def _set_site_pkg_path(self: Base) -> None: - """USe the interpreter to find the site packages path.""" - command = ( - f"{self.interpreter} -c" - " 'import json,site; print(json.dumps(site.getsitepackages()))'" - ) - msg = f"Running command: {command}" - logger.debug(msg) - try: - proc = subprocess.run( - command, - check=True, - capture_output=True, - shell=True, # noqa: S602 - text=True, - ) - except subprocess.CalledProcessError as exc: - err = f"Failed to find site packages path: {exc}" - logger.critical(err) - - try: - site_pkg_dirs = json.loads(proc.stdout) - except json.JSONDecodeError as exc: - err = f"Failed to decode json: {exc}" - logger.critical(err) - - if not site_pkg_dirs: - err = "Failed to find site packages path." - logger.critical(err) - - msg = f"Found site packages path: {site_pkg_dirs[0]}" - logger.debug(msg) - - self.site_pkg_path = Path(site_pkg_dirs[0]) diff --git a/src/pip4a/cli.py b/src/pip4a/cli.py index c88eccf..1bc65b1 100644 --- a/src/pip4a/cli.py +++ b/src/pip4a/cli.py @@ -9,12 +9,11 @@ from pathlib import Path from typing import TYPE_CHECKING -from .app import App from .arg_parser import parse +from .config import Config from .installer import Installer from .logger import ColoredFormatter, ExitOnExceptionHandler from .uninstaller import UnInstaller -from .utils import get_galaxy if TYPE_CHECKING: @@ -27,7 +26,7 @@ class Cli: def __init__(self: Cli) -> None: """Initialize the CLI and parse CLI args.""" self.args: Namespace - self.app: App + self.config: Config def parse_args(self: Cli) -> None: """Parse the command line arguments.""" @@ -43,6 +42,7 @@ def init_logger(self: Cli) -> None: ) ch.setFormatter(cf) logger.addHandler(ch) + if self.args.verbose: logger.setLevel(logging.DEBUG) else: @@ -92,23 +92,21 @@ def ensure_isolated(self: Cli) -> None: def run(self: Cli) -> None: """Run the application.""" logger = logging.getLogger("pip4a") - collection_name, dependencies = get_galaxy() - self.app = App( - args=self.args, - collection_name=collection_name, - collection_dependencies=dependencies, - ) - if "," in self.app.args.collection_specifier: + self.config = Config(args=self.args) + + if "," in self.config.args.collection_specifier: err = "Multiple optional dependencies are not supported at this time." logger.critical(err) - if self.app.args.subcommand == "install": - installer = Installer(self.app) + if self.config.args.subcommand == "install": + self.config.init(create_venv=True) + installer = Installer(self.config) installer.run() - if self.app.args.subcommand == "uninstall": - uninstaller = UnInstaller(self.app) + if self.config.args.subcommand == "uninstall": + self.config.init() + uninstaller = UnInstaller(self.config) uninstaller.run() diff --git a/src/pip4a/config.py b/src/pip4a/config.py new file mode 100644 index 0000000..a5d244a --- /dev/null +++ b/src/pip4a/config.py @@ -0,0 +1,238 @@ +"""Constants, for now, for pip4a.""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import subprocess +import sys + +from pathlib import Path +from typing import TYPE_CHECKING + +import yaml + +from .utils import subprocess_run + + +if TYPE_CHECKING: + from argparse import Namespace + +logger = logging.getLogger(__name__) + + +class Config: + """The application configuration.""" + + def __init__( + self: Config, args: Namespace, + ) -> None: + """Initialize the configuration.""" + self._create_venv: bool + self.args: Namespace = args + self.bindir: Path + self.c_name: str + self.c_namespace: str + self.python_path: Path + self.site_pkg_path: Path + self.venv_interpreter: Path + + + def init(self: Config, create_venv: bool = False) -> None: # noqa: FBT001, FBT002 + """Initialize the configuration. + + Args: + create_venv: Create a virtual environment. Defaults to False. + """ + if self.args.subcommand == "install": + self._get_galaxy() + elif self.args.subcommand == "uninstall": + parts = self.args.collection_specifier.split(".") + if len(parts) != 2: + err = ( + "The collection specifier must be in the form of" + " 'namespace.collection'" + ) + logger.critical(err) + self.c_namespace = parts[0] + self.c_name = parts[1] + self._create_venv = create_venv + self._set_interpreter() + self._set_site_pkg_path() + + @property + def collection_path(self: Config) -> Path: + """Set the collection root directory.""" + spec = self.args.collection_specifier.split("[")[0] + specp = Path(spec).expanduser().resolve() + if specp.is_dir(): + return specp + err = f"Cannot find collection root directory. {specp}" + logger.critical(err) + sys.exit(1) + + @property + def collection_name(self: Config) -> str: + """Return the collection name.""" + return f"{self.c_namespace}.{self.c_name}" + + @property + def cache_dir(self: Config) -> Path: + """Return the cache directory.""" + cache_home = os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache") + return (Path(cache_home) / "pip4a").resolve() + + @property + def venv(self: Config) -> Path: + """Return the virtual environment path.""" + if self.args.venv: + return Path(self.args.venv).expanduser().resolve() + venv_str = os.environ.get("VIRTUAL_ENV") + if venv_str: + return Path(venv_str).expanduser().resolve() + err = "Failed to find a virtual environment." + logger.critical(err) + sys.exit(1) + + @property + def vuuid(self: Config) -> str: + """Return the virtual environment uuid.""" + result = hashlib.md5(str(self.venv).encode()) # noqa: S324 + return result.hexdigest() + + @property + def venv_cache_dir(self: Config) -> Path: + """Return the virtual environment cache directory.""" + return self.cache_dir / self.vuuid + + @property + def collection_cache_dir(self: Config) -> Path: + """Return the collection cache directory.""" + return self.venv_cache_dir / self.collection_name + + @property + def collection_build_dir(self: Config) -> Path: + """Return the collection cache directory.""" + return self.collection_cache_dir / "build" + + @property + def discovered_python_reqs(self: Config) -> Path: + """Return the discovered python requirements file.""" + return self.collection_cache_dir / "discovered_requirements.txt" + + @property + def discovered_bindep_reqs(self: Config) -> Path: + """Return the discovered system package requirements file.""" + return self.collection_cache_dir / "discovered_bindep.txt" + + @property + def installed_collections(self: Config) -> Path: + """Return the installed collections file.""" + return self.collection_cache_dir / "installed_collections.txt" + + @property + def site_pkg_collection_path(self: Config) -> Path: + """Return the site packages collection path.""" + return self.site_pkg_path / "ansible_collections" / self.c_namespace / self.c_name + + @property + def venv_bindir(self: Config) -> Path: + """Return the virtual environment bin directory.""" + return self.venv / "bin" + + @property + def interpreter(self: Config) -> Path: + """Return the current interpreter.""" + return Path(sys.executable) + + def _get_galaxy(self: Config) -> None: + """Retrieve the collection name from the galaxy.yml file. + + Returns: + str: The collection name and dependencies + + Raises: + SystemExit: If the collection name is not found + """ + file_name = self.collection_path / "galaxy.yml" + if not file_name.exists(): + err = f"Failed to find {file_name} in {self.collection_path}" + logger.critical(err) + + with file_name.open(encoding="utf-8") as fileh: + try: + yaml_file = yaml.safe_load(fileh) + except yaml.YAMLError as exc: + err = f"Failed to load yaml file: {exc}" + logger.critical(err) + + try: + self.c_namespace = yaml_file["namespace"] + self.c_name = yaml_file["name"] + msg = f"Found collection name: {self.collection_name} from {file_name}." + logger.info(msg) + except KeyError as exc: + err = f"Failed to find collection name in {file_name}: {exc}" + logger.critical(err) + else: + return + raise SystemExit(1) # We shouldn't be here + + def _set_interpreter( + self: Config, + ) -> None: + """Set the interpreter.""" + if not self.venv.exists(): + if self._create_venv: + msg = f"Creating virtual environment: {self.venv}" + logger.info(msg) + command = f"python -m venv {self.venv}" + try: + subprocess_run(command=command, verbose=self.args.verbose) + except subprocess.CalledProcessError as exc: + err = f"Failed to create virtual environment: {exc}" + logger.critical(err) + else: + err = f"Cannot find virtual environment: {self.venv}." + logger.critical(err) + msg = f"Virtual environment: {self.venv}" + logger.info(msg) + venv_interpreter = self.venv / "bin" / "python" + if not venv_interpreter.exists(): + err = f"Cannot find interpreter: {venv_interpreter}." + logger.critical(err) + + msg = f"Virtual envrionment interpreter: {venv_interpreter}" + logger.info(msg) + self.venv_interpreter = venv_interpreter + + + def _set_site_pkg_path(self: Config) -> None: + """USe the interpreter to find the site packages path.""" + command = ( + f"{self.venv_interpreter} -c" + " 'import json,site; print(json.dumps(site.getsitepackages()))'" + ) + try: + proc = subprocess_run(command=command, verbose=self.args.verbose) + except subprocess.CalledProcessError as exc: + err = f"Failed to find site packages path: {exc}" + logger.critical(err) + + try: + site_pkg_dirs = json.loads(proc.stdout) + except json.JSONDecodeError as exc: + err = f"Failed to decode json: {exc}" + logger.critical(err) + + if not site_pkg_dirs: + err = "Failed to find site packages path." + logger.critical(err) + + msg = f"Found site packages path: {site_pkg_dirs[0]}" + logger.debug(msg) + + self.site_pkg_path = Path(site_pkg_dirs[0]) + diff --git a/src/pip4a/constants.py b/src/pip4a/constants.py deleted file mode 100644 index 0ce81a0..0000000 --- a/src/pip4a/constants.py +++ /dev/null @@ -1,13 +0,0 @@ -"""Constants, for now, for pip4a.""" - -from pathlib import Path - - -class Constants: - """Constants, for now, for pip4a.""" - - WORKING_DIR = Path("./.pip4a_cache").resolve() - COLLECTION_BUILD_DIR = WORKING_DIR / "src" - DISCOVERED_PYTHON_REQS = WORKING_DIR / "discovered_requirements.txt" - DISCOVERED_BINDEP_REQS = WORKING_DIR / "discovered_bindep.txt" - INSTALLED_COLLECTIONS = WORKING_DIR / "installed_collections.txt" diff --git a/src/pip4a/installer.py b/src/pip4a/installer.py index 4c027cd..dd5fd4d 100644 --- a/src/pip4a/installer.py +++ b/src/pip4a/installer.py @@ -9,64 +9,69 @@ import subprocess from pathlib import Path +from typing import TYPE_CHECKING -from .base import Base -from .constants import Constants as C # noqa: N817 from .utils import opt_deps_to_files, oxford_join, subprocess_run +if TYPE_CHECKING: + from .config import Config + + logger = logging.getLogger(__name__) -class Installer(Base): +class Installer: """The installer class.""" - def run(self: Installer) -> None: - """Run the installer.""" - if not self.app.args.collection_specifier.startswith("."): - err = "Only local collections are supported at this time. ('.' or .[test])]" - logger.critical(err) + def __init__(self: Installer, config: Config) -> None: + """Initialize the installer. - self._set_interpreter(create=True) - self._set_bindir() - self._set_site_pkg_path() + Args: + config: The application configuration. + """ + self._config = config + def run(self: Installer) -> None: + """Run the installer.""" self._install_core() self._install_collection() - if self.app.args.editable: + if self._config.args.editable: self._swap_editable_collection() self._discover_deps() self._pip_install() self._check_bindep() - if self.app.args.venv and (self.python_path != self.interpreter): + if self._config.args.venv and ( + self._config.interpreter != self._config.venv_interpreter + ): msg = "A virtual environment was specified but has not been activated." logger.warning(msg) msg = ( "Please activate the virtual environment:" - f"\nsource {self.app.args.venv}/bin/activate" + f"\nsource {self._config.args.venv}/bin/activate" ) logger.warning(msg) def _init_build_dir(self: Installer) -> None: """Initialize the build directory.""" - msg = f"Initializing build directory: {C.COLLECTION_BUILD_DIR}" + msg = f"Initializing build directory: {self._config.collection_build_dir}" logger.info(msg) - if C.COLLECTION_BUILD_DIR.exists(): - shutil.rmtree(C.COLLECTION_BUILD_DIR) - C.COLLECTION_BUILD_DIR.mkdir(parents=True) + if self._config.collection_build_dir.exists(): + shutil.rmtree(self._config.collection_build_dir) + self._config.collection_build_dir.mkdir(parents=True) def _install_core(self: Installer) -> None: """Install ansible-core if not installed already.""" - core = self.bindir / "ansible" + core = self._config.venv_bindir / "ansible" if core.exists(): return msg = "Installing ansible-core." logger.info(msg) - command = f"{self.interpreter} -m pip install ansible-core" + command = f"{self._config.venv_interpreter} -m pip install ansible-core" try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = f"Failed to install ansible-core: {exc}" logger.critical(err) @@ -74,42 +79,53 @@ def _install_core(self: Installer) -> None: def _discover_deps(self: Installer) -> None: """Discover the dependencies.""" command = ( - f"ansible-builder introspect {self.site_pkg_path}" - f" --write-pip {C.DISCOVERED_PYTHON_REQS}" - f" --write-bindep {C.DISCOVERED_BINDEP_REQS}" + f"ansible-builder introspect {self._config.site_pkg_path}" + f" --write-pip {self._config.discovered_python_reqs}" + f" --write-bindep {self._config.discovered_bindep_reqs}" " --sanitize" ) - opt_deps = re.match(r".*\[(.*)\]", self.app.args.collection_specifier) + opt_deps = re.match(r".*\[(.*)\]", self._config.args.collection_specifier) if opt_deps: - for dep in opt_deps_to_files(opt_deps.group(1)): - command += f" --user-pip {dep}" - msg = f"Writing discovered python requirements to: {C.DISCOVERED_PYTHON_REQS}" + dep_paths = opt_deps_to_files( + collection_path=self._config.collection_path, + dep_str=opt_deps.group(1), + ) + for dep_path in dep_paths: + command += f" --user-pip {dep_path}" + msg = f"Writing discovered python requirements to: {self._config.discovered_python_reqs}" logger.info(msg) - msg = f"Writing discovered system package requirements to: {C.DISCOVERED_BINDEP_REQS}" + msg = f"Writing discovered system requirements to: {self._config.discovered_bindep_reqs}" logger.info(msg) try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: - err = f"Failed to discover requirements: {exc}" + err = f"Failed to discover requirements: {exc} {exc.stderr}" logger.critical(err) def _install_collection(self: Installer) -> None: """Install the collection from the build directory.""" self._init_build_dir() - command = f"cp -r --parents $(git ls-files 2> /dev/null || ls) {C.COLLECTION_BUILD_DIR}" + command = ( + "cp -r --parents $(git ls-files 2> /dev/null || ls)" + f" {self._config.collection_build_dir}" + ) msg = "Copying collection to build directory using git ls-files." logger.info(msg) try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run( + command=command, + cwd=self._config.collection_path, + verbose=self._config.args.verbose, + ) except subprocess.CalledProcessError as exc: - err = f"Failed to copy collection to build directory: {exc}" + err = f"Failed to copy collection to build directory: {exc} {exc.stderr}" logger.critical(err) command = ( - f"cd {C.COLLECTION_BUILD_DIR} &&" - f" {self.bindir / 'ansible-galaxy'} collection build" - f" --output-path {C.COLLECTION_BUILD_DIR}" + f"cd {self._config.collection_build_dir} &&" + f" {self._config.venv_bindir / 'ansible-galaxy'} collection build" + f" --output-path {self._config.collection_build_dir}" " --force" ) @@ -117,51 +133,44 @@ def _install_collection(self: Installer) -> None: logger.info(msg) try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = f"Failed to build collection: {exc} {exc.stderr}" logger.critical(err) built = [ f - for f in Path(C.COLLECTION_BUILD_DIR).iterdir() + for f in Path(self._config.collection_build_dir).iterdir() if f.is_file() and f.name.endswith(".tar.gz") ] if len(built) != 1: err = ( "Expected to find one collection tarball in" - f"{C.COLLECTION_BUILD_DIR}, found {len(built)}" + f"{self._config.collection_build_dir}, found {len(built)}" ) raise RuntimeError(err) tarball = built[0] - # Remove installed collection if it exists - site_pkg_collection_path = ( - self.site_pkg_path - / "ansible_collections" - / self.app.collection_name.split(".")[0] - / self.app.collection_name.split(".")[1] - ) - if site_pkg_collection_path.exists(): - msg = f"Removing installed {site_pkg_collection_path}" + if self._config.site_pkg_collection_path.exists(): + msg = f"Removing installed {self._config.site_pkg_collection_path}" logger.debug(msg) - if site_pkg_collection_path.is_symlink(): - site_pkg_collection_path.unlink() + if self._config.site_pkg_collection_path.is_symlink(): + self._config.site_pkg_collection_path.unlink() else: - shutil.rmtree(site_pkg_collection_path) + shutil.rmtree(self._config.site_pkg_collection_path) command = ( - f"{self.bindir / 'ansible-galaxy'} collection" - f" install {tarball} -p {self.site_pkg_path}" + f"{self._config.venv_bindir / 'ansible-galaxy'} collection" + f" install {tarball} -p {self._config.site_pkg_path}" " --force" ) env = os.environ - if not self.app.args.verbose: + if not self._config.args.verbose: env["ANSIBLE_GALAXY_COLLECTIONS_PATH_WARNING"] = "false" msg = "Running ansible-galaxy to install collection and it's dependencies." logger.info(msg) try: - proc = subprocess_run(command=command, verbose=self.app.args.verbose) + proc = subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = f"Failed to install collection: {exc} {exc.stderr}" logger.critical(err) @@ -169,50 +178,49 @@ def _install_collection(self: Installer) -> None: installed = re.findall(r"(\w+\.\w+):.*installed", proc.stdout) msg = f"Installed collections: {oxford_join(installed)}" logger.info(msg) - with C.INSTALLED_COLLECTIONS.open(mode="w") as f: + with self._config.installed_collections.open(mode="w") as f: f.write("\n".join(installed)) def _swap_editable_collection(self: Installer) -> None: """Swap the installed collection with the current working directory.""" - site_pkg_collection_path = ( - self.site_pkg_path - / "ansible_collections" - / self.app.collection_name.split(".")[0] - / self.app.collection_name.split(".")[1] - ) - - msg = f"Removing installed {site_pkg_collection_path}" + msg = f"Removing installed {self._config.site_pkg_collection_path}" logger.info(msg) - if site_pkg_collection_path.exists(): - if site_pkg_collection_path.is_symlink(): - site_pkg_collection_path.unlink() + if self._config.site_pkg_collection_path.exists(): + if self._config.site_pkg_collection_path.is_symlink(): + self._config.site_pkg_collection_path.unlink() else: - shutil.rmtree(site_pkg_collection_path) + shutil.rmtree(self._config.site_pkg_collection_path) cwd = Path.cwd() - msg = f"Symlinking {site_pkg_collection_path} to {cwd}" + msg = f"Symlinking {self._config.site_pkg_collection_path} to {cwd}" logger.info(msg) - site_pkg_collection_path.symlink_to(cwd) + self._config.site_pkg_collection_path.symlink_to(cwd) def _pip_install(self: Installer) -> None: """Install the dependencies.""" - command = f"{self.interpreter} -m pip install -r {C.DISCOVERED_PYTHON_REQS}" + command = ( + f"{self._config.venv_interpreter} -m pip install" + f" -r {self._config.discovered_python_reqs}" + ) - msg = f"Installing python requirements from {C.DISCOVERED_PYTHON_REQS}" + msg = ( + f"Installing python requirements from {self._config.discovered_python_reqs}" + ) logger.info(msg) try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = ( - f"Failed to install requirements from {C.DISCOVERED_PYTHON_REQS}: {exc}" + "Failed to install requirements from" + f" {self._config.discovered_python_reqs}: {exc}" ) raise RuntimeError(err) from exc def _check_bindep(self: Installer) -> None: """Check the bindep file.""" - command = f"{self.bindir / 'bindep'} -b -f {C.DISCOVERED_BINDEP_REQS}" + command = f"bindep -b -f {self._config.discovered_bindep_reqs}" try: - subprocess_run(command=command, verbose=self.app.args.verbose) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: lines = exc.stdout.splitlines() msg = ( diff --git a/src/pip4a/uninstaller.py b/src/pip4a/uninstaller.py index 1d0c140..9118e3e 100644 --- a/src/pip4a/uninstaller.py +++ b/src/pip4a/uninstaller.py @@ -6,54 +6,53 @@ import shutil import subprocess -from .base import Base -from .constants import Constants as C # noqa: N817 +from typing import TYPE_CHECKING + +from .utils import subprocess_run + + +if TYPE_CHECKING: + from .config import Config logger = logging.getLogger(__name__) -class UnInstaller(Base): +class UnInstaller: """The uninstaller class.""" + def __init__(self: UnInstaller, config: Config) -> None: + """Initialize the installer. + + Args: + config: The application configuration. + """ + self._config = config + def run(self: UnInstaller) -> None: - """Run the installer.""" - if self.app.args.collection_specifier not in (self.app.collection_name, "."): - err = ( - "Invalid requirement: {self.app.args.collection_specifier} ignored -" - " the uninstall command expects the name of the local collection." - ) - logger.warning(err) + """Run the uninstaller.""" + if not self._config.collection_cache_dir.exists(): err = ( - "You must give at least one requirement" - f" to uninstall (e.g. {self.app.collection_name})." + f"Either the collection '{self._config.collection_name}' was" + " previously uninstalled or was not initially installed using pip4a." ) logger.critical(err) - if not C.WORKING_DIR.exists(): - msg = f"Failed to find working directory '{C.WORKING_DIR}', nothing to do." - logger.warning(msg) - return - - self._set_interpreter() - self._set_bindir() - self._set_site_pkg_path() - self._pip_uninstall() self._remove_collections() - shutil.rmtree(C.WORKING_DIR, ignore_errors=True) + shutil.rmtree(self._config.collection_cache_dir) def _remove_collections(self: UnInstaller) -> None: # noqa: C901, PLR0912, PLR0915 """Remove the collection and dependencies.""" - if not C.INSTALLED_COLLECTIONS.exists(): - msg = f"Failed to find {C.INSTALLED_COLLECTIONS}" + if not self._config.installed_collections.exists(): + msg = f"Failed to find {self._config}" logger.warning(msg) return - with C.INSTALLED_COLLECTIONS.open() as f: + with self._config.installed_collections.open() as f: collection_names = f.read().splitlines() collection_namespace_roots = [] - collection_root = self.site_pkg_path / "ansible_collections" + collection_root = self._config.site_pkg_path / "ansible_collections" for collection_name in collection_names: namespace, name = collection_name.split(".") @@ -118,30 +117,23 @@ def _remove_collections(self: UnInstaller) -> None: # noqa: C901, PLR0912, PLR0 def _pip_uninstall(self: UnInstaller) -> None: """Uninstall the dependencies.""" - if not C.DISCOVERED_PYTHON_REQS.exists(): - msg = f"Failed to find {C.DISCOVERED_PYTHON_REQS}" + if not self._config.discovered_python_reqs.exists(): + msg = f"Failed to find {self._config.discovered_python_reqs}" logger.warning(msg) return command = ( - f"{self.interpreter} -m pip uninstall -r {C.DISCOVERED_PYTHON_REQS} -y" + f"{self._config.venv_interpreter} -m pip uninstall" + f" -r {self._config.discovered_python_reqs} -y" ) - msg = f"Uninstalling python requirements from {C.DISCOVERED_PYTHON_REQS}" + msg = f"Uninstalling python requirements from {self._config.discovered_python_reqs}" logger.info(msg) - msg = f"Running command: {command}" - logger.debug(msg) try: - subprocess.run( - command, - check=True, - capture_output=not self.app.args.verbose, - shell=True, # noqa: S602 - text=True, - ) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = ( - f"Failed to uninstall requirements from {C.DISCOVERED_PYTHON_REQS}:" + f"Failed to uninstall requirements from {self._config.discovered_python_reqs}:" f" {exc} - {exc.stderr}" ) logger.critical(err) @@ -149,36 +141,20 @@ def _pip_uninstall(self: UnInstaller) -> None: # Repair the core and builder installs if needed. msg = "Repairing the core and builder installs if needed." logger.info(msg) - command = f"{self.interpreter} -m pip check" - msg = f"Running command: {command}" - logger.debug(msg) + command = f"{self._config.venv_interpreter} -m pip check" try: - subprocess.run( - command, - check=True, - capture_output=True, - shell=True, # noqa: S602 - text=True, - ) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError: pass else: return command = ( - f"{self.interpreter} -m pip install --upgrade" + f"{self._config.venv_interpreter} -m pip install --upgrade" " --force-reinstall ansible-core ansible-builder" ) - msg = f"Running command: {command}" - logger.debug(msg) try: - subprocess.run( - command, - check=True, - capture_output=True, - shell=True, # noqa: S602 - text=True, - ) + subprocess_run(command=command, verbose=self._config.args.verbose) except subprocess.CalledProcessError as exc: err = f"Failed to repair the core and builder installs: {exc}" logger.critical(err) diff --git a/src/pip4a/utils.py b/src/pip4a/utils.py index 14380e1..a8c6cdf 100644 --- a/src/pip4a/utils.py +++ b/src/pip4a/utils.py @@ -5,51 +5,22 @@ import logging import subprocess -from pathlib import Path +from typing import TYPE_CHECKING import subprocess_tee -import yaml -logger = logging.getLogger(__name__) - +if TYPE_CHECKING: + from pathlib import Path -def get_galaxy() -> tuple[str, dict[str, str]]: - """Retrieve the collection name from the galaxy.yml file. - Returns: - str: The collection name and dependencies - - Raises: - SystemExit: If the collection name is not found - """ - file_name = Path("galaxy.yml").resolve() - if not file_name.exists(): - err = f"Failed to find {file_name}, please run from the collection root." - logger.critical(err) - - with file_name.open(encoding="utf-8") as fileh: - try: - yaml_file = yaml.safe_load(fileh) - except yaml.YAMLError as exc: - err = f"Failed to load yaml file: {exc}" - logger.critical(err) - - dependencies = yaml_file.get("dependencies", []) - try: - collection_name = yaml_file["namespace"] + "." + yaml_file["name"] - msg = f"Found collection name: {collection_name} from {file_name}." - logger.info(msg) - return yaml_file["namespace"] + "." + yaml_file["name"], dependencies - except KeyError as exc: - err = f"Failed to find collection name in {file_name}: {exc}" - logger.critical(err) - raise SystemExit(1) # We shouldn't be here +logger = logging.getLogger(__name__) def subprocess_run( - verbose: bool, # noqa: FBT001 command: str, + verbose: bool, # noqa: FBT001 + cwd: Path | None = None, ) -> subprocess.CompletedProcess[str]: """Run a subprocess command.""" msg = f"Running command: {command}" @@ -58,12 +29,14 @@ def subprocess_run( return subprocess_tee.run( command, check=True, + cwd=cwd, shell=True, # noqa: S604 text=True, ) return subprocess.run( command, check=True, + cwd=cwd, shell=True, # noqa: S602 capture_output=True, text=True, @@ -85,7 +58,7 @@ def oxford_join(words: list[str]) -> str: return ", ".join(words[:-1]) + ", and " + words[-1] -def opt_deps_to_files(dep_str: str) -> list[Path]: +def opt_deps_to_files(collection_path: Path, dep_str: str) -> list[Path]: """Convert a string of optional dependencies to a list of files. :param dep_str: A string of optional dependencies @@ -95,11 +68,11 @@ def opt_deps_to_files(dep_str: str) -> list[Path]: files = [] for dep in deps: _dep = dep.strip() - variant1 = Path.cwd() / f"{_dep}-requirements.txt" + variant1 = collection_path / f"{_dep}-requirements.txt" if variant1.exists(): files.append(variant1) continue - variant2 = Path.cwd() / f"requirements-{_dep}.txt" + variant2 = collection_path / f"requirements-{_dep}.txt" if variant2.exists(): files.append(variant2) continue diff --git a/tests/integration/test_basic.py b/tests/integration/test_basic.py index 5981066..445d95e 100644 --- a/tests/integration/test_basic.py +++ b/tests/integration/test_basic.py @@ -1,6 +1,29 @@ """Basic smoke tests.""" +from pathlib import Path -def test() -> None: +import pytest + +from pip4a.cli import main +from pip4a.utils import subprocess_run + + +def test_venv( + caplog: pytest.LogCaptureFixture, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: """Basic smoke test.""" - assert True + command = ( + "git clone https://github.com/ansible-collections/cisco.nxos.git" + f" {tmp_path/ 'cisco.nxos'}" + ) + subprocess_run(command=command, verbose=True) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + "sys.argv", + ["pip4a", "install", str(tmp_path / "cisco.nxos"), "--venv=venv"], + ) + main() + string = "Installed collections: cisco.nxos, ansible.netcommon, and ansible.utils" + assert string in caplog.text