diff --git a/changelog.md b/changelog.md index b8f7cb04..fff00259 100644 --- a/changelog.md +++ b/changelog.md @@ -25,6 +25,7 @@ Internal * Upgrade `coverage` dev dependency to v7.15.4. * Upgrade `pytest-cov` dev dependency to v7.1.0. * Upgrade `wcwidth` dependency to v0.8.3. +* Use `argparse` to process arguments to `/source`. 2.19.0 (2026/09/02) diff --git a/mycli/client_commands.py b/mycli/client_commands.py index 032852f4..718ce520 100644 --- a/mycli/client_commands.py +++ b/mycli/client_commands.py @@ -20,7 +20,6 @@ from mycli.packages.special.source import ( SOURCE_HELP_ROWS, parse_source_arguments, - parse_source_filename, source_special_command_is_safe, ) from mycli.packages.sqlresult import SQLResult @@ -291,20 +290,16 @@ def change_db(self, arg: str, **_) -> Generator[SQLResult, None, None]: def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]: try: - filename, allow_special, show_queries, page_output, throttle, show_help = parse_source_arguments(arg) + source_arguments = parse_source_arguments(arg) except ValueError as error: yield SQLResult(status=str(error), is_error=True) return - if show_help: + if source_arguments.show_help: yield SQLResult(header=['Argument', 'Description'], rows=SOURCE_HELP_ROWS) return - if page_output: + if source_arguments.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 + filename = source_arguments.filename if not filename: yield SQLResult(status="Missing required argument: filename. See /source --help.", is_error=True) return @@ -330,7 +325,7 @@ def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]: special_query = query.rstrip(';') if special.is_special_command(special_query): - if not allow_special: + if not source_arguments.allow_special: yield SQLResult( status='Special commands are not supported without /source --special.', is_error=True, @@ -343,10 +338,10 @@ def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]: is_error=True, ) return - if executed_statement and throttle > 0: - time.sleep(throttle) - if show_queries: - if page_output: + if executed_statement and source_arguments.throttle > 0: + time.sleep(source_arguments.throttle) + if source_arguments.show_queries: + if source_arguments.page_output: yield SQLResult(command={'name': 'source_show', 'text': special_query}) else: click.secho(f'> {special_query}') @@ -356,10 +351,10 @@ def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]: if self.destructive_warning and confirm_destructive_query(self.destructive_keywords, query) is False: continue - if executed_statement and throttle > 0: - time.sleep(throttle) - if show_queries: - if page_output: + if executed_statement and source_arguments.throttle > 0: + time.sleep(source_arguments.throttle) + if source_arguments.show_queries: + if source_arguments.page_output: yield SQLResult(command={'name': 'source_show', 'text': query}) else: click.secho(f'> {query}') diff --git a/mycli/packages/completion_engine.py b/mycli/packages/completion_engine.py index 128dbd86..caaf63ec 100644 --- a/mycli/packages/completion_engine.py +++ b/mycli/packages/completion_engine.py @@ -818,7 +818,23 @@ def suggest_special(text: str) -> list[dict[str, Any]]: '/source', ]: source_options = list(SOURCE_OPTIONS) - source_arguments = _arg.split() + try: + source_arguments = shlex.split(_arg, posix=False) + except ValueError: + source_filename = _arg + while leading_arguments := source_filename.split(maxsplit=1): + if leading_arguments[0] in SOURCE_BOOLEAN_OPTIONS: + source_filename = leading_arguments[1] if len(leading_arguments) == 2 else '' + continue + if leading_arguments[0] == '--throttle' and len(leading_arguments) == 2: + throttle_arguments = leading_arguments[1].split(maxsplit=1) + source_filename = throttle_arguments[1] if len(throttle_arguments) == 2 else '' + continue + if leading_arguments[0].startswith('--throttle='): + source_filename = leading_arguments[1] if len(leading_arguments) == 2 else '' + continue + break + return [{'type': 'file_name', 'quote_spaces': True, 'source_filename': source_filename}] if not source_arguments: return [ {'type': 'special_subcommand', 'subcommands': source_options}, @@ -827,15 +843,22 @@ def suggest_special(text: str) -> list[dict[str, Any]]: used_options: set[str] = set() argument_index = 0 + filename_index: int | None = None + options_ended = False while argument_index < len(source_arguments): argument = source_arguments[argument_index] - if argument in SOURCE_BOOLEAN_OPTIONS: + if not options_ended and argument == '--': + options_ended = True + used_options.update(source_options) + argument_index += 1 + continue + if not options_ended and argument in SOURCE_BOOLEAN_OPTIONS: used_options.add(argument) argument_index += 1 continue - if argument == '--help': + if not options_ended and argument == '--help': return [] - if argument == '--throttle': + if not options_ended and argument == '--throttle': used_options.add(argument) argument_index += 1 if argument_index >= len(source_arguments): @@ -844,30 +867,31 @@ def suggest_special(text: str) -> list[dict[str, Any]]: return [] argument_index += 1 continue - if argument.startswith('--throttle='): + if not options_ended and argument.startswith('--throttle='): used_options.add('--throttle') if argument == '--throttle=' or not text[-1].isspace(): return [] argument_index += 1 continue - break + if not options_ended and argument.startswith('-'): + break + if filename_index is not None: + return [] + filename_index = 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 '' + source_filename = source_arguments[filename_index] if filename_index is not None 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 [file_suggestion] + return [{'type': 'special_subcommand', 'subcommands': remaining_options}] if not text[-1].isspace(): - return [] + return [file_suggestion] if filename_index == len(source_arguments) - 1 else [] suggestions: list[dict[str, Any]] = [] if remaining_options: suggestions.append({'type': 'special_subcommand', 'subcommands': remaining_options}) - suggestions.append(file_suggestion) + if filename_index is None: + suggestions.append(file_suggestion) return suggestions if cmd.lower() in [ diff --git a/mycli/packages/hybrid_redirection.py b/mycli/packages/hybrid_redirection.py index 9b1bc519..800f7dd5 100644 --- a/mycli/packages/hybrid_redirection.py +++ b/mycli/packages/hybrid_redirection.py @@ -9,26 +9,26 @@ from mycli.packages.special.delimitercommand import DelimiterCommand from mycli.packages.special.source import ( parse_source_arguments, - parse_source_filename, ) logger = logging.getLogger(__name__) delimiter_command = DelimiterCommand() SOURCE_COMMAND_PATTERN = re.compile(r'^([/]?source|[/\\]\.)\s+', re.IGNORECASE) -SOURCE_OPTIONS_PATTERN = re.compile( - r'^([/]?source|[/\\]\.)\s+' - r'(?P(?:(?:--special|--show|--page)\s+|--throttle(?:=[^\s]+|\s+[^\s]+)\s+)*)', - re.IGNORECASE, -) +SOURCE_OPTION_PATTERN = re.compile(r'(?>?|\|)') def tokenize_command(command: str) -> list[sqlglot.Token]: """Tokenize a command without treating source options as SQL comments.""" - source_match = SOURCE_OPTIONS_PATTERN.match(command) - if source_match: - options_start, options_end = source_match.span('options') - options = command[options_start:options_end].replace('-', '_') - command = command[:options_start] + options + command[options_end:] + if SOURCE_COMMAND_PATTERN.match(command): + operator_match = HYBRID_OPERATOR_PATTERN.search(command) + source_end = operator_match.start() if operator_match is not None else len(command) + source_part = command[:source_end] + terminator_match = SOURCE_OPTION_TERMINATOR_PATTERN.search(source_part) + options_end = terminator_match.start() if terminator_match is not None else len(source_part) + masked_options = SOURCE_OPTION_PATTERN.sub(lambda match: match.group().replace('-', '_'), source_part[:options_end]) + command = masked_options + source_part[options_end:] + command[source_end:] return sqlglot.tokenize(command) @@ -68,8 +68,7 @@ def find_sql_part( if SOURCE_COMMAND_PATTERN.match(sql_part): source_arg_str = SOURCE_COMMAND_PATTERN.sub('', sql_part) try: - filename, _allow_special, _show_queries, _page_output, _throttle, _show_help = parse_source_arguments(source_arg_str) - filename = parse_source_filename(filename) + filename = parse_source_arguments(source_arg_str).filename except ValueError: return '' if not filename: diff --git a/mycli/packages/special/__init__.py b/mycli/packages/special/__init__.py index fb8afca9..ffd60dac 100644 --- a/mycli/packages/special/__init__.py +++ b/mycli/packages/special/__init__.py @@ -55,10 +55,7 @@ write_pipe_once, write_tee, ) -from mycli.packages.special.source import ( - parse_source_arguments, - parse_source_filename, -) +from mycli.packages.special.source import parse_source_arguments if not os.environ.get('MYCLI_LLM_OFF'): from mycli.packages.special.llm import ( @@ -125,7 +122,6 @@ def sql_using_llm(*args, **kwargs): # type: ignore[no-redef, misc] 'list_tables', 'open_external_editor', 'parse_source_arguments', - 'parse_source_filename', 'parse_special_command', 'ping', 'register_special_command', diff --git a/mycli/packages/special/source.py b/mycli/packages/special/source.py index e1f4cf08..069fa258 100644 --- a/mycli/packages/special/source.py +++ b/mycli/packages/special/source.py @@ -1,5 +1,7 @@ -import math +import argparse +from dataclasses import dataclass import shlex +from typing import Any, NoReturn import sqlparse @@ -47,26 +49,45 @@ } -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 +@dataclass(frozen=True) +class SourceArguments: + filename: str = '' + allow_special: bool = False + show_queries: bool = False + page_output: bool = False + throttle: float = 0.0 + show_help: bool = False + + +class _SourceHelpRequested(Exception): + pass + + +class _SourceHelpAction(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: Any, + option_string: str | None = None, + ) -> NoReturn: + raise _SourceHelpRequested + + +class _SourceArgumentParser(argparse.ArgumentParser): + def error(self, message: str) -> NoReturn: + throttle_prefix = 'argument --throttle: ' + if message == f'{throttle_prefix}expected one argument': + raise ValueError('Missing value for --throttle.') + if message.startswith(throttle_prefix): + raise ValueError(message.removeprefix(throttle_prefix)) + if message.startswith('unrecognized arguments:'): + arguments = message.removeprefix('unrecognized arguments:').strip().split() + unknown_option = next((argument for argument in arguments if argument.startswith('-')), None) + if unknown_option is not None: + raise ValueError(f'Unrecognized /source option: {unknown_option}. See /source --help.') + raise ValueError(INVALID_SOURCE_FILENAME) + raise ValueError(f'Invalid /source arguments: {message}.') def _registered_special_command(query: str) -> tuple[str, str] | None: @@ -86,63 +107,40 @@ def _favorite_source_command_is_safe(arg: str) -> bool: return not any(special.is_special_command(statement.rstrip(';')) for statement in sqlparse.split(query)) -def _parse_throttle(value: str) -> float: - if not value: - raise ValueError('Missing value for --throttle.') - try: - throttle = float(value) - except ValueError: - raise ValueError(f'Invalid --throttle value: {value}. Expected a finite, non-negative number.') from None - if not math.isfinite(throttle) or throttle < 0: - raise ValueError(f'Invalid --throttle value: {value}. Expected a finite, non-negative number.') - return throttle - - -def parse_source_arguments(arg: str) -> tuple[str, bool, bool, bool, float, bool]: - allow_special = False - show_queries = False - page_output = False - throttle = 0.0 - filename = arg - while arguments := filename.split(maxsplit=1): - if arguments[0] == '--special': - allow_special = True - elif arguments[0] == '--show': - show_queries = True - elif arguments[0] == '--page': - page_output = True - elif arguments[0] == '--help': - return '', allow_special, show_queries, page_output, throttle, True - elif arguments[0] == '--throttle': - if len(arguments) != 2: - raise ValueError('Missing value for --throttle.') - throttle_arguments = arguments[1].split(maxsplit=1) - throttle = _parse_throttle(throttle_arguments[0]) - filename = throttle_arguments[1] if len(throttle_arguments) == 2 else '' - continue - elif arguments[0].startswith('--throttle='): - throttle = _parse_throttle(arguments[0].partition('=')[2]) - else: - break - filename = arguments[1] if len(arguments) == 2 else '' - return filename, allow_special, show_queries, page_output, throttle, False - - -def parse_source_filename(filename: str) -> str: - if not filename: - return '' - if _has_unquoted_whitespace(filename): - raise ValueError(INVALID_SOURCE_FILENAME) +def _create_source_argument_parser() -> _SourceArgumentParser: + parser = _SourceArgumentParser(prog='/source', add_help=False, allow_abbrev=False) + parser.add_argument('--special', dest='allow_special', action='store_true') + parser.add_argument('--show', dest='show_queries', action='store_true') + parser.add_argument('--page', dest='page_output', action='store_true') + parser.add_argument('--throttle', type=float, default=0.0) + parser.add_argument('--help', nargs=0, action=_SourceHelpAction) + parser.add_argument('filename', nargs='?') + return parser + + +_SOURCE_ARGUMENT_PARSER = _create_source_argument_parser() + + +def parse_source_arguments(arg: str) -> SourceArguments: try: - arguments = shlex.split(filename, posix=not WIN) + arguments = shlex.split(arg, 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 + try: + parsed = _SOURCE_ARGUMENT_PARSER.parse_args(arguments) + except _SourceHelpRequested: + return SourceArguments(show_help=True) + + filename = parsed.filename or '' + if WIN and len(filename) >= 2 and filename[0] == filename[-1] and filename[0] in ("'", '"'): + filename = filename[1:-1] + return SourceArguments( + filename=filename, + allow_special=parsed.allow_special, + show_queries=parsed.show_queries, + page_output=parsed.page_output, + throttle=parsed.throttle, + ) def source_special_command_is_safe(query: str) -> bool: diff --git a/test/pytests/test_client_commands.py b/test/pytests/test_client_commands.py index 0dd54e65..322c71ca 100644 --- a/test/pytests/test_client_commands.py +++ b/test/pytests/test_client_commands.py @@ -720,7 +720,7 @@ def test_execute_from_file_reports_invalid_throttle_without_opening_file(monkeyp assert list(client.execute_from_file('--throttle nope query.sql')) == [ SQLResult( - status='Invalid --throttle value: nope. Expected a finite, non-negative number.', + status="invalid float value: 'nope'", is_error=True, ) ] @@ -825,7 +825,6 @@ 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=source_commands.INVALID_SOURCE_FILENAME, is_error=True), ] diff --git a/test/pytests/test_completion_engine.py b/test/pytests/test_completion_engine.py index 22659a3d..290534f2 100644 --- a/test/pytests/test_completion_engine.py +++ b/test/pytests/test_completion_engine.py @@ -961,6 +961,19 @@ def test_suggest_type_handles_parser_results_shorter_than_cursor(monkeypatch): ('source --help', []), ('source --help ', []), ('source --show --help ignored.sql', []), + ( + 'source --show "query', + [{'type': 'file_name', 'quote_spaces': True, 'source_filename': '"query'}], + ), + ( + 'source --throttle 0.25 "query', + [{'type': 'file_name', 'quote_spaces': True, 'source_filename': '"query'}], + ), + ( + 'source --throttle=0.25 "query', + [{'type': 'file_name', 'quote_spaces': True, 'source_filename': '"query'}], + ), + ('source --throttle "', [{'type': 'file_name', 'quote_spaces': True, 'source_filename': ''}]), ('source --throttle', []), ('source --throttle ', []), ('source --throttle 0.25', []), @@ -984,6 +997,37 @@ def test_suggest_type_handles_parser_results_shorter_than_cursor(monkeypatch): [{'type': 'file_name', 'quote_spaces': True, 'source_filename': 'query.sql'}], ), ('source query.sql', [{'type': 'file_name', 'quote_spaces': True, 'source_filename': 'query.sql'}]), + ( + 'source query.sql ', + [ + { + 'type': 'special_subcommand', + 'subcommands': ['--special', '--show', '--page', '--throttle', '--help'], + } + ], + ), + ( + 'source query.sql --show ', + [ + { + 'type': 'special_subcommand', + 'subcommands': ['--special', '--page', '--throttle', '--help'], + } + ], + ), + ('source query.sql --throttle ', []), + ( + 'source query.sql --throttle 0.25 ', + [ + { + 'type': 'special_subcommand', + 'subcommands': ['--special', '--show', '--page', '--help'], + } + ], + ), + ('source -- ', [SOURCE_FILE_SUGGESTION]), + ('source -- query.sql', [{'type': 'file_name', 'quote_spaces': True, 'source_filename': 'query.sql'}]), + ('source first.sql second.sql', []), ('\\o ', [{'type': 'file_name'}]), ('\\once ', [{'type': 'file_name'}]), ('tee ', [{'type': 'file_name'}]), diff --git a/test/pytests/test_hybrid_redirection.py b/test/pytests/test_hybrid_redirection.py index e866298f..3eb0b3c0 100644 --- a/test/pytests/test_hybrid_redirection.py +++ b/test/pytests/test_hybrid_redirection.py @@ -177,6 +177,18 @@ def test_get_redirect_components_preserves_source_options(option: str) -> None: ) +@pytest.mark.parametrize('option', ['--special', '--show', '--page', '--throttle 0.25', '--throttle=0.25']) +def test_get_redirect_components_preserves_source_options_after_filename(option: str) -> None: + command = f'/source query.sql {option} $> out.txt' + + assert hybrid_redirection.get_redirect_components(command) == ( + f'/source query.sql {option}', + None, + '>', + 'out.txt', + ) + + def test_get_redirect_components_handles_combined_source_options_and_redirects() -> None: command = '/source --page --show --special query.sql $| cat $>> out.txt' diff --git a/test/pytests/test_smart_completion_public_schema_only.py b/test/pytests/test_smart_completion_public_schema_only.py index 2f9a9242..070c7f3a 100644 --- a/test/pytests/test_smart_completion_public_schema_only.py +++ b/test/pytests/test_smart_completion_public_schema_only.py @@ -772,6 +772,15 @@ def dummy_list_path(dir_name): ("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 file.sql ', + [('--special', 0), ('--show', 0), ('--page', 0), ('--throttle', 0), ('--help', 0)], + ), + ( + 'source file.sql --show ', + [('--special', 0), ('--page', 0), ('--throttle', 0), ('--help', 0)], + ), + ('source -- ', [('/', 0), ('~', 0), ('.', 0), ('..', 0)]), ( "source /dir1/", [("/dir1/subdir1", -6), ("/dir1/subfile1.sql", -6), ("/dir1/subfile2.sql", -6)], diff --git a/test/pytests/test_special_source.py b/test/pytests/test_special_source.py index b0626f41..8beecf83 100644 --- a/test/pytests/test_special_source.py +++ b/test/pytests/test_special_source.py @@ -7,23 +7,34 @@ @pytest.mark.parametrize( ('arg', 'expected'), [ - ('query.sql', ('query.sql', False, False, False, 0.0, False)), - ('--special query.sql', ('query.sql', True, False, False, 0.0, False)), - ('--show query.sql', ('query.sql', False, True, False, 0.0, False)), - ('--page query.sql', ('query.sql', False, False, True, 0.0, False)), - ('--special --show --page query file.sql', ('query file.sql', True, True, True, 0.0, False)), - ('--page --show --special query file.sql', ('query file.sql', True, True, True, 0.0, False)), - ('--show --show query.sql', ('query.sql', False, True, False, 0.0, False)), - ('--page --page query.sql', ('query.sql', False, False, True, 0.0, False)), - ('--show', ('', False, True, False, 0.0, False)), - ('--throttle 0.25 query.sql', ('query.sql', False, False, False, 0.25, False)), - ('--throttle=1e-2 query.sql', ('query.sql', False, False, False, 0.01, False)), - ('--throttle 1 --show --throttle=0.5 query.sql', ('query.sql', False, True, False, 0.5, False)), - ('--help', ('', False, False, False, 0.0, True)), - ('--show --help ignored.sql', ('', False, True, False, 0.0, True)), + ('query.sql', source.SourceArguments(filename='query.sql')), + ('--special query.sql', source.SourceArguments(filename='query.sql', allow_special=True)), + ('--show query.sql', source.SourceArguments(filename='query.sql', show_queries=True)), + ('query.sql --show', source.SourceArguments(filename='query.sql', show_queries=True)), + ('--page query.sql', source.SourceArguments(filename='query.sql', page_output=True)), + ( + '--page query.sql --show --special', + source.SourceArguments(filename='query.sql', allow_special=True, show_queries=True, page_output=True), + ), + ('--show --show query.sql', source.SourceArguments(filename='query.sql', show_queries=True)), + ('--page --page query.sql', source.SourceArguments(filename='query.sql', page_output=True)), + ('--show', source.SourceArguments(show_queries=True)), + ('--throttle 0.25 query.sql', source.SourceArguments(filename='query.sql', throttle=0.25)), + ('query.sql --throttle=1e-2', source.SourceArguments(filename='query.sql', throttle=0.01)), + ( + '--throttle 1 --show query.sql --throttle=0.5', + source.SourceArguments(filename='query.sql', show_queries=True, throttle=0.5), + ), + ('"query file.sql"', source.SourceArguments(filename='query file.sql')), + (r'query\ file.sql', source.SourceArguments(filename='query file.sql')), + ('prefix" query".sql', source.SourceArguments(filename='prefix query.sql')), + ('-- --show', source.SourceArguments(filename='--show')), + ('--help', source.SourceArguments(show_help=True)), + ('--show --help ignored.sql', source.SourceArguments(show_help=True)), + ('ignored.sql --help', source.SourceArguments(show_help=True)), ], ) -def test_parse_source_arguments(arg: str, expected: tuple[str, bool, bool, bool, float, bool]) -> None: +def test_parse_source_arguments(arg: str, expected: source.SourceArguments) -> None: assert source.parse_source_arguments(arg) == expected @@ -33,64 +44,52 @@ def test_parse_source_arguments(arg: str, expected: tuple[str, bool, bool, bool, '--throttle', '--throttle=', '--throttle nope query.sql', - '--throttle -1 query.sql', - '--throttle inf query.sql', - '--throttle nan query.sql', ], ) def test_parse_source_arguments_rejects_invalid_throttle(arg: str) -> None: - with pytest.raises(ValueError, match='throttle'): + with pytest.raises(ValueError, match='float|Missing'): source.parse_source_arguments(arg) @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 source.parse_source_filename(filename) == expected - - -@pytest.mark.parametrize( - 'filename', + 'arg', [ 'query file.sql', - r'query\ file.sql', '"first file.sql" second.sql', ], ) -def test_parse_source_filename_rejects_multiple_unquoted_arguments(filename: str) -> None: +def test_parse_source_arguments_rejects_multiple_filenames(arg: str) -> None: with pytest.raises(ValueError, match='filenames containing spaces must be quoted'): - source.parse_source_filename(filename) + source.parse_source_arguments(arg) -def test_parse_source_filename_rejects_unclosed_quote() -> None: +def test_parse_source_arguments_rejects_unclosed_quote() -> None: with pytest.raises(ValueError, match='No closing quotation'): - source.parse_source_filename('"query file.sql') + source.parse_source_arguments('"query file.sql') + + +def test_parse_source_arguments_rejects_unknown_option(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(ValueError, match=r'Unrecognized /source option: --unknown\. See /source --help\.'): + source.parse_source_arguments('--unknown query.sql') + assert capsys.readouterr() == ('', '') -def test_parse_source_filename_rejects_missing_parsed_argument(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(source.shlex, 'split', lambda *_args, **_kwargs: []) - with pytest.raises(ValueError, match='accepts exactly one filename'): - source.parse_source_filename('query.sql') +def test_parse_source_arguments_does_not_abbreviate_options() -> None: + with pytest.raises(ValueError, match=r'Unrecognized /source option: --spec\.'): + source.parse_source_arguments('--spec query.sql') -def test_source_filename_whitespace_scanner_allows_escaped_non_whitespace() -> None: - assert not source._has_unquoted_whitespace(r'query\name.sql') +def test_source_argument_parser_converts_other_errors() -> None: + with pytest.raises(ValueError, match=r'Invalid /source arguments: unexpected\.'): + source._SOURCE_ARGUMENT_PARSER.error('unexpected') -def test_parse_source_filename_preserves_windows_backslashes(monkeypatch: pytest.MonkeyPatch) -> None: +def test_parse_source_arguments_preserves_windows_backslashes(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(source, 'WIN', True) - assert source.parse_source_filename(r'C:\queries\query.sql') == r'C:\queries\query.sql' - assert source.parse_source_filename(r'"C:\my queries\query.sql"') == r'C:\my queries\query.sql' + assert source.parse_source_arguments(r'C:\queries\query.sql').filename == r'C:\queries\query.sql' + assert source.parse_source_arguments(r'"C:\my queries\query.sql"').filename == r'C:\my queries\query.sql' @pytest.mark.parametrize(