diff --git a/changelog.md b/changelog.md index bbda1ac7..6f5c5b2f 100644 --- a/changelog.md +++ b/changelog.md @@ -12,6 +12,7 @@ Features Bug Fixes --------- * Show CLI error on invalid `--execute` containing `/source`. +* Require `/source` filenames containing spaces to be quoted. 2.15.0 (2026/08/20) diff --git a/mycli/client_commands.py b/mycli/client_commands.py index a0e04673..3a89068c 100644 --- a/mycli/client_commands.py +++ b/mycli/client_commands.py @@ -4,11 +4,13 @@ import logging import os import re +import shlex from typing import TYPE_CHECKING, Any, cast import click import sqlparse +from mycli.compat import WIN from mycli.config import write_default_config from mycli.main_modes.repl import set_all_external_titles from mycli.packages import special @@ -33,6 +35,7 @@ DSN_CONFIG_VALUE = object() FAVORITES_CONFIG_VALUE = object() HIDDEN_CONFIG_SECTIONS = frozenset({'alias_dsn', 'favorite_queries'}) +INVALID_SOURCE_FILENAME = 'Source accepts exactly one filename; filenames containing spaces must be quoted.' SOURCE_SAFE_SPECIAL_COMMANDS = frozenset({ 'connect', 'fd', @@ -81,6 +84,45 @@ def _parse_source_arguments(arg: str) -> tuple[str, bool, bool, bool]: return filename, allow_special, show_queries, page_output +def _has_unquoted_whitespace(value: str) -> bool: + quote: str | None = None + escaped = False + for character in value: + if escaped: + if quote is None and character.isspace(): + return True + escaped = False + continue + if not WIN and character == '\\' and quote != "'": + escaped = True + continue + if character in ("'", '"'): + if quote is None: + quote = character + elif quote == character: + quote = None + elif quote is None and character.isspace(): + return True + return False + + +def _parse_source_filename(filename: str) -> str: + if not filename: + return '' + if _has_unquoted_whitespace(filename): + raise ValueError(INVALID_SOURCE_FILENAME) + try: + arguments = shlex.split(filename, posix=not WIN) + except ValueError as error: + raise ValueError(f'Invalid source filename: {error}.') from None + if len(arguments) != 1: + raise ValueError(INVALID_SOURCE_FILENAME) + parsed_filename = arguments[0] + if WIN and len(parsed_filename) >= 2 and parsed_filename[0] == parsed_filename[-1] and parsed_filename[0] in ("'", '"'): + parsed_filename = parsed_filename[1:-1] + return parsed_filename + + def _registered_special_command(query: str) -> tuple[str, str] | None: command, _verbosity, arg = special.parse_special_command(query) registered = special_main.COMMANDS.get(command) @@ -355,8 +397,13 @@ def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]: filename, allow_special, show_queries, page_output = _parse_source_arguments(arg) if page_output: yield SQLResult(command={'name': 'source_page'}) + try: + filename = _parse_source_filename(filename) + except ValueError as error: + yield SQLResult(status=str(error), is_error=True) + return if not filename: - yield SQLResult(status="Missing required argument: filename.") + yield SQLResult(status="Missing required argument: filename.", is_error=True) return try: diff --git a/mycli/packages/completion_engine.py b/mycli/packages/completion_engine.py index 1a6576e6..e0083ad5 100644 --- a/mycli/packages/completion_engine.py +++ b/mycli/packages/completion_engine.py @@ -703,6 +703,10 @@ def suggest_type(full_text: str, text_before_cursor: str) -> list[dict[str, Any] A scope for a column category will be a list of tables. """ + stripped_text = text_before_cursor.lstrip() + if re.match(r'^(?:source|/source|\\\.|/\.)\s', stripped_text, re.IGNORECASE): + return suggest_special(text_before_cursor) + word_before_cursor = last_word(text_before_cursor, include="many_punctuations") identifier: Identifier | None = None @@ -817,7 +821,7 @@ def suggest_special(text: str) -> list[dict[str, Any]]: if not source_arguments: return [ {'type': 'special_subcommand', 'subcommands': source_options}, - {'type': 'file_name'}, + {'type': 'file_name', 'quote_spaces': True, 'source_filename': ''}, ] used_options: set[str] = set() @@ -826,17 +830,22 @@ def suggest_special(text: str) -> list[dict[str, Any]]: used_options.add(source_arguments[argument_index]) argument_index += 1 remaining_options = [option for option in source_options if option not in used_options] + source_filename = _arg + for _index in range(argument_index): + parsed_argument = source_filename.split(maxsplit=1) + source_filename = parsed_argument[1] if len(parsed_argument) == 2 else '' + file_suggestion = {'type': 'file_name', 'quote_spaces': True, 'source_filename': source_filename} if argument_index < len(source_arguments): if source_arguments[argument_index].startswith('-'): return [{'type': 'special_subcommand', 'subcommands': remaining_options}] - return [{'type': 'file_name'}] + return [file_suggestion] if not text[-1].isspace(): return [] - suggestions = [] + suggestions: list[dict[str, Any]] = [] if remaining_options: suggestions.append({'type': 'special_subcommand', 'subcommands': remaining_options}) - suggestions.append({'type': 'file_name'}) + suggestions.append(file_suggestion) return suggestions if cmd.lower() in [ diff --git a/mycli/packages/filepaths.py b/mycli/packages/filepaths.py index 5d67582c..ef9f770f 100644 --- a/mycli/packages/filepaths.py +++ b/mycli/packages/filepaths.py @@ -85,7 +85,7 @@ def suggest_path(root_dir: str) -> list[str]: *list_path(os.curdir), ] - if root_dir[0] not in ('/', '~') and root_dir[0:1] != './': + if root_dir[0] not in ('/', '~') and root_dir[0:2] != './': return list_path(os.curdir) if "~" in root_dir: diff --git a/mycli/sqlcompleter.py b/mycli/sqlcompleter.py index a779a9c6..68b3d259 100644 --- a/mycli/sqlcompleter.py +++ b/mycli/sqlcompleter.py @@ -3,7 +3,10 @@ from collections import Counter from enum import IntEnum import logging +import os import re +import shlex +import subprocess from typing import Any, Collection, Generator, Iterable, Literal from jinja2 import TemplateError @@ -12,6 +15,7 @@ from pygments.lexers._mysql_builtins import MYSQL_DATATYPES, MYSQL_FUNCTIONS, MYSQL_KEYWORDS import rapidfuzz +from mycli.compat import WIN from mycli.packages.completion_engine import is_inside_quotes, suggest_type from mycli.packages.filepaths import complete_path, parse_path, suggest_path from mycli.packages.special import llm @@ -1454,6 +1458,7 @@ def get_completions( suggestions = suggest_type(document.text, document.text_before_cursor) rigid_sort = False length_based_on_path = False + source_file_completion_length: int | None = None config_property_length: int | None = None completion_filter_text = text_for_len @@ -1719,7 +1724,26 @@ def get_completions( completions.extend([(*x, rank) for x in formats_m]) elif suggestion["type"] == "file_name": - file_names_m = self.find_files(word_before_cursor) + source_filename = suggestion.get('source_filename') + if source_filename is None: + file_names_m = self.find_files(word_before_cursor) + else: + source_file_completion_length = len(source_filename) + quote = source_filename[0] if source_filename[:1] in ("'", '"') else None + partial_path = source_filename[1:] if quote else source_filename + if quote and partial_path.endswith(quote): + partial_path = partial_path[:-1] + base_path, _last_path, _position = parse_path(partial_path) + file_names_m = ( + ( + self._quote_source_path( + os.path.join(base_path, path) if base_path and not path.startswith('~') else path, + quote, + ), + fuzziness, + ) + for path, fuzziness in self.find_files(partial_path) + ) completions.extend([(*x, rank) for x in file_names_m]) # for filenames we _really_ want directories to go last rigid_sort = True @@ -1805,6 +1829,8 @@ def completion_sort_key(item: tuple[str, int, int], text_for_len: str): if config_property_length is not None: return (Completion(x, -config_property_length) for x in uniq_completions_str) + elif source_file_completion_length is not None: + return (Completion(x, -source_file_completion_length) for x in uniq_completions_str) elif length_based_on_path: return ( Completion( @@ -1842,6 +1868,19 @@ def find_files(self, word: str) -> Generator[tuple[str, int], None, None]: if suggestion: yield (suggestion, Fuzziness.PERFECT) + @staticmethod + def _quote_source_path(path: str, quote: str | None) -> str: + is_directory = path.endswith(('/', os.sep)) + if is_directory and any(character.isspace() for character in path) and not path.startswith(('/', '~', './')): + path = f'./{path}' + if quote: + return f'{quote}{path}' if is_directory else f'{quote}{path}{quote}' + if not any(character.isspace() for character in path): + return path + if is_directory: + return f'"{path}' if WIN else f"'{path}" + return subprocess.list2cmdline([path]) if WIN else shlex.quote(path) + def populate_scoped_cols(self, scoped_tbls: list[tuple[str | None, str, str | None]]) -> list[str]: """Find all columns in a set of scoped_tables :param scoped_tbls: list of (schema, table, alias) tuples diff --git a/test/pytests/test_client_commands.py b/test/pytests/test_client_commands.py index c6d71caf..7dcab4d4 100644 --- a/test/pytests/test_client_commands.py +++ b/test/pytests/test_client_commands.py @@ -113,6 +113,55 @@ def test_parse_source_arguments(arg: str, expected: tuple[str, bool, bool, bool] assert client_commands._parse_source_arguments(arg) == expected +@pytest.mark.parametrize( + ('filename', 'expected'), + [ + ('query.sql', 'query.sql'), + ('"query file.sql"', 'query file.sql'), + ("'query file.sql'", 'query file.sql'), + ('prefix" query".sql', 'prefix query.sql'), + ], +) +def test_parse_source_filename(filename: str, expected: str) -> None: + assert client_commands._parse_source_filename(filename) == expected + + +@pytest.mark.parametrize( + 'filename', + [ + 'query file.sql', + r'query\ file.sql', + '"first file.sql" second.sql', + ], +) +def test_parse_source_filename_rejects_multiple_unquoted_arguments(filename: str) -> None: + with pytest.raises(ValueError, match='filenames containing spaces must be quoted'): + client_commands._parse_source_filename(filename) + + +def test_parse_source_filename_rejects_unclosed_quote() -> None: + with pytest.raises(ValueError, match='No closing quotation'): + client_commands._parse_source_filename('"query file.sql') + + +def test_parse_source_filename_rejects_missing_parsed_argument(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(client_commands.shlex, 'split', lambda *_args, **_kwargs: []) + + with pytest.raises(ValueError, match='accepts exactly one filename'): + client_commands._parse_source_filename('query.sql') + + +def test_source_filename_whitespace_scanner_allows_escaped_non_whitespace() -> None: + assert not client_commands._has_unquoted_whitespace(r'query\name.sql') + + +def test_parse_source_filename_preserves_windows_backslashes(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(client_commands, 'WIN', True) + + assert client_commands._parse_source_filename(r'C:\queries\query.sql') == r'C:\queries\query.sql' + assert client_commands._parse_source_filename(r'"C:\my queries\query.sql"') == r'C:\my queries\query.sql' + + def test_register_special_commands_registers_expected_commands(monkeypatch: pytest.MonkeyPatch) -> None: client = DummyClient() calls: list[tuple[Any, ...]] = [] @@ -502,7 +551,7 @@ def test_change_db_without_argument_reports_error(monkeypatch: pytest.MonkeyPatc def test_execute_from_file_requires_filename() -> None: client = DummyClient() - assert list(client.execute_from_file('')) == [SQLResult(status='Missing required argument: filename.')] + assert list(client.execute_from_file('')) == [SQLResult(status='Missing required argument: filename.', is_error=True)] def test_execute_from_file_reports_open_errors() -> None: @@ -706,21 +755,46 @@ def open_file(path: str) -> IteratedFile: return file_h monkeypatch.setattr(client_commands, 'open', open_file, raising=False) + monkeypatch.setattr(client_commands.os.path, 'expanduser', lambda path: f'/expanded/{path.removeprefix("~/")}') - assert result_statuses(client.execute_from_file('--special query file.sql')) == ['ran select 1;'] - assert opened_paths == ['query file.sql'] + assert result_statuses(client.execute_from_file('--special "~/query file.sql"')) == ['ran select 1;'] + assert opened_paths == ['/expanded/query file.sql'] @pytest.mark.parametrize('options', ['--special', '--show', '--page', '--special --show --page']) def test_execute_from_file_reports_missing_filename_after_options(options: str) -> None: client = DummyClient() - expected = [SQLResult(status='Missing required argument: filename.')] + expected = [SQLResult(status='Missing required argument: filename.', is_error=True)] if '--page' in options: expected.insert(0, SQLResult(command={'name': 'source_page'})) assert list(client.execute_from_file(options)) == expected +def test_execute_from_file_rejects_unquoted_filename_with_spaces(monkeypatch: pytest.MonkeyPatch) -> None: + client = DummyClient() + opened_paths: list[str] = [] + monkeypatch.setattr(client_commands, 'open', lambda path: opened_paths.append(path), raising=False) + + assert list(client.execute_from_file('query file.sql')) == [SQLResult(status=client_commands.INVALID_SOURCE_FILENAME, is_error=True)] + assert opened_paths == [] + + +def test_execute_from_file_pages_invalid_filename_error() -> None: + client = DummyClient() + + assert list(client.execute_from_file('--page query file.sql')) == [ + SQLResult(command={'name': 'source_page'}), + SQLResult(status=client_commands.INVALID_SOURCE_FILENAME, is_error=True), + ] + + +def test_execute_from_file_treats_empty_quotes_as_missing_filename() -> None: + client = DummyClient() + + assert list(client.execute_from_file('""')) == [SQLResult(status='Missing required argument: filename.', is_error=True)] + + def test_execute_from_file_runs_permitted_special_commands(capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: client = DummyClient() sql_file = tmp_path / 'query.sql' diff --git a/test/pytests/test_completion_engine.py b/test/pytests/test_completion_engine.py index e6374082..70343e71 100644 --- a/test/pytests/test_completion_engine.py +++ b/test/pytests/test_completion_engine.py @@ -59,6 +59,8 @@ suggest_type, ) +SOURCE_FILE_SUGGESTION = {'type': 'file_name', 'quote_spaces': True, 'source_filename': ''} + def sorted_dicts(dicts): """input is a list of dicts.""" @@ -889,31 +891,34 @@ def test_suggest_type_handles_parser_results_shorter_than_cursor(monkeypatch): ('\\dt+ ', [{'type': 'table', 'schema': []}, {'type': 'view', 'schema': []}, {'type': 'schema'}]), ( '\\. ', - [{'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}, {'type': 'file_name'}], + [{'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}, SOURCE_FILE_SUGGESTION], ), ( 'source ', - [{'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}, {'type': 'file_name'}], + [{'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}, SOURCE_FILE_SUGGESTION], ), ('source --s', [{'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}]), ('source --special', []), ( 'source --special ', - [{'type': 'special_subcommand', 'subcommands': ['--show', '--page']}, {'type': 'file_name'}], + [{'type': 'special_subcommand', 'subcommands': ['--show', '--page']}, SOURCE_FILE_SUGGESTION], ), ('source --special --s', [{'type': 'special_subcommand', 'subcommands': ['--show', '--page']}]), ('source --show', []), ( 'source --show ', - [{'type': 'special_subcommand', 'subcommands': ['--special', '--page']}, {'type': 'file_name'}], + [{'type': 'special_subcommand', 'subcommands': ['--special', '--page']}, SOURCE_FILE_SUGGESTION], ), ( 'source --show --special ', - [{'type': 'special_subcommand', 'subcommands': ['--page']}, {'type': 'file_name'}], + [{'type': 'special_subcommand', 'subcommands': ['--page']}, SOURCE_FILE_SUGGESTION], + ), + ('source --show --special --page ', [SOURCE_FILE_SUGGESTION]), + ( + 'source --special query.sql', + [{'type': 'file_name', 'quote_spaces': True, 'source_filename': 'query.sql'}], ), - ('source --show --special --page ', [{'type': 'file_name'}]), - ('source --special query.sql', [{'type': 'file_name'}]), - ('source query.sql', [{'type': 'file_name'}]), + ('source query.sql', [{'type': 'file_name', 'quote_spaces': True, 'source_filename': 'query.sql'}]), ('\\o ', [{'type': 'file_name'}]), ('\\once ', [{'type': 'file_name'}]), ('tee ', [{'type': 'file_name'}]), @@ -1869,7 +1874,7 @@ def test_source_is_file(expression): suggestions = suggest_type(expression, expression) assert suggestions == [ {'type': 'special_subcommand', 'subcommands': ['--special', '--show', '--page']}, - {'type': 'file_name'}, + SOURCE_FILE_SUGGESTION, ] diff --git a/test/pytests/test_filepaths.py b/test/pytests/test_filepaths.py index 5acada46..97c69a51 100644 --- a/test/pytests/test_filepaths.py +++ b/test/pytests/test_filepaths.py @@ -91,6 +91,7 @@ def test_suggest_path_branches(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) nested.mkdir() (nested / 'inside.sql').write_text('select 1\n', encoding='utf-8') assert filepaths.suggest_path(str(nested / 'missing.sql')) == ['inside.sql'] + assert filepaths.suggest_path('./nested/missing.sql') == ['inside.sql'] def test_dir_path_exists(tmp_path: Path) -> None: diff --git a/test/pytests/test_smart_completion_public_schema_only.py b/test/pytests/test_smart_completion_public_schema_only.py index 3e4a0ca9..84423b94 100644 --- a/test/pytests/test_smart_completion_public_schema_only.py +++ b/test/pytests/test_smart_completion_public_schema_only.py @@ -726,11 +726,14 @@ def dummy_list_path(dir_name): ('source --show ', [('--special', 0), ('--page', 0), ('/', 0), ('~', 0), ('.', 0), ('..', 0)]), ('source --special --show ', [('--page', 0), ('/', 0), ('~', 0), ('.', 0), ('..', 0)]), ('source --special --show --page ', [('/', 0), ('~', 0), ('.', 0), ('..', 0)]), - ("source /", [("dir1", 0), ("file1.sql", 0), ("file2.sql", 0)]), - ('source --special /', [('dir1', 0), ('file1.sql', 0), ('file2.sql', 0)]), - ('source --show /', [('dir1', 0), ('file1.sql', 0), ('file2.sql', 0)]), - ("source /dir1/", [("subdir1", 0), ("subfile1.sql", 0), ("subfile2.sql", 0)]), - ("source /dir1/subdir1/", [("lastfile.sql", 0)]), + ("source /", [("/dir1", -1), ("/file1.sql", -1), ("/file2.sql", -1)]), + ('source --special /', [('/dir1', -1), ('/file1.sql', -1), ('/file2.sql', -1)]), + ('source --show /', [('/dir1', -1), ('/file1.sql', -1), ('/file2.sql', -1)]), + ( + "source /dir1/", + [("/dir1/subdir1", -6), ("/dir1/subfile1.sql", -6), ("/dir1/subfile2.sql", -6)], + ), + ("source /dir1/subdir1/", [("/dir1/subdir1/lastfile.sql", -14)]), ], ) @pytest.mark.skipif(os.name == 'nt', reason='todo: unknown') @@ -803,6 +806,63 @@ def test_source_eager_completion(completer, complete_event, tmp_path, monkeypatc raise AssertionError(error) +@pytest.mark.skipif(os.name == 'nt', reason='POSIX quoting expectations') +def test_source_completion_quotes_paths_with_spaces(completer, complete_event, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / 'spaced query.sql').touch() + (tmp_path / 'spaced dir').mkdir() + (tmp_path / 'spaced dir' / 'file.sql').touch() + special.register_special_command( + ..., + 'source', + '\\. ', + 'Execute commands from file.', + aliases=[special.SpecialCommandAlias('\\.', case_sensitive=False)], + ) + + text = 'source spaced' + result = list(completer.get_completions(Document(text=text, cursor_position=len(text)), complete_event)) + assert result == [ + Completion(text="'spaced query.sql'", start_position=-6), + Completion(text="'./spaced dir/", start_position=-6), + ] + + text = 'source "spaced' + result = list(completer.get_completions(Document(text=text, cursor_position=len(text)), complete_event)) + assert result == [ + Completion(text='"spaced query.sql"', start_position=-7), + Completion(text='"./spaced dir/', start_position=-7), + ] + + text = 'source "spaced query.sql"' + result = list(completer.get_completions(Document(text=text, cursor_position=len(text)), complete_event)) + assert result == [Completion(text='"spaced query.sql"', start_position=-18)] + + text = "source './spaced dir/fi" + result = list(completer.get_completions(Document(text=text, cursor_position=len(text)), complete_event)) + assert result == [Completion(text="'./spaced dir/file.sql'", start_position=-16)] + + +def test_non_source_file_completion_uses_current_path_token(completer, complete_event, monkeypatch): + import mycli.sqlcompleter as sqlcompleter + + monkeypatch.setattr(sqlcompleter, 'suggest_type', lambda *_args: [{'type': 'file_name'}]) + monkeypatch.setattr(completer, 'find_files', lambda _path: [('file.sql', 0)]) + + result = list(completer.get_completions(Document('fi'), complete_event)) + + assert result == [Completion(text='file.sql', start_position=-2)] + + +def test_source_path_completion_uses_windows_quotes(monkeypatch: pytest.MonkeyPatch) -> None: + import mycli.sqlcompleter as sqlcompleter + + monkeypatch.setattr(sqlcompleter, 'WIN', True) + + assert sqlcompleter.SQLCompleter._quote_source_path(r'C:\my queries\query.sql', None) == r'"C:\my queries\query.sql"' + assert sqlcompleter.SQLCompleter._quote_source_path('my directory/', None) == '"./my directory/' + + def test_source_leading_dot_suggestions_completion(completer, complete_event, tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) os.mkdir('doc') @@ -823,8 +883,8 @@ def test_source_leading_dot_suggestions_completion(completer, complete_event, tm error = 'unknown' try: assert [x.text for x in result] == [ - script_filename, - 'doc/', + f'./{script_filename}', + './doc/', ] except AssertionError as e: success = False