xonsh/tests/test_vox.py

"""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)