Add generic binary-help subcommand resolver
Consult a PATH-resolved binary's own --help (mtime-cached, LC_ALL=C, 2s timeout with kill) when a segment's first non-flag token is not a listed subcommand, and splice the unique close match (ratio >= 0.8, same first character). Ambiguity, sudo heads, flat-fallback artifact heads, valid subcommands and nested levels all decline to the rest of the chain; only <binary> --help is ever run.
This commit is contained in:
Alexander
@@ -0,0 +1,157 @@
|
||||
"""Tests for the generic binary-help subcommand resolver.
|
||||
|
||||
The harness runs REAL fake binaries (shell scripts on a tmp PATH) that
|
||||
append their argv to a log file, so every claim about spawning (or not
|
||||
spawning) `binary --help` is grounded in the argv log, not in the
|
||||
resolver's return value alone.
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from thefuck import shell_ast
|
||||
from thefuck.resolvers.help_resolver import get_help_correction
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.skipif(
|
||||
not shell_ast.AST_AVAILABLE, reason='bashlex is not available'),
|
||||
pytest.mark.skipif(
|
||||
os.name == 'nt', reason='fake binaries are POSIX shell scripts'),
|
||||
]
|
||||
|
||||
|
||||
def _make_fake(path, argv_log, commands, exec_sleep=0):
|
||||
"""Writes an executable fake binary.
|
||||
|
||||
It appends its argv to `argv_log`, records its pid, optionally
|
||||
`exec`s a `sleep` (replacing itself, so a kill leaves no orphan)
|
||||
before printing a `Commands:` section built from `commands`.
|
||||
"""
|
||||
body = [
|
||||
'#!/bin/sh',
|
||||
"printf '%s\\n' \"$*\" >> {}".format(argv_log),
|
||||
'echo $$ > {}'.format(path + '.pid'),
|
||||
]
|
||||
if exec_sleep:
|
||||
body.append('exec sleep {}'.format(exec_sleep))
|
||||
if commands is not None:
|
||||
body.append("cat <<'EOF'")
|
||||
body.append('Usage: {} COMMAND [arg...]'.format(
|
||||
os.path.basename(path)))
|
||||
body.append('')
|
||||
body.append('Commands:')
|
||||
for command in commands:
|
||||
body.append(' {}'.format(command))
|
||||
body.append('')
|
||||
body.append("Run '{} COMMAND --help' for more information.".format(
|
||||
os.path.basename(path)))
|
||||
body.append('EOF')
|
||||
with open(path, 'w') as script:
|
||||
script.write('\n'.join(body) + '\n')
|
||||
os.chmod(path, 0o755)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_bin(tmpdir, os_environ, no_memoize):
|
||||
"""Puts a tmp bin dir first on PATH; returns a binary factory."""
|
||||
bindir = tmpdir.mkdir('bin')
|
||||
os_environ['PATH'] = str(bindir) + os.pathsep + os_environ['PATH']
|
||||
|
||||
def _factory(name, commands=('build', 'clean', 'pull'), exec_sleep=0):
|
||||
path = str(bindir.join(name))
|
||||
_make_fake(path, str(tmpdir.join('argv.log')), commands, exec_sleep)
|
||||
return path
|
||||
|
||||
return _factory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def argv_log(tmpdir):
|
||||
return str(tmpdir.join('argv.log'))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def enable_cache(no_cache, monkeypatch, tmpdir, os_environ):
|
||||
"""Re-enables the mtime cache that the autouse no_cache disables."""
|
||||
os_environ['XDG_CACHE_HOME'] = str(tmpdir.mkdir('cache'))
|
||||
monkeypatch.setattr('thefuck.utils.cache.disabled', False)
|
||||
|
||||
|
||||
def _spawned(argv_log):
|
||||
if not os.path.exists(argv_log):
|
||||
return []
|
||||
with open(argv_log) as log:
|
||||
return log.read().splitlines()
|
||||
|
||||
|
||||
def test_returns_correction_when_subcommand_misspelled(
|
||||
fake_bin, argv_log, enable_cache):
|
||||
fake_bin('fake')
|
||||
assert get_help_correction('fake bilud .') == 'fake build .'
|
||||
# The mtime-keyed cache suppresses the second `--help` spawn.
|
||||
assert get_help_correction('fake bilud .') == 'fake build .'
|
||||
assert _spawned(argv_log) == ['--help']
|
||||
|
||||
|
||||
def test_respawns_when_binary_mtime_changes(
|
||||
fake_bin, argv_log, enable_cache):
|
||||
path = fake_bin('fake')
|
||||
assert get_help_correction('fake bilud .') == 'fake build .'
|
||||
stats = os.stat(path)
|
||||
os.utime(path, (stats.st_atime, stats.st_mtime + 5))
|
||||
assert get_help_correction('fake bilud .') == 'fake build .'
|
||||
assert _spawned(argv_log) == ['--help', '--help']
|
||||
|
||||
|
||||
def test_declines_when_help_times_out(fake_bin, argv_log, enable_cache):
|
||||
path = fake_bin('sleeper', commands=None, exec_sleep=5)
|
||||
started = time.time()
|
||||
assert get_help_correction('sleeper whatever') is None
|
||||
elapsed = time.time() - started
|
||||
# The spawn really happened and the process was killed, not waited out.
|
||||
assert _spawned(argv_log) == ['--help']
|
||||
with open(path + '.pid') as pid_file:
|
||||
pid = int(pid_file.read())
|
||||
with pytest.raises(OSError):
|
||||
os.kill(pid, 0)
|
||||
assert elapsed < 3
|
||||
|
||||
|
||||
def test_declines_when_token_is_flag(fake_bin, argv_log):
|
||||
fake_bin('fake')
|
||||
assert get_help_correction('fake --bilud') is None
|
||||
assert _spawned(argv_log) == []
|
||||
|
||||
|
||||
def test_declines_when_subcommand_is_listed(
|
||||
fake_bin, argv_log, enable_cache):
|
||||
fake_bin('fake')
|
||||
assert get_help_correction('fake build .') is None
|
||||
assert get_help_correction('fake build .') is None
|
||||
assert _spawned(argv_log) == ['--help']
|
||||
|
||||
|
||||
def test_declines_when_head_not_on_path(fake_bin, argv_log):
|
||||
assert get_help_correction('no-such-binary-anywhere xyz') is None
|
||||
assert _spawned(argv_log) == []
|
||||
|
||||
|
||||
def test_declines_when_match_is_ambiguous(fake_bin, argv_log):
|
||||
fake_bin('fake', commands=('clean', 'cleat', 'build', 'pull'))
|
||||
assert get_help_correction('fake clea') is None
|
||||
|
||||
|
||||
def test_declines_when_head_is_sudo(fake_bin, argv_log):
|
||||
fake_bin('fake')
|
||||
assert get_help_correction('sudo fake bilud') is None
|
||||
assert _spawned(argv_log) == []
|
||||
|
||||
|
||||
def test_declines_when_script_unparseable(fake_bin, argv_log):
|
||||
fake_bin('fake')
|
||||
# bashlex cannot parse `case`; the flat fallback's head token then
|
||||
# fails the resolver's gates and nothing is spawned.
|
||||
assert get_help_correction('case $x in a) fake bilud;; esac') is None
|
||||
assert _spawned(argv_log) == []
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Generic subcommand typos fixed from the binary's own `--help`.
|
||||
|
||||
Trigger: a pipeline segment whose head resolves on $PATH but whose
|
||||
first non-flag token is not a listed subcommand of that binary. Only
|
||||
the first subcommand level is considered (`git remote addd` is out).
|
||||
|
||||
Caching choice: the docker rule wraps its loader in
|
||||
`utils.cache(which('docker'))` at import time because its binary is
|
||||
static; here the binary is only known at call time, so every resolved
|
||||
binary path gets its own `cache(binary_path)`-wrapped loader stored in
|
||||
the module-level `_command_loaders` dict. That reuses the mtime-keyed
|
||||
`thefuck.utils.Cache` machinery (a changed binary re-runs `--help`)
|
||||
and honors `cache.disabled` without calling `utils._cache` by hand.
|
||||
|
||||
Personal fork only: depends on bashlex (GPL-3+), do not distribute.
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
from difflib import get_close_matches
|
||||
|
||||
from thefuck import shell_ast
|
||||
from thefuck.utils import cache, which
|
||||
|
||||
|
||||
_CUTOFF = 0.8
|
||||
_TIMEOUT = 2
|
||||
_MATCHES = 5
|
||||
_SECTION_HEADERS = ('Commands:', 'Management Commands:',
|
||||
'Available Commands:')
|
||||
|
||||
_command_loaders = {}
|
||||
|
||||
|
||||
def get_help_correction(script):
|
||||
"""Returns the script with misspelled subcommands fixed, or None."""
|
||||
if not shell_ast.AST_AVAILABLE:
|
||||
return None
|
||||
replacements = []
|
||||
for segment in shell_ast.parse(script):
|
||||
replacement = _segment_replacement(segment)
|
||||
if replacement is not None:
|
||||
replacements.append(replacement)
|
||||
if not replacements:
|
||||
return None
|
||||
return shell_ast.splice(script, replacements)
|
||||
|
||||
|
||||
def _segment_replacement(segment):
|
||||
"""Returns a (start, end, token) fix for one segment, or None."""
|
||||
head = segment.head
|
||||
# sudo is out of scope (single-level fix); `=`/`/`/`.` heads are
|
||||
# flat-fallback artifacts like `VAR=value` or `./tool`, not names
|
||||
# this resolver should consult --help for.
|
||||
if head == 'sudo' or '=' in head or '/' in head or '.' in head:
|
||||
return None
|
||||
binary_path = which(head)
|
||||
if binary_path is None:
|
||||
return None
|
||||
for word in segment.words[1:]:
|
||||
token, _, _ = word
|
||||
if not token or token.startswith('-'):
|
||||
continue
|
||||
return _token_replacement(word, head, binary_path)
|
||||
return None
|
||||
|
||||
|
||||
def _token_replacement(word, binary_name, binary_path):
|
||||
"""Returns the (start, end, match) fix for one subcommand token."""
|
||||
commands = _get_commands(binary_path, binary_name)
|
||||
if not commands:
|
||||
return None
|
||||
token, start, end = word
|
||||
if token in commands:
|
||||
return None
|
||||
matches = [match for match in get_close_matches(
|
||||
token, commands, n=_MATCHES, cutoff=_CUTOFF)
|
||||
if match.startswith(token[0])]
|
||||
if len(matches) != 1:
|
||||
return None
|
||||
return start, end, matches[0]
|
||||
|
||||
|
||||
def _get_commands(binary_path, binary_name):
|
||||
loader = _command_loaders.get(binary_path)
|
||||
if loader is None:
|
||||
loader = cache(binary_path)(_load_commands)
|
||||
_command_loaders[binary_path] = loader
|
||||
return loader(binary_name)
|
||||
|
||||
|
||||
def _load_commands(binary_name):
|
||||
"""Spawns `<binary> --help`; returns its subcommands or None.
|
||||
|
||||
The resolved binary path that keys the mtime cache is bound by the
|
||||
`cache(binary_path)` factory in `_get_commands`, not by an unused
|
||||
parameter here.
|
||||
"""
|
||||
env = dict(os.environ, LC_ALL='C', LANG='C')
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
[binary_name, '--help'], stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE, env=env)
|
||||
except OSError:
|
||||
return None
|
||||
try:
|
||||
stdout, stderr = proc.communicate(timeout=_TIMEOUT)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
proc.communicate()
|
||||
return None
|
||||
lines = stdout.decode('utf-8', 'replace').splitlines()
|
||||
if not lines:
|
||||
lines = stderr.decode('utf-8', 'replace').splitlines()
|
||||
return _parse_commands(lines)
|
||||
|
||||
|
||||
def _parse_commands(lines):
|
||||
"""First tokens of indented entries under command-section headers.
|
||||
|
||||
A section ends at the first blank or non-indented line; help with
|
||||
no known header yields None so the caller declines.
|
||||
"""
|
||||
commands = []
|
||||
found_section = False
|
||||
in_section = False
|
||||
for line in lines:
|
||||
stripped = line.strip()
|
||||
if not line[:1].isspace() and stripped in _SECTION_HEADERS:
|
||||
found_section = True
|
||||
in_section = True
|
||||
elif not stripped:
|
||||
in_section = False
|
||||
elif in_section and line[:1].isspace():
|
||||
commands.append(stripped.split()[0])
|
||||
else:
|
||||
in_section = False
|
||||
return commands if found_section else None
|
||||
Reference in New Issue
Block a user