mirrors/xonsh
1
0
Fork 0
mirror of https://github.com/xonsh/xonsh.git synced 2025-03-04 00:14:41 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Fix vox env dir (#4736)

Browse source 
* refactor: use stdout passed from function

better handling of errors

* test: no need to skip these on windows

* test: load local plugin

* test: update vox-runin

* test: speedup vox tests

we dont need to test venv, only calls made to it via subprocess

* test: pip.exe on windows
This commit is contained in:
Noorhteen Raja NJ 2022-03-27 17:12:52 +05:30 • committed by GitHub
parent 5ab9812f82
commit 011c5c00e9
Failed to generate hash of commit
5 changed files with 286 additions and 203 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/conftest.py
View file

@ -0,0 +1,2 @@
# when xonsh is not installed in the current env
# pytest_plugins = ("xonsh.pytest.plugin",)

2
tests/test_aliases.py
View file

@ -6,7 +6,6 @@ import os
import pytest import pytest
from xonsh.aliases import Aliases, ExecAlias from xonsh.aliases import Aliases, ExecAlias
from xonsh.pytest.tools import skip_if_on_windows
def cd(args, stdin=None, **kwargs): def cd(args, stdin=None, **kwargs):
@ -52,7 +51,6 @@ def test_eval_recursive(xession):
assert ales.get("color_ls") == ["ls", "- -", "--color=true"] assert ales.get("color_ls") == ["ls", "- -", "--color=true"]
@skip_if_on_windows
def test_eval_recursive_callable_partial(xonsh_execer, xession): def test_eval_recursive_callable_partial(xonsh_execer, xession):
ales = make_aliases() ales = make_aliases()
xession.env["HOME"] = os.path.expanduser("~") xession.env["HOME"] = os.path.expanduser("~")

337
tests/test_vox.py
View file

