diff --git a/sqlit/cli.py b/sqlit/cli.py index d6933b69..83329f26 100644 --- a/sqlit/cli.py +++ b/sqlit/cli.py @@ -58,7 +58,7 @@ def _extract_project_dir(argv: list[str]) -> tuple[Path | None, list[str]]: Returns (project_dir, remaining_argv). Exits with an error if the user passed a path that doesn't resolve to an existing directory. """ - subcommands = {"connections", "connection", "connect", "query", "docker"} + subcommands = {"connections", "connection", "connect", "query", "docker", "alerts"} result_argv: list[str] = [] project_dir: Path | None = None @@ -587,6 +587,11 @@ def main() -> int: add_schema_arguments(provider_parser, schema, include_name=True, name_required=True) provider_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password") provider_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password") + provider_parser.add_argument( + "--alert", + metavar="MODE", + help="Per-connection query alert mode: off|delete|write", + ) edit_parser = conn_subparsers.add_parser("edit", help="Edit an existing connection") edit_parser.add_argument("connection_name", help="Name of connection to edit") @@ -606,6 +611,11 @@ def main() -> int: edit_parser.add_argument("--file-path", help="Database file path (SQLite only)") edit_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password") edit_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password") + edit_parser.add_argument( + "--alert", + metavar="MODE", + help="Per-connection query alert mode: off|delete|write|unset (clears the override)", + ) delete_parser = conn_subparsers.add_parser("delete", help="Delete a connection") delete_parser.add_argument("connection_name", help="Name of connection to delete") @@ -622,6 +632,11 @@ def main() -> int: add_schema_arguments(provider_parser, schema, include_name=True, name_required=False) provider_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password") provider_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password") + provider_parser.add_argument( + "--alert", + metavar="MODE", + help="Per-connection query alert mode: off|delete|write", + ) query_parser = subparsers.add_parser("query", help="Execute a SQL query") query_parser.add_argument("--connection", "-c", required=True, help="Connection name to use") @@ -648,6 +663,44 @@ def main() -> int: docker_subparsers = docker_parser.add_subparsers(dest="docker_command", help="Docker commands") docker_subparsers.add_parser("list", help="List detected database containers") + # Query alert management + alerts_parser = subparsers.add_parser( + "alerts", + help="Manage query confirmation alert overrides", + description=( + "Manage query alert overrides. Hierarchy: database > connection > global." + ), + ) + alerts_subparsers = alerts_parser.add_subparsers(dest="alerts_command", help="Alert commands") + alerts_subparsers.add_parser("list", help="Show global and all configured overrides") + + alerts_set = alerts_subparsers.add_parser("set", help="Set an alert mode at a scope") + alerts_set.add_argument("mode", help="off | delete | write") + alerts_set.add_argument( + "--connection", + "-c", + help="Target a specific saved connection (omit for global)", + ) + alerts_set.add_argument( + "--database", + "-d", + help="Target a database on the connection (requires --connection)", + ) + + alerts_unset = alerts_subparsers.add_parser( + "unset", help="Clear an alert override (falls back to the next scope)" + ) + alerts_unset.add_argument( + "--connection", + "-c", + help="Connection whose override to clear", + ) + alerts_unset.add_argument( + "--database", + "-d", + help="Database whose override to clear (requires --connection)", + ) + log_startup_step("cli_parser_end") with startup_span("cli_parse_args"): @@ -758,6 +811,13 @@ def main() -> int: print(f"Error: {exc}") return 1 + from sqlit.domains.connections.cli.commands import _apply_alert_option + + alert_error = _apply_alert_option(temp_config, getattr(args, "alert", None)) + if alert_error: + print(f"Error: {alert_error}") + return 1 + app = SSMSTUI(services=services, startup_connection=temp_config) return _run_app(app) @@ -786,6 +846,22 @@ def main() -> int: docker_parser.print_help() return 1 + if args.command == "alerts": + from sqlit.domains.query.cli.alerts_commands import ( + cmd_alerts_list, + cmd_alerts_set, + cmd_alerts_unset, + ) + + if args.alerts_command == "list": + return cmd_alerts_list(args, services=services) + if args.alerts_command == "set": + return cmd_alerts_set(args, services=services) + if args.alerts_command == "unset": + return cmd_alerts_unset(args, services=services) + alerts_parser.print_help() + return 1 + parser.print_help() return 1 diff --git a/sqlit/domains/connections/cli/commands.py b/sqlit/domains/connections/cli/commands.py index 17e15090..4358d6cb 100644 --- a/sqlit/domains/connections/cli/commands.py +++ b/sqlit/domains/connections/cli/commands.py @@ -19,12 +19,37 @@ get_database_type_labels, ) from sqlit.domains.connections.providers.catalog import get_provider_schema +from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + format_alert_mode, + parse_alert_mode, +) from sqlit.shared.app.runtime import RuntimeConfig from sqlit.shared.app.services import AppServices, build_app_services from .helpers import build_connection_config_from_args +def _apply_alert_option(config: ConnectionConfig, raw: str | None) -> str | None: + """Apply an --alert value (mode token or "unset") to a connection. + + Returns an error message on invalid input, otherwise None. + """ + if raw is None: + return None + token = raw.strip().lower() + if not token: + return None + if token in {"unset", "clear", "default"}: + config.options.pop(CONNECTION_ALERT_OPTION, None) + return None + mode = parse_alert_mode(token) + if mode is None: + return f"Invalid --alert value '{raw}'. Use off|delete|write|unset." + config.set_option(CONNECTION_ALERT_OPTION, format_alert_mode(mode)) + return None + + def _find_connection_index(connections: list[ConnectionConfig], name: str) -> int | None: for idx, conn in enumerate(connections): if conn.name == name: @@ -204,6 +229,11 @@ def cmd_connection_create(args: Any, *, services: AppServices | None = None) -> print(f"Error: {exc}") return 1 + alert_error = _apply_alert_option(config, getattr(args, "alert", None)) + if alert_error: + print(f"Error: {alert_error}") + return 1 + connections.append(config) _ensure_password_storage(services, config) _save_connections(services, connections) @@ -270,6 +300,11 @@ def cmd_connection_edit(args: Any, *, services: AppServices | None = None) -> in conn.endpoint = FileEndpoint(path=file_path) + alert_error = _apply_alert_option(conn, getattr(args, "alert", None)) + if alert_error: + print(f"Error: {alert_error}") + return 1 + _ensure_password_storage(services, conn) _save_connections(services, connections) diff --git a/sqlit/domains/query/app/alerts.py b/sqlit/domains/query/app/alerts.py index 8d47ba59..7093e6fe 100644 --- a/sqlit/domains/query/app/alerts.py +++ b/sqlit/domains/query/app/alerts.py @@ -4,10 +4,17 @@ import re from enum import IntEnum +from typing import Any from sqlit.domains.query.app.multi_statement import split_statements +# Settings / option keys for the alert mode hierarchy. +GLOBAL_ALERT_SETTING = "query_alert_mode" +DATABASE_ALERT_SETTING = "query_alert_modes_by_database" +CONNECTION_ALERT_OPTION = "query_alert_mode" + + class AlertMode(IntEnum): """Alert mode thresholds (user-configured).""" @@ -129,3 +136,43 @@ def _coerce_alert_mode(value: int) -> AlertMode | None: if mode in {AlertMode.OFF, AlertMode.DELETE, AlertMode.WRITE}: return mode return None + + +def make_db_alert_key(connection_name: str, database: str) -> str: + """Compose the key used to store per-database alert overrides.""" + return f"{connection_name}::{database}" + + +def resolve_alert_mode( + *, + global_mode: Any = None, + connection_option: Any = None, + database_override: Any = None, +) -> tuple[AlertMode, str]: + """Resolve the effective alert mode given the three scopes. + + Returns a (mode, source) tuple where source is one of + ``"database"``, ``"connection"``, or ``"global"`` — describing which + scope supplied the resolved value. + """ + mode = parse_alert_mode(database_override) + if mode is not None: + return mode, "database" + mode = parse_alert_mode(connection_option) + if mode is not None: + return mode, "connection" + mode = parse_alert_mode(global_mode) + if mode is not None: + return mode, "global" + return AlertMode.OFF, "global" + + +def lookup_database_override( + overrides: Any, + connection_name: str | None, + database: str | None, +) -> Any: + """Look up a per-database alert override from a stored mapping.""" + if not connection_name or not database or not isinstance(overrides, dict): + return None + return overrides.get(make_db_alert_key(connection_name, database)) diff --git a/sqlit/domains/query/cli/alerts_commands.py b/sqlit/domains/query/cli/alerts_commands.py new file mode 100644 index 00000000..0cddc612 --- /dev/null +++ b/sqlit/domains/query/cli/alerts_commands.py @@ -0,0 +1,150 @@ +"""CLI handlers for the `sqlit alerts` subcommand.""" + +from __future__ import annotations + +from typing import Any + +from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, + GLOBAL_ALERT_SETTING, + AlertMode, + format_alert_mode, + make_db_alert_key, + parse_alert_mode, +) +from sqlit.shared.app.runtime import RuntimeConfig +from sqlit.shared.app.services import AppServices, build_app_services + + +def _services(services: AppServices | None) -> AppServices: + return services or build_app_services(RuntimeConfig.from_env()) + + +def _format(value: Any) -> str: + parsed = parse_alert_mode(value) + return format_alert_mode(parsed) if parsed is not None else f"" + + +def _resolve_global_mode(services: AppServices) -> AlertMode: + raw = services.settings_store.get(GLOBAL_ALERT_SETTING) + parsed = parse_alert_mode(raw) + return parsed if parsed is not None else AlertMode.OFF + + +def _resolve_db_overrides(services: AppServices) -> dict[str, Any]: + raw = services.settings_store.get(DATABASE_ALERT_SETTING) + return dict(raw) if isinstance(raw, dict) else {} + + +def cmd_alerts_list(args: Any, *, services: AppServices | None = None) -> int: + """Show global, per-connection, and per-database alert overrides.""" + services = _services(services) + + global_mode = _resolve_global_mode(services) + print(f"Global: {format_alert_mode(global_mode)}") + + connections = services.connection_store.load_all(load_credentials=False) + conn_overrides = [ + (c.name, c.get_option(CONNECTION_ALERT_OPTION)) + for c in connections + if c.get_option(CONNECTION_ALERT_OPTION) is not None + ] + if conn_overrides: + print("\nPer-connection:") + for name, value in conn_overrides: + print(f" {name:<30} {_format(value)}") + else: + print("\nPer-connection: (none)") + + db_overrides = _resolve_db_overrides(services) + if db_overrides: + print("\nPer-database:") + for key, value in sorted(db_overrides.items()): + print(f" {key:<30} {_format(value)}") + else: + print("\nPer-database: (none)") + return 0 + + +def cmd_alerts_set(args: Any, *, services: AppServices | None = None) -> int: + """Set an alert mode at the requested scope.""" + services = _services(services) + + mode = parse_alert_mode(getattr(args, "mode", None)) + if mode is None: + print(f"Error: invalid mode '{getattr(args, 'mode', '')}'. Use off, delete, or write.") + return 1 + + connection = getattr(args, "connection", None) + database = getattr(args, "database", None) + + if database and not connection: + print("Error: --database requires --connection.") + return 1 + + if not connection: + services.settings_store.set(GLOBAL_ALERT_SETTING, int(mode)) + print(f"Global alerts set to {format_alert_mode(mode)}") + return 0 + + connections = services.connection_store.load_all(load_credentials=False) + target = next((c for c in connections if c.name == connection), None) + if target is None: + print(f"Error: Connection '{connection}' not found.") + return 1 + + if database: + overrides = _resolve_db_overrides(services) + overrides[make_db_alert_key(connection, database)] = format_alert_mode(mode) + services.settings_store.set(DATABASE_ALERT_SETTING, overrides) + print( + f"Alerts for database '{database}' on '{connection}' set to {format_alert_mode(mode)}" + ) + return 0 + + target.set_option(CONNECTION_ALERT_OPTION, format_alert_mode(mode)) + services.connection_store.save_all(connections) + print(f"Connection '{connection}' alerts set to {format_alert_mode(mode)}") + return 0 + + +def cmd_alerts_unset(args: Any, *, services: AppServices | None = None) -> int: + """Clear an alert override at the requested scope.""" + services = _services(services) + + connection = getattr(args, "connection", None) + database = getattr(args, "database", None) + + if database and not connection: + print("Error: --database requires --connection.") + return 1 + + if not connection: + services.settings_store.delete(GLOBAL_ALERT_SETTING) + print("Global alert mode cleared (defaults to off)") + return 0 + + if database: + overrides = _resolve_db_overrides(services) + key = make_db_alert_key(connection, database) + if key not in overrides: + print(f"No alert override for database '{database}' on '{connection}'.") + return 0 + overrides.pop(key) + services.settings_store.set(DATABASE_ALERT_SETTING, overrides) + print(f"Alert override for database '{database}' on '{connection}' cleared") + return 0 + + connections = services.connection_store.load_all(load_credentials=False) + target = next((c for c in connections if c.name == connection), None) + if target is None: + print(f"Error: Connection '{connection}' not found.") + return 1 + if CONNECTION_ALERT_OPTION not in target.options: + print(f"Connection '{connection}' has no alert override.") + return 0 + target.options.pop(CONNECTION_ALERT_OPTION, None) + services.connection_store.save_all(connections) + print(f"Alert override for connection '{connection}' cleared") + return 0 diff --git a/sqlit/domains/query/ui/mixins/query_execution.py b/sqlit/domains/query/ui/mixins/query_execution.py index 293f9ef9..2370053d 100644 --- a/sqlit/domains/query/ui/mixins/query_execution.py +++ b/sqlit/domains/query/ui/mixins/query_execution.py @@ -182,18 +182,44 @@ def _maybe_confirm_query(self: QueryMixinHost, query: str, proceed: Callable[[], from sqlit.domains.query.app.alerts import ( AlertMode, AlertSeverity, + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, classify_query_alert, format_alert_mode, + lookup_database_override, + resolve_alert_mode, should_confirm, ) from sqlit.domains.query.app.multi_statement import get_executable_sql from sqlit.shared.ui.screens.confirm import ConfirmScreen - raw_mode = getattr(self.services.runtime, "query_alert_mode", 0) or 0 + global_mode = getattr(self.services.runtime, "query_alert_mode", 0) or 0 + connection_option = None + config = getattr(self, "current_config", None) + if config is not None: + connection_option = config.get_option(CONNECTION_ALERT_OPTION) + database: str | None = None + get_db = getattr(self, "_get_effective_database", None) + if callable(get_db): + try: + value = get_db() + database = str(value) if value else None + except Exception: + database = None + overrides: dict[str, Any] = {} try: - mode = AlertMode(int(raw_mode)) - except ValueError: - mode = AlertMode.OFF + raw_overrides = self.services.settings_store.get(DATABASE_ALERT_SETTING) + if isinstance(raw_overrides, dict): + overrides = raw_overrides + except Exception: + overrides = {} + connection_name = getattr(config, "name", None) if config is not None else None + database_override = lookup_database_override(overrides, connection_name, database) + mode, _ = resolve_alert_mode( + global_mode=global_mode, + connection_option=connection_option, + database_override=database_override, + ) if mode == AlertMode.OFF: proceed() diff --git a/sqlit/domains/shell/app/commands/alert.py b/sqlit/domains/shell/app/commands/alert.py index 3dc8e962..45269d50 100644 --- a/sqlit/domains/shell/app/commands/alert.py +++ b/sqlit/domains/shell/app/commands/alert.py @@ -4,47 +4,281 @@ from typing import Any -from sqlit.domains.query.app.alerts import AlertMode, format_alert_mode, parse_alert_mode +from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, + GLOBAL_ALERT_SETTING, + AlertMode, + format_alert_mode, + lookup_database_override, + make_db_alert_key, + parse_alert_mode, + resolve_alert_mode, +) from .router import register_command_handler +_USAGE = ( + "Usage: :alert [global|connection|database] off|delete|write|unset" +) + + def _handle_alert_command(app: Any, cmd: str, args: list[str]) -> bool: if cmd not in {"alert", "alerts"}: return False - value = args[0].lower() if args else "" - if not value: + if not args: _show_alert_status(app) return True - mode = parse_alert_mode(value) + scope, mode_token = _parse_scope_and_mode(args) + if scope is None: + app.notify(_USAGE, severity="warning") + return True + + if mode_token is None: + # Show current value for the named scope. + _show_scope_status(app, scope) + return True + + if mode_token in {"unset", "clear", "default"}: + _unset_scope(app, scope) + return True + + mode = parse_alert_mode(mode_token) if mode is None: - app.notify("Usage: :alert off|delete|write", severity="warning") + app.notify(_USAGE, severity="warning") return True - _set_alert_mode(app, mode) + _set_scope(app, scope, mode) return True +def _parse_scope_and_mode(args: list[str]) -> tuple[str | None, str | None]: + first = args[0].lower() + if first in {"global", "connection", "database", "db"}: + scope = "database" if first == "db" else first + mode_token = args[1].lower() if len(args) > 1 else None + return scope, mode_token + # No explicit scope -> implicit global. + return "global", first + + def _show_alert_status(app: Any) -> None: - mode = _get_alert_mode(app) - label = format_alert_mode(mode) - app.notify(f"Query alerts: {label}") + global_mode, connection_value, database_value, effective_mode, source = _collect_state(app) + parts = [f"effective: {format_alert_mode(effective_mode)} (from {source})"] + parts.append(f"global: {format_alert_mode(global_mode)}") + if connection_value is not None: + parts.append(f"connection: {_format_value(connection_value)}") + else: + parts.append("connection: unset") + if database_value is not None: + parts.append(f"database: {_format_value(database_value)}") + else: + parts.append("database: unset") + app.notify(" | ".join(parts)) + + +def _show_scope_status(app: Any, scope: str) -> None: + global_mode, connection_value, database_value, _, _ = _collect_state(app) + if scope == "global": + app.notify(f"Global alert mode: {format_alert_mode(global_mode)}") + return + if scope == "connection": + if connection_value is None: + app.notify("Connection alert override: unset (falls back to global)") + return + app.notify(f"Connection alert override: {_format_value(connection_value)}") + return + if scope == "database": + if database_value is None: + app.notify("Database alert override: unset (falls back to connection/global)") + return + app.notify(f"Database alert override: {_format_value(database_value)}") + + +def _set_scope(app: Any, scope: str, mode: AlertMode) -> None: + if scope == "global": + _set_global(app, mode) + return + if scope == "connection": + _set_connection(app, mode) + return + if scope == "database": + _set_database(app, mode) + +def _unset_scope(app: Any, scope: str) -> None: + if scope == "global": + # "Unset" on global means OFF — there's no level above it. + _set_global(app, AlertMode.OFF) + return + if scope == "connection": + _clear_connection(app) + return + if scope == "database": + _clear_database(app) -def _set_alert_mode(app: Any, mode: AlertMode) -> None: + +# --------------------------------------------------------------------------- +# Global scope + + +def _set_global(app: Any, mode: AlertMode) -> None: app.services.runtime.query_alert_mode = int(mode) try: - app.services.settings_store.set("query_alert_mode", int(mode)) + app.services.settings_store.set(GLOBAL_ALERT_SETTING, int(mode)) + except Exception: + pass + app.notify(f"Global query alerts set to {format_alert_mode(mode)}") + + +# --------------------------------------------------------------------------- +# Connection scope + + +def _set_connection(app: Any, mode: AlertMode) -> None: + config = getattr(app, "current_config", None) + if config is None: + app.notify("No active connection — connect first to set a connection alert", severity="warning") + return + config.set_option(CONNECTION_ALERT_OPTION, format_alert_mode(mode)) + _persist_connection(app, config) + app.notify(f"Connection '{config.name}' alerts set to {format_alert_mode(mode)}") + + +def _clear_connection(app: Any) -> None: + config = getattr(app, "current_config", None) + if config is None: + app.notify("No active connection — connect first to clear a connection alert", severity="warning") + return + if CONNECTION_ALERT_OPTION not in config.options: + app.notify(f"Connection '{config.name}' has no alert override") + return + config.options.pop(CONNECTION_ALERT_OPTION, None) + _persist_connection(app, config) + app.notify(f"Connection '{config.name}' alert override cleared") + + +def _persist_connection(app: Any, config: Any) -> None: + """Save the connection back to the store, if it's a saved one.""" + try: + connections = list(getattr(app, "connections", None) or []) + except Exception: + connections = [] + if not connections: + return + if not any(getattr(c, "name", None) == getattr(config, "name", None) for c in connections): + # Temporary / direct connection — keep in memory only. + return + try: + app.services.connection_store.save_all(connections) except Exception: pass - app.notify(f"Query alerts set to {format_alert_mode(mode)}") -def _get_alert_mode(app: Any) -> AlertMode: - raw = getattr(app.services.runtime, "query_alert_mode", 0) or 0 - return AlertMode(int(raw)) +# --------------------------------------------------------------------------- +# Database scope + + +def _set_database(app: Any, mode: AlertMode) -> None: + config = getattr(app, "current_config", None) + if config is None: + app.notify("No active connection — connect first to set a database alert", severity="warning") + return + database = _current_database(app) + if not database: + app.notify("No active database — open a database first to set a database alert", severity="warning") + return + overrides = _load_db_overrides(app) + overrides[make_db_alert_key(config.name, database)] = format_alert_mode(mode) + _save_db_overrides(app, overrides) + app.notify(f"Database '{database}' (on '{config.name}') alerts set to {format_alert_mode(mode)}") + + +def _clear_database(app: Any) -> None: + config = getattr(app, "current_config", None) + if config is None: + app.notify("No active connection — connect first to clear a database alert", severity="warning") + return + database = _current_database(app) + if not database: + app.notify("No active database — open a database first to clear a database alert", severity="warning") + return + overrides = _load_db_overrides(app) + key = make_db_alert_key(config.name, database) + if key not in overrides: + app.notify(f"Database '{database}' has no alert override") + return + overrides.pop(key, None) + _save_db_overrides(app, overrides) + app.notify(f"Database '{database}' alert override cleared") + + +def _current_database(app: Any) -> str | None: + get_db = getattr(app, "_get_effective_database", None) + if not callable(get_db): + return None + try: + value = get_db() + except Exception: + return None + return str(value) if value else None + + +def _load_db_overrides(app: Any) -> dict[str, Any]: + try: + raw = app.services.settings_store.get(DATABASE_ALERT_SETTING) + except Exception: + return {} + return dict(raw) if isinstance(raw, dict) else {} + + +def _save_db_overrides(app: Any, overrides: dict[str, Any]) -> None: + try: + app.services.settings_store.set(DATABASE_ALERT_SETTING, overrides) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Status collection + + +def _collect_state( + app: Any, +) -> tuple[AlertMode, Any, Any, AlertMode, str]: + raw_global = getattr(app.services.runtime, "query_alert_mode", 0) or 0 + try: + global_mode = AlertMode(int(raw_global)) + except (TypeError, ValueError): + global_mode = AlertMode.OFF + + config = getattr(app, "current_config", None) + connection_value = None + if config is not None: + connection_value = config.get_option(CONNECTION_ALERT_OPTION) + + database_value = None + database = _current_database(app) + overrides = _load_db_overrides(app) + connection_name = getattr(config, "name", None) if config is not None else None + database_value = lookup_database_override(overrides, connection_name, database) + + effective_mode, source = resolve_alert_mode( + global_mode=int(global_mode), + connection_option=connection_value, + database_override=database_value, + ) + return global_mode, connection_value, database_value, effective_mode, source + + +def _format_value(value: Any) -> str: + parsed = parse_alert_mode(value) + if parsed is None: + return str(value) + return format_alert_mode(parsed) register_command_handler(_handle_alert_command) diff --git a/sqlit/shared/core/protocols.py b/sqlit/shared/core/protocols.py index 2505382e..bbf65e96 100644 --- a/sqlit/shared/core/protocols.py +++ b/sqlit/shared/core/protocols.py @@ -121,3 +121,15 @@ def load_all(self) -> dict: def save_all(self, settings: dict) -> None: """Save settings.""" ... + + def get(self, key: str, default: object = None) -> object: + """Get a setting by key.""" + ... + + def set(self, key: str, value: object) -> None: + """Set a setting by key.""" + ... + + def delete(self, key: str) -> bool: + """Delete a setting by key.""" + ... diff --git a/tests/ui/mocks.py b/tests/ui/mocks.py index 53f6cc7f..48b03fca 100644 --- a/tests/ui/mocks.py +++ b/tests/ui/mocks.py @@ -104,6 +104,12 @@ def get(self, key: str, default: Any = None) -> Any: def set(self, key: str, value: Any) -> None: self.settings[key] = value + def delete(self, key: str) -> bool: + if key in self.settings: + del self.settings[key] + return True + return False + def build_test_services( *, diff --git a/tests/unit/test_alert_command_handler.py b/tests/unit/test_alert_command_handler.py new file mode 100644 index 00000000..b3c4043d --- /dev/null +++ b/tests/unit/test_alert_command_handler.py @@ -0,0 +1,258 @@ +"""Tests for the :alert shell command handler.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from sqlit.domains.connections.domain.config import ConnectionConfig +from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, + GLOBAL_ALERT_SETTING, + AlertMode, + make_db_alert_key, +) +from sqlit.domains.shell.app.commands.alert import _handle_alert_command + + +class _FakeRuntime: + def __init__(self, mode: int = 0) -> None: + self.query_alert_mode = mode + + +class _FakeSettingsStore: + def __init__(self) -> None: + self.data: dict[str, Any] = {} + + def load_all(self) -> dict: + return dict(self.data) + + def save_all(self, settings: dict) -> None: + self.data = dict(settings) + + def get(self, key: str, default: Any = None) -> Any: + return self.data.get(key, default) + + def set(self, key: str, value: Any) -> None: + self.data[key] = value + + def delete(self, key: str) -> bool: + return self.data.pop(key, None) is not None + + +class _FakeConnectionStore: + def __init__(self) -> None: + self.saved: list[ConnectionConfig] = [] + + def save_all(self, connections: list[ConnectionConfig]) -> None: + self.saved = list(connections) + + def load_all(self, load_credentials: bool = True) -> list[ConnectionConfig]: + return list(self.saved) + + +@dataclass +class _FakeServices: + runtime: _FakeRuntime = field(default_factory=_FakeRuntime) + settings_store: _FakeSettingsStore = field(default_factory=_FakeSettingsStore) + connection_store: _FakeConnectionStore = field(default_factory=_FakeConnectionStore) + + +class _FakeApp: + """Minimal duck-typed app stand-in used by _handle_alert_command.""" + + def __init__( + self, + *, + current_config: ConnectionConfig | None = None, + database: str | None = None, + connections: list[ConnectionConfig] | None = None, + ) -> None: + self.services = _FakeServices() + self.current_config = current_config + self._database = database + self.connections = list(connections or []) + self.notifications: list[tuple[str, str | None]] = [] + + def notify(self, message: str, *, severity: str | None = None) -> None: + self.notifications.append((message, severity)) + + def _get_effective_database(self) -> str | None: + return self._database + + # Helpers for assertions. + def last_notification(self) -> tuple[str, str | None]: + assert self.notifications, "expected at least one notification" + return self.notifications[-1] + + +# --------------------------------------------------------------------------- +# Command routing + + +def test_handler_rejects_unrelated_command() -> None: + app = _FakeApp() + assert _handle_alert_command(app, "theme", ["dark"]) is False + assert _handle_alert_command(app, "alert", []) is True + assert _handle_alert_command(app, "alerts", []) is True + + +# --------------------------------------------------------------------------- +# Implicit-global form (back-compat: `:alert delete`) + + +def test_implicit_global_set() -> None: + app = _FakeApp() + _handle_alert_command(app, "alert", ["delete"]) + assert app.services.runtime.query_alert_mode == int(AlertMode.DELETE) + assert app.services.settings_store.get(GLOBAL_ALERT_SETTING) == int(AlertMode.DELETE) + + +def test_implicit_global_invalid_warns() -> None: + app = _FakeApp() + _handle_alert_command(app, "alert", ["garbage"]) + msg, severity = app.last_notification() + assert "Usage" in msg + assert severity == "warning" + + +# --------------------------------------------------------------------------- +# Explicit scopes + + +def test_global_scope_set_and_unset() -> None: + app = _FakeApp() + _handle_alert_command(app, "alert", ["global", "write"]) + assert app.services.runtime.query_alert_mode == int(AlertMode.WRITE) + + _handle_alert_command(app, "alert", ["global", "unset"]) + assert app.services.runtime.query_alert_mode == int(AlertMode.OFF) + + +def test_connection_scope_requires_active_connection() -> None: + app = _FakeApp(current_config=None) + _handle_alert_command(app, "alert", ["connection", "delete"]) + msg, severity = app.last_notification() + assert "No active connection" in msg + assert severity == "warning" + + +def test_connection_scope_set_writes_to_options_and_persists() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, connections=[conn]) + _handle_alert_command(app, "alert", ["connection", "delete"]) + assert conn.get_option(CONNECTION_ALERT_OPTION) == "delete" + # Saved connections were persisted because it was a known one. + assert app.services.connection_store.saved[0].get_option(CONNECTION_ALERT_OPTION) == "delete" + + +def test_connection_scope_temp_connection_not_persisted() -> None: + """A connection not in app.connections (e.g. ad-hoc) should not be saved.""" + conn = ConnectionConfig(name="temp") + app = _FakeApp(current_config=conn, connections=[]) # not in list + _handle_alert_command(app, "alert", ["connection", "write"]) + assert conn.get_option(CONNECTION_ALERT_OPTION) == "write" + # No persistence call happened. + assert app.services.connection_store.saved == [] + + +def test_connection_scope_unset_clears_option() -> None: + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "delete") + app = _FakeApp(current_config=conn, connections=[conn]) + _handle_alert_command(app, "alert", ["connection", "unset"]) + assert CONNECTION_ALERT_OPTION not in conn.options + + +def test_database_scope_requires_database() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, database=None, connections=[conn]) + _handle_alert_command(app, "alert", ["database", "delete"]) + msg, severity = app.last_notification() + assert "No active database" in msg + assert severity == "warning" + + +def test_database_scope_set_writes_to_settings() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, database="warehouse", connections=[conn]) + _handle_alert_command(app, "alert", ["database", "write"]) + overrides = app.services.settings_store.get(DATABASE_ALERT_SETTING) + assert overrides == {make_db_alert_key("prod", "warehouse"): "write"} + + +def test_database_scope_unset_removes_key() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, database="warehouse", connections=[conn]) + app.services.settings_store.set( + DATABASE_ALERT_SETTING, + {make_db_alert_key("prod", "warehouse"): "delete"}, + ) + _handle_alert_command(app, "alert", ["database", "unset"]) + overrides = app.services.settings_store.get(DATABASE_ALERT_SETTING) + assert overrides == {} + + +def test_db_scope_alias_maps_to_database() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, database="warehouse", connections=[conn]) + _handle_alert_command(app, "alert", ["db", "delete"]) + overrides = app.services.settings_store.get(DATABASE_ALERT_SETTING) + assert overrides == {make_db_alert_key("prod", "warehouse"): "delete"} + + +# --------------------------------------------------------------------------- +# Status / inspection + + +def test_status_with_no_args_shows_all_scopes() -> None: + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "delete") + app = _FakeApp(current_config=conn, database="warehouse", connections=[conn]) + app.services.runtime.query_alert_mode = int(AlertMode.WRITE) + app.services.settings_store.set( + DATABASE_ALERT_SETTING, + {make_db_alert_key("prod", "warehouse"): "off"}, + ) + + _handle_alert_command(app, "alert", []) + msg, _ = app.last_notification() + # Effective mode is "off" because database scope wins. + assert "effective: off (from database)" in msg + assert "global: write" in msg + assert "connection: delete" in msg + assert "database: off" in msg + + +def test_status_with_scope_only_reports_that_scope() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, connections=[conn]) + _handle_alert_command(app, "alert", ["connection"]) + msg, _ = app.last_notification() + assert "Connection alert override: unset" in msg + + +# --------------------------------------------------------------------------- +# Invalid input + + +def test_invalid_mode_warns() -> None: + conn = ConnectionConfig(name="prod") + app = _FakeApp(current_config=conn, connections=[conn]) + _handle_alert_command(app, "alert", ["connection", "supernuke"]) + msg, severity = app.last_notification() + assert "Usage" in msg + assert severity == "warning" + + +def test_clear_alias_unset_alias_default_alias() -> None: + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "delete") + app = _FakeApp(current_config=conn, connections=[conn]) + _handle_alert_command(app, "alert", ["connection", "clear"]) + assert CONNECTION_ALERT_OPTION not in conn.options + + conn.set_option(CONNECTION_ALERT_OPTION, "delete") + _handle_alert_command(app, "alert", ["connection", "default"]) + assert CONNECTION_ALERT_OPTION not in conn.options diff --git a/tests/unit/test_query_alerts_hierarchy.py b/tests/unit/test_query_alerts_hierarchy.py new file mode 100644 index 00000000..a206e2db --- /dev/null +++ b/tests/unit/test_query_alerts_hierarchy.py @@ -0,0 +1,308 @@ +"""Tests for the alert mode resolution hierarchy and CLI/shell helpers.""" + +from __future__ import annotations + +import argparse + +from sqlit.domains.query.app.alerts import ( + AlertMode, + lookup_database_override, + make_db_alert_key, + resolve_alert_mode, +) + + +def test_resolve_falls_back_to_global() -> None: + mode, source = resolve_alert_mode( + global_mode="delete", + connection_option=None, + database_override=None, + ) + assert mode == AlertMode.DELETE + assert source == "global" + + +def test_resolve_connection_overrides_global() -> None: + mode, source = resolve_alert_mode( + global_mode="off", + connection_option="write", + database_override=None, + ) + assert mode == AlertMode.WRITE + assert source == "connection" + + +def test_resolve_database_overrides_everything() -> None: + mode, source = resolve_alert_mode( + global_mode="write", + connection_option="delete", + database_override="off", + ) + assert mode == AlertMode.OFF + assert source == "database" + + +def test_resolve_invalid_values_fall_through() -> None: + # Invalid db override should fall through to connection. + mode, source = resolve_alert_mode( + global_mode="off", + connection_option="delete", + database_override="not-a-mode", + ) + assert mode == AlertMode.DELETE + assert source == "connection" + + +def test_resolve_default_off_when_nothing_set() -> None: + mode, source = resolve_alert_mode() + assert mode == AlertMode.OFF + assert source == "global" + + +def test_lookup_database_override_key() -> None: + overrides = {make_db_alert_key("MyConn", "warehouse"): "delete"} + assert lookup_database_override(overrides, "MyConn", "warehouse") == "delete" + assert lookup_database_override(overrides, "MyConn", "other") is None + assert lookup_database_override(overrides, None, "warehouse") is None + assert lookup_database_override(overrides, "MyConn", None) is None + assert lookup_database_override(None, "MyConn", "warehouse") is None + + +def test_apply_alert_option_sets_and_clears() -> None: + from sqlit.domains.connections.cli.commands import _apply_alert_option + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.app.alerts import CONNECTION_ALERT_OPTION + + config = ConnectionConfig(name="x") + assert _apply_alert_option(config, "delete") is None + assert config.get_option(CONNECTION_ALERT_OPTION) == "delete" + + assert _apply_alert_option(config, "unset") is None + assert CONNECTION_ALERT_OPTION not in config.options + + err = _apply_alert_option(config, "garbage") + assert err is not None + assert "Invalid --alert" in err + + +def _make_services(settings: dict | None = None): + from sqlit.shared.app.runtime import RuntimeConfig + from tests.ui.mocks import ( + MockConnectionStore, + MockHistoryStore, + MockSettingsStore, + build_test_services, + ) + + return build_test_services( + runtime=RuntimeConfig(), + connection_store=MockConnectionStore(), + settings_store=MockSettingsStore(settings), + history_store=MockHistoryStore(), + ) + + +def test_alerts_cli_set_global_and_list(capsys) -> None: + from sqlit.domains.query.cli.alerts_commands import ( + cmd_alerts_list, + cmd_alerts_set, + ) + + services = _make_services() + rc = cmd_alerts_set( + argparse.Namespace(mode="write", connection=None, database=None), + services=services, + ) + assert rc == 0 + assert services.settings_store.get("query_alert_mode") == int(AlertMode.WRITE) + + cmd_alerts_list(argparse.Namespace(), services=services) + output = capsys.readouterr().out + assert "Global: write" in output + + +def test_alerts_cli_set_database_requires_connection() -> None: + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_set + + services = _make_services() + rc = cmd_alerts_set( + argparse.Namespace(mode="delete", connection=None, database="prod"), + services=services, + ) + assert rc == 1 + + +def test_alerts_cli_set_connection_persists_option() -> None: + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.app.alerts import CONNECTION_ALERT_OPTION + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_set, cmd_alerts_unset + + services = _make_services() + services.connection_store.save_all([ConnectionConfig(name="prod")]) + + rc = cmd_alerts_set( + argparse.Namespace(mode="delete", connection="prod", database=None), + services=services, + ) + assert rc == 0 + saved = services.connection_store.load_all(load_credentials=False) + assert saved[0].get_option(CONNECTION_ALERT_OPTION) == "delete" + + rc = cmd_alerts_unset( + argparse.Namespace(connection="prod", database=None), + services=services, + ) + assert rc == 0 + saved = services.connection_store.load_all(load_credentials=False) + assert CONNECTION_ALERT_OPTION not in saved[0].options + + +def test_alerts_cli_database_override_roundtrip() -> None: + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.app.alerts import ( + DATABASE_ALERT_SETTING, + make_db_alert_key, + ) + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_set, cmd_alerts_unset + + services = _make_services() + services.connection_store.save_all([ConnectionConfig(name="prod")]) + + rc = cmd_alerts_set( + argparse.Namespace(mode="write", connection="prod", database="warehouse"), + services=services, + ) + assert rc == 0 + overrides = services.settings_store.get(DATABASE_ALERT_SETTING) + assert overrides == {make_db_alert_key("prod", "warehouse"): "write"} + + rc = cmd_alerts_unset( + argparse.Namespace(connection="prod", database="warehouse"), + services=services, + ) + assert rc == 0 + overrides = services.settings_store.get(DATABASE_ALERT_SETTING) + assert isinstance(overrides, dict) + assert make_db_alert_key("prod", "warehouse") not in overrides + + +# --------------------------------------------------------------------------- +# CLI edge cases + + +def test_alerts_set_rejects_invalid_mode() -> None: + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_set + + services = _make_services() + rc = cmd_alerts_set( + argparse.Namespace(mode="explode", connection=None, database=None), + services=services, + ) + assert rc == 1 + # Nothing got written. + assert services.settings_store.load_all() == {} + + +def test_alerts_set_db_on_missing_connection_errors() -> None: + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_set + + services = _make_services() + rc = cmd_alerts_set( + argparse.Namespace(mode="delete", connection="ghost", database="x"), + services=services, + ) + assert rc == 1 + + +def test_alerts_unset_database_without_connection_errors() -> None: + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_unset + + services = _make_services() + rc = cmd_alerts_unset( + argparse.Namespace(connection=None, database="warehouse"), + services=services, + ) + assert rc == 1 + + +def test_alerts_unset_global_clears_setting() -> None: + from sqlit.domains.query.app.alerts import GLOBAL_ALERT_SETTING + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_unset + + services = _make_services({GLOBAL_ALERT_SETTING: int(AlertMode.WRITE)}) + rc = cmd_alerts_unset( + argparse.Namespace(connection=None, database=None), + services=services, + ) + assert rc == 0 + assert GLOBAL_ALERT_SETTING not in services.settings_store.load_all() + + +def test_alerts_unset_missing_db_override_succeeds_quietly() -> None: + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_unset + + services = _make_services() + services.connection_store.save_all([ConnectionConfig(name="prod")]) + rc = cmd_alerts_unset( + argparse.Namespace(connection="prod", database="absent"), + services=services, + ) + # Idempotent — unsetting an absent override is fine. + assert rc == 0 + + +def test_alerts_unset_missing_connection_override_succeeds_quietly() -> None: + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_unset + + services = _make_services() + services.connection_store.save_all([ConnectionConfig(name="prod")]) + rc = cmd_alerts_unset( + argparse.Namespace(connection="prod", database=None), + services=services, + ) + assert rc == 0 + + +def test_alerts_list_shows_mixed_scopes(capsys) -> None: + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, + GLOBAL_ALERT_SETTING, + make_db_alert_key, + ) + from sqlit.domains.query.cli.alerts_commands import cmd_alerts_list + + services = _make_services( + { + GLOBAL_ALERT_SETTING: int(AlertMode.DELETE), + DATABASE_ALERT_SETTING: {make_db_alert_key("prod", "warehouse"): "off"}, + } + ) + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "write") + services.connection_store.save_all([conn]) + + cmd_alerts_list(argparse.Namespace(), services=services) + out = capsys.readouterr().out + assert "Global: delete" in out + assert "Per-connection:" in out + assert "prod" in out + assert "write" in out + assert "Per-database:" in out + assert "prod::warehouse" in out + assert "off" in out + + +def test_apply_alert_option_treats_blank_as_noop() -> None: + from sqlit.domains.connections.cli.commands import _apply_alert_option + from sqlit.domains.connections.domain.config import ConnectionConfig + from sqlit.domains.query.app.alerts import CONNECTION_ALERT_OPTION + + config = ConnectionConfig(name="x") + config.set_option(CONNECTION_ALERT_OPTION, "delete") + # Whitespace should not clear the option, only "unset" does. + assert _apply_alert_option(config, " ") is None + assert config.get_option(CONNECTION_ALERT_OPTION) == "delete" diff --git a/tests/unit/test_query_confirm_integration.py b/tests/unit/test_query_confirm_integration.py new file mode 100644 index 00000000..68cc9c77 --- /dev/null +++ b/tests/unit/test_query_confirm_integration.py @@ -0,0 +1,195 @@ +"""Integration test: drive _maybe_confirm_query through each scope. + +This exercises the actual code path in +``sqlit/domains/query/ui/mixins/query_execution.py`` rather than the +isolated resolver helper. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from sqlit.domains.connections.domain.config import ConnectionConfig +from sqlit.domains.query.app.alerts import ( + CONNECTION_ALERT_OPTION, + DATABASE_ALERT_SETTING, + AlertMode, + make_db_alert_key, +) +from sqlit.domains.query.ui.mixins.query_execution import QueryExecutionMixin + + +class _Runtime: + def __init__(self, mode: int = 0) -> None: + self.query_alert_mode = mode + + +class _Settings: + def __init__(self, data: dict[str, Any] | None = None) -> None: + self.data: dict[str, Any] = dict(data or {}) + + def load_all(self) -> dict: + return dict(self.data) + + def save_all(self, settings: dict) -> None: + self.data = dict(settings) + + def get(self, key: str, default: Any = None) -> Any: + return self.data.get(key, default) + + def set(self, key: str, value: Any) -> None: + self.data[key] = value + + def delete(self, key: str) -> bool: + return self.data.pop(key, None) is not None + + +@dataclass +class _Services: + runtime: _Runtime = field(default_factory=_Runtime) + settings_store: _Settings = field(default_factory=_Settings) + + +class _Host: + """Bare host that satisfies the attribute surface _maybe_confirm_query touches.""" + + def __init__( + self, + *, + runtime_mode: int = 0, + connection: ConnectionConfig | None = None, + database: str | None = None, + db_overrides: dict[str, Any] | None = None, + ) -> None: + self.services = _Services( + runtime=_Runtime(runtime_mode), + settings_store=_Settings( + {DATABASE_ALERT_SETTING: db_overrides} if db_overrides else None + ), + ) + self.current_config = connection + self._database = database + self.pushed_screens: list[Any] = [] + self.notifications: list[tuple[str, str | None]] = [] + + def _get_effective_database(self) -> str | None: + return self._database + + def push_screen(self, screen: Any, callback: Any) -> None: + self.pushed_screens.append((screen, callback)) + + def notify(self, msg: str, *, severity: str | None = None, **kwargs: Any) -> None: + self.notifications.append((msg, severity)) + + +def _confirm(host: _Host, sql: str) -> tuple[bool, _Host]: + """Invoke the real _maybe_confirm_query; return (proceed_called, host).""" + called = {"v": False} + + def _proceed() -> None: + called["v"] = True + + QueryExecutionMixin._maybe_confirm_query(host, sql, _proceed) # type: ignore[arg-type] + return called["v"], host + + +# --------------------------------------------------------------------------- +# Hierarchy: each scope shadows the looser one + + +def test_global_off_skips_confirmation() -> None: + host = _Host(runtime_mode=int(AlertMode.OFF)) + proceeded, host = _confirm(host, "DELETE FROM users") + assert proceeded is True + assert host.pushed_screens == [] + + +def test_global_delete_triggers_confirm_for_delete() -> None: + host = _Host(runtime_mode=int(AlertMode.DELETE)) + proceeded, host = _confirm(host, "DELETE FROM users") + assert proceeded is False # waiting for confirm + assert len(host.pushed_screens) == 1 + + +def test_global_delete_does_not_prompt_on_pure_select() -> None: + host = _Host(runtime_mode=int(AlertMode.DELETE)) + proceeded, host = _confirm(host, "SELECT 1") + assert proceeded is True + assert host.pushed_screens == [] + + +def test_connection_override_relaxes_stricter_global() -> None: + """Global = delete (confirm on delete) but connection sets off → skip.""" + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "off") + host = _Host(runtime_mode=int(AlertMode.DELETE), connection=conn) + proceeded, host = _confirm(host, "DELETE FROM users") + assert proceeded is True, "connection-level off should override global delete" + assert host.pushed_screens == [] + + +def test_connection_override_tightens_looser_global() -> None: + """Global = off but a specific connection wants write-level alerts.""" + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "write") + host = _Host(runtime_mode=int(AlertMode.OFF), connection=conn) + proceeded, host = _confirm(host, "UPDATE users SET active = 0") + assert proceeded is False + assert len(host.pushed_screens) == 1 + + +def test_database_override_beats_connection_and_global() -> None: + """Strictest scope (db) takes precedence.""" + conn = ConnectionConfig(name="prod") + conn.set_option(CONNECTION_ALERT_OPTION, "off") + host = _Host( + runtime_mode=int(AlertMode.OFF), + connection=conn, + database="warehouse", + db_overrides={make_db_alert_key("prod", "warehouse"): "delete"}, + ) + proceeded, host = _confirm(host, "DELETE FROM events") + assert proceeded is False + assert len(host.pushed_screens) == 1 + + +def test_database_override_can_silence_strict_global() -> None: + """Db scope can also relax a strict global setting.""" + conn = ConnectionConfig(name="prod") + host = _Host( + runtime_mode=int(AlertMode.WRITE), + connection=conn, + database="sandbox", + db_overrides={make_db_alert_key("prod", "sandbox"): "off"}, + ) + proceeded, host = _confirm(host, "INSERT INTO t VALUES (1)") + assert proceeded is True + assert host.pushed_screens == [] + + +def test_db_override_only_matches_its_own_database() -> None: + """An override keyed at db A must not apply when active db is B.""" + conn = ConnectionConfig(name="prod") + host = _Host( + runtime_mode=int(AlertMode.OFF), + connection=conn, + database="reports", # active db is reports + db_overrides={make_db_alert_key("prod", "warehouse"): "delete"}, + ) + proceeded, host = _confirm(host, "DELETE FROM stuff") + # No matching db override and no connection override and global is off. + assert proceeded is True + + +def test_db_override_only_matches_its_own_connection() -> None: + """An override keyed at connection A must not apply when active conn is B.""" + conn = ConnectionConfig(name="other") + host = _Host( + runtime_mode=int(AlertMode.OFF), + connection=conn, + database="warehouse", + db_overrides={make_db_alert_key("prod", "warehouse"): "delete"}, + ) + proceeded, host = _confirm(host, "DELETE FROM stuff") + assert proceeded is True