diff --git a/changelog.md b/changelog.md index a696e925e..72ff653f4 100644 --- a/changelog.md +++ b/changelog.md @@ -1,6 +1,11 @@ Upcoming (TBD) ============== +Features +--------- +* Add alternative interface `/favorite` for favorite queries. + + Bug Fixes --------- * Lower Boundary tunnel stabilization pause to 0.15 sec. diff --git a/mycli/TIPS b/mycli/TIPS index 9dfbbc217..af012d837 100644 --- a/mycli/TIPS +++ b/mycli/TIPS @@ -64,6 +64,8 @@ edit a query in an external editor using /edit ! Favorite queries support Jinja templates and named arguments! +Manage favorite queries with /favorite! + /l lists databases! /once appends the next result to ! diff --git a/mycli/clibuffer.py b/mycli/clibuffer.py index 507ead829..8d320e4d8 100644 --- a/mycli/clibuffer.py +++ b/mycli/clibuffer.py @@ -29,7 +29,7 @@ def _multiline_exception(text: str) -> bool: # Multi-statement favorite query is a special case. Because there will # be a semicolon separating statements, we can't consider semicolon an # EOL. Let's consider an empty line an EOL instead. - if first_word.startswith(("\\fs", "/fs")): + if iocommands.is_favorite_save_command(text): return orig.endswith("\n") return ( diff --git a/mycli/client.py b/mycli/client.py index 15d447624..9ffdbb4a3 100644 --- a/mycli/client.py +++ b/mycli/client.py @@ -122,6 +122,7 @@ def __init__( self.config, myclirc, c['main'].get('shared_favorites_file'), + system_config_files=self.system_config_files, ) DsnAliases.instance = DsnAliases.from_config( self.config, diff --git a/mycli/config.py b/mycli/config.py index 5df3c8c1d..b19c77fa3 100644 --- a/mycli/config.py +++ b/mycli/config.py @@ -48,6 +48,7 @@ def read_config_file( f: str | IO[str], list_values: bool = True, preserve_quotes: bool = False, + raise_errors: bool = False, ) -> ConfigObj | LimiitedQuotePreservingConfigObj | None: """Read a config file. @@ -57,6 +58,9 @@ def read_config_file( not unquoted. We are disabling list_values when reading MySQL config files so we can correctly interpret commas in passwords. + Set *raise_errors* to propagate parsing and I/O errors instead of logging + them and returning a partial config or ``None``. + """ if isinstance(f, str): @@ -68,10 +72,14 @@ def read_config_file( else: config = ConfigObj(f, interpolation=False, encoding="utf8", list_values=list_values) except ConfigObjError as e: + if raise_errors: + raise log(logger, logging.WARNING, "Unable to parse line {0} of config file '{1}'.".format(e.line_number, f)) log(logger, logging.WARNING, "Using successfully parsed config values.") return e.config except (IOError, OSError) as e: + if raise_errors: + raise log(logger, logging.WARNING, "You don't have permission to read config file '{0}'.".format(e.filename)) return None diff --git a/mycli/main_modes/repl.py b/mycli/main_modes/repl.py index b2bf0a377..8c4c2cdc6 100644 --- a/mycli/main_modes/repl.py +++ b/mycli/main_modes/repl.py @@ -105,6 +105,7 @@ class ReplState: iterations: int = 0 mutating: bool = False + buffer_text: str | None = None @Condition @@ -449,6 +450,9 @@ def _output_results( mycli.logger.debug('status: %r', result.status) mycli.logger.debug('command: %r', result.command) threshold = 1000 + if result.command is not None and result.command['name'] == 'set_buffer': + state.buffer_text = str(result.command['text']) + continue if result.command is not None and result.command['name'] == 'watch': if watch_count > 0: try: @@ -630,10 +634,19 @@ def _one_iteration( try: assert mycli.prompt_session is not None loaded_message_fn = partial(_get_prompt_message, mycli, mycli.prompt_session.app) - text = mycli.prompt_session.prompt( - inputhook=inputhook, - message=loaded_message_fn, - ) + if state.buffer_text is None: + text = mycli.prompt_session.prompt( + inputhook=inputhook, + message=loaded_message_fn, + ) + else: + default = state.buffer_text + state.buffer_text = None + text = mycli.prompt_session.prompt( + default=default, + inputhook=inputhook, + message=loaded_message_fn, + ) except KeyboardInterrupt: return diff --git a/mycli/packages/completion_engine.py b/mycli/packages/completion_engine.py index 9bc111259..d445adbd6 100644 --- a/mycli/packages/completion_engine.py +++ b/mycli/packages/completion_engine.py @@ -8,6 +8,7 @@ from sqlparse.sql import Comparison, Identifier, Token, Where from mycli.packages.special.dsn_aliases import DSN_SUBCOMMANDS +from mycli.packages.special.favoritequeries import FAVORITE_SUBCOMMANDS from mycli.packages.special.main import COMMANDS as SPECIAL_COMMANDS from mycli.packages.special.main import parse_special_command from mycli.packages.sql_utils import extract_tables, find_prev_keyword, last_word @@ -831,6 +832,36 @@ def suggest_special(text: str) -> list[dict[str, Any]]: if cmd in ["\\llm", "/llm", "\\ai", "/ai"]: return [{"type": "llm"}] + if cmd.lower() in (r'\favorite', '/favorite'): + favorite_arguments = _arg.split(maxsplit=1) + if favorite_arguments and favorite_arguments[0].lower() in ('run', 'eval'): + if len(favorite_arguments) == 1 and not text[-1].isspace(): + return [] + expansion_arg = favorite_arguments[1] if len(favorite_arguments) == 2 else '' + return suggest_favorite_query_with_template(text, expansion_arg) + if favorite_arguments and favorite_arguments[0].lower() == 'save': + if len(favorite_arguments) == 1 and not text[-1].isspace(): + return [] + if len(favorite_arguments) == 1: + return [{'type': 'favoritequery'}] + save_arguments = favorite_arguments[1].split(maxsplit=1) + if len(save_arguments) == 1 and not text[-1].isspace(): + return [{'type': 'favoritequery'}] + query = save_arguments[1] if len(save_arguments) == 2 else '' + if query and text[-1].isspace(): + query += ' ' + return suggest_type(query, query) + if favorite_arguments and favorite_arguments[0].lower() in ('edit', 'delete'): + if len(favorite_arguments) == 1: + return [] if not text[-1].isspace() else [{'type': 'favoritequery'}] + target_arguments = favorite_arguments[1].split() + if len(target_arguments) == 1 and not text[-1].isspace(): + return [{'type': 'favoritequery'}] + return [] + if favorite_arguments and favorite_arguments[0].lower() in FAVORITE_SUBCOMMANDS: + return [] + return [{'type': 'special_subcommand', 'subcommands': list(FAVORITE_SUBCOMMANDS)}] + if cmd.lower() in (r'\config', '/config'): config_arguments = _arg.split(maxsplit=1) config_subcommands = ['help', 'get', 'search', 'edit'] diff --git a/mycli/packages/special/favoritequeries.py b/mycli/packages/special/favoritequeries.py index e88909cc8..cdd086313 100644 --- a/mycli/packages/special/favoritequeries.py +++ b/mycli/packages/special/favoritequeries.py @@ -1,23 +1,85 @@ from __future__ import annotations +from collections.abc import Mapping import logging import os import re -from typing import Any +from typing import IO, Any +from configobj import ConfigObjError from jinja2 import meta, nodes from jinja2.sandbox import SandboxedEnvironment -from mycli.config import log, read_config_file +from mycli.config import log, read_config_file, read_config_files logger = logging.getLogger(__name__) MISSING = object() +FAVORITE_SUBCOMMANDS = ('help', 'list', 'reload', 'run', 'eval', 'save', 'edit', 'delete') +FAVORITE_COMMAND_HELP = ''' +Favorite Queries are a way to save frequently used queries +with a short name. +Examples: + + # Save a new favorite query. + > /favorite save simple SELECT * FROM abc WHERE a IS NOT NULL; + + # When multi-line mode is on, pressing Return twice is needed to save. + # This supports multi-statement favorites. + + # List all favorite queries. + > /favorite list + ╒═══════════╤══════════════════════════════════════════════════╕ + │ Name │ Query │ + ╞═══════════╪══════════════════════════════════════════════════╡ + │ simple │ SELECT * FROM abc WHERE a IS NOT NULL │ + │ find_user │ SELECT * FROM users WHERE name = '{{ kv.name }}' │ + ╘═══════════╧══════════════════════════════════════════════════╛ + + # Run a favorite query. + > /favorite run simple + ╒════════╤════════╕ + │ a │ b │ + ╞════════╪════════╡ + │ 日本語 │ 日本語 │ + ╘════════╧════════╛ + + # Run a favorite query containing {{ kv.name }} in the template. + > /favorite run find_user --name=henry + > /favorite run find_user --name henry + + # Run a favorite query containing positional parameter $1 in the + # template. + > /favorite run find_user henry + + # Use -- to disambiguate positional parameters such as $1, especially + # if the positional value starts with a dash. + > /favorite run query --key=value -- positional-value + > /favorite run query -- --positional-value-which-looks-like-a-flag-- + + # Expand a favorite query into the command-line buffer without running it. + > /favorite eval find_user --name=henry + + # Edit a favorite query in an external editor. + > /favorite edit simple + + # Delete a favorite query. + > /favorite delete simple + simple: Deleted. + + # Reload favorite queries from the configuration files. + > /favorite reload + + See also the alternative interface /f, /fs, /fd.''' favorite_query_template_environment = SandboxedEnvironment(autoescape=False) favorite_query_variable_pattern = re.compile(r'^[A-Za-z_][A-Za-z0-9_-]*$') +class FavoriteQueryReloadError(Exception): + pass + + def analyze_favorite_query_template(query: str) -> tuple[set[str], bool]: """Return statically referenced keys and whether ``kv`` is used dynamically.""" parsed_template = favorite_query_template_environment.parse(query) @@ -116,14 +178,24 @@ class FavoriteQueries: # Delete a favorite query. > /fd simple simple: Deleted. + + See also the alternative interface /favorite. """ # Class-level variable, for convenience to use as a singleton. instance: FavoriteQueries - def __init__(self, config: Any, config_file: str | None = None) -> None: + def __init__( + self, + config: Any, + config_file: str | None = None, + shared_favorites_file: str | None = None, + system_config_files: list[str | IO[str]] | None = None, + ) -> None: self.config = config self.config_file = config_file + self.shared_favorites_file = shared_favorites_file + self.system_config_files = list(system_config_files or []) @classmethod def from_config( @@ -131,10 +203,10 @@ def from_config( config: Any, config_file: str | None = None, shared_favorites_file: str | None = None, + system_config_files: list[str | IO[str]] | None = None, ) -> FavoriteQueries: - favorites = cls(config, config_file) if not shared_favorites_file: - return favorites + return cls(config, config_file, system_config_files=system_config_files) shared_favorites_file = os.path.expanduser(shared_favorites_file) if not os.path.isabs(shared_favorites_file): @@ -143,7 +215,9 @@ def from_config( logging.WARNING, f"Shared favorites file path must be absolute: '{shared_favorites_file}'.", ) - return favorites + return cls(config, config_file, system_config_files=system_config_files) + + favorites = cls(config, config_file, shared_favorites_file, system_config_files) if not os.path.isfile(shared_favorites_file): log( @@ -164,6 +238,45 @@ def from_config( config[cls.section_name].update(configured_queries) return favorites + def _reload_queries(self, path: str, description: str) -> dict[str, str]: + expanded_path = os.path.expanduser(path) + if not os.path.isfile(expanded_path): + raise FavoriteQueryReloadError(f"unable to read {description} file '{expanded_path}'") + try: + config = read_config_file(expanded_path, raise_errors=True) + except (ConfigObjError, OSError, UnicodeError) as exc: + raise FavoriteQueryReloadError( + f"unable to read {description} file '{expanded_path}': {exc}", + ) from exc + + assert config is not None + configured_queries = config.get(self.section_name, {}) + if not isinstance(configured_queries, Mapping) or any( + not isinstance(name, str) or not isinstance(query, str) for name, query in configured_queries.items() + ): + raise FavoriteQueryReloadError( + f"invalid [{self.section_name}] section in {description} file '{expanded_path}'", + ) + return dict(configured_queries) + + def reload(self) -> None: + if self.config_file is None: + raise FavoriteQueryReloadError('no user configuration file is configured') + + user_queries = self._reload_queries(self.config_file, 'user configuration') + system_config = read_config_files(self.system_config_files, ignore_package_defaults=True) + system_queries = system_config.get(self.section_name, {}) + if not isinstance(system_queries, Mapping) or any( + not isinstance(name, str) or not isinstance(query, str) for name, query in system_queries.items() + ): + raise FavoriteQueryReloadError(f'invalid [{self.section_name}] section in system configuration files') + queries: dict[str, str] = {} + if self.shared_favorites_file is not None: + queries.update(self._reload_queries(self.shared_favorites_file, 'shared favorites')) + queries.update(system_queries) + queries.update(user_queries) + self.config[self.section_name] = queries + def _clean_query(self, query: str | None) -> str | None: if not query: return query diff --git a/mycli/packages/special/iocommands.py b/mycli/packages/special/iocommands.py index 01ae3d7f2..1bbcee1b0 100644 --- a/mycli/packages/special/iocommands.py +++ b/mycli/packages/special/iocommands.py @@ -7,7 +7,7 @@ import shlex import subprocess from time import sleep -from typing import Any, Generator +from typing import Any, Generator, Iterable from uuid import uuid4 import click @@ -23,7 +23,9 @@ from mycli.packages.special.delimitercommand import DelimiterCommand from mycli.packages.special.dsn_aliases import INVALID_DSN_ALIAS_ERROR, DsnAliases, is_valid_dsn_alias from mycli.packages.special.favoritequeries import ( + FAVORITE_COMMAND_HELP, FavoriteQueries, + FavoriteQueryReloadError, analyze_favorite_query_template, favorite_query_template_environment, favorite_query_variable_pattern, @@ -348,6 +350,61 @@ def set_redirect(command_part: str | None, file_operator_part: str | None, file_ return set_once(file_part) +@special_command( + r'\favorite', + '/favorite ', + 'Alternative favorite query interface. See /favorite help.', + arg_type=ArgType.PARSED_QUERY, + case_sensitive=False, +) +def favorite(arg: str, cur: Cursor | None = None, **_) -> Iterable[SQLResult]: + args = arg.strip().split(maxsplit=1) + if len(args) == 1 and args[0].lower() == 'list': + return list_favorite_queries(include_usage=False) + if args and args[0].lower() == 'reload': + if len(args) != 1: + return [SQLResult(status='Syntax: /favorite reload.')] + try: + FavoriteQueries.instance.reload() + except FavoriteQueryReloadError as exc: + return [SQLResult(status=f'Error: Unable to reload favorite queries: {exc}.')] + return [SQLResult(status='Favorite queries reloaded.')] + if args and args[0].lower() == 'eval': + eval_arg = args[1] if len(args) == 2 else '' + if not eval_arg: + return [SQLResult(status='Syntax: /favorite eval [args..] [--key=value].')] + query, error = expand_favorite_query(eval_arg) + if query is None: + return [SQLResult(status=error)] + query = _terminate_favorite_eval_query(query) + return [ + SQLResult( + status='Error: /favorite eval is only available in the interactive REPL.', + command={'name': 'set_buffer', 'text': query}, + ) + ] + if args and args[0].lower() == 'run': + run_arg = args[1] if len(args) == 2 else '' + if not run_arg: + return [SQLResult(status='Syntax: /favorite run [args..] [--key=value].')] + assert cur is not None + return execute_favorite_query(cur, run_arg) + if args and args[0].lower() == 'save': + save_arg = args[1] if len(args) == 2 else '' + usage = 'Syntax: /favorite save .' + return _save_favorite_query(save_arg, usage) + if args and args[0].lower() == 'edit': + edit_arg = args[1] if len(args) == 2 else '' + if not edit_arg: + return [SQLResult(status='Syntax: /favorite edit .')] + return _edit_favorite_query(edit_arg) + if args and args[0].lower() == 'delete': + delete_arg = args[1] if len(args) == 2 else '' + usage = 'Syntax: /favorite delete .' + return _delete_favorite_query(delete_arg, usage) + return [SQLResult(preamble=FAVORITE_COMMAND_HELP)] + + @special_command( "\\f", "/f [name [args..] [--key=value]]", @@ -360,55 +417,31 @@ def execute_favorite_query(cur: Cursor, arg: str, **_) -> Generator[SQLResult, N yield from list_favorite_queries() return - # Parse out favorite name and optional substitution parameters - name, _separator, arg_str = arg.partition(" ") - try: - args, template_values = parse_favorite_query_args(arg_str) - except ValueError as exc: - yield SQLResult(status=f'Invalid favorite query arguments: {exc}') + query, error = expand_favorite_query(arg) + if query is None: + yield SQLResult(status=error) return - query = FavoriteQueries.instance.get(name) - if query is None: - message = f"No favorite query: {name}" - yield SQLResult(status=message) - else: - query, positional_values, arg_error = prepare_favorite_query_args(query, args) - if query is None: - yield SQLResult(status=arg_error) + for sql in sqlparse.split(query): + sql = sql.rstrip(";") + preamble = f"> {sql}" if is_show_favorite_query() else None + is_special = False + for special in SPECIAL_COMMANDS: + if sql.lower().startswith(special.lower()): + is_special = True + break + if is_special: + for result in special_execute(cur, sql): + result.preamble = preamble + # special_execute() already returns a SQLResult + yield result else: - try: - query = render_favorite_query(query, template_values) - except TemplateError as exc: - yield SQLResult(status=f'Favorite query template error: {exc}') - return - except FavoriteQueryArgumentError as exc: - yield SQLResult(status=f'Invalid favorite query arguments: {exc}') - return - except Exception as exc: - yield SQLResult(status=f'Favorite query template error: {exc}') - return - query = restore_favorite_query_args(query, positional_values) - for sql in sqlparse.split(query): - sql = sql.rstrip(";") - preamble = f"> {sql}" if is_show_favorite_query() else None - is_special = False - for special in SPECIAL_COMMANDS: - if sql.lower().startswith(special.lower()): - is_special = True - break - if is_special: - for result in special_execute(cur, sql): - result.preamble = preamble - # special_execute() already returns a SQLResult - yield result - else: - cur.execute(sql) - if cur.description: - header = [x[0] for x in cur.description] - yield SQLResult(preamble=preamble, header=header, rows=cur) - else: - yield SQLResult(preamble=preamble) + cur.execute(sql) + if cur.description: + header = [x[0] for x in cur.description] + yield SQLResult(preamble=preamble, header=header, rows=cur) + else: + yield SQLResult(preamble=preamble) def parse_favorite_query_args(arg_str: str) -> tuple[list[str], dict[str, str]]: @@ -454,14 +487,52 @@ def render_favorite_query(query: str, template_values: dict[str, str]) -> str: return favorite_query_template_environment.from_string(query).render(kv=template_values) -def list_favorite_queries() -> list[SQLResult]: +def expand_favorite_query(arg: str) -> tuple[str | None, str | None]: + """Expand favorite query arguments without executing the query.""" + name, _separator, arg_str = arg.partition(" ") + try: + args, template_values = parse_favorite_query_args(arg_str) + except ValueError as exc: + return None, f'Invalid favorite query arguments: {exc}' + + query = FavoriteQueries.instance.get(name) + if query is None: + return None, f"No favorite query: {name}" + + query, positional_values, error = prepare_favorite_query_args(query, args) + if query is None: + return None, error + + try: + query = render_favorite_query(query, template_values) + except TemplateError as exc: + return None, f'Favorite query template error: {exc}' + except FavoriteQueryArgumentError as exc: + return None, f'Invalid favorite query arguments: {exc}' + except Exception as exc: + return None, f'Favorite query template error: {exc}' + + return restore_favorite_query_args(query, positional_values), None + + +def _terminate_favorite_eval_query(query: str) -> str: + query = query.rstrip() + delimiter = get_current_delimiter() + if query.endswith((delimiter, r'\G', r'\g', r'\x')): + return query + return query + delimiter + + +def list_favorite_queries(include_usage: bool = True) -> list[SQLResult]: """List of all favorite queries.""" header = ["Name", "Query"] rows = [(r, FavoriteQueries.instance.get(r)) for r in FavoriteQueries.instance.list()] if not rows: - status = "\nNo favorite queries found." + FavoriteQueries.instance.usage + status = "\nNo favorite queries found." + if include_usage: + status += FavoriteQueries.instance.usage else: status = "" return [SQLResult(header=header, rows=rows, status=status)] @@ -516,6 +587,10 @@ def save_favorite_query(arg: str, **_) -> list[SQLResult]: """Save a new favorite query.""" usage = "Syntax: \\fs name query.\n\n" + FavoriteQueries.instance.usage + return _save_favorite_query(arg, usage) + + +def _save_favorite_query(arg: str, usage: str) -> list[SQLResult]: if not arg: return [SQLResult(status=usage)] @@ -529,6 +604,34 @@ def save_favorite_query(arg: str, **_) -> list[SQLResult]: return [SQLResult(status="Saved.")] +def _edit_favorite_query(name: str) -> list[SQLResult]: + query = FavoriteQueries.instance.get(name) + if query is None: + return [SQLResult(status=f'No favorite query: {name}')] + + try: + edited_query = click.edit(query, extension='.sql') + if edited_query is None: + return [SQLResult(status=f'{name}: Not Changed.')] + FavoriteQueries.instance.save(name, edited_query) + except KeyboardInterrupt: + return [SQLResult(status=f'{name}: Edit Cancelled.')] + except (click.ClickException, OSError) as error: + return [SQLResult(status=f'Unable to edit favorite "{name}": {error}')] + + return [SQLResult(status=f'{name}: Edited.')] + + +def is_favorite_save_command(statement: str) -> bool: + """Return whether a statement saves a favorite query.""" + parts = statement.lstrip().split(maxsplit=2) + if not parts: + return False + if parts[0].startswith((r'\fs', '/fs')): + return True + return len(parts) >= 2 and parts[0].lower() in (r'\favorite', '/favorite') and parts[1].lower() == 'save' + + @special_command( "\\fd", "/fd ", @@ -537,6 +640,10 @@ def save_favorite_query(arg: str, **_) -> list[SQLResult]: def delete_favorite_query(arg: str, **_) -> list[SQLResult]: """Delete an existing favorite query.""" usage = "Syntax: \\fd name.\n\n" + FavoriteQueries.instance.usage + return _delete_favorite_query(arg, usage) + + +def _delete_favorite_query(arg: str, usage: str) -> list[SQLResult]: if not arg: return [SQLResult(status=usage)] diff --git a/mycli/sqlexecute.py b/mycli/sqlexecute.py index 9b780557c..847bb79a1 100644 --- a/mycli/sqlexecute.py +++ b/mycli/sqlexecute.py @@ -365,7 +365,7 @@ def run(self, statement: str) -> Generator[SQLResult, None, None]: # Split the sql into separate queries and run each one. # Unless it's saving a favorite query, in which case we # want to save them all together. - if statement.startswith(("\\fs", "/fs")): + if iocommands.is_favorite_save_command(statement): components: Iterable[str] = [statement] else: components = iocommands.split_queries(statement) diff --git a/test/features/fixture_data/help_commands.txt b/test/features/fixture_data/help_commands.txt index 01c5799d3..9882dd066 100644 --- a/test/features/fixture_data/help_commands.txt +++ b/test/features/fixture_data/help_commands.txt @@ -11,6 +11,7 @@ | /edit | /e | /edit | \edit | Edit query with editor (uses $VISUAL or $EDITOR). | | /exit | /q | /exit | Exit. | | /f | | /f [name [args..] [--key=value]] | List or execute favorite queries. | +| /favorite | | /favorite | Alternative favorite query interface. See /favorite help. | | /fd | | /fd | Delete a favorite query. | | /fs | | /fs | Save a favorite query. | | \g | | \g | Display query results (mnemonic: go). | diff --git a/test/pytests/test_clibuffer.py b/test/pytests/test_clibuffer.py index f777abc1e..12a831c7d 100644 --- a/test/pytests/test_clibuffer.py +++ b/test/pytests/test_clibuffer.py @@ -34,9 +34,12 @@ def make_app_for_text(text: str) -> tuple[SimpleNamespace, DummyLayout]: return SimpleNamespace(layout=layout), layout -def test_multiline_exception_handles_favorite_queries_only_after_blank_line() -> None: - assert clibuffer._multiline_exception(r'\fs demo select 1; select 2') is False - assert clibuffer._multiline_exception('\\fs demo select 1; select 2\n') is True +@pytest.mark.parametrize('command', [r'\fs', '/fs', r'\favorite save', '/favorite save']) +def test_multiline_exception_handles_favorite_queries_only_after_blank_line(command: str) -> None: + text = f'{command} demo select 1; select 2' + + assert clibuffer._multiline_exception(text) is False + assert clibuffer._multiline_exception(f'{text}\n') is True @pytest.mark.parametrize( diff --git a/test/pytests/test_client_commands.py b/test/pytests/test_client_commands.py index 312f60ebc..c3cca4413 100644 --- a/test/pytests/test_client_commands.py +++ b/test/pytests/test_client_commands.py @@ -616,7 +616,16 @@ def test_execute_from_file_runs_file_query(tmp_path: Path) -> None: assert client.sqlexecute.runs == ['select 1;'] -@pytest.mark.parametrize('command', ['/fs report select 1; select 2;', '\\fs report select 1; select 2;', 'pager;']) +@pytest.mark.parametrize( + 'command', + [ + '/fs report select 1; select 2;', + '\\fs report select 1; select 2;', + '/favorite save report select 1; select 2;', + '\\favorite save report select 1; select 2;', + 'pager;', + ], +) def test_execute_from_file_rejects_special_commands(command: str, tmp_path: Path) -> None: client = DummyClient() sql_file = tmp_path / 'query.sql' diff --git a/test/pytests/test_client_query.py b/test/pytests/test_client_query.py index 8adfa5919..f89e265f7 100644 --- a/test/pytests/test_client_query.py +++ b/test/pytests/test_client_query.py @@ -305,6 +305,25 @@ def test_run_query_writes_checkpoint(monkeypatch, tmp_path) -> None: assert state['checkpoint_path'].read_text(encoding='utf-8') == 'select 1;\n' +def test_run_query_displays_set_buffer_fallback_outside_repl(monkeypatch) -> None: + cli = make_bare_mycli() + status = 'Error: /favorite eval is only available in the interactive REPL.' + result = SQLResult(status=status, command={'name': 'set_buffer', 'text': 'select 1'}) + echoed: list[str] = [] + cli.sqlexecute = SimpleNamespace(run=lambda query: [result]) + cli.log_query = lambda query: None + cli.log_output = lambda line: None + cli.format_sqlresult = lambda result, **kwargs: [result.status_plain] + monkeypatch.setattr(client_query.special, 'is_expanded_output', lambda: False) + monkeypatch.setattr(client_query.special, 'is_redirected', lambda: False) + monkeypatch.setattr(client_query.special, 'is_show_warnings_enabled', lambda: False) + monkeypatch.setattr(client_query.click, 'echo', lambda line, nl=True: echoed.append(line)) + + main.MyCli.run_query(cli, '/favorite eval report') + + assert echoed == [status] + + def test_get_last_query_returns_none() -> None: cli = make_bare_mycli() diff --git a/test/pytests/test_completion_engine.py b/test/pytests/test_completion_engine.py index 9ab21cf20..5412338b9 100644 --- a/test/pytests/test_completion_engine.py +++ b/test/pytests/test_completion_engine.py @@ -10,6 +10,7 @@ from mycli.packages import completion_engine, special from mycli.packages.completion_engine import ( DSN_SUBCOMMANDS, + FAVORITE_SUBCOMMANDS, _aliases, _build_suggest_context, _charset_suggestion, @@ -910,6 +911,51 @@ def test_suggest_type_handles_parser_results_shorter_than_cursor(monkeypatch): ('/config get main.show_warnings ', []), ('/config search', []), ('/config search ', []), + ('/favorite ', [{'type': 'special_subcommand', 'subcommands': list(FAVORITE_SUBCOMMANDS)}]), + ('/favorite h', [{'type': 'special_subcommand', 'subcommands': list(FAVORITE_SUBCOMMANDS)}]), + ('\\favorite l', [{'type': 'special_subcommand', 'subcommands': list(FAVORITE_SUBCOMMANDS)}]), + ('/favorite help', []), + ('/favorite help ', []), + ('/favorite list', []), + ('/favorite list ', []), + ('/favorite reload', []), + ('/favorite reload ', []), + ('/favorite run', []), + ('/favorite run ', [{'type': 'favoritequery'}]), + ('/favorite run rep', [{'type': 'favoritequery'}]), + ('/favorite run report ', [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': set()}]), + ('\\favorite RUN report --u', [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': set()}]), + ('/favorite run report --user=', []), + ( + '/favorite run report --user=henry ', + [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': {'user'}}], + ), + ('/favorite eval', []), + ('/favorite eval ', [{'type': 'favoritequery'}]), + ('/favorite eval rep', [{'type': 'favoritequery'}]), + ('/favorite eval report ', [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': set()}]), + ('\\favorite EVAL report --u', [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': set()}]), + ('/favorite eval report --user=', []), + ( + '/favorite eval report --user=henry ', + [{'type': 'favoritequery_template_key', 'name': 'report', 'used_keys': {'user'}}], + ), + ('/favorite save', []), + ('/favorite save ', [{'type': 'favoritequery'}]), + ('/favorite save rep', [{'type': 'favoritequery'}]), + ('\\favorite SAVE ', [{'type': 'favoritequery'}]), + ('/favorite edit', []), + ('/favorite edit ', [{'type': 'favoritequery'}]), + ('/favorite edit rep', [{'type': 'favoritequery'}]), + ('/favorite edit report ', []), + ('/favorite edit report extra', []), + ('\\favorite EDIT ', [{'type': 'favoritequery'}]), + ('/favorite delete', []), + ('/favorite delete ', [{'type': 'favoritequery'}]), + ('/favorite delete rep', [{'type': 'favoritequery'}]), + ('/favorite delete report ', []), + ('/favorite delete report extra', []), + ('\\favorite DELETE ', [{'type': 'favoritequery'}]), ('/dsn ', [{'type': 'special_subcommand', 'subcommands': list(DSN_SUBCOMMANDS)}]), ('/dsn delete ', [{'type': 'dsn_alias'}]), ('/dsn delete pro', [{'type': 'dsn_alias'}]), @@ -940,6 +986,28 @@ def test_suggest_special(text, expected): assert suggest_special(text) == expected +@pytest.mark.parametrize( + ('text', 'query'), + [ + ('/favorite save report ', ''), + ('/favorite save report SELECT * FROM ', 'SELECT * FROM '), + (r'\favorite SAVE report SELECT col', 'SELECT col'), + ], +) +def test_suggest_special_routes_favorite_save_query_to_sql_completion(monkeypatch, text: str, query: str) -> None: + suggestions = [{'type': 'sql-query'}] + calls: list[tuple[str, str]] = [] + + def suggest_sql(full_text: str, text_before_cursor: str) -> list[dict[str, str]]: + calls.append((full_text, text_before_cursor)) + return suggestions + + monkeypatch.setattr(completion_engine, 'suggest_type', suggest_sql) + + assert suggest_special(text) == suggestions + assert calls == [(query, query)] + + @pytest.mark.parametrize( ('token', 'text_before_cursor', 'word_before_cursor', 'full_text', 'expected'), [ diff --git a/test/pytests/test_config.py b/test/pytests/test_config.py index d02a3f8ed..54b408424 100644 --- a/test/pytests/test_config.py +++ b/test/pytests/test_config.py @@ -10,6 +10,7 @@ from tempfile import NamedTemporaryFile from types import SimpleNamespace +from configobj import ConfigObjError import pytest from mycli import config as config_module @@ -272,6 +273,28 @@ def raise_oserror(*_args, **_kwargs): assert "You don't have permission to read config file '/tmp/test.cnf'." in caplog.text +def test_read_config_file_can_raise_parse_errors(tmp_path) -> None: + invalid_path = tmp_path / 'invalid.cnf' + invalid_path.write_text('[main\nfoo=bar\n', encoding='utf8') + + with pytest.raises(ConfigObjError): + read_config_file(str(invalid_path), raise_errors=True) + + +def test_read_config_file_can_raise_io_errors(monkeypatch) -> None: + error = OSError(13, 'denied', '/tmp/test.cnf') + + def raise_oserror(*_args, **_kwargs): + raise error + + monkeypatch.setattr(config_module, 'ConfigObj', raise_oserror) + + with pytest.raises(OSError) as exc_info: + read_config_file('/tmp/test.cnf', raise_errors=True) + + assert exc_info.value is error + + def test_create_and_write_default_config(tmp_path) -> None: default_config = create_default_config() assert 'main' in default_config diff --git a/test/pytests/test_favoritequeries.py b/test/pytests/test_favoritequeries.py index 985fae091..d330c6bb6 100644 --- a/test/pytests/test_favoritequeries.py +++ b/test/pytests/test_favoritequeries.py @@ -5,7 +5,7 @@ import pytest import mycli.packages.special.favoritequeries as favoritequeries_module -from mycli.packages.special.favoritequeries import FavoriteQueries +from mycli.packages.special.favoritequeries import FavoriteQueries, FavoriteQueryReloadError class DummyConfig(dict): @@ -133,6 +133,182 @@ def test_from_config_uses_successfully_parsed_shared_queries( assert 'Unable to parse line 3 of config file' in caplog.text +def test_reload_readds_user_and_shared_favorites_atomically(tmp_path: Path) -> None: + user_file = tmp_path / 'myclirc' + shared_file = tmp_path / 'shared-myclirc' + user_file.write_text('[favorite_queries]\nlocal = select 1\nremoved = select 2\n', encoding='utf-8') + shared_file.write_text('[favorite_queries]\nshared = select 3\noverridden = select 4\n', encoding='utf-8') + config = DummyConfig({ + 'main': {'setting': 'unchanged'}, + 'favorite_queries': {'local': 'select 1', 'removed': 'select 2', 'runtime': 'select 5'}, + }) + favorites = FavoriteQueries.from_config(config, str(user_file), str(shared_file)) + + user_file.write_text('[favorite_queries]\nlocal = select 10\noverridden = select 40\n', encoding='utf-8') + shared_file.write_text('[favorite_queries]\nshared = select 30\noverridden = select 4\n', encoding='utf-8') + + favorites.reload() + + assert config['main'] == {'setting': 'unchanged'} + assert config['favorite_queries'] == { + 'shared': 'select 30', + 'overridden': 'select 40', + 'local': 'select 10', + } + + +def test_reload_readds_system_favorites_with_startup_precedence(tmp_path: Path) -> None: + user_file = tmp_path / 'myclirc' + system_file = tmp_path / 'system-myclirc' + shared_file = tmp_path / 'shared-myclirc' + user_file.write_text('[favorite_queries]\nuser = select 1\noverridden = select user\n', encoding='utf-8') + system_file.write_text('[favorite_queries]\nsystem = select 2\noverridden = select system\n', encoding='utf-8') + shared_file.write_text('[favorite_queries]\nshared = select 3\noverridden = select shared\n', encoding='utf-8') + config = DummyConfig({'favorite_queries': {'runtime': 'select 4'}}) + favorites = FavoriteQueries.from_config( + config, + str(user_file), + str(shared_file), + system_config_files=[str(system_file)], + ) + + favorites.reload() + + assert config['favorite_queries'] == { + 'shared': 'select 3', + 'overridden': 'select user', + 'system': 'select 2', + 'user': 'select 1', + } + + +def test_reload_invalid_system_favorites_preserves_runtime_favorites(tmp_path: Path) -> None: + user_file = tmp_path / 'myclirc' + system_file = tmp_path / 'system-myclirc' + user_file.write_text('[favorite_queries]\nuser = select 1\n', encoding='utf-8') + system_file.write_text('favorite_queries = invalid\n', encoding='utf-8') + config = DummyConfig({'favorite_queries': {'runtime': 'select 2'}}) + favorites = FavoriteQueries.from_config( + config, + str(user_file), + system_config_files=[str(system_file)], + ) + + with pytest.raises(FavoriteQueryReloadError, match=r'invalid \[favorite_queries\] section in system'): + favorites.reload() + + assert config['favorite_queries'] == {'runtime': 'select 2'} + + +def test_reload_keeps_startup_shared_favorites_path(tmp_path: Path) -> None: + user_file = tmp_path / 'myclirc' + startup_shared_file = tmp_path / 'startup-shared-myclirc' + replacement_shared_file = tmp_path / 'replacement-shared-myclirc' + user_file.write_text('[favorite_queries]\nlocal = select 1\n', encoding='utf-8') + startup_shared_file.write_text('[favorite_queries]\nshared = select 2\n', encoding='utf-8') + replacement_shared_file.write_text('[favorite_queries]\nreplacement = select 3\n', encoding='utf-8') + favorites = FavoriteQueries.from_config(DummyConfig(), str(user_file), str(startup_shared_file)) + user_file.write_text( + f'[main]\nshared_favorites_file = {replacement_shared_file}\n[favorite_queries]\nlocal = select 10\n', + encoding='utf-8', + ) + + favorites.reload() + + assert favorites.get('shared') == 'select 2' + assert favorites.get('replacement') is None + assert favorites.get('local') == 'select 10' + + +@pytest.mark.parametrize('broken_source', ['user', 'shared']) +def test_reload_failure_preserves_runtime_favorites(tmp_path: Path, broken_source: str) -> None: + user_file = tmp_path / 'myclirc' + shared_file = tmp_path / 'shared-myclirc' + user_file.write_text('[favorite_queries]\nlocal = select 1\n', encoding='utf-8') + shared_file.write_text('[favorite_queries]\nshared = select 2\n', encoding='utf-8') + config = DummyConfig({'favorite_queries': {'runtime': 'select 3'}}) + favorites = FavoriteQueries.from_config(config, str(user_file), str(shared_file)) + before_reload = dict(config['favorite_queries']) + broken_file = user_file if broken_source == 'user' else shared_file + broken_file.write_text('[favorite_queries\ninvalid = select 4\n', encoding='utf-8') + + with pytest.raises(FavoriteQueryReloadError, match=f'unable to read {broken_source}'): + favorites.reload() + + assert config['favorite_queries'] == before_reload + + +@pytest.mark.parametrize( + ('contents', 'error_pattern'), + [ + (b'[favorite_queries]\ninvalid = \xff\n', 'unable to read user configuration'), + (b'favorite_queries = invalid\n', r'invalid \[favorite_queries\] section'), + (b'[favorite_queries]\ninvalid = select 1, select 2\n', r'invalid \[favorite_queries\] section'), + ], +) +def test_reload_invalid_config_preserves_runtime_favorites( + tmp_path: Path, + contents: bytes, + error_pattern: str, +) -> None: + user_file = tmp_path / 'myclirc' + user_file.write_bytes(contents) + config = DummyConfig({'favorite_queries': {'runtime': 'select 3'}}) + favorites = FavoriteQueries(config, str(user_file)) + + with pytest.raises(FavoriteQueryReloadError, match=error_pattern): + favorites.reload() + + assert config['favorite_queries'] == {'runtime': 'select 3'} + + +@pytest.mark.parametrize('missing_source', ['user', 'shared']) +def test_reload_missing_file_preserves_runtime_favorites(tmp_path: Path, missing_source: str) -> None: + user_file = tmp_path / 'myclirc' + shared_file = tmp_path / 'shared-myclirc' + user_file.write_text('[favorite_queries]\nlocal = select 1\n', encoding='utf-8') + shared_file.write_text('[favorite_queries]\nshared = select 2\n', encoding='utf-8') + config = DummyConfig({'favorite_queries': {'runtime': 'select 3'}}) + favorites = FavoriteQueries.from_config(config, str(user_file), str(shared_file)) + before_reload = dict(config['favorite_queries']) + missing_file = user_file if missing_source == 'user' else shared_file + missing_file.unlink() + + with pytest.raises(FavoriteQueryReloadError, match=f'unable to read {missing_source}'): + favorites.reload() + + assert config['favorite_queries'] == before_reload + + +def test_reload_unreadable_file_preserves_runtime_favorites( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + user_file = tmp_path / 'myclirc' + user_file.write_text('[favorite_queries]\nlocal = select 1\n', encoding='utf-8') + config = DummyConfig({'favorite_queries': {'runtime': 'select 3'}}) + favorites = FavoriteQueries.from_config(config, str(user_file)) + + def deny_read(_path: str, **_kwargs: object) -> None: + raise OSError(13, 'Permission denied', str(user_file)) + + monkeypatch.setattr(favoritequeries_module, 'read_config_file', deny_read) + + with pytest.raises(FavoriteQueryReloadError, match='Permission denied'): + favorites.reload() + + assert config['favorite_queries'] == {'runtime': 'select 3'} + + +def test_reload_requires_user_config_file() -> None: + favorites = FavoriteQueries(DummyConfig({'favorite_queries': {'runtime': 'select 1'}})) + + with pytest.raises(FavoriteQueryReloadError, match='no user configuration file is configured'): + favorites.reload() + + assert favorites.get('runtime') == 'select 1' + + def test_list_and_get_use_favorite_queries_section() -> None: config = DummyConfig({ 'favorite_queries': { diff --git a/test/pytests/test_main_modes_repl.py b/test/pytests/test_main_modes_repl.py index 401cab92a..c7c1734f3 100644 --- a/test/pytests/test_main_modes_repl.py +++ b/test/pytests/test_main_modes_repl.py @@ -821,6 +821,34 @@ def format_sqlresult_with_width(result: SQLResult, **kwargs: Any) -> Iterator[st ) +def test_output_results_moves_set_buffer_command_to_repl_state() -> None: + cli = make_repl_cli(SimpleNamespace()) + state = repl_mode.ReplState() + result = SQLResult( + status='Error: /favorite eval is only available in the interactive REPL.', + command={'name': 'set_buffer', 'text': 'select 1'}, + ) + + repl_mode._output_results(cli, state, iter([result]), start=0.0) + + assert state.buffer_text == 'select 1' + assert cli.output_calls == [] + assert cli.echo_calls == [] + + +def test_one_iteration_prefills_and_clears_pending_buffer_text(monkeypatch: pytest.MonkeyPatch) -> None: + patch_repl_runtime_defaults(monkeypatch) + cli = make_repl_cli(SimpleNamespace()) + cli.prompt_session = FakePromptSession(['']) + state = repl_mode.ReplState(buffer_text='select 1') + + repl_mode._one_iteration(cli, state) + + assert cli.prompt_session.prompt_calls[0]['default'] == 'select 1' + assert state.buffer_text is None + assert cli.query_history == [] + + def test_keepalive_hook_covers_threshold_and_errors() -> None: cli = make_repl_cli(SimpleNamespace(conn=FakeConnection())) repl_mode._keepalive_hook(cli, None) diff --git a/test/pytests/test_special_iocommands.py b/test/pytests/test_special_iocommands.py index cea1b69e7..a73e2a1c5 100644 --- a/test/pytests/test_special_iocommands.py +++ b/test/pytests/test_special_iocommands.py @@ -19,7 +19,11 @@ import mycli.packages.special from mycli.packages.special import iocommands -from mycli.packages.special.favoritequeries import analyze_favorite_query_template, find_favorite_query_template_keys +from mycli.packages.special.favoritequeries import ( + FavoriteQueryReloadError, + analyze_favorite_query_template, + find_favorite_query_template_keys, +) from mycli.packages.sqlresult import SQLResult from test.utils import TEMPFILE_PREFIX, db_connection, dbtest @@ -31,6 +35,7 @@ def __init__(self, queries: dict[str, str] | None = None) -> None: self.queries = {} if queries is None else dict(queries) self.saved: list[tuple[str, str]] = [] self.deleted: list[str] = [] + self.reload_calls = 0 def list(self) -> list[str]: return list(self.queries) @@ -46,6 +51,9 @@ def delete(self, name: str) -> str: self.deleted.append(name) return f'{name}: Deleted.' + def reload(self) -> None: + self.reload_calls += 1 + class FakeDsnAliases: usage = '\nFAKE DSN USAGE' @@ -651,6 +659,329 @@ def test_execute_favorite_query_list_missing_and_bad_args(monkeypatch) -> None: assert bad_args[0].status == 'missing substitution for $1 in query:\n select $1' +@pytest.mark.parametrize('arg', ['', 'help', 'HELP', 'unknown', 'list extra']) +def test_favorite_command_shows_help_for_non_list_forms(monkeypatch, arg: str) -> None: + favorite_queries = FakeFavoriteQueries({'demo': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert iocommands.favorite(arg=arg) == [SQLResult(preamble=iocommands.FAVORITE_COMMAND_HELP)] + assert iocommands.FAVORITE_COMMAND_HELP != favorite_queries.usage + + +@pytest.mark.parametrize('arg', ['list', 'LIST']) +def test_favorite_command_delegates_list(monkeypatch, arg: str) -> None: + listed = SQLResult(status='listed') + include_usage_values: list[bool] = [] + + def list_favorite_queries(include_usage: bool = True) -> list[SQLResult]: + include_usage_values.append(include_usage) + return [listed] + + monkeypatch.setattr(iocommands, 'list_favorite_queries', list_favorite_queries) + + assert iocommands.favorite(arg=arg) == [listed] + assert include_usage_values == [False] + + +def test_favorite_list_empty_does_not_show_legacy_help(monkeypatch) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + result = iocommands.favorite(arg='list')[0] + + assert result.status == '\nNo favorite queries found.' + assert favorite_queries.usage not in result.status + + +@pytest.mark.parametrize('command', ['/favorite reload', r'\favorite RELOAD']) +def test_favorite_reload_command(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert mycli.packages.special.execute(FakeCursor(), command) == [SQLResult(status='Favorite queries reloaded.')] + assert favorite_queries.reload_calls == 1 + + +def test_favorite_reload_command_reports_errors(monkeypatch) -> None: + favorite_queries = FakeFavoriteQueries() + + def fail_reload() -> None: + raise FavoriteQueryReloadError('unable to read user configuration file') + + favorite_queries.reload = fail_reload + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert iocommands.favorite(arg='reload') == [ + SQLResult(status='Error: Unable to reload favorite queries: unable to read user configuration file.') + ] + + +def test_favorite_reload_command_rejects_arguments() -> None: + assert iocommands.favorite(arg='reload extra') == [SQLResult(status='Syntax: /favorite reload.')] + + +def test_favorite_help_documents_reload() -> None: + assert '> /favorite reload' in iocommands.FAVORITE_COMMAND_HELP + + +@pytest.mark.parametrize( + 'command', + [ + '/favorite run report value --user=henry', + r'\favorite RUN report value --user=henry', + ], +) +def test_favorite_run_command_executes_with_arguments(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select $1; select {{ kv.user }}'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + cursor = FakeCursor() + + results = list(mycli.packages.special.execute(cursor, command)) + + assert [result.preamble for result in results] == ['> select value', '> select henry'] + assert cursor.executed == ['select value', 'select henry'] + + +def test_favorite_command_delegates_run_lazily(monkeypatch) -> None: + cursor = FakeCursor() + calls: list[tuple[FakeCursor, str]] = [] + + def execute(cur: FakeCursor, arg: str): + calls.append((cur, arg)) + yield SQLResult(status='ran') + + monkeypatch.setattr(iocommands, 'execute_favorite_query', execute) + + results = iocommands.favorite(cur=cursor, arg='run report positional --user=henry') + + assert calls == [] + assert list(results) == [SQLResult(status='ran')] + assert calls == [(cursor, 'report positional --user=henry')] + + +def test_favorite_command_reports_run_usage() -> None: + assert iocommands.favorite(arg='run') == [SQLResult(status='Syntax: /favorite run [args..] [--key=value].')] + + +@pytest.mark.parametrize( + 'command', + [ + '/favorite eval report value --user=henry', + r'\favorite EVAL report value --user=henry', + ], +) +def test_favorite_eval_command_expands_without_execution(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select $1; select {{ kv.user }}'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + cursor = FakeCursor() + + assert mycli.packages.special.execute(cursor, command) == [ + SQLResult( + status='Error: /favorite eval is only available in the interactive REPL.', + command={'name': 'set_buffer', 'text': 'select value; select henry;'}, + ) + ] + assert cursor.executed == [] + + +def test_favorite_command_reports_eval_usage() -> None: + assert iocommands.favorite(arg='eval') == [SQLResult(status='Syntax: /favorite eval [args..] [--key=value].')] + + +@pytest.mark.parametrize( + ('query', 'delimiter', 'expected'), + [ + ('select 1', ';', 'select 1;'), + ('select 1;', ';', 'select 1;'), + ('select 1\\G', ';', 'select 1\\G'), + ('select 1\\g', ';', 'select 1\\g'), + ('select 1\\x', ';', 'select 1\\x'), + ('select 1 ', ';', 'select 1;'), + ('select 1//', '//', 'select 1//'), + ('select 1;', '//', 'select 1;//'), + ('', ';', ';'), + ], +) +def test_favorite_eval_uses_active_terminator(monkeypatch, query: str, delimiter: str, expected: str) -> None: + monkeypatch.setattr(iocommands, 'get_current_delimiter', lambda: delimiter) + + assert iocommands._terminate_favorite_eval_query(query) == expected + + +def test_favorite_eval_termination_does_not_change_shared_expansion(monkeypatch) -> None: + monkeypatch.setattr( + iocommands.FavoriteQueries, + 'instance', + FakeFavoriteQueries({'report': 'select 1'}), + raising=False, + ) + + assert iocommands.expand_favorite_query('report') == ('select 1', None) + assert iocommands.favorite(arg='eval report')[0].command == {'name': 'set_buffer', 'text': 'select 1;'} + + +@pytest.mark.parametrize('arg', ['save report select 1; select 2', 'SAVE report select 1; select 2']) +def test_favorite_command_saves_query(monkeypatch, arg: str) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert iocommands.favorite(arg=arg) == [SQLResult(status='Saved.')] + assert favorite_queries.saved == [('report', 'select 1; select 2')] + + +@pytest.mark.parametrize('command', ['/favorite save report select 1', r'\favorite SAVE report select 1']) +def test_favorite_save_command_is_registered(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert mycli.packages.special.execute(None, command) == [SQLResult(status='Saved.')] + assert favorite_queries.saved == [('report', 'select 1')] + + +@pytest.mark.parametrize('arg', ['save', 'save report']) +def test_favorite_command_reports_save_usage(monkeypatch, arg: str) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + usage = 'Syntax: /favorite save .' + + result = iocommands.favorite(arg=arg)[0] + + if arg == 'save': + assert result.status == usage + else: + assert result.status == usage + ' Err: Both name and query are required.' + + +@pytest.mark.parametrize('command', ['/favorite edit report', r'\favorite EDIT report']) +def test_favorite_edit_command_edits_and_saves_query(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + edit_calls: list[tuple[str, str]] = [] + + def edit(query: str, extension: str) -> str: + edit_calls.append((query, extension)) + return 'select 2\n' + + monkeypatch.setattr(iocommands.click, 'edit', edit) + + assert mycli.packages.special.execute(None, command) == [SQLResult(status='report: Edited.')] + assert edit_calls == [('select 1', '.sql')] + assert favorite_queries.saved == [('report', 'select 2\n')] + + +def test_favorite_edit_command_reports_missing_and_unknown_names(monkeypatch) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + edit_calls: list[str] = [] + monkeypatch.setattr(iocommands.click, 'edit', lambda query, extension: edit_calls.append(query)) + + assert iocommands.favorite(arg='edit') == [SQLResult(status='Syntax: /favorite edit .')] + assert iocommands.favorite(arg='edit unknown') == [SQLResult(status='No favorite query: unknown')] + assert edit_calls == [] + assert favorite_queries.saved == [] + + +@pytest.mark.parametrize( + ('edited_query', 'expected_status'), + [ + (None, 'report: Not Changed.'), + ('', 'report: Edited.'), + ], +) +def test_favorite_edit_command_handles_unchanged_and_empty_queries( + monkeypatch, + edited_query: str | None, + expected_status: str, +) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + monkeypatch.setattr(iocommands.click, 'edit', lambda query, extension: edited_query) + + assert iocommands.favorite(arg='edit report') == [SQLResult(status=expected_status)] + assert favorite_queries.saved == ([] if edited_query is None else [('report', '')]) + + +@pytest.mark.parametrize( + ('error', 'expected_status'), + [ + (KeyboardInterrupt(), 'report: Edit Cancelled.'), + (iocommands.click.ClickException('editor failed'), 'Unable to edit favorite "report": editor failed'), + (OSError('editor unavailable'), 'Unable to edit favorite "report": editor unavailable'), + ], +) +def test_favorite_edit_command_reports_editor_errors(monkeypatch, error: BaseException, expected_status: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + def edit(query: str, extension: str) -> str: + raise error + + monkeypatch.setattr(iocommands.click, 'edit', edit) + + assert iocommands.favorite(arg='edit report') == [SQLResult(status=expected_status)] + assert favorite_queries.saved == [] + + +def test_favorite_edit_command_reports_save_error(monkeypatch) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + monkeypatch.setattr(iocommands.click, 'edit', lambda query, extension: 'select 2\n') + + def save(name: str, query: str) -> None: + raise OSError('write failed') + + monkeypatch.setattr(favorite_queries, 'save', save) + + assert iocommands.favorite(arg='edit report') == [SQLResult(status='Unable to edit favorite "report": write failed')] + + +def test_favorite_edit_command_creates_local_override_for_shared_query(monkeypatch, tmp_path: Path) -> None: + shared_file = tmp_path / 'shared-myclirc' + shared_contents = '[favorite_queries]\nreport = select 1\n' + shared_file.write_text(shared_contents, encoding='utf-8') + config_file = tmp_path / 'myclirc' + config_file.write_text('# User config.\n', encoding='utf-8') + favorite_queries = iocommands.FavoriteQueries.from_config( + iocommands.ConfigObj(), + str(config_file), + str(shared_file), + ) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + monkeypatch.setattr(iocommands.click, 'edit', lambda query, extension: 'select 2\n') + + assert iocommands.favorite(arg='edit report') == [SQLResult(status='report: Edited.')] + assert shared_file.read_text(encoding='utf-8') == shared_contents + assert 'report = select 2' in config_file.read_text(encoding='utf-8') + assert favorite_queries.get('report') == 'select 2' + + +@pytest.mark.parametrize('arg', ['delete report', 'DELETE report']) +def test_favorite_command_deletes_query(monkeypatch, arg: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert iocommands.favorite(arg=arg) == [SQLResult(status='report: Deleted.')] + assert favorite_queries.deleted == ['report'] + + +@pytest.mark.parametrize('command', ['/favorite delete report', r'\favorite DELETE report']) +def test_favorite_delete_command_is_registered(monkeypatch, command: str) -> None: + favorite_queries = FakeFavoriteQueries({'report': 'select 1'}) + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + + assert mycli.packages.special.execute(None, command) == [SQLResult(status='report: Deleted.')] + assert favorite_queries.deleted == ['report'] + + +def test_favorite_command_reports_delete_usage(monkeypatch) -> None: + favorite_queries = FakeFavoriteQueries() + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) + usage = 'Syntax: /favorite delete .' + + assert iocommands.favorite(arg='delete') == [SQLResult(status=usage)] + assert favorite_queries.deleted == [] + + def test_execute_favorite_query_special_and_plain_sql(monkeypatch) -> None: favorite_queries = FakeFavoriteQueries({'combo': 'help demo; select 1'}) monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', favorite_queries, raising=False) @@ -849,12 +1180,34 @@ def test_execute_favorite_query_reports_template_argument_errors_without_executi cursor = FakeCursor() results = list(iocommands.execute_favorite_query(cursor, arg)) + expanded_query, expansion_error = iocommands.expand_favorite_query(arg) assert results[0].status is not None assert results[0].status.startswith(status_prefix) + assert expanded_query is None + assert expansion_error == results[0].status assert cursor.executed == [] +@pytest.mark.parametrize( + ('query', 'arg'), + [ + (None, 'unknown'), + ('select $1', 'report'), + ('select 1', 'report "'), + ], +) +def test_favorite_eval_errors_match_execution(monkeypatch, query: str | None, arg: str) -> None: + queries = {} if query is None else {'report': query} + monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', FakeFavoriteQueries(queries), raising=False) + + execution_result = next(iocommands.execute_favorite_query(FakeCursor(), arg)) + eval_result = iocommands.favorite(arg=f'eval {arg}')[0] + + assert eval_result.status == execution_result.status + assert eval_result.command is None + + def test_list_substitute_save_delete_and_redirect_state(tmp_path: Path, monkeypatch) -> None: empty_favorites = FakeFavoriteQueries() monkeypatch.setattr(iocommands.FavoriteQueries, 'instance', empty_favorites, raising=False) diff --git a/test/pytests/test_sqlexecute.py b/test/pytests/test_sqlexecute.py index 294cc9264..4436193d5 100644 --- a/test/pytests/test_sqlexecute.py +++ b/test/pytests/test_sqlexecute.py @@ -911,9 +911,17 @@ def fake_split_queries(statement: str): assert split_inputs == [''] -def test_run_does_not_split_favorite_query(monkeypatch) -> None: +@pytest.mark.parametrize( + 'favorite_sql', + [ + '\\fs test-name select 1; select 2', + '/fs test-name select 1; select 2', + '\\favorite save test-name select 1; select 2', + '/favorite save test-name select 1; select 2', + ], +) +def test_run_does_not_split_favorite_query(monkeypatch, favorite_sql: str) -> None: favorite_results = [SQLResult(status='Saved.')] - favorite_sql = '\\fs test-name select 1; select 2' cursor = FakeQueryCursor() execute_calls: list[str] = []