diff --git a/README.md b/README.md index ebd8784..b2dece1 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,6 @@ A lightweight, context‑aware shell for developers --- -TruShell is not a full replacement for bash or zsh. It is a small -utility shell that sits next to your normal terminal and helps you -track tasks, check times, set alarms, and run ordinary commands. - It is written in Python and uses a SQLite database for todos. When you type a command TruShell does not recognise, it passes it directly to the host system’s shell (bash, cmd, etc.). diff --git a/pyproject.toml b/pyproject.toml index 5ed31ed..baf3a30 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ dependencies = [ "pyjokes>=0.6.0", "cowsay>=4.0", "pytz>=2024.1", + "tzdata>=2024.1", "platformdirs>=3.0.0", "textual>=0.86.0", "psutil>=5.9.0", diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..7328514 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,11 @@ +"""Pytest configuration and shared fixtures for TruShell tests.""" + +import pytest +from trushell.core.plugin_manager import PluginManager + + +@pytest.fixture(autouse=True) +def reset_plugin_manager(): + """Reset PluginManager singleton after each test to ensure isolation.""" + yield + PluginManager.reset() diff --git a/trushell/__init__.py b/trushell/__init__.py index 231865a..cf47ca5 100644 --- a/trushell/__init__.py +++ b/trushell/__init__.py @@ -2,4 +2,4 @@ __all__ = ["__version__"] -__version__ = "0.1.0" +__version__ = "0.1.2" diff --git a/trushell/cli.py b/trushell/cli.py index 80604c2..ae6f9bc 100644 --- a/trushell/cli.py +++ b/trushell/cli.py @@ -8,9 +8,11 @@ import time import typer from pathlib import Path +from rich.console import Console from textual.app import App, ComposeResult from textual.binding import Binding from textual.widgets import Footer, Header, TextArea +from prompt_toolkit import prompt as toolkit_prompt try: import psutil @@ -21,15 +23,17 @@ from .core.trukernel import EXIT_SENTINEL, get_kernel app = typer.Typer(name="trushell", help="TruShell manifest-driven launcher.") +console = Console() def app_with_lower() -> None: """Entry point that normalizes the first argument to lowercase for case-insensitive invocation.""" - argv = sys.argv.copy() - if len(argv) > 1: - argv[1] = argv[1].lower() - if argv[1] not in {"--help", "-h", "version"}: - raw = " ".join(argv[1:]) + if len(sys.argv) > 1: + # Create a local copy to avoid mutating the global sys.argv + argv_copy = sys.argv.copy() + argv_copy[1] = argv_copy[1].lower() + if argv_copy[1] not in {"--help", "-h", "version"}: + raw = " ".join(argv_copy[1:]) get_kernel().execute_command(raw) return @@ -48,20 +52,17 @@ def _split_command(user_input: str) -> tuple[str, str]: def _prompt_command() -> tuple[str, str, str]: + prompt_text = f"trushell {os.getcwd()} ❯ " try: - try: - from prompt_toolkit import prompt as prompt_toolkit_prompt - except ImportError: - prompt_toolkit_prompt = None - - if prompt_toolkit_prompt is not None: - raw_command = prompt_toolkit_prompt(f"trushell {os.getcwd()} ❯ ") - else: - raw_command = input("trushell> ") - except UnicodeEncodeError: - raw_command = input("trushell> ") - - raw_command = raw_command.strip() + # Try to use the emoji prompt with UTF-8 encoding support + raw_command = toolkit_prompt(prompt_text, enable_history_search=True).strip() + except (UnicodeEncodeError, UnicodeDecodeError): + # Fallback to plain ASCII prompt if emojis fail (e.g., on Windows with limited encoding) + raw_command = input("trushell> ").strip() + except Exception: + # Further fallback to standard input if prompt_toolkit unavailable or fails + raw_command = input("trushell> ").strip() + command, argument = _split_command(raw_command) return raw_command, command, argument @@ -123,18 +124,18 @@ def _run_external_command( ) -> subprocess.CompletedProcess[str]: process = subprocess.Popen(command, shell=shell, cwd=cwd) monitor = None - if psutil is not None: - try: - monitor = psutil.Process(process.pid) - monitor.cpu_percent(None) - except Exception: - monitor = None - peak_rss = 0 peak_cpu = 0.0 start = time.perf_counter() - + try: + if psutil is not None: + try: + monitor = psutil.Process(process.pid) + monitor.cpu_percent(None) + except Exception: + monitor = None + while True: try: process.wait(timeout=0.05) @@ -146,18 +147,17 @@ def _run_external_command( peak_cpu = max(peak_cpu, monitor.cpu_percent(None)) except (Exception, OSError): break + + if monitor is not None: + try: + peak_rss = max(peak_rss, monitor.memory_info().rss) + peak_cpu = max(peak_cpu, monitor.cpu_percent(None)) + except (Exception, OSError): + pass finally: - try: + # Ensure the process is waited on, even if exceptions occur + if process.returncode is None: process.wait() - except Exception: - pass - - if monitor is not None: - try: - peak_rss = max(peak_rss, monitor.memory_info().rss) - peak_cpu = max(peak_cpu, monitor.cpu_percent(None)) - except (Exception, OSError): - pass elapsed = time.perf_counter() - start if peak_rss or peak_cpu: @@ -290,7 +290,7 @@ def _handle_cd_command(raw_command: str) -> bool: try: os.chdir(target) - typer.echo(os.getcwd()) + console.print(f"[green]→ {os.getcwd()}[/green]") except (FileNotFoundError, NotADirectoryError, PermissionError) as error: typer.secho(f"❌ Cannot navigate: {error}", fg=typer.colors.RED) except OSError as error: diff --git a/trushell/commands/core.py b/trushell/commands/core.py index 8fb8d49..5556632 100644 --- a/trushell/commands/core.py +++ b/trushell/commands/core.py @@ -7,29 +7,35 @@ def run_help(args: str) -> None: + """Display available commands by reading the manifest registry. + + If args contains a command name, print its docstring if available. + Otherwise, list all available commands. + """ + # Import here to avoid circular dependency if trukernel imports core """Display available commands or the docstring for a specific command.""" from trushell.core.trukernel import get_kernel kernel = get_kernel() - command = args.strip().lower() if args else "" - - if command: - entry = kernel.registry.get(command) - if entry is not None: - try: - module_name = entry["path"].replace("/", ".").removesuffix(".py") - module = importlib.import_module(module_name) - func = getattr(module, entry["function"]) - doc = inspect.getdoc(func) - if doc: - print(doc) - return - except Exception: - pass - - print(f"No help available for '{command}'.") + + command_name = args.strip().lower() if args else None + + # If a specific command is requested, show its help + if command_name and command_name in kernel.registry: + entry = kernel.registry[command_name] + try: + module = kernel._import_module(entry["path"]) + func = getattr(module, entry["function"], None) + if func and func.__doc__: + print(f"Help for '{command_name}':") + print(func.__doc__) + return + except Exception: + pass + print(f"No detailed help available for '{command_name}'.") return - + + # Otherwise, list all commands cmds = sorted(kernel.registry.keys()) print("Available commands:") col_width = max(len(c) for c in cmds) + 2 diff --git a/trushell/commands/settings.py b/trushell/commands/settings.py index 6a9c37a..42214df 100644 --- a/trushell/commands/settings.py +++ b/trushell/commands/settings.py @@ -76,6 +76,8 @@ def __init__(self, **kwargs) -> None: self.settings = self.manager.load() self.dirty_settings = dict(self.settings) self.selected_category = "General" + # Track changes to settings without losing data when switching categories + self.dirty_settings: dict = {} def compose(self) -> ComposeResult: yield Header(show_clock=False) @@ -112,6 +114,10 @@ def on_button_pressed(self, event: Button.Pressed) -> None: elif event.button.id == "exit_button": self.exit() + def _track_change(self, key: str, value) -> None: + """Track changes to settings without losing data when switching categories.""" + self.dirty_settings[key] = value + def action_save_settings(self) -> None: self._save_settings() @@ -169,6 +175,8 @@ def add_field(label_text: str, widget: Static | Input | Select | Switch) -> None id="theme_selector", ), ) + select_widget.watch_value(lambda value: self._track_change("theme", value)) + add_field("Theme:", select_widget) elif current_cat == "Navigation": add_field( "Show git status:", @@ -212,6 +220,8 @@ def _save_settings(self) -> None: self.manager.set(key, value) self.manager.save() + # Clear dirty settings after save + self.dirty_settings.clear() self._apply_theme(theme) self.notify("Settings saved.", severity="success") diff --git a/trushell/config/plugins.md b/trushell/config/plugins.md index 393315a..d62e8ba 100644 --- a/trushell/config/plugins.md +++ b/trushell/config/plugins.md @@ -2,4 +2,3 @@ # Plugins can optionally run immediately with {lifecycle: on_load}. {cmd: gstatus}; "plugin_init()"; [trushell/plugins/git_enhancer/main.py]; {lifecycle: on_load}; -{cmd: theme}; "load_theme()"; [~/.trushell/plugins/theme_engine/loader.py]; {version:1.2}; diff --git a/trushell/core/database.py b/trushell/core/database.py index 4e819ad..4fe9ed6 100644 --- a/trushell/core/database.py +++ b/trushell/core/database.py @@ -2,26 +2,46 @@ import sqlite3 from pathlib import Path -from typing import List +from typing import List, Optional from platformdirs import user_data_dir from trushell.core.models import Todo -APP_NAME = "TruShell" -APP_AUTHOR = "AkshajSinghal" -DATA_DIR = Path(user_data_dir(APP_NAME, APP_AUTHOR)) -DATA_DIR.mkdir(parents=True, exist_ok=True) -DB_PATH = DATA_DIR / "todos.db" - - -def get_db_connection() -> sqlite3.Connection: - return sqlite3.connect(DB_PATH, check_same_thread=False) - - -def _create_table() -> None: - with get_db_connection() as conn: - conn.execute( +# Global state to track initialization +_INITIALIZED = False +_DB_PATH: Optional[Path] = None + + +def _get_db_path() -> Path: + """Get the path to the database file.""" + global _DB_PATH + if _DB_PATH is None: + data_dir = Path(user_data_dir("TruShell", "AkshajSinghal")) + data_dir.mkdir(parents=True, exist_ok=True) + _DB_PATH = data_dir / "todos.db" + return _DB_PATH + + +def _ensure_initialized() -> None: + """ + Create the database and tables if they don't exist. + This function is idempotent and safe to call multiple times. + Does NOT call get_db_connection() to avoid infinite recursion. + """ + global _INITIALIZED + + if _INITIALIZED: + return + + db_path = _get_db_path() + + # Open a direct connection to initialize. + # We do NOT use get_db_connection() here to avoid recursion. + conn = sqlite3.connect(str(db_path), check_same_thread=False) + try: + cursor = conn.cursor() + cursor.execute( """CREATE TABLE IF NOT EXISTS todos ( task TEXT, category TEXT, @@ -31,9 +51,20 @@ def _create_table() -> None: position INTEGER )""" ) + conn.commit() + _INITIALIZED = True + finally: + conn.close() -_create_table() +def get_db_connection() -> sqlite3.Connection: + """ + Return a connection to the SQLite database. + Ensures the database is initialized before returning the connection. + """ + _ensure_initialized() + db_path = _get_db_path() + return sqlite3.connect(str(db_path), check_same_thread=False) def insert_todo(todo: Todo) -> None: diff --git a/trushell/core/plugin_manager.py b/trushell/core/plugin_manager.py index 84860e9..714bc86 100644 --- a/trushell/core/plugin_manager.py +++ b/trushell/core/plugin_manager.py @@ -43,6 +43,12 @@ def instance(cls, kernel=None) -> "PluginManager": _instance = PluginManager(kernel) return _instance + @classmethod + def reset(cls) -> None: + """Reset the singleton instance. Used for testing.""" + global _instance + _instance = None + def discover(self) -> List[Path]: candidates: List[Path] = [] # user plugin directory (~/.trushell/plugins) diff --git a/trushell/core/trukernel.py b/trushell/core/trukernel.py index 9a1929b..5ff90b5 100644 --- a/trushell/core/trukernel.py +++ b/trushell/core/trukernel.py @@ -2,8 +2,10 @@ import importlib.util import os +import shlex import subprocess import sys +import threading from pathlib import Path from typing import Any @@ -16,6 +18,7 @@ _kernel_instance: TruKernel | None = None +_kernel_lock = threading.Lock() class TruKernel: """Central manifest-driven dispatch engine for TruShell.""" @@ -242,16 +245,23 @@ def _os_passthrough(self, command: str, args: str) -> bool: """Try to run an unregistered command via the host OS shell. Special-cases 'cd' because it must change the Python process CWD. + Blocks shell metacharacters for security. """ full_cmd = f"{command} {args}".strip() if command == "cd": return self._handle_cd(args) + # Block dangerous shell metacharacters + if "|" in full_cmd or ">" in full_cmd or "<" in full_cmd: + print("Error: pipes and redirects are not allowed") + self.logger.warning("Blocked shell metacharacter in command: %s", full_cmd) + return False + try: result = subprocess.run( - full_cmd, - shell=True, + shlex.split(full_cmd), + shell=False, capture_output=True, text=True, ) @@ -306,6 +316,7 @@ def get_kernel() -> TruKernel: Used by commands like help that need a live registry. """ global _kernel_instance - if _kernel_instance is None: - _kernel_instance = TruKernel() + with _kernel_lock: + if _kernel_instance is None: + _kernel_instance = TruKernel() return _kernel_instance \ No newline at end of file