@ -1,4 +1,5 @@
"""Vox tests""" """Vox tests"""
import io
import os import os
import pathlib import pathlib
import stat import stat
@ -8,21 +9,75 @@ import types
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import pytest import pytest
from py.path import local
from xonsh.platform import ON_WINDOWS from xonsh.platform import ON_WINDOWS
from xonsh.pytest.tools import skip_if_on_conda, skip_if_on_msys from xonsh.pytest.tools import skip_if_on_conda, skip_if_on_msys
from xontrib.voxapi import Vox from xontrib.voxapi import Vox, _get_vox_default_interpreter
if TYPE_CHECKING: if TYPE_CHECKING:
from pytest_subprocess import FakeProcess
from xontrib.vox import VoxHandler from xontrib.vox import VoxHandler
@pytest.fixture @pytest.fixture
def vox(xession, tmpdir, load_xontrib) -> "VoxHandler": def venv_home(tmpdir, xession):
"""vox Alias function""" """Path where VENVs are created"""
home = tmpdir / "venvs"
home.ensure_dir()
# Set up an isolated venv home # Set up an isolated venv home
xession.env["VIRTUALENV_HOME"] = str(tmpdir) xession.env["VIRTUALENV_HOME"] = str(home)
return home
@pytest.fixture
def venv_proc(fake_process: "FakeProcess", venv_home):
def version_handle(process):
ver = str(sys.version).split()[0]
process.stdout.write(f"Python {ver}")
def venv_handle(process):
env_path = local(process.args[3])
(env_path / "lib").ensure_dir()
bin_path = env_path / ("Scripts" if ON_WINDOWS else "bin")
(bin_path / "python").write("", ensure=True)
(bin_path / "python.exe").write("", ensure=True)
for file in bin_path.listdir():
st = os.stat(str(file))
os.chmod(str(file), st.st_mode | stat.S_IEXEC)
for pip_name in ["pip", "pip.exe"]:
fake_process.register(
[str(bin_path / pip_name), "freeze", fake_process.any()], stdout=""
)
# will be used by `vox runin`
fake_process.register(
[pip_name, "--version"],
stdout=f"pip 22.0.4 from {env_path}/lib/python3.10/site-packages/pip (python 3.10)",
)
fake_process.keep_last_process(True)
return env_path
def get_interpreters():
interpreter = _get_vox_default_interpreter()
yield interpreter
if sys.executable != interpreter:
yield sys.executable
for cmd in get_interpreters():
fake_process.register([cmd, "--version"], callback=version_handle)
venv = (cmd, "-m", "venv")
fake_process.register([*venv, fake_process.any(min=1)], callback=venv_handle)
fake_process.keep_last_process(True)
return fake_process
@pytest.fixture
def vox(xession, load_xontrib, venv_proc) -> "VoxHandler":
"""vox Alias function"""
# Set up enough environment for xonsh to function # Set up enough environment for xonsh to function
xession.env["PWD"] = os.getcwd() xession.env["PWD"] = os.getcwd()
@ -31,8 +86,8 @@ def vox(xession, tmpdir, load_xontrib) -> "VoxHandler":
xession.env["XONSH_SHOW_TRACEBACK"] = True xession.env["XONSH_SHOW_TRACEBACK"] = True
load_xontrib("vox") load_xontrib("vox")
vox = xession.aliases["vox"]
yield xession.aliases["vox"] return vox
@pytest.fixture @pytest.fixture
@ -55,46 +110,67 @@ def record_events(xession):
yield Listener() yield Listener()
@skip_if_on_msys def test_vox_flow(xession, vox, record_events, venv_home):
@skip_if_on_conda
def test_vox_flow(xession, vox, record_events, tmpdir):
""" """
Creates a virtual environment, gets it, enumerates it, and then deletes it. Creates a virtual environment, gets it, enumerates it, and then deletes it.
""" """
xession.env["VIRTUALENV_HOME"] = str(tmpdir)
record_events( record_events(
"vox_on_create", "vox_on_delete", "vox_on_activate", "vox_on_deactivate" "vox_on_create", "vox_on_delete", "vox_on_activate", "vox_on_deactivate"
) )
vox(["create", "spam"]) vox(["create", "spam"])
assert stat.S_ISDIR(tmpdir.join("spam").stat().mode) assert stat.S_ISDIR(venv_home.join("spam").stat().mode)
assert record_events.last == ("vox_on_create", "spam") assert record_events.last == ("vox_on_create", "spam")
ve = vox.vox["spam"] ve = vox.vox["spam"]
assert ve.env == str(tmpdir.join("spam")) assert ve.env == str(venv_home.join("spam"))
assert os.path.isdir(ve.bin) assert os.path.isdir(ve.bin)
assert "spam" in vox.vox assert "spam" in vox.vox
assert "spam" in list(vox.vox) assert "spam" in list(vox.vox)
# activate/deactivate # activate
vox(["activate", "spam"]) vox(["activate", "spam"])
assert xession.env["VIRTUAL_ENV"] == vox.vox["spam"].env assert xession.env["VIRTUAL_ENV"] == vox.vox["spam"].env
assert record_events.last == ("vox_on_activate", "spam", str(ve.env)) assert record_events.last == ("vox_on_activate", "spam", str(ve.env))
vox.deactivate()
out = io.StringIO()
# info
vox(["info"], stdout=out)
assert "spam" in out.getvalue()
out.seek(0)
# list
vox(["list"], stdout=out)
print(out.getvalue())
assert "spam" in out.getvalue()
out.seek(0)
# wipe
vox(["wipe"], stdout=out)
print(out.getvalue())
assert "Nothing to remove" in out.getvalue()
out.seek(0)
# deactivate
vox(["deactivate"])
assert "VIRTUAL_ENV" not in xession.env assert "VIRTUAL_ENV" not in xession.env
assert record_events.last == ("vox_on_deactivate", "spam", str(ve.env)) assert record_events.last == ("vox_on_deactivate", "spam", str(ve.env))
# runin
vox(["runin", "spam", "pip", "--version"], stdout=out)
print(out.getvalue())
assert "spam" in out.getvalue()
out.seek(0)
# removal # removal
vox(["rm", "spam", "--force"]) vox(["rm", "spam", "--force"])
assert not tmpdir.join("spam").check() assert not venv_home.join("spam").check()
assert record_events.last == ("vox_on_delete", "spam") assert record_events.last == ("vox_on_delete", "spam")
@skip_if_on_msys def test_activate_non_vox_venv(xession, vox, record_events, venv_proc, venv_home):
@skip_if_on_conda
def test_activate_non_vox_venv(xession, vox, record_events, tmpdir):
""" """
Create a virtual environment using Python's built-in venv module Create a virtual environment using Python's built-in venv module
(not in VIRTUALENV_HOME) and verify that vox can activate it correctly. (not in VIRTUALENV_HOME) and verify that vox can activate it correctly.
@ -103,7 +179,7 @@ def test_activate_non_vox_venv(xession, vox, record_events, tmpdir):
record_events("vox_on_activate", "vox_on_deactivate") record_events("vox_on_activate", "vox_on_deactivate")
with tmpdir.as_cwd(): with venv_home.as_cwd():
venv_dirname = "venv" venv_dirname = "venv"
sp.run([sys.executable, "-m", "venv", venv_dirname]) sp.run([sys.executable, "-m", "venv", venv_dirname])
vox(["activate", venv_dirname]) vox(["activate", venv_dirname])
@ -117,7 +193,7 @@ def test_activate_non_vox_venv(xession, vox, record_events, tmpdir):
assert record_events.last == ( assert record_events.last == (
"vox_on_activate", "vox_on_activate",
venv_dirname, venv_dirname,
str(pathlib.Path(str(tmpdir)) / venv_dirname), str(pathlib.Path(str(venv_home)) / venv_dirname),
) )
vox(["deactivate"]) vox(["deactivate"])
@ -126,13 +202,13 @@ def test_activate_non_vox_venv(xession, vox, record_events, tmpdir):
assert record_events.last == ( assert record_events.last == (
"vox_on_deactivate", "vox_on_deactivate",
venv_dirname, venv_dirname,
str(pathlib.Path(str(tmpdir)) / venv_dirname), str(pathlib.Path(str(venv_home)) / venv_dirname),
) )
@skip_if_on_msys @skip_if_on_msys
@skip_if_on_conda @skip_if_on_conda
def test_path(xession, vox, tmpdir, a_venv): def test_path(xession, vox, a_venv):
""" """
Test to make sure Vox properly activates and deactivates by examining $PATH Test to make sure Vox properly activates and deactivates by examining $PATH
""" """
@ -146,20 +222,17 @@ def test_path(xession, vox, tmpdir, a_venv):
assert oldpath == xession.env["PATH"] assert oldpath == xession.env["PATH"]
@skip_if_on_msys def test_crud_subdir(xession, venv_home, venv_proc):
@skip_if_on_conda
def test_crud_subdir(xession, tmpdir):
""" """
Creates a virtual environment, gets it, enumerates it, and then deletes it. Creates a virtual environment, gets it, enumerates it, and then deletes it.
""" """
xession.env["VIRTUALENV_HOME"] = str(tmpdir)
vox = Vox(force_removals=True) vox = Vox(force_removals=True)
vox.create("spam/eggs") vox.create("spam/eggs")
assert stat.S_ISDIR(tmpdir.join("spam", "eggs").stat().mode) assert stat.S_ISDIR(venv_home.join("spam", "eggs").stat().mode)
ve = vox["spam/eggs"] ve = vox["spam/eggs"]
assert ve.env == str(tmpdir.join("spam", "eggs")) assert ve.env == str(venv_home.join("spam", "eggs"))
assert os.path.isdir(ve.bin) assert os.path.isdir(ve.bin)
assert "spam/eggs" in vox assert "spam/eggs" in vox
@ -170,16 +243,14 @@ def test_crud_subdir(xession, tmpdir):
del vox["spam/eggs"] del vox["spam/eggs"]
assert not tmpdir.join("spam", "eggs").check() assert not venv_home.join("spam", "eggs").check()
@skip_if_on_msys def test_crud_path(xession, tmpdir, venv_proc):
@skip_if_on_conda
def test_crud_path(xession, tmpdir):
""" """
Creates a virtual environment, gets it, enumerates it, and then deletes it. Creates a virtual environment, gets it, enumerates it, and then deletes it.
""" """
tmp = pathlib.Path(str(tmpdir)) tmp = str(tmpdir)
vox = Vox(force_removals=True) vox = Vox(force_removals=True)
vox.create(tmp) vox.create(tmp)
@ -217,9 +288,7 @@ def test_reserved_names(xession, tmpdir):
@pytest.mark.parametrize("registered", [False, True]) @pytest.mark.parametrize("registered", [False, True])
@skip_if_on_msys def test_autovox(xession, vox, a_venv, load_xontrib, registered):
@skip_if_on_conda
def test_autovox(xession, tmpdir, vox, a_venv, load_xontrib, registered):
""" """
Tests that autovox works Tests that autovox works
""" """
@ -248,39 +317,33 @@ def test_autovox(xession, tmpdir, vox, a_venv, load_xontrib, registered):
@pytest.fixture @pytest.fixture
def venv_home(tmpdir): def create_venv(venv_proc):
"""Path where VENVs are created""" vox = Vox(force_removals=True)
return tmpdir
def wrapped(name):
vox.create(name)
return local(vox[name].env)
return wrapped
@pytest.fixture @pytest.fixture
def venvs(venv_home): def venvs(venv_home, create_venv):
"""Create virtualenv with names venv0, venv1""" """Create virtualenv with names venv0, venv1"""
from xonsh.dirstack import popd, pushd from xonsh.dirstack import popd, pushd
pushd([str(venv_home)]) pushd([str(venv_home)])
paths = [] yield [create_venv(f"venv{idx}") for idx in range(2)]
for idx in range(2):
env_path = venv_home / f"venv{idx}"
bin_path = env_path / ("Scripts" if ON_WINDOWS else "bin")
paths.append(env_path)
(bin_path / "python").write("", ensure=True)
(bin_path / "python.exe").write("", ensure=True)
for file in bin_path.listdir():
st = os.stat(str(file))
os.chmod(str(file), st.st_mode | stat.S_IEXEC)
yield paths
popd([]) popd([])
@pytest.fixture @pytest.fixture
def a_venv(venvs): def a_venv(create_venv):
return venvs[0] return create_venv("venv0")
@pytest.fixture @pytest.fixture
def patched_cmd_cache(xession, vox, venvs, monkeypatch): def patched_cmd_cache(xession, vox, monkeypatch):
cc = xession.commands_cache cc = xession.commands_cache
def no_change(self, *_): def no_change(self, *_):
@ -289,7 +352,7 @@ def patched_cmd_cache(xession, vox, venvs, monkeypatch):
monkeypatch.setattr(cc, "_check_changes", types.MethodType(no_change, cc)) monkeypatch.setattr(cc, "_check_changes", types.MethodType(no_change, cc))
monkeypatch.setattr(cc, "_update_cmds_cache", types.MethodType(no_change, cc)) monkeypatch.setattr(cc, "_update_cmds_cache", types.MethodType(no_change, cc))
bins = {path: (path, False) for path in _PY_BINS} bins = {path: (path, False) for path in _PY_BINS}
cc._cmds_cache.update(bins) cc._cmds_cache = bins
yield cc yield cc
@ -308,7 +371,6 @@ _VOX_NEW_OPTS = {
"--system-site-packages", "--system-site-packages",
"--without-pip", "--without-pip",
}.union(_HELP_OPTS) }.union(_HELP_OPTS)
_VOX_NEW_EXP = _PY_BINS.union(_VOX_NEW_OPTS)
if ON_WINDOWS: if ON_WINDOWS:
_VOX_NEW_OPTS.add("--symlinks") _VOX_NEW_OPTS.add("--symlinks")
@ -318,77 +380,92 @@ else:
_VOX_RM_OPTS = {"-f", "--force"}.union(_HELP_OPTS) _VOX_RM_OPTS = {"-f", "--force"}.union(_HELP_OPTS)
@pytest.mark.parametrize( class TestVoxCompletions:
"args, positionals, opts", @pytest.fixture
[ def check(self, check_completer, xession, vox):
( def wrapped(cmd, positionals, options=None):
"vox", for k in list(xession.completers):
{ if k != "alias":
"delete", xession.completers.pop(k)
"new", assert check_completer(cmd) == positionals
"remove", xession.env["ALIAS_COMPLETIONS_OPTIONS_BY_DEFAULT"] = True
"del", if options:
"workon", assert check_completer(cmd) == positionals.union(options)
"list",
"exit", return wrapped
"info",
"ls", @pytest.mark.parametrize(
"rm", "args, positionals, opts",
"deactivate", [
"activate", (
"enter", "vox",
"create",
"project-get",
"project-set",
"runin",
"runin-all",
"toggle-ssp",
"wipe",
"upgrade",
},
_HELP_OPTS,
),
(
"vox create",
set(),
_VOX_NEW_OPTS.union(
{ {
"-a", "delete",
"--activate", "new",
"--wp", "remove",
"--without-pip", "del",
"-p", "workon",
"--interpreter", "list",
"-i", "exit",
"--install", "info",
"-l", "ls",
"--link", "rm",
"--link-project", "deactivate",
"-r", "activate",
"--requirements", "enter",
"-t", "create",
"--temp", "project-get",
"--prompt", "project-set",
} "runin",
"runin-all",
"toggle-ssp",
"wipe",
"upgrade",
},
_HELP_OPTS,
), ),
), (
("vox activate", _VENV_NAMES, _HELP_OPTS.union({"-n", "--no-cd"})), "vox create",
("vox rm", _VENV_NAMES, _VOX_RM_OPTS), set(),
("vox rm venv1", _VENV_NAMES, _VOX_RM_OPTS), # pos nargs: one or more _VOX_NEW_OPTS.union(
("vox rm venv1 venv2", _VENV_NAMES, _VOX_RM_OPTS), # pos nargs: two or more {
("vox new --activate --interpreter", _PY_BINS, set()), # option after option "-a",
("vox new --interpreter", _PY_BINS, set()), # "option: first "--activate",
("vox new --activate env1 --interpreter", _PY_BINS, set()), # option after pos "--wp",
("vox new env1 --interpreter", _PY_BINS, set()), # "option: at end" "--without-pip",
("vox new env1 --interpreter=", _PY_BINS, set()), # "option: at end with "-p",
], "--interpreter",
) "-i",
def test_vox_completer( "--install",
args, check_completer, positionals, opts, xession, patched_cmd_cache, venv_home "-l",
): "--link",
xession.env["XONSH_DATA_DIR"] = venv_home "--link-project",
if positionals: "-r",
assert check_completer(args) == positionals "--requirements",
xession.env["ALIAS_COMPLETIONS_OPTIONS_BY_DEFAULT"] = True "-t",
if opts: "--temp",
assert check_completer(args) == positionals.union(opts) "--prompt",
}
),
),
("vox activate", _VENV_NAMES, _HELP_OPTS.union({"-n", "--no-cd"})),
("vox rm", _VENV_NAMES, _VOX_RM_OPTS),
("vox rm venv1", _VENV_NAMES, _VOX_RM_OPTS), # pos nargs: one or more
("vox rm venv1 venv2", _VENV_NAMES, _VOX_RM_OPTS), # pos nargs: two or more
],
)
def test_vox_commands(self, args, positionals, opts, check, venvs):
check(args, positionals, opts)
@pytest.mark.parametrize(
"args",
[
"vox new --activate --interpreter", # option after option
"vox new --interpreter", # "option: first
"vox new --activate env1 --interpreter", # option after pos
"vox new env1 --interpreter", # "option: at end"
"vox new env1 --interpreter=", # "option: at end with
],
)
def test_interpreter(self, check, args, patched_cmd_cache):
check(args, _PY_BINS)

