xonsh/xontrib/autovox.py
Noorhteen Raja NJ 38295a1dd9
Remove globals (#4280)
* refactor: remove usage of global variables in abbrevs.py

* chore: add flake8-mutable to prevent mutable defaults

* fix: abbrevs expand test

* refactor: add xonsh session singleton

* refactor: fix circular errors when using xonshSession as singleton

* refactor: remove black magicked builtin attributes

* style: black format tests as well

* refactor: update tests to use xonsh-session singleton

* refactor: update abbrevs to not use builtins

* test: remove DummyCommandsCache and patch orig class

* fix: failing test_command_completers

* test: use monkeypatch to update xession fixture

* fix: failing test_pipelines

* fix: failing test_main

* chore: run test suit as single invocation

* test: fix tests/test_xonsh.xsh

* refactor: remove builtins from docs/conf.py

* fix: mypy error in jobs

* fix: test error from test_main

* test: close xession error in test_command_completers

* chore: use pytest-cov for reporting coverage

this will include subprocess calls, and will increase coverage

* style:
2021-05-20 13:14:26 +03:00

107 lines
2.6 KiB
Python

"""
A framework for automatic vox.
This coordinates multiple automatic vox policies and deals with some of the
mechanics of venv searching and chdir handling.
This provides no interface for end users.
Developers should look at XSH.builtins.events.autovox_policy
"""
import itertools
from pathlib import Path
import xontrib.voxapi as voxapi
import warnings
from xonsh.built_ins import XSH
__all__ = ()
_policies = []
XSH.builtins.events.doc(
"autovox_policy",
"""
autovox_policy(path: pathlib.Path) -> Union[str, pathlib.Path, None]
Register a policy with autovox.
A policy is a function that takes a Path and returns the venv associated with it,
if any.
NOTE: The policy should only return a venv for this path exactly, not for
parent paths. Parent walking is handled by autovox so that all policies can
be queried at each level.
""",
)
class MultipleVenvsWarning(RuntimeWarning):
pass
def get_venv(vox, dirpath):
# Search up the directory tree until a venv is found, or none
for path in itertools.chain((dirpath,), dirpath.parents):
venvs = [
vox[p]
for p in XSH.builtins.events.autovox_policy.fire(path=path)
if p is not None and p in vox # Filter out venvs that don't exist
]
if len(venvs) == 0:
continue
else:
if len(venvs) > 1:
warnings.warn(
MultipleVenvsWarning(
"Found {numvenvs} venvs for {path}; using the first".format(
numvenvs=len(venvs), path=path
)
)
)
return venvs[0]
def check_for_new_venv(curdir, olddir):
vox = voxapi.Vox()
if olddir is ... or olddir is None:
try:
oldve = vox[...]
except KeyError:
oldve = None
else:
oldve = get_venv(vox, olddir)
newve = get_venv(vox, curdir)
if oldve != newve:
if newve is None:
vox.deactivate()
else:
vox.activate(newve.env)
# Core mechanism: Check for venv when the current directory changes
@XSH.builtins.events.on_chdir
def cd_handler(newdir, olddir, **_):
check_for_new_venv(Path(newdir), Path(olddir))
# Recalculate when venvs are created or destroyed
@XSH.builtins.events.vox_on_create
def create_handler(**_):
check_for_new_venv(Path.cwd(), ...)
@XSH.builtins.events.vox_on_destroy
def destroy_handler(**_):
check_for_new_venv(Path.cwd(), ...)
# Initial activation before first prompt
@XSH.builtins.events.on_post_init
def load_handler(**_):
check_for_new_venv(Path.cwd(), None)