diff --git a/src/antsibull_core/venv.py b/src/antsibull_core/venv.py index 6932a58..524b343 100644 --- a/src/antsibull_core/venv.py +++ b/src/antsibull_core/venv.py @@ -7,11 +7,23 @@ from __future__ import annotations +import asyncio import os import venv +from collections.abc import MutableSequence +from typing import TYPE_CHECKING import sh +from antsibull_core import subprocess_util + +if TYPE_CHECKING: + import subprocess + from logging import Logger as StdLogger + + from _typeshed import StrPath + from twiggy.logger import Logger as TwiggyLogger # type: ignore[import] + def get_clean_environment() -> dict[str, str]: env = os.environ.copy() @@ -35,7 +47,11 @@ class VenvRunner: * :python:mod:`venv` """ - def __init__(self, name: str, top_dir: str) -> None: + name: str + top_dir: StrPath + venv_dir: str + + def __init__(self, name: str, top_dir: StrPath) -> None: """ Create a venv. @@ -46,13 +62,14 @@ def __init__(self, name: str, top_dir: str) -> None: self.top_dir = top_dir self.venv_dir: str = os.path.join(top_dir, name) venv.create(self.venv_dir, clear=True, symlinks=True, with_pip=True) - self._python = self.get_command('python') # Upgrade pip to the latest version. # Note that cryptography stopped building manylinux1 wheels (the only ship manylinux2010) so # we need pip19+ in order to work now. RHEL8 and Ubuntu 18.04 contain a pip that's older # than that so we must upgrade to something even if it's not latest. - self._python('-m', 'pip', 'install', '--upgrade', 'pip') + + # pyre thinks a list is not a MutableSequence when it very much is. + self.log_run(['pip', 'install', '--upgrade', 'pip']) # pyre-ignore[6] def get_command(self, executable_name) -> sh.Command: """ @@ -60,18 +77,67 @@ def get_command(self, executable_name) -> sh.Command: :arg executable_name: Program to return a command for. :returns: An :obj:`sh.Command` that will invoke the program. + + .. deprecated:: 2.0.0 + This method is deprecated in favor of :method:`asnyc_log_run` and + :method:`log_run`. It will be removed in antsibull_core 3.0.0. """ return sh.Command(os.path.join(self.venv_dir, 'bin', executable_name)) - def install_package(self, package_name: str) -> sh.RunningCommand: + def install_package(self, package_name: str) -> subprocess.CompletedProcess: """ Install a python package into the venv. :arg package_name: This can be a bare package name or a path to a file. It's passed directly to :command:`pip install`. - :returns: An :sh:obj:`sh.RunningCommand` for the pip output. + :returns: An :obj:`subprocess.CompletedProcess` for the pip output. """ - return self._python('-m', 'pip', 'install', package_name, _env=get_clean_environment()) + return self.log_run(["pip", "install", package_name]) # pyre-ignore[6] + + async def async_log_run( + self, + args: MutableSequence[StrPath], + logger: TwiggyLogger | StdLogger | None = None, + stdout_loglevel: str | None = None, + stderr_loglevel: str | None = 'debug', + check: bool = True, + **kwargs, + ) -> subprocess.CompletedProcess: + """ + This method asynchronously runs a command in a subprocess and logs + its output. It calls `antsibull_core.subprocess_util.async_log_run` to + do the heavy lifting. `args[0]` must be a filename that's installed in + the venv. If it's not, a `ValueError` will be raised. + """ + kwargs.setdefault("env", get_clean_environment()) + basename = args[0] + if os.path.isabs(basename): + raise ValueError(f'{basename!r} must not be an absolute path!') + path = os.path.join(self.venv_dir, 'bin', basename) + if not os.path.exists(path): + raise ValueError(f'{path!r} does not exist!') + args[0] = path + return await subprocess_util.async_log_run( + args, logger, stdout_loglevel, stderr_loglevel, check, **kwargs + ) + + def log_run( + self, + args: MutableSequence[StrPath], + logger: TwiggyLogger | StdLogger | None = None, + stdout_loglevel: str | None = None, + stderr_loglevel: str | None = 'debug', + check: bool = True, + **kwargs, + ) -> subprocess.CompletedProcess: + """ + See :method:`async_log_run` + """ + return asyncio.run( + self.async_log_run( + args, logger, stdout_loglevel, stderr_loglevel, check, **kwargs + ) + ) class FakeVenvRunner: @@ -83,6 +149,9 @@ class FakeVenvRunner: * :python:mod:`venv` """ + log_run = staticmethod(subprocess_util.log_run) + async_log_run = staticmethod(subprocess_util.async_log_run) + @staticmethod def get_command(executable_name) -> sh.Command: """ @@ -90,6 +159,10 @@ def get_command(executable_name) -> sh.Command: :arg executable_name: Program to return a command for. :returns: An :obj:`sh.Command` that will invoke the program. + + .. deprecated:: 2.0.0 + This function is deprecated in favor of :method:`asnyc_log_run` and + :method:`log_run`. It will be removed in antsibull_core 3.0.0. """ return sh.Command(executable_name) diff --git a/tests/functional/test_venv.py b/tests/functional/test_venv.py new file mode 100644 index 0000000..99a0a62 --- /dev/null +++ b/tests/functional/test_venv.py @@ -0,0 +1,48 @@ +# Copyright (C) 2023 Maxwell G +# SPDX-License-Identifier: GPL-3.0-or-later +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or +# https://www.gnu.org/licenses/gpl-3.0.txt) + +import os.path +from unittest import mock + +import pytest + +from antsibull_core import subprocess_util +from antsibull_core.venv import FakeVenvRunner, VenvRunner, get_clean_environment + + +def test_venv_clean_env(monkeypatch): + monkeypatch.setenv("PYTHONPATH", "/jfjfjfjfjfjfjfjfj") + assert "PYTHONPATH" not in get_clean_environment() + + +def test_venv_run_init(tmp_path): + with mock.patch("antsibull_core.subprocess_util.async_log_run") as log_run: + runner = VenvRunner("asdfgh", tmp_path) + assert runner.name == "asdfgh" + assert runner.top_dir == tmp_path + assert runner.venv_dir == str(tmp_path / "asdfgh") + pip = os.path.join(runner.venv_dir, "bin", "pip") + log_run.assert_called_once_with( + [pip, "install", "--upgrade", "pip"], + None, + None, + "debug", + True, + env=get_clean_environment(), + ) + + +def test_venv_log_run_error(tmp_path): + runner = VenvRunner("zxcvb", tmp_path) + echo = os.path.join(runner.venv_dir, "bin", "echo") + with pytest.raises(ValueError, match=rf"^{echo!r} does not exist!"): + runner.log_run(["echo", "This won't work!"]) + + +def test_venv_log_run_error2(tmp_path): + runner = VenvRunner("zxcvb", tmp_path) + echo = "/usr/bin/echo" + with pytest.raises(ValueError, match=rf"^{echo!r} must not be an absolute path!"): + runner.log_run([echo, "This also won't work!"])