Skip to content

Commit

Permalink
feat: use static arguments instaed of **kwargs (#815)
Browse files Browse the repository at this point in the history
Signed-off-by: Henry Schreiner <henryschreineriii@gmail.com>
  • Loading branch information
henryiii committed Apr 12, 2024
1 parent bc21883 commit d3dd1f8
Show file tree
Hide file tree
Showing 5 changed files with 219 additions and 59 deletions.
22 changes: 18 additions & 4 deletions nox/command.py
Expand Up @@ -17,18 +17,23 @@
import os
import shlex
import shutil
import subprocess
import sys
from collections.abc import Iterable, Mapping, Sequence
from typing import Any

from nox.logger import logger
from nox.popen import popen
from nox.popen import DEFAULT_INTERRUPT_TIMEOUT, DEFAULT_TERMINATE_TIMEOUT, popen

if sys.version_info < (3, 8):
from typing_extensions import Literal
else:
from typing import Literal

TYPE_CHECKING = False

if TYPE_CHECKING:
from typing import IO

ExternalType = Literal["error", True, False]


Expand Down Expand Up @@ -81,7 +86,10 @@ def run(
success_codes: Iterable[int] | None = None,
log: bool = True,
external: ExternalType = False,
**popen_kws: Any,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> str | bool:
"""Run a command-line program."""

Expand Down Expand Up @@ -119,7 +127,13 @@ def run(

try:
return_code, output = popen(
[cmd_path, *str_args], silent=silent, env=env, **popen_kws
[cmd_path, *str_args],
silent=silent,
env=env,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

if return_code not in success_codes:
Expand Down
7 changes: 5 additions & 2 deletions nox/popen.py
Expand Up @@ -21,6 +21,9 @@
from collections.abc import Mapping, Sequence
from typing import IO

DEFAULT_INTERRUPT_TIMEOUT = 0.3
DEFAULT_TERMINATE_TIMEOUT = 0.2


def shutdown_process(
proc: subprocess.Popen[bytes],
Expand Down Expand Up @@ -64,8 +67,8 @@ def popen(
silent: bool = False,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = 0.3,
terminate_timeout: float | None = 0.2,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> tuple[int, str]:
if silent and stdout is not None:
raise ValueError(
Expand Down
175 changes: 143 additions & 32 deletions nox/sessions.py
Expand Up @@ -21,6 +21,7 @@
import os
import pathlib
import re
import subprocess
import sys
import unicodedata
from collections.abc import (
Expand All @@ -41,9 +42,13 @@
import nox.virtualenv
from nox._decorators import Func
from nox.logger import logger
from nox.popen import DEFAULT_INTERRUPT_TIMEOUT, DEFAULT_TERMINATE_TIMEOUT
from nox.virtualenv import UV, CondaEnv, PassthroughEnv, ProcessEnv, VirtualEnv

if TYPE_CHECKING:
from typing import IO

from nox.command import ExternalType
from nox.manifest import Manifest


Expand Down Expand Up @@ -269,13 +274,11 @@ def chdir(self, dir: str | os.PathLike[str]) -> _WorkingDirContext:
cd = chdir
"""An alias for :meth:`chdir`."""

def _run_func(
self, func: Callable[..., Any], args: Iterable[Any], kwargs: Mapping[str, Any]
) -> Any:
def _run_func(self, func: Callable[..., Any], args: Iterable[Any]) -> Any:
"""Legacy support for running a function through :func`run`."""
self.log(f"{func}(args={args!r}, kwargs={kwargs!r})")
self.log(f"{func}(args={args!r})")
try:
return func(*args, **kwargs)
return func(*args)
except Exception as e:
logger.exception(f"Function {func!r} raised {e!r}.")
raise nox.command.CommandFailed() from e
Expand All @@ -285,7 +288,14 @@ def run(
*args: str | os.PathLike[str],
env: Mapping[str, str | None] | None = None,
include_outer_env: bool = True,
**kwargs: Any,
silent: bool = False,
success_codes: Iterable[int] | None = None,
log: bool = True,
external: ExternalType | None = None,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> Any | None:
"""Run a command.
Expand Down Expand Up @@ -402,15 +412,29 @@ def run(
*args,
env=env,
include_outer_env=include_outer_env,
**kwargs,
silent=silent,
success_codes=success_codes,
log=log,
external=external,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def run_install(
self,
*args: str | os.PathLike[str],
env: Mapping[str, str | None] | None = None,
include_outer_env: bool = True,
**kwargs: Any,
silent: bool = False,
success_codes: Iterable[int] | None = None,
log: bool = True,
external: ExternalType | None = None,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> Any | None:
"""Run a command in the install step.
Expand Down Expand Up @@ -470,62 +494,116 @@ def run_install(
*args,
env=env,
include_outer_env=include_outer_env,
**kwargs,
silent=silent,
success_codes=success_codes,
log=log,
external=external,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def run_always(
self,
*args: str | os.PathLike[str],
env: Mapping[str, str | None] | None = None,
include_outer_env: bool = True,
**kwargs: Any,
silent: bool = False,
success_codes: Iterable[int] | None = None,
log: bool = True,
external: ExternalType | None = None,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> Any | None:
"""This is an alias to ``run_install``, which better describes the use case.
:meta private:
"""

return self.run_install(
*args, env=env, include_outer_env=include_outer_env, **kwargs
*args,
env=env,
include_outer_env=include_outer_env,
silent=silent,
success_codes=success_codes,
log=log,
external=external,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def _run(
self,
*args: str | os.PathLike[str],
env: Mapping[str, str | None] | None = None,
include_outer_env: bool = True,
**kwargs: Any,
include_outer_env: bool,
silent: bool,
success_codes: Iterable[int] | None,
log: bool,
external: ExternalType | None,
stdout: int | IO[str] | None,
stderr: int | IO[str],
interrupt_timeout: float | None,
terminate_timeout: float | None,
) -> Any:
"""Like run(), except that it runs even if --install-only is provided."""
# Legacy support - run a function given.
if callable(args[0]):
return self._run_func(args[0], args[1:], kwargs) # type: ignore[unreachable]
return self._run_func(args[0], args[1:]) # type: ignore[unreachable]

# Combine the env argument with our virtualenv's env vars.
if include_outer_env:
overlay_env = env or {}
env = {**self.env, **overlay_env}

# If --error-on-external-run is specified, error on external programs.
if self._runner.global_config.error_on_external_run:
kwargs.setdefault("external", "error")
if self._runner.global_config.error_on_external_run and external is None:
external = "error"

# Allow all external programs when running outside a sandbox.
if not self.virtualenv.is_sandboxed:
kwargs["external"] = True
if (
not self.virtualenv.is_sandboxed
or args[0] in self.virtualenv.allowed_globals
):
external = True

if args[0] in self.virtualenv.allowed_globals:
kwargs["external"] = True
if external is None:
external = False

# Run a shell command.
return nox.command.run(args, env=env, paths=self.bin_paths, **kwargs)
return nox.command.run(
args,
env=env,
paths=self.bin_paths,
silent=silent,
success_codes=success_codes,
log=log,
external=external,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def conda_install(
self,
*args: str,
auto_offline: bool = True,
channel: str | Sequence[str] = "",
**kwargs: Any,
env: Mapping[str, str] | None = None,
include_outer_env: bool = True,
silent: bool | None = None,
success_codes: Iterable[int] | None = None,
log: bool = True,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> None:
"""Install invokes `conda install`_ to install packages inside of the
session's environment.
Expand Down Expand Up @@ -584,8 +662,8 @@ def conda_install(
# Escape args that should be (conda-specific; pip install does not need this)
args = _dblquote_pkg_install_args(args)

if "silent" not in kwargs:
kwargs["silent"] = True
if silent is None:
silent = True

extraopts: list[str] = []
if auto_offline and venv.is_offline():
Expand All @@ -608,11 +686,32 @@ def conda_install(
*extraopts,
*prefix_args,
*args,
env=env,
include_outer_env=include_outer_env,
silent=silent,
success_codes=success_codes,
log=log,
external="error",
**kwargs,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def install(self, *args: str, **kwargs: Any) -> None:
def install(
self,
*args: str,
env: Mapping[str, str] | None = None,
include_outer_env: bool = True,
silent: bool | None = None,
success_codes: Iterable[int] | None = None,
log: bool = True,
external: ExternalType | None = None,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] = subprocess.STDOUT,
interrupt_timeout: float | None = DEFAULT_INTERRUPT_TIMEOUT,
terminate_timeout: float | None = DEFAULT_TERMINATE_TIMEOUT,
) -> None:
"""Install invokes `pip`_ to install packages inside of the session's
virtualenv.
Expand Down Expand Up @@ -666,15 +765,27 @@ def install(self, *args: str, **kwargs: Any) -> None:
if self._runner.global_config.no_install and venv._reused:
return

if "silent" not in kwargs:
kwargs["silent"] = True
if silent is None:
silent = True

if isinstance(venv, VirtualEnv) and venv.venv_backend == "uv":
self._run(UV, "pip", "install", *args, external="error", **kwargs)
cmd = [UV, "pip", "install"]
else:
self._run(
"python", "-m", "pip", "install", *args, external="error", **kwargs
)
cmd = ["python", "-m", "pip", "install"]
self._run(
*cmd,
*args,
env=env,
include_outer_env=include_outer_env,
external="error",
silent=silent,
success_codes=success_codes,
log=log,
stdout=stdout,
stderr=stderr,
interrupt_timeout=interrupt_timeout,
terminate_timeout=terminate_timeout,
)

def notify(
self,
Expand Down
10 changes: 9 additions & 1 deletion tests/test_command.py
Expand Up @@ -195,7 +195,15 @@ def test_run_path_existent(tmp_path: Path):
with mock.patch("nox.command.popen") as mock_command:
mock_command.return_value = (0, "")
nox.command.run([executable_name], silent=True, paths=[str(tmp_path)])
mock_command.assert_called_with([str(executable)], env=mock.ANY, silent=True)
mock_command.assert_called_with(
[str(executable)],
env=None,
silent=True,
stdout=None,
stderr=subprocess.STDOUT,
interrupt_timeout=0.3,
terminate_timeout=0.2,
)


def test_run_external_warns(tmpdir, caplog):
Expand Down

0 comments on commit d3dd1f8

Please sign in to comment.