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

cmd cache refactor - optimize cache usage (#4954)

Browse source 
* 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
This commit is contained in:
Noorhteen Raja NJ 2022-11-14 23:52:10 +05:30 • committed by GitHub
parent f61a6d3da0
commit c63f75efd0
Failed to generate hash of commit
13 changed files with 220 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
news/cmds-cache.rst Normal file
View file

@ -0,0 +1,25 @@
**Added:**
* ``$XDG_CACHE_HOME``, ``$XONSH_CACHE_DIR`` are now available inside ``Xonsh``
**Changed:**
* <news item>
**Deprecated:**
* <news item>
**Removed:**
* ``$COMMANDS_CACHE_SIZE_WARNING`` is removed. When ``$COMMANDS_CACHE_SAVE_INTERMEDIATE`` is enabled,
the cache file size is optimized.
**Fixed:**
* fixed stale results when ``$COMMANDS_CACHE_SAVE_INTERMEDIATE`` is enabled.
**Security:**
* <news item>

4
tests/completers/test_base_completer.py
View file

@ -10,11 +10,11 @@ CUR_DIR = (
@pytest.fixture(autouse=True)
def setup(xession, xonsh_execer, monkeypatch, patch_commands_cache_bins):
def setup(xession, xonsh_execer, monkeypatch, mock_executables_in):
xession.env["COMMANDS_CACHE_SAVE_INTERMEDIATE"] = False
xession.env["COMPLETION_QUERY_LIMIT"] = 2000
patch_commands_cache_bins(["cool"])
mock_executables_in(["cool"])
def test_empty_line(check_completer):

95
tests/test_commands_cache.py
View file

@ -1,13 +1,12 @@
import os
import pickle
import time
from unittest.mock import MagicMock
import pytest
from xonsh.commands_cache import (
SHELL_PREDICTOR_PARSER,
CommandsCache,
_Commands,
predict_false,
predict_shell,
predict_true,
@ -27,56 +26,38 @@ def test_predict_threadable_unknown_command(xession):
assert isinstance(result, bool)
@pytest.fixture
def commands_cache_tmp(xession, tmp_path, monkeypatch, patch_commands_cache_bins):
class TestCommandsCacheSaveIntermediate:
"""test behavior when $COMMANDS_CACHE_SAVE_INTERMEDIATE=True"""
@pytest.fixture
def exin_mock(self, xession, mock_executables_in):
xession.env["COMMANDS_CACHE_SAVE_INTERMEDIATE"] = True
return patch_commands_cache_bins(["bin1", "bin2"])
return mock_executables_in(["bin1", "bin2"])
def test_commands_cached_between_runs(commands_cache_tmp, tmp_path, tmpdir):
# 1. no pickle file
# 2. return empty result first and create a thread to populate result
# 3. once the result is available then next call to cc.all_commands returns
cc = commands_cache_tmp
# wait for thread to end
cnt = 0 # timeout waiting for thread
while True:
if cc.all_commands or cnt > 10:
break
cnt += 1
time.sleep(0.1)
assert [b.lower() for b in cc.all_commands.keys()] == ["bin1", "bin2"]
def test_caching_to_file(self, exin_mock, xession, tmp_path):
assert [b.lower() for b in xession.commands_cache.all_commands.keys()] == [
"bin1",
"bin2",
]
files = tmp_path.glob("*.pickle")
assert len(list(files)) == 1
exin_mock.assert_called_once()
# cleanup dir
for file in files:
os.remove(file)
def test_commands_cache_uses_pickle_file(commands_cache_tmp, tmp_path, monkeypatch):
cc = commands_cache_tmp
update_cmds_cache = MagicMock()
monkeypatch.setattr(cc, "_update_cmds_cache", update_cmds_cache)
def test_loading_cache(self, exin_mock, tmp_path, xession):
cc = xession.commands_cache
file = tmp_path / CommandsCache.CACHE_FILE
bins = {
"bin1": (
"/some-path/bin1",
None,
),
"bin2": (
"/some-path/bin2",
None,
),
file.touch()
cached = {
str(tmp_path): _Commands(
mtime=tmp_path.stat().st_mtime, cmds=("bin1", "bin2")
)
}
file.write_bytes(pickle.dumps(bins))
file.write_bytes(pickle.dumps(cached))
assert str(cc.cache_file) == str(file)
assert cc.all_commands == bins
assert cc._loaded_pickled
assert [b.lower() for b in cc.all_commands.keys()] == ["bin1", "bin2"]
exin_mock.assert_not_called()
TRUE_SHELL_ARGS = [
@ -99,7 +80,7 @@ def test_predict_shell_parser(args):
@pytest.mark.parametrize("args", TRUE_SHELL_ARGS)
def test_predict_shell_true(args):
assert predict_shell(args)
assert predict_shell(args, None)
FALSE_SHELL_ARGS = [[], ["-c"], ["-i"], ["-i", "-l"]]
@ -107,7 +88,7 @@ FALSE_SHELL_ARGS = [[], ["-c"], ["-i"], ["-i", "-l"]]
@pytest.mark.parametrize("args", FALSE_SHELL_ARGS)
def test_predict_shell_false(args):
assert not predict_shell(args)
assert not predict_shell(args, None)
PATTERN_BIN_USING_TTY_OR_NOT = [
@ -174,25 +155,21 @@ def test_commands_cache_predictor_default(args, xession, tmp_path):
assert result == expected
@skip_if_on_windows
def test_cd_is_only_functional_alias(xession):
class Test_is_only_functional_alias:
@skip_if_on_windows
def test_cd(self, xession):
xession.aliases["cd"] = lambda args: os.chdir(args[0])
xession.env["PATH"] = []
assert xession.commands_cache.is_only_functional_alias("cd")
def test_non_exist_is_only_functional_alias(xession):
assert not xession.commands_cache.is_only_functional_alias(
def test_non_exist(self, xession):
assert (
xession.commands_cache.is_only_functional_alias(
"<not really a command name>"
)
is False
)
@skip_if_on_windows
def test_bash_is_only_functional_alias(xession):
assert not xession.commands_cache.is_only_functional_alias("bash")
@skip_if_on_windows
def test_bash_and_is_alias_is_only_functional_alias(xession):
xession.aliases["bash"] = lambda args: os.chdir(args[0])
assert not xession.commands_cache.is_only_functional_alias("bash")
def test_bash_and_is_alias_is_only_functional_alias(self, xession):
xession.aliases["git"] = lambda args: os.chdir(args[0])
assert xession.commands_cache.is_only_functional_alias("git") is False

2
tests/test_pipelines.py
View file

@ -25,7 +25,7 @@ def patched_events(monkeypatch, xonsh_events, xonsh_session):
) # capture output of ![]
if ON_WINDOWS:
monkeypatch.setattr(
xonsh_session,
xonsh_session.commands_cache,
"aliases",
{
"echo": "cmd /c echo".split(),

4
tests/test_ptk_completer.py
View file

@ -38,7 +38,7 @@ def test_rich_completion(completion, lprefix, ptk_completion, monkeypatch, xessi
document_mock.current_line = ""
document_mock.cursor_position_col = 0
monkeypatch.setattr(xession, "aliases", Aliases())
monkeypatch.setattr(xession.commands_cache, "aliases", Aliases())
completions = list(ptk_completer.get_completions(document_mock, MagicMock()))
if isinstance(completion, RichCompletion) and not ptk_completion:
@ -216,7 +216,7 @@ def test_alias_expansion(code, index, expected_args, monkeypatch, xession):
ptk_completer.reserve_space = lambda: None
ptk_completer.suggestion_completion = lambda _, __: None
monkeypatch.setattr(xession, "aliases", Aliases(gb=["git branch"]))
monkeypatch.setattr(xession.commands_cache, "aliases", Aliases(gb=["git branch"]))
list(ptk_completer.get_completions(Document(code, index), MagicMock()))
mock_call = xonsh_completer_mock.complete.call_args

5
tests/test_vox.py
View file

@ -347,12 +347,11 @@ def patched_cmd_cache(xession, vox, monkeypatch):
cc = xession.commands_cache
def no_change(self, *_):
return False, False, False
return False, False
monkeypatch.setattr(cc, "_check_changes", 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}
cc._cmds_cache = bins
monkeypatch.setattr(cc, "_cmds_cache", bins)
yield cc

4
xonsh/aliases.py
View file

@ -5,6 +5,7 @@ import functools
import inspect
import os
import re
import shutil
import sys
import types
import typing as tp
@ -898,7 +899,8 @@ def make_default_aliases():
# Add aliases specific to the Anaconda python distribution.
default_aliases["activate"] = ["source-cmd", "activate.bat"]
default_aliases["deactivate"] = ["source-cmd", "deactivate.bat"]
if not locate_binary("sudo"):
if shutil.which("sudo", path=XSH.env.get_detyped("PATH")):
# XSH.commands_cache is not available during setup
import xonsh.winutils as winutils
def sudo(args):

30
xonsh/built_ins.py
View file

@ -551,7 +551,12 @@ class XonshSession:
self.completers = None
self.builtins = None
self._initial_builtin_names = None
self.aliases = None
@property
def aliases(self):
if self.commands_cache is None:
return
return self.commands_cache.aliases
def _disable_python_exit(self):
# Disable Python interactive quit/exit
@ -597,16 +602,9 @@ class XonshSession:
self._disable_python_exit()
self.execer = execer
self.commands_cache = (
kwargs.pop("commands_cache")
if "commands_cache" in kwargs
else CommandsCache()
)
self.modules_cache = {}
self.all_jobs = {}
self.completers = default_completers(self.commands_cache)
self.builtins = get_default_builtins(execer)
self._initial_builtin_names = frozenset(vars(self.builtins))
@ -614,8 +612,14 @@ class XonshSession:
for attr, value in kwargs.items():
if hasattr(self, attr):
setattr(self, attr, value)
self.link_builtins(aliases_given)
self.commands_cache = (
kwargs.pop("commands_cache")
if "commands_cache" in kwargs
else CommandsCache(self.env, aliases_given)
)
self.link_builtins()
self.builtins_loaded = True
self.completers = default_completers(self.commands_cache)
def flush_on_exit(s=None, f=None):
if self.history is not None:
@ -627,9 +631,7 @@ class XonshSession:
for sig in AT_EXIT_SIGNALS:
resetting_signal_handle(sig, flush_on_exit)
def link_builtins(self, aliases=None):
from xonsh.aliases import Aliases, make_default_aliases
def link_builtins(self):
# public built-ins
for refname in self._initial_builtin_names:
objname = f"__xonsh__.builtins.{refname}"
@ -639,9 +641,7 @@ class XonshSession:
# sneak the path search functions into the aliases
# Need this inline/lazy import here since we use locate_binary that
# relies on __xonsh__.env in default aliases
if aliases is None:
aliases = Aliases(make_default_aliases())
self.aliases = builtins.default_aliases = builtins.aliases = aliases
builtins.default_aliases = builtins.aliases = self.aliases
def unlink_builtins(self):
for name in self._initial_builtin_names:

216
xonsh/commands_cache.py
View file

@ -7,21 +7,22 @@ True) or must be run the foreground (returns False).
"""
import argparse
import collections.abc as cabc
import functools
import os
import pickle
import sys
import threading
import time
import typing as tp
from pathlib import Path
from xonsh.built_ins import XSH
from xonsh.lazyasd import lazyobject
from xonsh.platform import ON_POSIX, ON_WINDOWS, pathbasename
from xonsh.tools import executables_in
class _Commands(tp.NamedTuple):
mtime: float
cmds: "tuple[str, ...]"
class CommandsCache(cabc.Mapping):
"""A lazy cache representing the commands available on the file system.
The keys are the command names and the values a tuple of (loc, has_alias)
@ -30,28 +31,37 @@ class CommandsCache(cabc.Mapping):
the command has an alias.
"""
CACHE_FILE = "commands-cache.pickle"
CACHE_FILE = "path-commands-cache.pickle"
def __init__(self):
self._cmds_cache = {}
self._path_checksum = None
self._alias_checksum = None
self._path_mtime = -1
def __init__(self, env, aliases=None) -> None:
# cache commands in path by mtime
self._paths_cache: "dict[str, _Commands]" = {}
# wrap aliases and commands in one place
self._cmds_cache: "dict[str, tuple[str, bool|None]]" = {}
self._alias_checksum: "int|None" = None
self.threadable_predictors = default_threadable_predictors()
self._loaded_pickled = False
# force it to load from env by setting it to None
# Path to the cache-file where all commands/aliases are cached for pre-loading"""
self.env = env
if aliases is None:
from xonsh.aliases import Aliases, make_default_aliases
self.aliases = Aliases(make_default_aliases())
else:
self.aliases = aliases
self._cache_file = None
@property
def cache_file(self):
"""Keeping a property that lies on instance-attribute"""
env = XSH.env or {}
env = self.env
# Path to the cache-file where all commands/aliases are cached for pre-loading
if self._cache_file is None:
if "XONSH_DATA_DIR" in env and env.get("COMMANDS_CACHE_SAVE_INTERMEDIATE"):
if "XONSH_CACHE_DIR" in env and env.get("COMMANDS_CACHE_SAVE_INTERMEDIATE"):
self._cache_file = (
Path(env["XONSH_DATA_DIR"]).joinpath(self.CACHE_FILE).resolve()
Path(env["XONSH_CACHE_DIR"]).joinpath(self.CACHE_FILE).resolve()
)
else:
# set a falsy value other than None
@ -87,13 +97,12 @@ class CommandsCache(cabc.Mapping):
"""Returns whether the cache is populated or not."""
return len(self._cmds_cache) == 0
@staticmethod
def get_possible_names(name):
def get_possible_names(self, name):
"""Generates the possible `PATHEXT` extension variants of a given executable
name on Windows as a list, conserving the ordering in `PATHEXT`.
Returns a list as `name` being the only item in it on other platforms."""
if ON_WINDOWS:
pathext = XSH.env.get("PATHEXT", [])
pathext = self.env.get("PATHEXT", [])
name = name.upper()
return [name + ext for ext in ([""] + pathext)]
else:
@ -108,126 +117,73 @@ class CommandsCache(cabc.Mapping):
if os.path.isdir(p):
yield p
def _check_changes(self, paths: tp.Tuple[str, ...], aliases):
def _check_changes(self, paths: tp.Tuple[str, ...]):
# did PATH change?
path_hash = hash(paths)
yield path_hash == self._path_checksum
self._path_checksum = path_hash
yield self._update_paths_cache(paths)
# did aliases change?
al_hash = hash(frozenset(aliases))
yield al_hash == self._alias_checksum
al_hash = hash(frozenset(self.aliases))
yield al_hash != self._alias_checksum
self._alias_checksum = al_hash
# did the contents of any directory in PATH change?
max_mtime = max(map(lambda path: os.stat(path).st_mtime, paths), default=0)
yield max_mtime <= self._path_mtime
self._path_mtime = max_mtime
@property
def all_commands(self):
self.update_cache()
return self._cmds_cache
def update_cache(self):
env = XSH.env or {}
paths = tuple(CommandsCache.remove_dups(env.get("PATH") or []))
# in case it is empty or unset
alss = {} if XSH.aliases is None else XSH.aliases
(
no_new_paths,
no_new_alias,
no_new_bins,
) = tuple(self._check_changes(paths, alss))
if no_new_paths and no_new_bins:
if not no_new_alias: # only aliases have changed
for cmd, alias in alss.items():
key = cmd.upper() if ON_WINDOWS else cmd
if key in self._cmds_cache:
self._cmds_cache[key] = (self._cmds_cache[key][0], alias)
else:
self._cmds_cache[key] = (cmd, True)
# save the changes to cache as well
self.set_cmds_cache(self._cmds_cache)
return self._cmds_cache
if self.cache_file and self.cache_file.exists():
# pickle the result only if XONSH_DATA_DIR is set
if not self._loaded_pickled:
# first time load the commands from cache-file
self._cmds_cache = self.get_cached_commands()
self._loaded_pickled = True
# also start a thread that updates the cache in the bg
worker = threading.Thread(
target=self._update_cmds_cache,
args=[paths, alss],
daemon=True,
)
worker.start()
else:
self._update_cmds_cache(paths, alss)
return self._cmds_cache
@staticmethod
@functools.lru_cache(maxsize=10)
def _get_all_cmds(paths: tp.Sequence[str]):
"""Cache results when possible
This will be helpful especially during tests where the PATH will be the same mostly.
"""
def _getter():
for path in reversed(paths):
env = self.env
# iterate backwards so that entries at the front of PATH overwrite
# entries at the back.
for cmd in executables_in(path):
yield cmd, os.path.join(path, cmd)
return dict(_getter())
def _update_cmds_cache(
self, paths: tp.Sequence[str], aliases: tp.Dict[str, str]
) -> tp.Dict[str, tp.Any]:
"""Update the cmds_cache variable in background without slowing down parseLexer"""
env = XSH.env or {} # type: ignore
allcmds = {}
for cmd, path in self._get_all_cmds(paths).items():
paths = tuple(reversed(tuple(self.remove_dups(env.get("PATH") or []))))
if any(self._check_changes(paths)):
all_cmds = {}
for cmd, path in self._iter_binaries(paths):
key = cmd.upper() if ON_WINDOWS else cmd
allcmds[key] = (path, aliases.get(key, None))
# None -> not in aliases
all_cmds[key] = (path, None)
warn_cnt = env.get("COMMANDS_CACHE_SIZE_WARNING")
if warn_cnt and len(allcmds) > warn_cnt:
print(
f"Warning! Found {len(allcmds):,} executable files in the PATH directories!",
file=sys.stderr,
)
for cmd in aliases:
if cmd not in allcmds:
# aliases override cmds
for cmd in self.aliases:
key = cmd.upper() if ON_WINDOWS else cmd
allcmds[key] = (cmd, True) # type: ignore
if key in all_cmds:
# (path, False) -> has same named alias
all_cmds[key] = (all_cmds[key][0], False)
else:
# True -> pure alias
all_cmds[key] = (cmd, True)
self._cmds_cache = all_cmds
return self._cmds_cache
return self.set_cmds_cache(allcmds)
def get_cached_commands(self) -> tp.Dict[str, str]:
if self.cache_file and self.cache_file.exists():
def _update_paths_cache(self, paths: tp.Sequence[str]) -> bool:
"""load cached results or update cache"""
if (not self._paths_cache) and self.cache_file and self.cache_file.exists():
# first time load the commands from cache-file if configured
try:
return pickle.loads(self.cache_file.read_bytes()) or {}
self._paths_cache = pickle.loads(self.cache_file.read_bytes()) or {}
except Exception:
# the file is corrupt
self.cache_file.unlink(missing_ok=True)
return {}
def set_cmds_cache(self, allcmds: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
"""write cmds to cache-file and instance-attribute"""
if self.cache_file:
self.cache_file.write_bytes(pickle.dumps(allcmds))
self._cmds_cache = allcmds
return allcmds
updated = False
for path in paths:
modified_time = os.stat(path).st_mtime
if (path not in self._paths_cache) or (
self._paths_cache[path].mtime != modified_time
):
updated = True
self._paths_cache[path] = _Commands(
modified_time, tuple(executables_in(path))
)
if updated and self.cache_file:
self.cache_file.write_bytes(pickle.dumps(self._paths_cache))
return updated
def _iter_binaries(self, paths):
for path in paths:
for cmd in self._paths_cache[path].cmds:
yield cmd, os.path.join(path, cmd)
def cached_name(self, name):
"""Returns the name that would appear in the cache, if it exists."""
@ -317,7 +273,7 @@ class CommandsCache(cabc.Mapping):
self.update_cache()
return self.lazy_is_only_functional_alias(name)
def lazy_is_only_functional_alias(self, name):
def lazy_is_only_functional_alias(self, name) -> bool:
"""Returns whether or not a command is only a functional alias, and has
no underlying executable. For example, the "cd" command is only available
as a functional alias. This search is performed lazily.
@ -334,7 +290,7 @@ class CommandsCache(cabc.Mapping):
thread, rather than the main thread.
"""
predictor = self.get_predictor_threadable(cmd[0])
return predictor(cmd[1:])
return predictor(cmd[1:], self)
def get_predictor_threadable(self, cmd0):
"""Return the predictor whether a command list is able to be run on a
@ -370,8 +326,7 @@ class CommandsCache(cabc.Mapping):
"""
# alias stuff
if not os.path.isabs(cmd0) and os.sep not in cmd0:
alss = getattr(XSH, "aliases", dict())
if cmd0 in alss:
if cmd0 in self.aliases:
return self.default_predictor_alias(cmd0)
# other default stuff
@ -387,9 +342,8 @@ class CommandsCache(cabc.Mapping):
10 # this limit is se to handle infinite loops in aliases definition
)
first_args = [] # contains in reverse order args passed to the aliased command
alss = getattr(XSH, "aliases", dict())
while cmd0 in alss:
alias_name = alss[cmd0]
while cmd0 in self.aliases:
alias_name = self.aliases
if isinstance(alias_name, (str, bytes)) or not isinstance(
alias_name, cabc.Sequence
):
@ -404,7 +358,7 @@ class CommandsCache(cabc.Mapping):
if alias_recursion_limit == 0:
return predict_true
predictor_cmd0 = self.get_predictor_threadable(cmd0)
return lambda cmd1: predictor_cmd0(first_args[::-1] + cmd1)
return lambda cmd1: predictor_cmd0(first_args[::-1] + cmd1, self)
def default_predictor_readbin(self, name, cmd0, timeout, failure):
"""Make a default predictor by
@ -463,12 +417,12 @@ class CommandsCache(cabc.Mapping):
#
def predict_true(args):
def predict_true(_, __):
"""Always say the process is threadable."""
return True
def predict_false(args):
def predict_false(_, __):
"""Never say the process is threadable."""
return False
@ -481,7 +435,7 @@ def SHELL_PREDICTOR_PARSER():
return p
def predict_shell(args):
def predict_shell(args, _):
"""Predict the backgroundability of the normal shell interface, which
comes down to whether it is being run in subproc mode.
"""
@ -503,7 +457,7 @@ def HELP_VER_PREDICTOR_PARSER():
return p
def predict_help_ver(args):
def predict_help_ver(args, _):
"""Predict the backgroundability of commands that have help & version
switches: -h, --help, -v, -V, --version. If either of these options is
present, the command is assumed to print to stdout normally and is therefore
@ -526,7 +480,7 @@ def HG_PREDICTOR_PARSER():
return p
def predict_hg(args):
def predict_hg(args, _):
"""Predict if mercurial is about to be run in interactive mode.
If it is interactive, predict False. If it isn't, predict True.
Also predict False for certain commands, such as split.
@ -538,7 +492,7 @@ def predict_hg(args):
return not ns.interactive
def predict_env(args):
def predict_env(args, cmd_cache: CommandsCache):
"""Predict if env is launching a threadable command or not.
The launched command is extracted from env args, and the predictor of
lauched command is used."""
@ -547,7 +501,7 @@ def predict_env(args):
if args[i] and args[i][0] != "-" and "=" not in args[i]:
# args[i] is the command and the following is its arguments
# so args[i:] is used to predict if the command is threadable
return XSH.commands_cache.predict_threadable(args[i:])
return cmd_cache.predict_threadable(args[i:])
return True

29
xonsh/environ.py
View file

@ -570,6 +570,14 @@ def xonsh_data_dir(env):
return xdd
@default_value
def xonsh_cache_dir(env):
"""Ensures and returns the $XONSH_CACHE_DIR"""
xdd = os.path.expanduser(os.path.join(env.get("XDG_CACHE_HOME"), "xonsh"))
os.makedirs(xdd, exist_ok=True)
return xdd
@default_value
def xonsh_config_dir(env):
"""``$XDG_CONFIG_HOME/xonsh``"""
@ -852,10 +860,6 @@ class GeneralSetting(Xettings):
"will print information about how to continue the stopped process.",
)
COMMANDS_CACHE_SIZE_WARNING = Var.with_default(
6000,
"Number of files on the PATH above which a warning is shown.",
)
COMMANDS_CACHE_SAVE_INTERMEDIATE = Var.with_default(
False,
"If enabled, the CommandsCache is saved between runs and can reduce the startup time.",
@ -1012,6 +1016,12 @@ class GeneralSetting(Xettings):
doc_default="``~/.local/share``",
type_str="str",
)
XDG_CACHE_HOME = Var.with_default(
os.path.expanduser(os.path.join("~", ".cache")),
"The base directory relative to which user-specific non-essential data files should be stored.",
doc_default="``~/.cache``",
type_str="str",
)
XDG_DATA_DIRS = Var.with_default(
xdg_data_dirs,
"A list of directories where system level data files are stored.",
@ -1100,6 +1110,12 @@ The file should contain a function with the signature
doc_default="``$XDG_DATA_HOME/xonsh``",
type_str="str",
)
XONSH_CACHE_DIR = Var.with_default(
xonsh_cache_dir,
"This is the location where cache files used by xonsh are stored, such as commands-cache...",
doc_default="``$XDG_CACHE_HOME/xonsh``",
type_str="str",
)
XONSH_ENCODING = Var.with_default(
DEFAULT_ENCODING,
"This is the encoding that xonsh should use for subprocess operations.",
@ -1901,7 +1917,12 @@ class Env(cabc.MutableMapping):
self._d["PATH"] = list(PATH_DEFAULT)
self._detyped = None
def get_detyped(self, key: str):
detyped = self.detype()
return detyped.get(key)
def detype(self):
"""return a dict that can be used as ``os.environ``"""
if self._detyped is not None:
return self._detyped
ctx = {}

2
xonsh/pyghooks.py
View file

@ -1653,7 +1653,7 @@ class XonshLexer(Python3Lexer):
pathext = os_environ.get("PATHEXT", [".EXE", ".BAT", ".CMD"])
XSH.env["PATHEXT"] = pathext.split(os.pathsep)
if getattr(XSH, "commands_cache", None) is None:
XSH.commands_cache = CommandsCache()
XSH.commands_cache = CommandsCache(XSH.env)
_ = XSH.commands_cache.all_commands # NOQA
super().__init__(*args, **kwargs)

19
xonsh/pytest/plugin.py
View file

@ -128,12 +128,12 @@ def xonsh_execer_parse(xonsh_execer):
@pytest.fixture
def patch_commands_cache_bins(xession, tmp_path, monkeypatch):
def mock_executables_in(xession, tmp_path, monkeypatch):
def _factory(binaries: tp.List[str]):
xession.env["PATH"] = [tmp_path]
exec_mock = MagicMock(return_value=binaries)
monkeypatch.setattr(commands_cache, "executables_in", exec_mock)
return xession.commands_cache
return exec_mock
return _factory
@ -207,10 +207,10 @@ def os_env(session_os_env):
@pytest.fixture
def env(tmpdir, session_env):
def env(tmp_path, session_env):
"""a mutable copy of session_env"""
env_copy = copy_env(session_env)
initial_vars = {"XONSH_DATA_DIR": str(tmpdir)}
initial_vars = {"XONSH_DATA_DIR": str(tmp_path), "XONSH_CACHE_DIR": str(tmp_path)}
env_copy.update(initial_vars)
return env_copy
@ -223,7 +223,6 @@ def xonsh_session(xonsh_events, session_execer, os_env, monkeypatch):
XSH.load(
ctx={},
execer=session_execer,
commands_cache=commands_cache.CommandsCache(),
env=os_env,
)
yield XSH
@ -238,12 +237,12 @@ def mock_xonsh_session(monkeypatch, xonsh_events, xonsh_session, env):
# make sure that all other fixtures call this mock only one time
session = []
def factory(*attrs: str):
def factory(*attrs_to_skip: str):
"""
Parameters
----------
attrs
attrs_to_skip
do not mock the given attributes
Returns
@ -254,16 +253,16 @@ def mock_xonsh_session(monkeypatch, xonsh_events, xonsh_session, env):
if session:
raise RuntimeError("The factory should be called only once per test")
aliases = None if "aliases" in attrs_to_skip else Aliases()
for attr, val in [
("env", env),
("shell", DummyShell()),
("help", lambda x: x),
("aliases", Aliases()),
("exit", False),
("history", DummyHistory()),
(
"commands_cache",
commands_cache.CommandsCache(),
commands_cache.CommandsCache(env, aliases),
), # since env,aliases change , patch cmds-cache
# ("subproc_captured", sp),
("subproc_uncaptured", sp),
@ -272,7 +271,7 @@ def mock_xonsh_session(monkeypatch, xonsh_events, xonsh_session, env):
("subproc_captured_object", sp),
("subproc_captured_hiddenobject", sp),
]:
if attr in attrs:
if attr in attrs_to_skip:
continue
monkeypatch.setattr(xonsh_session, attr, val)

2
xonsh/tools.py
View file

@ -838,7 +838,7 @@ def _executables_in_windows(path):
return
def executables_in(path):
def executables_in(path) -> tp.Iterable[str]:
"""Returns a generator of files in path that the user could execute."""
if ON_WINDOWS:
func = _executables_in_windows