mirror of
https://github.com/xonsh/xonsh.git
synced 2025-03-04 08:24:40 +01:00

* todo: * refactor: remove usage of singleton in cmd-cache * refactor: pass aliases/env to commands_cache on creation reduce the usage of singletons * fix: setting up mockups for tests * feat: add $XONSH_CACHE_DIR * refactor: use cache path to stire cmds-cache * chore: todo * feat: efficient cache commands per path and modified time * refactor: move aliases to commands cache now XonshSession is not needed to setup cmds-cache * refactor: update tests * fix: type annotation * refactor: init aliases at commands-cache * test: update failing tests * fix: handle paths mtime changing * docs: add news item * refactor: remove $COMMANDS_CACHE_SIZE_WARNING * fix: loading on windows fails because of setup sequence * fix: failing tests
470 lines
12 KiB
Python
470 lines
12 KiB
Python
"""Vox tests"""
|
|
import io
|
|
import os
|
|
import pathlib
|
|
import stat
|
|
import subprocess as sp
|
|
import sys
|
|
import types
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
from py.path import local
|
|
|
|
from xonsh.platform import ON_WINDOWS
|
|
from xonsh.pytest.tools import skip_if_on_conda, skip_if_on_msys
|
|
from xontrib.voxapi import Vox, _get_vox_default_interpreter
|
|
|
|
if TYPE_CHECKING:
|
|
from pytest_subprocess import FakeProcess
|
|
|
|
from xontrib.vox import VoxHandler
|
|
|
|
|
|
@pytest.fixture
|
|
def venv_home(tmpdir, xession):
|
|
"""Path where VENVs are created"""
|
|
home = tmpdir / "venvs"
|
|
home.ensure_dir()
|
|
# Set up an isolated venv home
|
|
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
|
|
xession.env["PWD"] = os.getcwd()
|
|
xession.env["DIRSTACK_SIZE"] = 10
|
|
xession.env["PATH"] = []
|
|
xession.env["XONSH_SHOW_TRACEBACK"] = True
|
|
|
|
load_xontrib("vox")
|
|
vox = xession.aliases["vox"]
|
|
return vox
|
|
|
|
|
|
@pytest.fixture
|
|
def record_events(xession):
|
|
class Listener:
|
|
def __init__(self):
|
|
self.last = None
|
|
|
|
def listener(self, name):
|
|
def _wrapper(**kwargs):
|
|
self.last = (name,) + tuple(kwargs.values())
|
|
|
|
return _wrapper
|
|
|
|
def __call__(self, *events: str):
|
|
for name in events:
|
|
event = getattr(xession.builtins.events, name)
|
|
event(self.listener(name))
|
|
|
|
yield Listener()
|
|
|
|
|
|
def test_vox_flow(xession, vox, record_events, venv_home):
|
|
"""
|
|
Creates a virtual environment, gets it, enumerates it, and then deletes it.
|
|
"""
|
|
|
|
record_events(
|
|
"vox_on_create", "vox_on_delete", "vox_on_activate", "vox_on_deactivate"
|
|
)
|
|
|
|
vox(["create", "spam"])
|
|
assert stat.S_ISDIR(venv_home.join("spam").stat().mode)
|
|
assert record_events.last == ("vox_on_create", "spam")
|
|
|
|
ve = vox.vox["spam"]
|
|
assert ve.env == str(venv_home.join("spam"))
|
|
assert os.path.isdir(ve.bin)
|
|
|
|
assert "spam" in vox.vox
|
|
assert "spam" in list(vox.vox)
|
|
|
|
# activate
|
|
vox(["activate", "spam"])
|
|
assert xession.env["VIRTUAL_ENV"] == vox.vox["spam"].env
|
|
assert record_events.last == ("vox_on_activate", "spam", str(ve.env))
|
|
|
|
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 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
|
|
vox(["rm", "spam", "--force"])
|
|
assert not venv_home.join("spam").check()
|
|
assert record_events.last == ("vox_on_delete", "spam")
|
|
|
|
|
|
def test_activate_non_vox_venv(xession, vox, record_events, venv_proc, venv_home):
|
|
"""
|
|
Create a virtual environment using Python's built-in venv module
|
|
(not in VIRTUALENV_HOME) and verify that vox can activate it correctly.
|
|
"""
|
|
xession.env["PATH"] = []
|
|
|
|
record_events("vox_on_activate", "vox_on_deactivate")
|
|
|
|
with venv_home.as_cwd():
|
|
venv_dirname = "venv"
|
|
sp.run([sys.executable, "-m", "venv", venv_dirname])
|
|
vox(["activate", venv_dirname])
|
|
vxv = vox.vox[venv_dirname]
|
|
|
|
env = xession.env
|
|
assert os.path.isabs(vxv.bin)
|
|
assert env["PATH"][0] == vxv.bin
|
|
assert os.path.isabs(vxv.env)
|
|
assert env["VIRTUAL_ENV"] == vxv.env
|
|
assert record_events.last == (
|
|
"vox_on_activate",
|
|
venv_dirname,
|
|
str(pathlib.Path(str(venv_home)) / venv_dirname),
|
|
)
|
|
|
|
vox(["deactivate"])
|
|
assert not env["PATH"]
|
|
assert "VIRTUAL_ENV" not in env
|
|
assert record_events.last == (
|
|
"vox_on_deactivate",
|
|
venv_dirname,
|
|
str(pathlib.Path(str(venv_home)) / venv_dirname),
|
|
)
|
|
|
|
|
|
@skip_if_on_msys
|
|
@skip_if_on_conda
|
|
def test_path(xession, vox, a_venv):
|
|
"""
|
|
Test to make sure Vox properly activates and deactivates by examining $PATH
|
|
"""
|
|
oldpath = list(xession.env["PATH"])
|
|
vox(["activate", a_venv.basename])
|
|
|
|
assert oldpath != xession.env["PATH"]
|
|
|
|
vox.deactivate()
|
|
|
|
assert oldpath == xession.env["PATH"]
|
|
|
|
|
|
def test_crud_subdir(xession, venv_home, venv_proc):
|
|
"""
|
|
Creates a virtual environment, gets it, enumerates it, and then deletes it.
|
|
"""
|
|
|
|
vox = Vox(force_removals=True)
|
|
vox.create("spam/eggs")
|
|
assert stat.S_ISDIR(venv_home.join("spam", "eggs").stat().mode)
|
|
|
|
ve = vox["spam/eggs"]
|
|
assert ve.env == str(venv_home.join("spam", "eggs"))
|
|
assert os.path.isdir(ve.bin)
|
|
|
|
assert "spam/eggs" in vox
|
|
assert "spam" not in vox
|
|
|
|
# assert 'spam/eggs' in list(vox) # This is NOT true on Windows
|
|
assert "spam" not in list(vox)
|
|
|
|
del vox["spam/eggs"]
|
|
|
|
assert not venv_home.join("spam", "eggs").check()
|
|
|
|
|
|
def test_crud_path(xession, tmpdir, venv_proc):
|
|
"""
|
|
Creates a virtual environment, gets it, enumerates it, and then deletes it.
|
|
"""
|
|
tmp = str(tmpdir)
|
|
|
|
vox = Vox(force_removals=True)
|
|
vox.create(tmp)
|
|
assert stat.S_ISDIR(tmpdir.join("lib").stat().mode)
|
|
|
|
ve = vox[tmp]
|
|
assert ve.env == str(tmp)
|
|
assert os.path.isdir(ve.bin)
|
|
|
|
del vox[tmp]
|
|
|
|
assert not tmpdir.check()
|
|
|
|
|
|
@skip_if_on_msys
|
|
@skip_if_on_conda
|
|
def test_reserved_names(xession, tmpdir):
|
|
"""
|
|
Tests that reserved words are disallowed.
|
|
"""
|
|
xession.env["VIRTUALENV_HOME"] = str(tmpdir)
|
|
|
|
vox = Vox()
|
|
with pytest.raises(ValueError):
|
|
if ON_WINDOWS:
|
|
vox.create("Scripts")
|
|
else:
|
|
vox.create("bin")
|
|
|
|
with pytest.raises(ValueError):
|
|
if ON_WINDOWS:
|
|
vox.create("spameggs/Scripts")
|
|
else:
|
|
vox.create("spameggs/bin")
|
|
|
|
|
|
@pytest.mark.parametrize("registered", [False, True])
|
|
def test_autovox(xession, vox, a_venv, load_xontrib, registered):
|
|
"""
|
|
Tests that autovox works
|
|
"""
|
|
from xonsh.dirstack import popd, pushd
|
|
|
|
# Makes sure that event handlers are registered
|
|
load_xontrib("autovox")
|
|
|
|
env_name = a_venv.basename
|
|
env_path = str(a_venv)
|
|
|
|
# init properly
|
|
assert vox.parser
|
|
|
|
def policy(path, **_):
|
|
if str(path) == env_path:
|
|
return env_name
|
|
|
|
if registered:
|
|
xession.builtins.events.autovox_policy(policy)
|
|
|
|
pushd([env_path])
|
|
value = env_name if registered else None
|
|
assert vox.vox.active() == value
|
|
popd([])
|
|
|
|
|
|
@pytest.fixture
|
|
def create_venv(venv_proc):
|
|
vox = Vox(force_removals=True)
|
|
|
|
def wrapped(name):
|
|
vox.create(name)
|
|
return local(vox[name].env)
|
|
|
|
return wrapped
|
|
|
|
|
|
@pytest.fixture
|
|
def venvs(venv_home, create_venv):
|
|
"""Create virtualenv with names venv0, venv1"""
|
|
from xonsh.dirstack import popd, pushd
|
|
|
|
pushd([str(venv_home)])
|
|
yield [create_venv(f"venv{idx}") for idx in range(2)]
|
|
popd([])
|
|
|
|
|
|
@pytest.fixture
|
|
def a_venv(create_venv):
|
|
return create_venv("venv0")
|
|
|
|
|
|
@pytest.fixture
|
|
def patched_cmd_cache(xession, vox, monkeypatch):
|
|
cc = xession.commands_cache
|
|
|
|
def no_change(self, *_):
|
|
return False, False
|
|
|
|
monkeypatch.setattr(cc, "_check_changes", types.MethodType(no_change, cc))
|
|
bins = {path: (path, False) for path in _PY_BINS}
|
|
monkeypatch.setattr(cc, "_cmds_cache", bins)
|
|
yield cc
|
|
|
|
|
|
_VENV_NAMES = {"venv1", "venv1/", "venv0/", "venv0"}
|
|
if ON_WINDOWS:
|
|
_VENV_NAMES = {"venv1\\", "venv0\\", "venv0", "venv1"}
|
|
|
|
_HELP_OPTS = {
|
|
"-h",
|
|
"--help",
|
|
}
|
|
_PY_BINS = {"/bin/python2", "/bin/python3"}
|
|
|
|
_VOX_NEW_OPTS = {
|
|
"--ssp",
|
|
"--system-site-packages",
|
|
"--without-pip",
|
|
}.union(_HELP_OPTS)
|
|
|
|
if ON_WINDOWS:
|
|
_VOX_NEW_OPTS.add("--symlinks")
|
|
else:
|
|
_VOX_NEW_OPTS.add("--copies")
|
|
|
|
_VOX_RM_OPTS = {"-f", "--force"}.union(_HELP_OPTS)
|
|
|
|
|
|
class TestVoxCompletions:
|
|
@pytest.fixture
|
|
def check(self, check_completer, xession, vox):
|
|
def wrapped(cmd, positionals, options=None):
|
|
for k in list(xession.completers):
|
|
if k != "alias":
|
|
xession.completers.pop(k)
|
|
assert check_completer(cmd) == positionals
|
|
xession.env["ALIAS_COMPLETIONS_OPTIONS_BY_DEFAULT"] = True
|
|
if options:
|
|
assert check_completer(cmd) == positionals.union(options)
|
|
|
|
return wrapped
|
|
|
|
@pytest.mark.parametrize(
|
|
"args, positionals, opts",
|
|
[
|
|
(
|
|
"vox",
|
|
{
|
|
"delete",
|
|
"new",
|
|
"remove",
|
|
"del",
|
|
"workon",
|
|
"list",
|
|
"exit",
|
|
"info",
|
|
"ls",
|
|
"rm",
|
|
"deactivate",
|
|
"activate",
|
|
"enter",
|
|
"create",
|
|
"project-get",
|
|
"project-set",
|
|
"runin",
|
|
"runin-all",
|
|
"toggle-ssp",
|
|
"wipe",
|
|
"upgrade",
|
|
},
|
|
_HELP_OPTS,
|
|
),
|
|
(
|
|
"vox create",
|
|
set(),
|
|
_VOX_NEW_OPTS.union(
|
|
{
|
|
"-a",
|
|
"--activate",
|
|
"--wp",
|
|
"--without-pip",
|
|
"-p",
|
|
"--interpreter",
|
|
"-i",
|
|
"--install",
|
|
"-l",
|
|
"--link",
|
|
"--link-project",
|
|
"-r",
|
|
"--requirements",
|
|
"-t",
|
|
"--temp",
|
|
"--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)
|