diff --git a/tests/resolvers/test_help_resolver.py b/tests/resolvers/test_help_resolver.py new file mode 100644 index 0000000..77e4080 --- /dev/null +++ b/tests/resolvers/test_help_resolver.py @@ -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) == [] diff --git a/thefuck/resolvers/__init__.py b/thefuck/resolvers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/thefuck/resolvers/help_resolver.py b/thefuck/resolvers/help_resolver.py new file mode 100644 index 0000000..b03dc35 --- /dev/null +++ b/thefuck/resolvers/help_resolver.py @@ -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 ` --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