diff --git a/sqlit/core/keymap.py b/sqlit/core/keymap.py index b6d7fe4d..0a1c535a 100644 --- a/sqlit/core/keymap.py +++ b/sqlit/core/keymap.py @@ -353,6 +353,9 @@ def _build_action_keys(self) -> list[ActionKeyDef]: ActionKeyDef("f", "refresh_tree", "tree"), ActionKeyDef("R", "refresh_tree", "tree", primary=False), ActionKeyDef("e", "edit_connection", "tree"), + ActionKeyDef("r", "rename_saved_query", "tree"), + ActionKeyDef("d", "delete_saved_query", "tree"), + ActionKeyDef("delete", "delete_saved_query", "tree", primary=False), ActionKeyDef("M", "rename_connection_folder", "tree"), ActionKeyDef("d", "delete_connection_folder", "tree"), ActionKeyDef("delete", "delete_connection_folder", "tree", primary=False), diff --git a/sqlit/domains/explorer/state/tree_on_folder.py b/sqlit/domains/explorer/state/tree_on_folder.py index e2fd3f63..643167f0 100644 --- a/sqlit/domains/explorer/state/tree_on_folder.py +++ b/sqlit/domains/explorer/state/tree_on_folder.py @@ -25,6 +25,18 @@ def is_connection_folder(app: InputContext) -> bool: label="Delete", help="Delete connection folder", ) + self.allows( + "rename_saved_query", + lambda app: app.tree_node_kind == "saved_query_folder", + label="Rename", + help="Rename saved-query folder", + ) + self.allows( + "delete_saved_query", + lambda app: app.tree_node_kind == "saved_query_folder", + label="Delete", + help="Delete saved-query folder", + ) def get_display_bindings(self, app: InputContext) -> tuple[list[DisplayBinding], list[DisplayBinding]]: left: list[DisplayBinding] = [] @@ -58,6 +70,22 @@ def get_display_bindings(self, app: InputContext) -> tuple[list[DisplayBinding], ) ) seen.add("delete_connection_folder") + elif app.tree_node_kind == "saved_query_folder": + left.extend( + ( + DisplayBinding( + key=resolve_display_key("rename_saved_query") or "r", + label="Rename", + action="rename_saved_query", + ), + DisplayBinding( + key=resolve_display_key("delete_saved_query") or "d", + label="Delete", + action="delete_saved_query", + ), + ) + ) + seen.update({"rename_saved_query", "delete_saved_query"}) if self.parent: parent_left, _ = self.parent.get_display_bindings(app) diff --git a/sqlit/domains/explorer/state/tree_on_object.py b/sqlit/domains/explorer/state/tree_on_object.py index 9a410e04..610a9355 100644 --- a/sqlit/domains/explorer/state/tree_on_object.py +++ b/sqlit/domains/explorer/state/tree_on_object.py @@ -13,6 +13,18 @@ class TreeOnObjectState(State): def _setup_actions(self) -> None: self.allows("select_table", label="Open", help="Open the selected object") + self.allows( + "rename_saved_query", + lambda app: app.tree_node_kind == "saved_query_file", + label="Rename", + help="Rename saved query", + ) + self.allows( + "delete_saved_query", + lambda app: app.tree_node_kind == "saved_query_file", + label="Delete", + help="Delete saved query", + ) def get_display_bindings(self, app: InputContext) -> tuple[list[DisplayBinding], list[DisplayBinding]]: left: list[DisplayBinding] = [] @@ -27,6 +39,22 @@ def get_display_bindings(self, app: InputContext) -> tuple[list[DisplayBinding], ) ) seen.add("select_table") + if app.tree_node_kind == "saved_query_file": + left.extend( + ( + DisplayBinding( + key=resolve_display_key("rename_saved_query") or "r", + label="Rename", + action="rename_saved_query", + ), + DisplayBinding( + key=resolve_display_key("delete_saved_query") or "d", + label="Delete", + action="delete_saved_query", + ), + ) + ) + seen.update({"rename_saved_query", "delete_saved_query"}) left.append( DisplayBinding( key=resolve_display_key("refresh_tree") or "f", diff --git a/sqlit/domains/explorer/ui/mixins/tree.py b/sqlit/domains/explorer/ui/mixins/tree.py index f53960f4..5e682134 100644 --- a/sqlit/domains/explorer/ui/mixins/tree.py +++ b/sqlit/domains/explorer/ui/mixins/tree.py @@ -6,7 +6,10 @@ from textual.widgets import Tree -from sqlit.domains.query.store.saved_queries import SavedQueryNameError +from sqlit.domains.query.store.saved_queries import ( + SavedQueryConflictError, + SavedQueryNameError, +) from sqlit.shared.ui.protocols import TreeMixinHost from ..tree import builder as tree_builder @@ -96,12 +99,7 @@ def on_tree_node_expanded(self: TreeMixinHost, event: Tree.NodeExpanded) -> None if self._get_node_kind(node) == "connection": config = getattr(data, "config", None) - if ( - config - and self.current_config - and self.current_config.name == config.name - and not list(node.children) - ): + if config and self.current_config and self.current_config.name == config.name and not list(node.children): tree_builder.populate_connected_tree(self) return @@ -242,6 +240,181 @@ def open_query() -> None: else: open_query() + def _saved_query_document_moved( + self: TreeMixinHost, + connection_name: str, + old_path: str, + new_path: str, + *, + folder: bool, + ) -> None: + document = getattr(self, "_query_document", None) + if document is None or document.connection_name != connection_name: + return + current = document.relative_path + if not current: + return + if folder: + prefix = f"{old_path}/" + if not current.startswith(prefix): + return + replacement = f"{new_path}/{current[len(prefix) :]}" + elif current == old_path: + replacement = new_path + else: + return + setter = getattr(self, "_set_query_document", None) + if callable(setter): + setter( + connection_name=connection_name, + relative_path=replacement, + saved_text=document.saved_text, + fingerprint=document.fingerprint, + ) + + def _saved_query_document_deleted( + self: TreeMixinHost, + connection_name: str, + path: str, + *, + folder: bool, + ) -> None: + document = getattr(self, "_query_document", None) + if document is None or document.connection_name != connection_name: + return + current = document.relative_path + matches = current == path or (folder and bool(current) and current.startswith(f"{path}/")) + if not matches: + return + setter = getattr(self, "_set_query_document", None) + if callable(setter): + setter( + connection_name=connection_name, + relative_path=None, + saved_text="", + fingerprint=None, + ) + + def action_rename_saved_query(self: TreeMixinHost) -> None: + node = self.object_tree.cursor_node + if not node or not node.data: + return + data = node.data + kind = self._get_node_kind(node) + if kind == "saved_query_file": + from sqlit.domains.query.ui.screens import SavedQueryNameScreen + + old_path = data.entry.relative_path + + def on_name(name: str | None) -> None: + if not name: + return + try: + entry = self.services.saved_query_store.rename_file( + data.connection_name, + old_path, + name, + expected=data.entry.fingerprint, + ) + except FileExistsError: + self.notify("A saved query already uses that name", severity="warning") + return + except (OSError, SavedQueryConflictError, SavedQueryNameError) as exc: + self.notify(f"Could not rename saved query: {exc}", severity="error") + return + self._saved_query_document_moved( + data.connection_name, + old_path, + entry.relative_path, + folder=False, + ) + self._refresh_connection_tree() + self.notify(f"Renamed to {entry.relative_path}") + + self.push_screen( + SavedQueryNameScreen(initial_name=old_path, title="Rename Saved Query"), + on_name, + ) + return + if kind != "saved_query_folder" or not data.relative_path: + return + from sqlit.domains.connections.ui.screens import FolderInputScreen + + old_path = data.relative_path + + def on_folder(name: str | None) -> None: + if not name: + return + try: + new_path = self.services.saved_query_store.rename_folder(data.connection_name, old_path, name) + except FileExistsError: + self.notify("A saved-query folder already uses that name", severity="warning") + return + except (OSError, SavedQueryNameError) as exc: + self.notify(f"Could not rename saved-query folder: {exc}", severity="error") + return + self._saved_query_document_moved(data.connection_name, old_path, new_path, folder=True) + self._refresh_connection_tree() + self.notify(f"Renamed folder to {new_path}") + + self.push_screen( + FolderInputScreen( + data.connection_name, + current_value=old_path, + title="Rename Saved-Query Folder", + description="New folder path (use / for nesting):", + ), + on_folder, + ) + + def action_delete_saved_query(self: TreeMixinHost) -> None: + node = self.object_tree.cursor_node + if not node or not node.data: + return + data = node.data + kind = self._get_node_kind(node) + if kind == "saved_query_file": + path = data.entry.relative_path + folder = False + description = "The query text will remain open if this file is active." + elif kind == "saved_query_folder" and data.relative_path: + path = data.relative_path + folder = True + count = sum(1 for entry in self.services.saved_query_store.list_for_connection(data.connection_name) if entry.relative_path.startswith(f"{path}/")) + description = f"This permanently deletes {count} saved query file{'s' if count != 1 else ''}." + else: + return + from sqlit.shared.ui.screens.confirm import ConfirmScreen + + def on_confirm(confirmed: bool | None) -> None: + if not confirmed: + return + try: + if folder: + self.services.saved_query_store.delete_folder(data.connection_name, path) + else: + self.services.saved_query_store.delete_file( + data.connection_name, + path, + expected=data.entry.fingerprint, + ) + except (OSError, SavedQueryConflictError, SavedQueryNameError) as exc: + self.notify(f"Could not delete saved query: {exc}", severity="error") + return + self._saved_query_document_deleted(data.connection_name, path, folder=folder) + self._refresh_connection_tree() + self.notify(f"Deleted {path}") + + self.push_screen( + ConfirmScreen( + f"Delete {'folder' if folder else 'saved query'} '{path}'?", + description, + yes_label="Delete", + no_label="Cancel", + ), + on_confirm, + ) + def on_tree_node_highlighted(self: TreeMixinHost, event: Tree.NodeHighlighted) -> None: """Update footer when tree selection changes.""" self._update_footer_bindings() @@ -445,9 +618,7 @@ async def work_async() -> None: incoming_fks: list[Any] = [] try: runtime = getattr(self.services, "runtime", None) - use_worker = bool(getattr(runtime, "process_worker", False)) and not bool( - getattr(getattr(runtime, "mock", None), "enabled", False) - ) + use_worker = bool(getattr(runtime, "process_worker", False)) and not bool(getattr(getattr(runtime, "mock", None), "enabled", False)) client = None if use_worker and hasattr(self, "_get_process_worker_client_async"): client = await self._get_process_worker_client_async() # type: ignore[attr-defined] @@ -473,28 +644,37 @@ async def work_async() -> None: schema_service = self._get_schema_service() if schema_service: if not columns: - columns = await asyncio.to_thread( - schema_service.list_columns, - database, - schema, - name, - ) or [] + columns = ( + await asyncio.to_thread( + schema_service.list_columns, + database, + schema, + name, + ) + or [] + ) try: - outgoing_fks = await asyncio.to_thread( - schema_service.list_foreign_keys, - database, - schema, - name, - ) or [] + outgoing_fks = ( + await asyncio.to_thread( + schema_service.list_foreign_keys, + database, + schema, + name, + ) + or [] + ) except Exception: outgoing_fks = [] try: - incoming_fks = await asyncio.to_thread( - schema_service.list_referencing_foreign_keys, - database, - schema, - name, - ) or [] + incoming_fks = ( + await asyncio.to_thread( + schema_service.list_referencing_foreign_keys, + database, + schema, + name, + ) + or [] + ) except Exception: incoming_fks = [] except Exception: @@ -530,11 +710,7 @@ def _apply_last_query_table_columns( table_info = getattr(self, "_last_query_table", None) if not table_info: return - if ( - table_info.get("database") != database - or table_info.get("schema") != schema - or table_info.get("name") != name - ): + if table_info.get("database") != database or table_info.get("schema") != schema or table_info.get("name") != name: return table_info["columns"] = columns if foreign_keys is not None: diff --git a/sqlit/domains/query/store/saved_queries.py b/sqlit/domains/query/store/saved_queries.py index e770d5d2..f71894ad 100644 --- a/sqlit/domains/query/store/saved_queries.py +++ b/sqlit/domains/query/store/saved_queries.py @@ -64,16 +64,12 @@ def _fingerprint(path: Path, data: bytes | None = None) -> FileFingerprint: while chunk := handle.read(64 * 1024): size += len(chunk) if size > _MAX_QUERY_BYTES: - raise OSError( - f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") digest.update(chunk) digest_value = digest.hexdigest() else: if len(data) > _MAX_QUERY_BYTES: - raise OSError( - f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") size = len(data) digest_value = hashlib.sha256(data).hexdigest() return FileFingerprint( @@ -88,15 +84,24 @@ def _normalize_relative_name(name: str) -> PurePosixPath: if not raw: raise SavedQueryNameError("Enter a query name") relative = PurePosixPath(raw) + if not relative.parts or relative.is_absolute() or any(part in {"", ".", ".."} for part in relative.parts) or any(part.startswith(".") for part in relative.parts): + raise SavedQueryNameError("Use a relative name inside the query library") + if relative.suffix.lower() != ".sql": + relative = relative.with_suffix(".sql") + return relative + + +def _normalize_relative_folder(name: str) -> PurePosixPath: + raw = name.strip().replace("\\", "/").strip("/") + relative = PurePosixPath(raw) if ( - not relative.parts + not raw + or raw in {".", ".."} or relative.is_absolute() or any(part in {"", ".", ".."} for part in relative.parts) or any(part.startswith(".") for part in relative.parts) ): - raise SavedQueryNameError("Use a relative name inside the query library") - if relative.suffix.lower() != ".sql": - relative = relative.with_suffix(".sql") + raise SavedQueryNameError("Use a relative folder inside the query library") return relative @@ -122,9 +127,7 @@ def _path_for_name(self, connection_name: str, name: str) -> tuple[Path, str]: def _read_entry(self, root: Path, path: Path) -> SavedQueryEntry: if self._secure_dir_fd_available(): relative_path = path.relative_to(root).as_posix() - directory_fd, filename = self._open_secure_parent( - root, relative_path, create=False - ) + directory_fd, filename = self._open_secure_parent(root, relative_path, create=False) try: try: file_fd = os.open( @@ -133,24 +136,16 @@ def _read_entry(self, root: Path, path: Path) -> SavedQueryEntry: dir_fd=directory_fd, ) except OSError as exc: - raise SavedQueryNameError( - "Symlinks inside a query library are not supported" - ) from exc + raise SavedQueryNameError("Symlinks inside a query library are not supported") from exc try: file_stat = os.fstat(file_fd) if file_stat.st_size > _MAX_QUERY_BYTES: - raise OSError( - "Query file is larger than " - f"{_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") payload = bytearray() while chunk := os.read(file_fd, 64 * 1024): payload.extend(chunk) if len(payload) > _MAX_QUERY_BYTES: - raise OSError( - "Query file is larger than " - f"{_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") finally: os.close(file_fd) finally: @@ -195,9 +190,7 @@ def list_for_connection(self, connection_name: str) -> list[SavedQueryEntry]: return [] entries: list[SavedQueryEntry] = [] try: - paths = sorted( - root.rglob("*"), key=lambda item: item.as_posix().lower() - ) + paths = sorted(root.rglob("*"), key=lambda item: item.as_posix().lower()) except OSError: return [] for path in paths: @@ -223,20 +216,117 @@ def current_fingerprint(self, connection_name: str, relative_path: str) -> FileF except OSError: return None + def _remove_empty_parents(self, root: Path, parent: Path) -> None: + while parent != root: + try: + parent.rmdir() + except OSError: + return + parent = parent.parent + + def delete_file( + self, + connection_name: str, + relative_path: str, + *, + expected: FileFingerprint | None = None, + ) -> None: + """Delete one saved query without following links or stale selections.""" + path, normalized = self._path_for_name(connection_name, relative_path) + root = self.connection_dir(connection_name) + if self._secure_dir_fd_available(): + directory_fd, filename = self._open_secure_parent(root, normalized, create=False) + try: + current = self._fingerprint_at(directory_fd, filename) + if current is None: + raise FileNotFoundError(normalized) + if expected is not None and current != expected: + raise SavedQueryConflictError(f"{normalized} changed on disk") + os.unlink(filename, dir_fd=directory_fd) + finally: + os.close(directory_fd) + else: + self._reject_nested_symlink(root, path) + current = _fingerprint(path) if path.exists() else None + if current is None: + raise FileNotFoundError(normalized) + if expected is not None and current != expected: + raise SavedQueryConflictError(f"{normalized} changed on disk") + path.unlink() + self._remove_empty_parents(root, path.parent) + + def delete_folder(self, connection_name: str, relative_path: str) -> int: + """Delete visible SQL files in a folder, preserving unrelated files.""" + relative = _normalize_relative_folder(relative_path) + prefix = f"{relative.as_posix()}/" + entries = [ + entry + for entry in self.list_for_connection(connection_name) + if entry.relative_path.startswith(prefix) + ] + for entry in entries: + self.delete_file( + connection_name, + entry.relative_path, + expected=entry.fingerprint, + ) + return len(entries) + + def rename_file( + self, + connection_name: str, + old_relative_path: str, + new_name: str, + *, + expected: FileFingerprint | None = None, + ) -> SavedQueryEntry: + """Rename or move one saved query without overwriting another file.""" + old_path, old_relative = self._path_for_name(connection_name, old_relative_path) + new_path, new_relative = self._path_for_name(connection_name, new_name) + root = self.connection_dir(connection_name) + if old_relative == new_relative: + return self.load(connection_name, old_relative) + if new_path.exists(): + raise FileExistsError(new_relative) + if expected is not None and self.current_fingerprint(connection_name, old_relative) != expected: + raise SavedQueryConflictError(f"{old_relative} changed on disk") + self._reject_nested_symlink(root, old_path) + self._reject_nested_symlink(root, new_path) + new_path.parent.mkdir(parents=True, exist_ok=True) + old_path.rename(new_path) + self._remove_empty_parents(root, old_path.parent) + return self._read_entry(root, new_path) + + def rename_folder( + self, + connection_name: str, + old_relative_path: str, + new_name: str, + ) -> str: + """Rename or move a saved-query folder without overwriting.""" + old_relative = _normalize_relative_folder(old_relative_path) + new_relative = _normalize_relative_folder(new_name) + root = self.connection_dir(connection_name) + old_path = root.joinpath(*old_relative.parts) + new_path = root.joinpath(*new_relative.parts) + if old_path == new_path: + return new_relative.as_posix() + if new_path.exists(): + raise FileExistsError(new_relative.as_posix()) + if new_path == old_path or old_path in new_path.parents: + raise SavedQueryNameError("A folder cannot be moved inside itself") + self._reject_nested_symlink(root, old_path) + self._reject_nested_symlink(root, new_path) + new_path.parent.mkdir(parents=True, exist_ok=True) + old_path.rename(new_path) + self._remove_empty_parents(root, old_path.parent) + return new_relative.as_posix() + @staticmethod def _secure_dir_fd_available() -> bool: - return ( - os.name != "nt" - and hasattr(os, "O_DIRECTORY") - and hasattr(os, "O_NOFOLLOW") - and os.open in os.supports_dir_fd - and os.mkdir in os.supports_dir_fd - and os.rename in os.supports_dir_fd - ) + return os.name != "nt" and hasattr(os, "O_DIRECTORY") and hasattr(os, "O_NOFOLLOW") and os.open in os.supports_dir_fd and os.mkdir in os.supports_dir_fd and os.rename in os.supports_dir_fd - def _open_secure_parent( - self, root: Path, relative_path: str, *, create: bool = True - ) -> tuple[int, str]: + def _open_secure_parent(self, root: Path, relative_path: str, *, create: bool = True) -> tuple[int, str]: """Open and pin every directory below an intentionally trusted root.""" if create: root.parent.mkdir(parents=True, exist_ok=True) @@ -262,9 +352,7 @@ def _open_secure_parent( dir_fd=directory_fd, ) except OSError as exc: - raise SavedQueryNameError( - "Symlinks inside a query library are not supported" - ) from exc + raise SavedQueryNameError("Symlinks inside a query library are not supported") from exc os.close(directory_fd) directory_fd = child_fd return directory_fd, parts[-1] @@ -283,23 +371,17 @@ def _fingerprint_at(directory_fd: int, filename: str) -> FileFingerprint | None: except FileNotFoundError: return None except OSError as exc: - raise SavedQueryNameError( - "Symlinks inside a query library are not supported" - ) from exc + raise SavedQueryNameError("Symlinks inside a query library are not supported") from exc try: file_stat = os.fstat(file_fd) if file_stat.st_size > _MAX_QUERY_BYTES: - raise OSError( - f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") digest = hashlib.sha256() size = 0 while chunk := os.read(file_fd, 64 * 1024): size += len(chunk) if size > _MAX_QUERY_BYTES: - raise OSError( - f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB" - ) + raise OSError(f"Query file is larger than {_MAX_QUERY_BYTES // (1024 * 1024)} MiB") digest.update(chunk) return FileFingerprint( digest=digest.hexdigest(), diff --git a/tests/ui/test_saved_query_documents.py b/tests/ui/test_saved_query_documents.py index 76233005..a35a64f9 100644 --- a/tests/ui/test_saved_query_documents.py +++ b/tests/ui/test_saved_query_documents.py @@ -12,7 +12,11 @@ from textual.widgets import Input from sqlit.domains.connections.app.credentials import CredentialsPersistError -from sqlit.domains.explorer.domain.tree_nodes import TableNode +from sqlit.domains.explorer.domain.tree_nodes import ( + SavedQueryFileNode, + SavedQueryFolderNode, + TableNode, +) from sqlit.domains.query.store.saved_queries import SavedQueryStore from sqlit.domains.query.ui.screens import ( ExternalQueryChangeScreen, @@ -568,3 +572,65 @@ async def test_deleting_active_connection_protects_dirty_document( assert app.services.connection_store.get_by_name(config.name) is not None assert app._document_is_dirty() + + +@pytest.mark.asyncio +async def test_explorer_rename_saved_query_updates_open_document( + tmp_path: Path, +) -> None: + app, store = _make_app(tmp_path) + entry = store.save("production-reporting", "reports/daily", "SELECT 1") + + async with app.run_test(size=(120, 40)) as pilot: + app._load_saved_query_entry(entry) + app._refresh_connection_tree = MagicMock() # type: ignore[method-assign] + node = app.object_tree.root.add_leaf("daily.sql") + node.data = SavedQueryFileNode("production-reporting", entry) + app.object_tree.root.expand() + app.object_tree.refresh(layout=True) + await pilot.pause() + app.object_tree.move_cursor(node) + + app.action_rename_saved_query() + await pilot.pause() + prompt = app.screen + assert isinstance(prompt, SavedQueryNameScreen) + prompt.query_one("#saved-query-name-input", Input).value = "reports/weekly" + prompt.action_save() + await pilot.pause() + + assert store.load("production-reporting", "reports/weekly.sql").query == ("SELECT 1") + assert app._get_query_document().relative_path == "reports/weekly.sql" + assert not app._document_is_dirty() + + +@pytest.mark.asyncio +async def test_explorer_delete_folder_keeps_open_query_as_scratch( + tmp_path: Path, +) -> None: + app, store = _make_app(tmp_path) + entry = store.save("production-reporting", "reports/daily", "SELECT 1") + + async with app.run_test(size=(120, 40)) as pilot: + app._load_saved_query_entry(entry) + app.query_input.text = "SELECT edited" + app._refresh_connection_tree = MagicMock() # type: ignore[method-assign] + node = app.object_tree.root.add("reports") + node.data = SavedQueryFolderNode("production-reporting", "reports") + app.object_tree.root.expand() + app.object_tree.refresh(layout=True) + await pilot.pause() + app.object_tree.move_cursor(node) + + app.action_delete_saved_query() + await pilot.pause() + confirm = app.screen + assert isinstance(confirm, ConfirmScreen) + assert "1 saved query file" in str(confirm.query_one("#confirm-description").render()) + confirm.action_yes() + await pilot.pause() + + assert store.list_for_connection("production-reporting") == [] + assert app.query_input.text == "SELECT edited" + assert app._get_query_document().relative_path is None + assert app._document_is_dirty() diff --git a/tests/unit/test_saved_query_store.py b/tests/unit/test_saved_query_store.py index 939e939a..ff2b2acf 100644 --- a/tests/unit/test_saved_query_store.py +++ b/tests/unit/test_saved_query_store.py @@ -74,9 +74,7 @@ def test_list_is_recursive_sorted_and_ignores_hidden_files( ] -def test_list_tolerates_directory_traversal_failure( - store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_list_tolerates_directory_traversal_failure(store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch) -> None: store.save("production", "report", "SELECT 1") def fail_traversal(_self: Path, _pattern: str): @@ -150,9 +148,7 @@ def test_fingerprint_bounds_externally_replaced_oversized_file( assert store.current_fingerprint("production", "report.sql") is None -def test_save_revalidates_fingerprint_immediately_before_replace( - store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_save_revalidates_fingerprint_immediately_before_replace(store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch) -> None: existing = store.save("production", "report", "SELECT original") changed = type(existing.fingerprint)( digest="changed", @@ -174,9 +170,7 @@ def test_save_revalidates_fingerprint_immediately_before_replace( assert not list(existing.path.parent.glob(".sqlit-save-*")) -def test_fallback_save_does_not_replace_concurrently_created_file( - store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_fallback_save_does_not_replace_concurrently_created_file(store: SavedQueryStore, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(store, "_secure_dir_fd_available", lambda: False) path, _ = store._path_for_name("production", "report") checks = 0 @@ -301,3 +295,87 @@ def fail_rename(_self: Path, _target: Path) -> None: assert store.rename_connection("old", "new") is False assert store.list_for_connection("old")[0].query == "SELECT 1" + + +def test_rename_file_moves_query_and_removes_empty_folder( + store: SavedQueryStore, +) -> None: + original = store.save("production", "old/report", "SELECT 1") + + renamed = store.rename_file( + "production", + original.relative_path, + "new/daily", + expected=original.fingerprint, + ) + + assert renamed.relative_path == "new/daily.sql" + assert renamed.query == "SELECT 1" + assert not (store.connection_dir("production") / "old").exists() + + +def test_rename_file_refuses_collision_and_external_change( + store: SavedQueryStore, +) -> None: + original = store.save("production", "first", "SELECT 1") + store.save("production", "second", "SELECT 2") + + with pytest.raises(FileExistsError): + store.rename_file("production", "first.sql", "second.sql") + + original.path.write_text("SELECT changed", encoding="utf-8") + with pytest.raises(SavedQueryConflictError): + store.rename_file( + "production", + "first.sql", + "renamed.sql", + expected=original.fingerprint, + ) + + +def test_delete_file_checks_fingerprint_and_removes_empty_parents( + store: SavedQueryStore, +) -> None: + entry = store.save("production", "reports/daily", "SELECT 1") + entry.path.write_text("SELECT changed", encoding="utf-8") + + with pytest.raises(SavedQueryConflictError): + store.delete_file("production", entry.relative_path, expected=entry.fingerprint) + + current = store.load("production", entry.relative_path) + store.delete_file("production", current.relative_path, expected=current.fingerprint) + assert store.list_for_connection("production") == [] + assert not (store.connection_dir("production") / "reports").exists() + + +def test_rename_and_delete_folder_update_all_nested_queries( + store: SavedQueryStore, +) -> None: + store.save("production", "reports/daily", "SELECT 1") + store.save("production", "reports/monthly/summary", "SELECT 2") + + new_path = store.rename_folder("production", "reports", "analytics") + + assert new_path == "analytics" + assert [entry.relative_path for entry in store.list_for_connection("production")] == ["analytics/daily.sql", "analytics/monthly/summary.sql"] + assert store.delete_folder("production", "analytics") == 2 + assert store.list_for_connection("production") == [] + + +def test_delete_folder_preserves_unrelated_files(store: SavedQueryStore) -> None: + store.save("production", "reports/daily", "SELECT 1") + notes = store.connection_dir("production") / "reports" / "README.md" + notes.write_text("keep me", encoding="utf-8") + + assert store.delete_folder("production", "reports") == 1 + + assert notes.read_text(encoding="utf-8") == "keep me" + + +@pytest.mark.parametrize("name", ["", ".", "..", "../outside", ".hidden"]) +def test_folder_operations_reject_unsafe_names(store: SavedQueryStore, name: str) -> None: + store.save("production", "reports/daily", "SELECT 1") + with pytest.raises(SavedQueryNameError): + store.rename_folder("production", "reports", name) + with pytest.raises(SavedQueryNameError): + store.delete_folder("production", name)