Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
8 changes: 8 additions & 0 deletions mycli/main_modes/repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down
1 change: 1 addition & 0 deletions mycli/myclirc
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions test/myclirc
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
92 changes: 92 additions & 0 deletions test/pytests/test_main_modes_repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -684,6 +685,97 @@ def test_maybe_html_escape() -> None:
assert repl_mode.maybe_html_escape('a&b<1>', True) == 'a&amp;b&lt;1&gt;'


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'\<html><b>\b</b>\</html>', '[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(
Expand Down
Loading