48
xonsh/cli_utils.py
View file

@ -582,6 +582,14 @@ class ArgParserAlias:
For usage please check ``xonsh.completers.completer.py`` module. For usage please check ``xonsh.completers.completer.py`` module.
""" """
class Error(Exception):
"""Special case, when raised, the traceback will not be shown.
Instead the process with exit with error code and message"""
def __init__(self, message: str, errno=1):
super().__init__(message)
self.errno = errno
def __init__(self, threadable=True, **kwargs) -> None: def __init__(self, threadable=True, **kwargs) -> None:
if not threadable: if not threadable:
from xonsh.tools import unthreadable from xonsh.tools import unthreadable
@ -646,11 +654,11 @@ class ArgParserAlias:
def err(self, *args, **kwargs): def err(self, *args, **kwargs):
"""Write text to error stream""" """Write text to error stream"""
return self.write_to("stderr", *args, **kwargs) self.write_to("stderr", *args, **kwargs)
def out(self, *args, **kwargs): def out(self, *args, **kwargs):
"""Write text to output stream""" """Write text to output stream"""
return self.write_to("stdout", *args, **kwargs) self.write_to("stdout", *args, **kwargs)
def __call__( def __call__(
self, self,
@ -664,22 +672,26 @@ class ArgParserAlias:
): ):
self.stdout = stdout self.stdout = stdout
self.stderr = stderr self.stderr = stderr
result = dispatch( try:
self.parser, result = dispatch(
args, self.parser,
_parser=self.parser, args,
_args=args, _parser=self.parser,
_stdin=stdin, _args=args,
_stdout=stdout, _stdin=stdin,
_stderr=stderr, _stdout=stdout,
_spec=spec, _stderr=stderr,
_stack=stack, _spec=spec,
**kwargs, _stack=stack,
) **kwargs,
)
# free the reference to input/output. Otherwise it will result in errors except self.Error as ex:
self.stdout = None self.err(f"Error: {ex}")
self.stderr = None sys.exit(getattr(ex, "errno", 1))
finally:
# free the reference to input/output. Otherwise it will result in errors
self.stdout = None
self.stderr = None
return result return result

