From 83c7c42a887bcef37e0d88d2f5ef5c5dd1288384 Mon Sep 17 00:00:00 2001 From: Peter Adams <18162810+Maxteabag@users.noreply.github.com> Date: Mon, 25 May 2026 13:18:24 +0200 Subject: [PATCH 1/2] feat(history): j/k vim-style navigation in query history screen j and k now move the cursor down/up in the history OptionList when the filter is closed; when the filter is open, they're typed into the search like any other character. --- .../domains/query/ui/screens/query_history.py | 11 ++ tests/ui/test_query_history.py | 132 ++++++++++++++++++ 2 files changed, 143 insertions(+) diff --git a/sqlit/domains/query/ui/screens/query_history.py b/sqlit/domains/query/ui/screens/query_history.py index a3b8223d..4623420b 100644 --- a/sqlit/domains/query/ui/screens/query_history.py +++ b/sqlit/domains/query/ui/screens/query_history.py @@ -287,6 +287,17 @@ def action_cancel(self) -> None: def on_key(self, event: Any) -> None: if not self._filter_active: + if event.key in ("j", "k"): + try: + option_list = self.query_one("#history-list", OptionList) + except Exception: + return + if event.key == "j": + option_list.action_cursor_down() + else: + option_list.action_cursor_up() + event.prevent_default() + event.stop() return key = event.key diff --git a/tests/ui/test_query_history.py b/tests/ui/test_query_history.py index 87a4b553..30ec5a86 100644 --- a/tests/ui/test_query_history.py +++ b/tests/ui/test_query_history.py @@ -294,3 +294,135 @@ def save_query(self, connection_name, query): option_list = screen.query_one("#history-list", OptionList) assert option_list.option_count == 1 assert all(entry.connection_name == "saved-db" for entry in screen._merged_entries) + + +class TestQueryHistoryVimNavigation: + """Tests for j/k vim-style navigation in the history screen.""" + + @pytest.mark.asyncio + async def test_j_moves_cursor_down_and_k_moves_up(self) -> None: + saved_conn = create_test_connection("saved-db", "sqlite") + entries = [ + QueryHistoryEntry( + query=f"select {i}", + timestamp=f"2026-01-0{i}T00:00:00", + connection_name="saved-db", + ) + for i in range(1, 4) + ] + + class StubHistoryStore: + def __init__(self, entries): + self._entries = entries + + def load_all(self): + return list(self._entries) + + def load_for_connection(self, connection_name): + return [e for e in self._entries if e.connection_name == connection_name] + + def delete_entry(self, connection_name, timestamp): + _ = connection_name + _ = timestamp + return False + + def save_query(self, connection_name, query): + _ = connection_name + _ = query + + history_store = StubHistoryStore(entries) + services = build_test_services( + connection_store=MockConnectionStore([saved_conn]), + settings_store=MockSettingsStore({"theme": "tokyo-night"}), + history_store=history_store, + ) + app = SSMSTUI(services=services) + + async with app.run_test(size=(100, 35)) as pilot: + app.connections = [saved_conn] + app.current_config = saved_conn + app.action_show_history() + await pilot.pause(0.2) + + screen = next( + (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)), + None, + ) + assert screen is not None + + option_list = screen.query_one("#history-list", OptionList) + assert option_list.option_count == 3 + assert option_list.highlighted == 0 + + await pilot.press("j") + assert option_list.highlighted == 1 + + await pilot.press("j") + assert option_list.highlighted == 2 + + await pilot.press("k") + assert option_list.highlighted == 1 + + @pytest.mark.asyncio + async def test_j_and_k_typeable_when_filter_active(self) -> None: + saved_conn = create_test_connection("saved-db", "sqlite") + entries = [ + QueryHistoryEntry( + query="select jk_marker", + timestamp="2026-01-01T00:00:00", + connection_name="saved-db", + ), + QueryHistoryEntry( + query="select other", + timestamp="2026-01-02T00:00:00", + connection_name="saved-db", + ), + ] + + class StubHistoryStore: + def __init__(self, entries): + self._entries = entries + + def load_all(self): + return list(self._entries) + + def load_for_connection(self, connection_name): + return [e for e in self._entries if e.connection_name == connection_name] + + def delete_entry(self, connection_name, timestamp): + _ = connection_name + _ = timestamp + return False + + def save_query(self, connection_name, query): + _ = connection_name + _ = query + + history_store = StubHistoryStore(entries) + services = build_test_services( + connection_store=MockConnectionStore([saved_conn]), + settings_store=MockSettingsStore({"theme": "tokyo-night"}), + history_store=history_store, + ) + app = SSMSTUI(services=services) + + async with app.run_test(size=(100, 35)) as pilot: + app.connections = [saved_conn] + app.current_config = saved_conn + app.action_show_history() + await pilot.pause(0.2) + + screen = next( + (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)), + None, + ) + assert screen is not None + + await pilot.press("slash") + await pilot.pause() + assert screen._filter_active + + await pilot.press("j") + await pilot.press("k") + await pilot.pause() + assert screen._filter_text == "jk" From 4ff2b539653363005366155c0e8d628a758d73a5 Mon Sep 17 00:00:00 2001 From: Peter Adams <18162810+Maxteabag@users.noreply.github.com> Date: Mon, 25 May 2026 13:38:16 +0200 Subject: [PATCH 2/2] fix(history): render brackets safely, collapse same-text rows Two bugs in the query history screen: 1. Bracket identifiers like [dbo].[ServiceFeatures] were truncated to "[dbo]." in the option list. rich.markup.escape only escapes brackets whose tag-name starts with [a-z#/@], so [ServiceFeatures] slipped through and Textual's markup parser ate it as an unknown tag. Build options (and the preview) as Rich Text instead of markup strings to sidestep parsing entirely. 2. Starring one row marked every same-text row as starred, because history dedup is per (connection, database, query_hash) but the starred store is keyed by (connection, query_text). Collapse rows with identical query text in _merge_entries (keeping the most-recent timestamp), so the display row count matches the starring model. --- .../domains/query/ui/screens/query_history.py | 61 ++++-- tests/ui/test_query_history.py | 180 ++++++++++++++++++ 2 files changed, 222 insertions(+), 19 deletions(-) diff --git a/sqlit/domains/query/ui/screens/query_history.py b/sqlit/domains/query/ui/screens/query_history.py index 4623420b..55062616 100644 --- a/sqlit/domains/query/ui/screens/query_history.py +++ b/sqlit/domains/query/ui/screens/query_history.py @@ -7,7 +7,7 @@ from typing import Any from textual.app import ComposeResult -from rich.markup import escape +from rich.text import Text from textual.binding import Binding from textual.containers import VerticalScroll from textual.screen import ModalScreen @@ -112,17 +112,32 @@ def __init__( self._filtered_entries: list[QueryHistoryEntry] = [] def _merge_entries(self) -> list[QueryHistoryEntry]: - """Merge history entries with starred-only queries.""" + """Merge history entries with starred-only queries. + + Same-text rows collapse to a single most-recent entry so the row + count matches the starring model (which is keyed by query text): + starring one row never marks two visually-identical rows. In + multi-connection (telescope) mode the dedupe key is + (connection_name, query); in single-connection mode it's just + the query. + """ result: list[QueryHistoryEntry] = [] if self._multi_connection: - history_queries_by_connection: dict[str, set[str]] = {} + seen: dict[tuple[str, str], QueryHistoryEntry] = {} for entry in self.history: starred_queries = self._starred_by_connection.get(entry.connection_name, set()) entry.is_starred = entry.query.strip() in starred_queries entry.is_starred_only = False - history_queries_by_connection.setdefault(entry.connection_name, set()).add(entry.query.strip()) - result.append(entry) + key = (entry.connection_name, entry.query.strip()) + existing = seen.get(key) + if existing is None or entry.timestamp > existing.timestamp: + seen[key] = entry + result.extend(seen.values()) + + history_queries_by_connection: dict[str, set[str]] = {} + for (conn_name, query_text) in seen: + history_queries_by_connection.setdefault(conn_name, set()).add(query_text) for connection_name, starred_queries in self._starred_by_connection.items(): history_queries = history_queries_by_connection.get(connection_name, set()) @@ -137,15 +152,17 @@ def _merge_entries(self) -> list[QueryHistoryEntry]: ) result.append(entry) else: - # Mark history entries that are starred - history_queries: set[str] = set() + seen_single: dict[str, QueryHistoryEntry] = {} for entry in self.history: entry.is_starred = entry.query.strip() in self.starred entry.is_starred_only = False - history_queries.add(entry.query.strip()) - result.append(entry) + key = entry.query.strip() + existing = seen_single.get(key) + if existing is None or entry.timestamp > existing.timestamp: + seen_single[key] = entry + result.extend(seen_single.values()) - # Add starred-only entries (queries that were starred but not in history) + history_queries = set(seen_single.keys()) for starred_query in self.starred: if starred_query not in history_queries: entry = QueryHistoryEntry( @@ -221,7 +238,7 @@ def _update_preview(self, idx: int) -> None: entries = self._get_display_entries() if idx < len(entries): preview = self.query_one("#history-preview", Static) - preview.update(escape(entries[idx].query)) + preview.update(Text(entries[idx].query)) def action_select(self) -> None: entries = self._get_display_entries() @@ -479,19 +496,25 @@ def _build_action_result(self, action: str, entry: QueryHistoryEntry) -> tuple[s return action, entry.query def _build_option(self, entry: QueryHistoryEntry) -> Option: - star = "[yellow]*[/] " if entry.is_starred else " " - # Collapse all whitespace to single spaces and strip + # Build a Rich Text (not a markup string) so identifiers like + # [dbo].[ServiceFeatures] aren't eaten by the markup parser: + # rich.markup.escape only escapes "[a-z#/@…]" tags, so uppercase + # bracket identifiers slip through and get consumed as fake tags. query_single_line = re.sub(r"\s+", " ", entry.query).strip() max_len = 40 if self._multi_connection else 55 if len(query_single_line) > max_len: query_preview = query_single_line[:max_len] + "..." else: query_preview = query_single_line - query_preview = escape(query_preview) + + text = Text() + if entry.is_starred: + text.append("* ", style="yellow") + else: + text.append(" ") if self._multi_connection: conn_label = self._truncate_label(self._entry_connection_label(entry), 24) - conn_label = escape(conn_label) - option_id = self._entry_option_id(entry) - if self._multi_connection: - return Option(f"{star}[dim]{conn_label}[/]/{query_preview}", id=option_id) - return Option(f"{star}{query_preview}", id=option_id) + text.append(conn_label, style="dim") + text.append("/") + text.append(query_preview) + return Option(text, id=self._entry_option_id(entry)) diff --git a/tests/ui/test_query_history.py b/tests/ui/test_query_history.py index 30ec5a86..0defb72a 100644 --- a/tests/ui/test_query_history.py +++ b/tests/ui/test_query_history.py @@ -296,6 +296,186 @@ def save_query(self, connection_name, query): assert all(entry.connection_name == "saved-db" for entry in screen._merged_entries) +class TestQueryHistoryStarringDuplicates: + """Regression tests for starring behavior when multiple history entries + share the same query text (e.g., same SQL run against multiple databases + on one connection — file-backed dedup is per-(connection, database)).""" + + def _make_screen( + self, + history: list[QueryHistoryEntry], + starred: set[str], + connection_name: str = "saved-db", + ) -> QueryHistoryScreen: + return QueryHistoryScreen( + history=history, + connection_name=connection_name, + starred=starred, + ) + + def test_same_query_across_databases_collapses_to_single_row(self) -> None: + history = [ + QueryHistoryEntry( + query="SELECT 1", + timestamp="2026-01-02T00:00:00", + connection_name="saved-db", + database="db_a", + ), + QueryHistoryEntry( + query="SELECT 1", + timestamp="2026-01-01T00:00:00", + connection_name="saved-db", + database="db_b", + ), + ] + screen = self._make_screen(history, starred={"SELECT 1"}) + + merged = screen._merge_entries() + + assert len(merged) == 1, ( + "Same query text from two databases should collapse to one row " + "in the history display (starring is keyed by text, so the row " + "count must match)." + ) + assert merged[0].is_starred + # Most-recent occurrence wins the row. + assert merged[0].database == "db_a" + + @pytest.mark.asyncio + async def test_starring_one_marks_duplicate_in_other_database(self) -> None: + saved_conn = create_test_connection("saved-db", "sqlite") + entries = [ + QueryHistoryEntry( + query="SELECT 1", + timestamp="2026-01-02T00:00:00", + connection_name="saved-db", + database="db_a", + ), + QueryHistoryEntry( + query="SELECT 1", + timestamp="2026-01-01T00:00:00", + connection_name="saved-db", + database="db_b", + ), + ] + + class StubHistoryStore: + def __init__(self, entries): + self._entries = entries + + def load_all(self): + return list(self._entries) + + def load_for_connection(self, connection_name): + return [e for e in self._entries if e.connection_name == connection_name] + + def delete_entry(self, connection_name, timestamp): + _ = connection_name + _ = timestamp + return False + + def save_query(self, connection_name, query, database=""): + _ = connection_name + _ = query + _ = database + + history_store = StubHistoryStore(entries) + services = build_test_services( + connection_store=MockConnectionStore([saved_conn]), + settings_store=MockSettingsStore({"theme": "tokyo-night"}), + history_store=history_store, + ) + app = SSMSTUI(services=services) + + async with app.run_test(size=(100, 35)) as pilot: + app.connections = [saved_conn] + app.current_config = saved_conn + + # Open history, star the first row (which corresponds to db_a), + # close — emulates: user opens history, presses '*' on one row. + app.action_show_history() + await pilot.pause(0.2) + screen = next( + (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)), + None, + ) + assert screen is not None + option_list = screen.query_one("#history-list", OptionList) + # The two same-text rows should already be collapsed to one. + assert option_list.option_count == 1 + assert option_list.highlighted == 0 + screen.action_toggle_star() + await pilot.pause(0.2) + + # Re-open: exactly one starred row, not two. + app.action_show_history() + await pilot.pause(0.2) + screen = next( + (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)), + None, + ) + assert screen is not None + starred_entries = [e for e in screen._merged_entries if e.is_starred] + try: + assert len(starred_entries) == 1, ( + f"Expected exactly one starred row after toggling one; " + f"got {len(starred_entries)}" + ) + finally: + # The shared test config dir is module-scoped (see + # tests/fixtures/utils.py), so any star we save here would + # leak to later tests. Clean up. + services.starred_store.clear_for_connection(saved_conn.name) + + +class TestQueryHistoryBracketIdentifierRendering: + """Regression tests for MSSQL bracket identifiers like [dbo].[ServiceFeatures] + being silently eaten by Textual markup when the bracket contents start with + an uppercase letter (rich.markup.escape only escapes [a-z#/@] tags).""" + + @pytest.mark.asyncio + async def test_bracket_identifier_with_uppercase_not_truncated(self) -> None: + from textual.app import App + from textual.widgets.option_list import Option + + # Drive the option directly through a minimal Textual app so we + # observe what _build_option's text actually renders to. + screen = QueryHistoryScreen( + history=[ + QueryHistoryEntry( + query="SELECT TOP 100 * FROM [dbo].[ServiceFeatures]", + timestamp="2026-01-01T00:00:00", + connection_name="saved-db", + ), + ], + connection_name="saved-db", + starred=set(), + ) + screen._merged_entries = screen._merge_entries() + built = screen._build_option(screen._merged_entries[0]) + + class _ProbeApp(App): + def compose(self): + yield OptionList(Option(built.prompt, id=built.id or "probe"), id="probe-list") + + app = _ProbeApp() + async with app.run_test(size=(120, 10)) as pilot: + await pilot.pause() + option_list = app.query_one("#probe-list", OptionList) + lines = option_list.render_lines(option_list.region) + rendered_text = "\n".join( + "".join(seg.text for seg in line) for line in lines + ) + + assert "[ServiceFeatures]" in rendered_text, ( + "Bracket identifier with uppercase first letter was silently " + "swallowed by Textual markup (rich.markup.escape only escapes " + "tags starting with [a-z#/@], so '[ServiceFeatures]' is left " + "unescaped and the markup parser eats it). Visible text:\n" + + rendered_text + ) + + class TestQueryHistoryVimNavigation: """Tests for j/k vim-style navigation in the history screen."""