diff --git a/changelog.md b/changelog.md index aad43e309..2783bce33 100644 --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,11 @@ +Upcoming (TBD) +============== + +Features +-------- +* Add `/ping` special command. + + 2.18.5 (2026/08/31) ============== diff --git a/mycli/TIPS b/mycli/TIPS index d078d57a9..390bd5b9f 100644 --- a/mycli/TIPS +++ b/mycli/TIPS @@ -118,6 +118,8 @@ use /dsn to manage saved DSNs! use /config to inspect persistent configuration from the REPL! +/ping checks the health of the connection! + ### ### environment variables ### diff --git a/mycli/client_commands.py b/mycli/client_commands.py index ce670a479..938a03365 100644 --- a/mycli/client_commands.py +++ b/mycli/client_commands.py @@ -44,6 +44,7 @@ 'help', 'l', 'nowarnings', + 'ping', 'prompt', 'redirectformat', 'rehash', diff --git a/mycli/packages/special/__init__.py b/mycli/packages/special/__init__.py index 33ef6f46b..361b3752c 100644 --- a/mycli/packages/special/__init__.py +++ b/mycli/packages/special/__init__.py @@ -10,6 +10,7 @@ from mycli.packages.special.dbcommands import ( list_databases, list_tables, + ping, status, ) from mycli.packages.special.iocommands import ( @@ -118,6 +119,7 @@ def sql_using_llm(*args, **kwargs): # type: ignore[no-redef, misc] 'list_tables', 'open_external_editor', 'parse_special_command', + 'ping', 'register_special_command', 'run_post_redirect_hook', 'set_delimiter', diff --git a/mycli/packages/special/dbcommands.py b/mycli/packages/special/dbcommands.py index 5ac43b0be..c2855dda5 100644 --- a/mycli/packages/special/dbcommands.py +++ b/mycli/packages/special/dbcommands.py @@ -2,7 +2,7 @@ import os import platform -from pymysql import ProgrammingError +from pymysql import Error, ProgrammingError from pymysql.cursors import Cursor from mycli import __version__ @@ -80,6 +80,24 @@ def list_databases(cur: Cursor, **_) -> list[SQLResult]: return [SQLResult()] +@special_command( + r'\ping', + '/ping', + 'Check the connection.', + arg_type=ArgType.PARSED_QUERY, + completion_snippet='check connection', +) +def ping(cur: Cursor, arg: str | None = None, **_) -> list[SQLResult]: + if arg: + return [SQLResult(status='Syntax: /ping.')] + + try: + cur.connection.ping(reconnect=False) + except Error: + return [SQLResult(status='Not connected')] + return [SQLResult(status='Connected')] + + @special_command( "status", "/status", diff --git a/test/features/fixture_data/help_commands.txt b/test/features/fixture_data/help_commands.txt index fc89562cb..8fa187cb0 100644 --- a/test/features/fixture_data/help_commands.txt +++ b/test/features/fixture_data/help_commands.txt @@ -24,6 +24,7 @@ | /nowarnings | /w | /nowarnings | Disable automatic warnings display. | | /once | /o | /once [-o] | Append next result to an output file (overwrite using -o). | | /pager | /P | /pager [command] | Set pager to [command]. Print query results via pager. | +| /ping | | /ping | Check the connection. | | /pipe_once | /| | /pipe_once | Send next result to a subprocess. | | /prompt | /R | /prompt [string] | Show or change prompt format. | | /quit | /q | /quit | Quit. | diff --git a/test/pytests/test_client_commands.py b/test/pytests/test_client_commands.py index a9942b733..ca5bed49e 100644 --- a/test/pytests/test_client_commands.py +++ b/test/pytests/test_client_commands.py @@ -891,6 +891,7 @@ def test_execute_from_file_requires_semicolon_for_special_commands(tmp_path: Pat ('command', 'arg', 'expected'), [ ('status', '', True), + ('ping', '', True), ('connect', 'db', True), ('config', 'get main.prompt', True), ('config', 'edit', False), diff --git a/test/pytests/test_special_dbcommands.py b/test/pytests/test_special_dbcommands.py index 2859e6544..d92ffd5f0 100644 --- a/test/pytests/test_special_dbcommands.py +++ b/test/pytests/test_special_dbcommands.py @@ -2,11 +2,14 @@ from unittest.mock import MagicMock -from pymysql import ProgrammingError +from pymysql import Error, ProgrammingError +import pytest from mycli.packages.completion_engine import suggest_type from mycli.packages.special import dbcommands -from mycli.packages.special.dbcommands import list_databases, list_tables, status +from mycli.packages.special import main as special_main +from mycli.packages.special.dbcommands import list_databases, list_tables, ping, status +from mycli.packages.sqlresult import SQLResult from test.pytests.test_completion_engine import sorted_dicts @@ -19,16 +22,24 @@ def __init__( host_info: str = 'Localhost via UNIX socket', unix_socket: str | None = None, thread_id_value: int = 42, + ping_error: Exception | None = None, ) -> None: self.host = host self.port = port self.host_info = host_info self.unix_socket = unix_socket self._thread_id_value = thread_id_value + self.ping_error = ping_error + self.ping_calls: list[bool] = [] def thread_id(self) -> int: return self._thread_id_value + def ping(self, reconnect: bool = True) -> None: + self.ping_calls.append(reconnect) + if self.ping_error is not None: + raise self.ping_error + class FakeCursor: def __init__( @@ -177,6 +188,48 @@ def test_list_databases_with_and_without_description() -> None: assert empty[0].rows is None +def test_ping_reports_connected_without_reconnecting() -> None: + connection = FakeConnection() + cursor = FakeCursor(query_results={}, connection=connection) + + assert ping(cursor) == [SQLResult(status='Connected')] + assert connection.ping_calls == [False] + + +def test_ping_reports_not_connected_on_pymysql_error() -> None: + connection = FakeConnection(ping_error=Error('connection lost')) + cursor = FakeCursor(query_results={}, connection=connection) + + assert ping(cursor) == [SQLResult(status='Not connected')] + assert connection.ping_calls == [False] + + +def test_ping_propagates_unrelated_errors() -> None: + connection = FakeConnection(ping_error=RuntimeError('unexpected')) + cursor = FakeCursor(query_results={}, connection=connection) + + with pytest.raises(RuntimeError, match='unexpected'): + ping(cursor) + + +def test_ping_rejects_arguments_without_contacting_server() -> None: + connection = FakeConnection() + cursor = FakeCursor(query_results={}, connection=connection) + + assert ping(cursor, arg='unexpected') == [SQLResult(status='Syntax: /ping.')] + assert connection.ping_calls == [] + + +def test_ping_command_registration() -> None: + command = special_main.COMMANDS[r'\ping'] + + assert command.handler is ping + assert command.usage == '/ping' + assert command.description == 'Check the connection.' + assert command.completion_snippet == 'check connection' + assert special_main.COMMANDS['/ping'].handler is ping + + def test_status_uses_global_queries_decodes_bytes_and_formats_stats(monkeypatch) -> None: monkeypatch.setattr(dbcommands, '__version__', '9.9.9') monkeypatch.setattr(dbcommands.platform, 'python_implementation', lambda: 'CPython') diff --git a/test/pytests/test_special_init.py b/test/pytests/test_special_init.py index 71f9cf522..31631c5cc 100644 --- a/test/pytests/test_special_init.py +++ b/test/pytests/test_special_init.py @@ -68,6 +68,7 @@ def test_special_init_reexports_dbcommands(load_special: Callable[[bool], Module assert special.list_databases is dbcommands.list_databases assert special.list_tables is dbcommands.list_tables + assert special.ping is dbcommands.ping assert special.status is dbcommands.status