diff --git a/changelog.rst b/changelog.rst index 0c3905e38..2c3932389 100644 --- a/changelog.rst +++ b/changelog.rst @@ -10,6 +10,8 @@ Bug fixes: Features: --------- +* Add ``\\cd`` and ``\\ls`` filesystem commands, with path completion for + filesystem-aware commands including ``\\e``, ``\\i``, ``\\log-file``, and ``\\o``. * Add ``\\ne `` to edit a named query in the external editor. Loads the named query's SQL into ``$EDITOR``; on save it is written back to the ``[named queries]`` section, creating it if it does not exist. Complements diff --git a/pgcli/main.py b/pgcli/main.py index 8a9b1b024..f17070f14 100644 --- a/pgcli/main.py +++ b/pgcli/main.py @@ -410,6 +410,18 @@ def refresh_callback(): arg_type=NO_QUERY, ) self.pgspecial.register(self.execute_from_file, "\\i", "\\i filename", "Execute commands from file.") + self.pgspecial.register( + self.change_directory, + "\\cd", + "\\cd [directory]", + "Change the current working directory.", + ) + self.pgspecial.register( + self.list_directory, + "\\ls", + "\\ls [path]", + "List files in a directory.", + ) self.pgspecial.register( self.write_to_file, "\\o", @@ -555,6 +567,28 @@ def execute_from_file(self, pattern, **_): explain_mode=self.explain_mode, ) + def change_directory(self, pattern, **_): + try: + directory = pathlib.Path(pattern or "~").expanduser() + os.chdir(directory) + except (OSError, RuntimeError) as e: + return [(None, None, None, str(e), "", False, True)] + + return [(None, None, None, str(pathlib.Path.cwd()), "", True, True)] + + def list_directory(self, pattern, **_): + try: + path = pathlib.Path(pattern or ".").expanduser() + path.stat() + entries = list(path.iterdir()) if path.is_dir() else [path] + entries_with_types = [(entry, entry.is_dir()) for entry in entries] + entries_with_types.sort(key=lambda item: (not item[1], item[0].name.casefold(), item[0].name)) + output = "\n".join(entry.name + (os.sep if is_directory else "") for entry, is_directory in entries_with_types) + except (OSError, RuntimeError) as e: + return [(None, None, None, str(e), "", False, True)] + + return [(None, None, None, output, "", True, True)] + def write_to_logfile(self, pattern, **_): if not pattern: self.log_file = None diff --git a/pgcli/packages/sqlcompletion.py b/pgcli/packages/sqlcompletion.py index 2bb57e3d2..0cf495b20 100644 --- a/pgcli/packages/sqlcompletion.py +++ b/pgcli/packages/sqlcompletion.py @@ -45,7 +45,11 @@ Datatype = namedtuple("Datatype", ["schema"]) Alias = namedtuple("Alias", ["aliases"]) -Path = namedtuple("Path", []) +Path = namedtuple("Path", ["only_directories"]) +Path.__new__.__defaults__ = (False,) + +PATH_COMMANDS = frozenset(("\\e", "\\i", "\\log-file", "\\ls", "\\o")) +DIRECTORY_COMMANDS = frozenset(("\\cd",)) class SqlStatement: @@ -124,8 +128,13 @@ def suggest_type(full_text, text_before_cursor): A scope for a column category will be a list of tables. """ - if full_text.startswith("\\i "): - return (Path(),) + command_text = text_before_cursor.lstrip() + command, _, _ = parse_special_command(command_text) + if command != command_text: + if command in PATH_COMMANDS: + return (Path(),) + if command in DIRECTORY_COMMANDS: + return (Path(only_directories=True),) # This is a temporary hack; the exception handling # here should be removed once sqlparse has been fixed diff --git a/pgcli/pgcompleter.py b/pgcli/pgcompleter.py index 284c6da48..07d68bf1d 100644 --- a/pgcli/pgcompleter.py +++ b/pgcli/pgcompleter.py @@ -5,6 +5,7 @@ import operator from collections import namedtuple, defaultdict, OrderedDict from cli_helpers.tabular_output import TabularOutputFormatter +from pgspecial.main import parse_special_command from pgspecial.namedqueries import NamedQueries from prompt_toolkit.completion import Completer, Completion, PathCompleter from prompt_toolkit.document import Document @@ -484,7 +485,8 @@ def get_completions(self, document, complete_event, smart_completion=None): # Map suggestion type to method # e.g. 'table' -> self.get_table_matches matcher = self.suggestion_matchers[suggestion_type] - matches.extend(matcher(self, suggestion, word_before_cursor)) + completion_text = document.text_before_cursor if suggestion_type is Path else word_before_cursor + matches.extend(matcher(self, suggestion, completion_text)) # Sort matches so highest priorities are first matches = sorted(matches, key=operator.attrgetter("priority"), reverse=True) @@ -840,9 +842,10 @@ def get_keyword_matches(self, suggestion, word_before_cursor): return self.find_matches(word_before_cursor, keywords, mode="strict", meta="keyword") - def get_path_matches(self, _, word_before_cursor): - completer = PathCompleter(expanduser=True) - document = Document(text=word_before_cursor, cursor_position=len(word_before_cursor)) + def get_path_matches(self, suggestion, text_before_cursor): + _, _, path = parse_special_command(text_before_cursor.lstrip()) + completer = PathCompleter(expanduser=True, only_directories=suggestion.only_directories) + document = Document(text=path, cursor_position=len(path)) for c in completer.get_completions(document, None): yield Match(completion=c, priority=(0,)) diff --git a/tests/test_main.py b/tests/test_main.py index 40e96ab05..9ca0a4861 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,4 +1,5 @@ import os +import pathlib import platform import re import tempfile @@ -335,6 +336,58 @@ def test_i_works(tmpdir, executor): run(executor, statement, pgspecial=cli.pgspecial) +def test_change_directory(tmp_path, monkeypatch): + cli = object.__new__(PGCli) + target = tmp_path / "target" + target.mkdir() + monkeypatch.chdir(tmp_path) + + result = cli.change_directory(str(target)) + + assert pathlib.Path.cwd() == target + assert result[0][3] == str(target) + + +def test_change_directory_reports_missing_path(tmp_path, monkeypatch): + cli = object.__new__(PGCli) + monkeypatch.chdir(tmp_path) + + result = cli.change_directory(str(tmp_path / "missing")) + + assert pathlib.Path.cwd() == tmp_path + assert result[0][5:] == (False, True) + assert "missing" in result[0][3] + + +@pytest.mark.parametrize("command", ["change_directory", "list_directory"]) +def test_filesystem_commands_report_expanduser_errors(command): + cli = object.__new__(PGCli) + with mock.patch.object(pathlib.Path, "expanduser", side_effect=RuntimeError("unknown home")): + result = getattr(cli, command)("~missing") + + assert result[0][3] == "unknown home" + assert result[0][5:] == (False, True) + + +def test_list_directory(tmp_path): + cli = object.__new__(PGCli) + (tmp_path / "folder").mkdir() + (tmp_path / "query.sql").write_text("select 1") + + result = cli.list_directory(str(tmp_path)) + + assert result[0][3].splitlines() == [f"folder{os.sep}", "query.sql"] + + +def test_list_directory_reports_missing_path(tmp_path): + cli = object.__new__(PGCli) + + result = cli.list_directory(str(tmp_path / "missing")) + + assert result[0][5:] == (False, True) + assert "missing" in result[0][3] + + @dbtest def test_toggle_verbose_errors(executor): cli = PGCli(pgexecute=executor) diff --git a/tests/test_pgcompleter.py b/tests/test_pgcompleter.py index 028d02524..bf38db236 100644 --- a/tests/test_pgcompleter.py +++ b/tests/test_pgcompleter.py @@ -2,6 +2,12 @@ import pytest from pgcli import pgcompleter import tempfile +from prompt_toolkit.document import Document + + +def path_completions(completer, text): + document = Document(text=text, cursor_position=len(text)) + return list(completer.get_completions(document, None)) def test_load_alias_map_file_missing_file(): @@ -19,6 +25,27 @@ def test_load_alias_map_file_invalid_json(tmp_path): pgcompleter.load_alias_map_file(str(fpath)) +def test_path_completion_filters_files_for_cd(tmp_path, monkeypatch): + (tmp_path / "folder").mkdir() + (tmp_path / "file.sql").touch() + monkeypatch.chdir(tmp_path) + completer = pgcompleter.PGCompleter() + + assert [completion.text for completion in path_completions(completer, r"\ls f")] == ["ile.sql", "older"] + assert [completion.text for completion in path_completions(completer, r"\cd f")] == ["older"] + + +def test_path_completion_handles_spaces(tmp_path, monkeypatch): + directory = tmp_path / "space directory" + directory.mkdir() + (directory / "query.sql").touch() + monkeypatch.chdir(tmp_path) + + completions = path_completions(pgcompleter.PGCompleter(), r"\i space directory/q") + + assert [completion.text for completion in completions] == ["uery.sql"] + + @pytest.mark.parametrize( "table_name, alias", [ diff --git a/tests/test_pgspecial.py b/tests/test_pgspecial.py index cd99e321b..322253659 100644 --- a/tests/test_pgspecial.py +++ b/tests/test_pgspecial.py @@ -8,6 +8,7 @@ View, Function, Datatype, + Path, ) @@ -60,6 +61,20 @@ def test_leading_whitespace_ok(): assert suggestions == suggest_type(cmd, cmd) +@pytest.mark.parametrize("command", [r"\i ", r"\e ", r"\ls ", r"\o ", r"\log-file "]) +def test_file_commands_suggest_paths(command): + assert suggest_type(command, command) == (Path(),) + + +def test_cd_suggests_directories(): + command = r"\cd " + assert suggest_type(command, command) == (Path(only_directories=True),) + + +def test_path_suggestion_uses_text_before_cursor(): + assert suggest_type(r"\ls folder", r"\l") == (Special(),) + + def test_dT_suggests_schema_or_datatypes(): text = "\\dT " suggestions = suggest_type(text, text)