diff --git a/changelog.md b/changelog.md index e7d0cdc6..eed302cc 100644 --- a/changelog.md +++ b/changelog.md @@ -8,6 +8,7 @@ Features * Make approximate-matching completion thresholds configurable. * Make regex matching completion thresholds configurable. * Make completion-candidate sorting tiebreaker configurable. +* Add a `\b` prompt format string to show transaction status. 2.23.0 (2026/09/09) diff --git a/mycli/main_modes/repl.py b/mycli/main_modes/repl.py index 86683939..5aa94c99 100644 --- a/mycli/main_modes/repl.py +++ b/mycli/main_modes/repl.py @@ -43,6 +43,7 @@ from prompt_toolkit.output import ColorDepth from prompt_toolkit.shortcuts import CompleteStyle, PromptSession import pymysql +from pymysql.constants.SERVER_STATUS import SERVER_STATUS_IN_TRANS from pymysql.cursors import Cursor import mycli as mycli_package @@ -343,6 +344,13 @@ def render_prompt_string( strings = [x.replace('\\_', ' ') for x in strings] checker_string = ' '.join(strings) + if r'\b' in checker_string: + connection = getattr(sqlexecute, 'conn', None) + if connection: + connection.ping(reconnect=False) + server_status = getattr(connection, 'server_status', 0) or 0 + transaction_indicator = '[TX]' if server_status & SERVER_STATUS_IN_TRANS else '' + strings = [x.replace(r'\b', transaction_indicator) for x in strings] if r'\e' in checker_string: if mycli.prompt_session: edit_mode = mycli.prompt_session.editing_mode.value.lower() diff --git a/mycli/myclirc b/mycli/myclirc index 16818269..c3bc44a3 100644 --- a/mycli/myclirc +++ b/mycli/myclirc @@ -200,6 +200,7 @@ wider_completion_menu = False # * \W - number of warnings, or the empty string (requires frequent trips to the server) # * \y - uptime in seconds (requires frequent trips to the server) # * \Y - uptime in words (requires frequent trips to the server) +# * \b - "[TX]" when in a transaction, otherwise empty (requires frequent trips to the server) # * \A - DSN alias # * \n - a newline # * \_ - a space diff --git a/test/myclirc b/test/myclirc index 6d7808fe..37c497d1 100644 --- a/test/myclirc +++ b/test/myclirc @@ -200,6 +200,7 @@ wider_completion_menu = False # * \W - number of warnings, or the empty string (requires frequent trips to the server) # * \y - uptime in seconds (requires frequent trips to the server) # * \Y - uptime in words (requires frequent trips to the server) +# * \b - "[TX]" when in a transaction, otherwise empty (requires frequent trips to the server) # * \A - DSN alias # * \n - a newline # * \_ - a space diff --git a/test/pytests/test_main_modes_repl.py b/test/pytests/test_main_modes_repl.py index badf448a..841bb13e 100644 --- a/test/pytests/test_main_modes_repl.py +++ b/test/pytests/test_main_modes_repl.py @@ -7,6 +7,7 @@ import os from types import SimpleNamespace from typing import Any, Literal, cast +from unittest.mock import Mock from prompt_toolkit.formatted_text import to_formatted_text, to_plain_text import pymysql @@ -684,6 +685,97 @@ def test_maybe_html_escape() -> None: assert repl_mode.maybe_html_escape('a&b<1>', True) == 'a&b<1>' +def make_transaction_prompt_cli(connection: Any) -> Any: + return make_repl_cli( + SimpleNamespace( + user='alice', + host='db.example.com', + dbname='nameprod', + port=3306, + socket=None, + server_info=None, + conn=connection, + ) + ) + + +@pytest.mark.parametrize(('status', 'expected'), [(0, ''), (1, '[TX]'), (2, ''), (3, '[TX]'), (9, '[TX]'), (None, '')]) +def test_transaction_prompt_reads_refreshed_flag(status: int | None, expected: str) -> None: + connection = SimpleNamespace(server_status=0 if expected else 1, cursor=pytest.fail) + + def ping(*, reconnect: bool) -> None: + assert reconnect is False + connection.server_status = status + + connection.ping = ping + cli = make_transaction_prompt_cli(connection) + + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 0)) == expected + + +@pytest.mark.parametrize('connection', [None, SimpleNamespace(ping=Mock())]) +def test_transaction_prompt_handles_unavailable_connection_status(connection: Any) -> None: + cli = make_transaction_prompt_cli(connection) + + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 0)) == '' + + +def test_transaction_prompt_handles_missing_connection_attribute() -> None: + cli = make_transaction_prompt_cli(None) + del cli.sqlexecute.conn + + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 0)) == '' + + +def test_transaction_prompt_updates_on_new_render() -> None: + connection = SimpleNamespace(server_status=0, ping=Mock()) + cli = make_transaction_prompt_cli(connection) + + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 0)) == '' + connection.server_status = 1 + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 1)) == '[TX]' + connection.server_status = 0 + assert to_plain_text(repl_mode.render_prompt_string(cli, r'\b', 2)) == '' + + +@pytest.mark.parametrize('status', [0, 1]) +@pytest.mark.parametrize( + ('format_string', 'active', 'idle'), + [ + (r'\b|\b', '[TX]|[TX]', '|'), + (r'\\b', r'\b', r'\b'), + (r'\x1b[31m\b\x1b[0m', '[TX]', ''), + (r'\\b\', '[TX]', ''), + ], +) +def test_transaction_prompt_preserves_formatting(status: int, format_string: str, active: str, idle: str) -> None: + cli = make_transaction_prompt_cli(SimpleNamespace(server_status=status, ping=Mock())) + + assert to_plain_text(repl_mode.render_prompt_string(cli, format_string, 0)) == (active if status else idle) + + +@pytest.mark.parametrize(('format_string', 'expected_calls'), [(r'\b|\b', 1), (r'\\b', 0), ('plain', 0)]) +def test_transaction_prompt_pings_only_for_active_escape(format_string: str, expected_calls: int) -> None: + ping = Mock() + cli = make_transaction_prompt_cli(SimpleNamespace(server_status=0, ping=ping)) + + repl_mode.render_prompt_string(cli, format_string, 0) + + assert ping.call_count == expected_calls + if expected_calls: + ping.assert_called_once_with(reconnect=False) + + +def test_transaction_prompt_reuses_cached_render_without_ping() -> None: + ping = Mock() + cli = make_transaction_prompt_cli(SimpleNamespace(server_status=0, ping=ping)) + + repl_mode.render_prompt_string(cli, r'\b', 0) + repl_mode.render_prompt_string(cli, r'\b', 0) + + ping.assert_called_once_with(reconnect=False) + + def test_render_prompt_string_includes_current_edit_mode() -> None: cli = make_repl_cli( SimpleNamespace(