100
xontrib/vox.py
View file

@ -1,7 +1,6 @@
"""Python virtual environment manager for xonsh.""" """Python virtual environment manager for xonsh."""
import os.path import os.path
import subprocess import subprocess
import sys
import tempfile import tempfile
import typing as tp import typing as tp
from pathlib import Path from pathlib import Path
@ -11,6 +10,7 @@ import xontrib.voxapi as voxapi
from xonsh.built_ins import XSH from xonsh.built_ins import XSH
from xonsh.dirstack import pushd_fn from xonsh.dirstack import pushd_fn
from xonsh.platform import ON_WINDOWS from xonsh.platform import ON_WINDOWS
from xonsh.tools import XonshError
__all__ = () __all__ = ()
@ -119,7 +119,7 @@ class VoxHandler(xcli.ArgParserAlias):
Provides an alternative prompt prefix for this environment. Provides an alternative prompt prefix for this environment.
""" """
print("Creating environment...") self.out("Creating environment...")
if temporary: if temporary:
path = tempfile.mkdtemp(prefix=f"vox-env-{name}") path = tempfile.mkdtemp(prefix=f"vox-env-{name}")
@ -150,9 +150,9 @@ class VoxHandler(xcli.ArgParserAlias):
if activate: if activate:
self.activate(name) self.activate(name)
print(f"Environment {name!r} created and activated.\n") self.out(f"Environment {name!r} created and activated.\n")
else: else:
print( self.out(
f'Environment {name!r} created. Activate it with "vox activate {name}".\n' f'Environment {name!r} created. Activate it with "vox activate {name}".\n'
) )
@ -179,12 +179,11 @@ class VoxHandler(xcli.ArgParserAlias):
try: try:
self.vox.activate(name) self.vox.activate(name)
except KeyError: except KeyError:
self.parser.error( raise self.Error(
f'This environment doesn\'t exist. Create it with "vox new {name}".\n', f'This environment doesn\'t exist. Create it with "vox new {name}"',
) )
return None
print(f'Activated "{name}".\n') self.out(f'Activated "{name}".\n')
if not no_cd: if not no_cd:
project_dir = self._get_project_dir(name) project_dir = self._get_project_dir(name)
if project_dir: if project_dir:
@ -202,16 +201,16 @@ class VoxHandler(xcli.ArgParserAlias):
""" """
if self.vox.active() is None: if self.vox.active() is None:
self.parser.error( raise self.Error(
'No environment currently active. Activate one with "vox activate".\n', 'No environment currently active. Activate one with "vox activate".\n',
) )
env_name = self.vox.deactivate() env_name = self.vox.deactivate()
if remove: if remove:
self.vox.force_removals = force self.vox.force_removals = force
del self.vox[env_name] del self.vox[env_name]
print(f'Environment "{env_name}" deactivated and removed.\n') self.out(f'Environment "{env_name}" deactivated and removed.\n')
else: else:
print(f'Environment "{env_name}" deactivated.\n') self.out(f'Environment "{env_name}" deactivated.\n')
def list(self): def list(self):
"""List available virtual environments.""" """List available virtual environments."""
@ -219,16 +218,15 @@ class VoxHandler(xcli.ArgParserAlias):
try: try:
envs = sorted(self.vox.keys()) envs = sorted(self.vox.keys())
except PermissionError: except PermissionError:
self.parser.error("No permissions on VIRTUALENV_HOME") raise self.Error("No permissions on VIRTUALENV_HOME")
return None
if not envs: if not envs:
self.parser.error( raise self.Error(
'No environments available. Create one with "vox new".\n', 'No environments available. Create one with "vox new".\n',
) )
print("Available environments:") self.out("Available environments:")
print("\n".join(envs)) self.out("\n".join(envs))
def remove( def remove(
self, self,
@ -253,28 +251,24 @@ class VoxHandler(xcli.ArgParserAlias):
try: try:
del self.vox[name] del self.vox[name]
except voxapi.EnvironmentInUse: except voxapi.EnvironmentInUse:
self.parser.error( raise self.Error(
f'The "{name}" environment is currently active. ' f'The "{name}" environment is currently active. '
'In order to remove it, deactivate it first with "vox deactivate".\n', 'In order to remove it, deactivate it first with "vox deactivate".\n',
) )
except KeyError: except KeyError:
self.parser.error(f'"{name}" environment doesn\'t exist.\n') raise self.Error(f'"{name}" environment doesn\'t exist.\n')
else: else:
print(f'Environment "{name}" removed.') self.out(f'Environment "{name}" removed.')
print() self.out()
def _in_venv(self, env_dir: str, command: str, *args, **kwargs): def _in_venv(self, env_dir: str, command: str, *args, **kwargs):
env = XSH.env.detype() env = {**XSH.env.detype(), "VIRTUAL_ENV": env_dir}
env["VIRTUAL_ENV"] = env_dir
env["PATH"] = os.pathsep.join( bin_path = os.path.join(env_dir, self.vox.sub_dirs[0])
[ env["PATH"] = os.pathsep.join([bin_path, env["PATH"]])
os.path.join(env_dir, self.vox.sub_dirs[0]),
env["PATH"],
]
)
for key in ("PYTHONHOME", "__PYVENV_LAUNCHER__"): for key in ("PYTHONHOME", "__PYVENV_LAUNCHER__"):
if key in env: env.pop(key, None)
del env[key]
try: try:
return subprocess.check_call( return subprocess.check_call(
@ -284,7 +278,7 @@ class VoxHandler(xcli.ArgParserAlias):
# won't inherit the PATH # won't inherit the PATH
except OSError as e: except OSError as e:
if e.errno == 2: if e.errno == 2:
self.parser.error(f"Unable to find {command}") raise self.Error(f"Unable to find {command}")
raise raise
def runin( def runin(
@ -310,7 +304,7 @@ class VoxHandler(xcli.ArgParserAlias):
""" """
env_dir = self._get_env_dir(venv) env_dir = self._get_env_dir(venv)
if not args: if not args:
self.parser.error("No command is passed") raise self.Error("No command is passed")
self._in_venv(env_dir, *args) self._in_venv(env_dir, *args)
def runin_all( def runin_all(
@ -326,19 +320,18 @@ class VoxHandler(xcli.ArgParserAlias):
""" """
errors = False errors = False
for env in self.vox: for env in self.vox:
print("\n%s:" % env) self.out("\n%s:" % env)
try: try:
self.runin(env, *args) self.runin(env, *args)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
errors = True errors = True
print(e, file=sys.stderr) self.err(e)
sys.exit(errors) self.parser.exit(errors)
def _sitepackages_dir(self, venv_path: str): def _sitepackages_dir(self, venv_path: str):
env_python = self.vox.get_binary_path("python", venv_path) env_python = self.vox.get_binary_path("python", venv_path)
if not os.path.exists(env_python): if not os.path.exists(env_python):
self.parser.error("ERROR: no virtualenv active") raise self.Error("no virtualenv active")
return
return Path( return Path(
subprocess.check_output( subprocess.check_output(
@ -352,17 +345,18 @@ class VoxHandler(xcli.ArgParserAlias):
) )
def _get_env_dir(self, venv=None): def _get_env_dir(self, venv=None):
venv = venv or "..." venv = venv or ...
try: try:
env_dir = self.vox[venv].env env_dir = self.vox[venv].env
except KeyError: except KeyError:
# check whether the venv is a valid path to an environment # check whether the venv is a valid path to an environment
if os.path.exists(venv) and os.path.exists( if (
self.vox.get_binary_path("python", venv) isinstance(venv, str)
and os.path.exists(venv)
and os.path.exists(self.vox.get_binary_path("python", venv))
): ):
return venv return venv
self.parser.error("No virtualenv is found") raise XonshError("No virtualenv is found")
return
return env_dir return env_dir
def toggle_ssp(self): def toggle_ssp(self):
@ -374,10 +368,10 @@ class VoxHandler(xcli.ArgParserAlias):
ngsp_file = site.parent / "no-global-site-packages.txt" ngsp_file = site.parent / "no-global-site-packages.txt"
if ngsp_file.exists(): if ngsp_file.exists():
ngsp_file.unlink() ngsp_file.unlink()
print("Enabled global site-packages") self.out("Enabled global site-packages")
else: else:
with ngsp_file.open("w"): with ngsp_file.open("w"):
print("Disabled global site-packages") self.out("Disabled global site-packages")
def project_set( def project_set(
self, self,
@ -397,9 +391,9 @@ class VoxHandler(xcli.ArgParserAlias):
project = os.path.abspath(project_path or ".") project = os.path.abspath(project_path or ".")
if not os.path.exists(env_dir): if not os.path.exists(env_dir):
self.parser.error(f"Environment '{env_dir}' doesn't exist.") raise self.Error(f"Environment '{env_dir}' doesn't exist.")
if not os.path.isdir(project): if not os.path.isdir(project):
self.parser.error(f"{project} does not exist") raise self.Error(f"{project} does not exist")
project_file = self._get_project_file() project_file = self._get_project_file()
project_file.write_text(project) project_file.write_text(project)
@ -428,10 +422,10 @@ class VoxHandler(xcli.ArgParserAlias):
""" """
project_dir = self._get_project_dir(venv) project_dir = self._get_project_dir(venv)
if project_dir: if project_dir:
print(project_dir) self.out(project_dir)
else: else:
project_file = self._get_project_file(venv) project_file = self._get_project_file(venv)
self.parser.error( raise self.Error(
f"Corrupted or outdated: {project_file}\nDirectory: {project_dir} doesn't exist." f"Corrupted or outdated: {project_file}\nDirectory: {project_dir} doesn't exist."
) )
@ -454,11 +448,11 @@ class VoxHandler(xcli.ArgParserAlias):
ignored = sorted(all_pkgs - pkgs) ignored = sorted(all_pkgs - pkgs)
to_remove = {p.split("==")[0] for p in pkgs} to_remove = {p.split("==")[0] for p in pkgs}
if to_remove: if to_remove:
print("Ignoring:\n %s" % "\n ".join(ignored)) self.out("Ignoring:\n %s" % "\n ".join(ignored))
print("Uninstalling packages:\n %s" % "\n ".join(to_remove)) self.out("Uninstalling packages:\n %s" % "\n ".join(to_remove))
return subprocess.run([pip_bin, "uninstall", "-y", *to_remove]) return subprocess.run([pip_bin, "uninstall", "-y", *to_remove])
else: else:
print("Nothing to remove") self.out("Nothing to remove")
def info(self, venv: _venv_option = None): def info(self, venv: _venv_option = None):
"""Prints the path for the supplied env """Prints the path for the supplied env
@ -468,7 +462,7 @@ class VoxHandler(xcli.ArgParserAlias):
venv venv
name of the venv name of the venv
""" """
print(self.vox[venv or ...]) self.out(self.vox[venv or ...])
def upgrade( def upgrade(
self, self,
@ -498,7 +492,7 @@ class VoxHandler(xcli.ArgParserAlias):
venv = self.vox.upgrade( venv = self.vox.upgrade(
name or ..., symlinks=symlinks, with_pip=with_pip, interpreter=interpreter name or ..., symlinks=symlinks, with_pip=with_pip, interpreter=interpreter
) )
print(venv) self.out(venv)
XSH.aliases["vox"] = VoxHandler() XSH.aliases["vox"] = VoxHandler()