diff --git a/README.md b/README.md index fe4c0f3..4f0ab58 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,181 @@ # tool-tasks + Zero-dependency Python task runner using pyproject.toml [tool.tasks] + +## Features + +- **Zero dependencies**: Uses only Python standard library (requires Python 3.11+ for `tomllib`) +- **Multiple task types**: + - Shell commands + - Task aliases + - Task chains + - Python module entry points +- **Simple configuration**: Define tasks in your `pyproject.toml` file +- **Automatic discovery**: Searches for `pyproject.toml` in current directory and parent directories + +## Installation + +```bash +pip install tool-tasks +``` + +## Usage + +### Basic Usage + +```bash +# List available tasks +task --list + +# Run a task +task + +# Run a task with arguments +task arg1 arg2 +``` + +### Configuration + +Define tasks in your `pyproject.toml` file under the `[tool.tasks]` section: + +```toml +[tool.tasks] +# Simple shell command +hello = "echo 'Hello, World!'" + +# Task alias (references another task) +greet = "hello" + +# Task chain (runs multiple tasks in sequence) +ci = ["test", "lint", "build"] + +# Shell command task +test = "pytest tests/" + +# Dict-style task with cmd +[tool.tasks.build] +cmd = "python -m build" + +# Dict-style task with chain +[tool.tasks.full-test] +chain = ["test", "lint"] + +# Python entry point (calls a function from a module) +[tool.tasks.version] +call = "mypackage:print_version" +``` + +## Task Types + +### 1. Shell Commands + +Execute shell commands directly: + +```toml +[tool.tasks] +test = "pytest tests/" +build = "python -m build" +``` + +### 2. Task Aliases + +Reference other tasks by name: + +```toml +[tool.tasks] +test = "pytest tests/" +t = "test" # Alias for test task +``` + +### 3. Task Chains + +Run multiple tasks in sequence. Stops on first failure: + +```toml +[tool.tasks] +test = "pytest tests/" +lint = "ruff check ." +ci = ["test", "lint"] +``` + +### 4. Python Entry Points + +Call Python functions directly: + +```toml +[tool.tasks.version] +call = "mypackage:print_version" + +# Can also call class methods +[tool.tasks.run-server] +call = "mypackage.server:Server.start" +``` + +The function/method should: +- Return an integer exit code (0 for success, non-zero for failure) +- Return `None` (treated as success) +- Access command-line arguments via `sys.argv` + +## Examples + +### Example pyproject.toml + +```toml +[tool.tasks] +# Development tasks +dev = "python -m myapp --dev" +test = "pytest tests/ -v" +lint = "ruff check ." +format = "ruff format ." + +# Build tasks +clean = "rm -rf dist/ build/ *.egg-info" +build = "python -m build" + +# Task chains +check = ["lint", "test"] +ci = ["check", "build"] + +# Python entry points +[tool.tasks.version] +call = "myapp:print_version" +``` + +### Running Tasks + +```bash +# Run tests +task test + +# Run CI pipeline (lint + test + build) +task ci + +# Run development server +task dev + +# Check version +task version +``` + +## Development + +### Requirements + +- Python 3.11 or higher (for `tomllib` from standard library) + +### Running Tests + +```bash +# Install development dependencies +pip install -e ".[dev]" + +# Run tests +pytest tests/ + +# Run tests with coverage +pytest tests/ --cov=src/tool_tasks --cov-report=term-missing +``` + +## License + +See LICENSE file for details. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7142992 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,47 @@ +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api" + +[tool.poetry] +name = "tool-tasks" +version = "0.1.0" +description = "Zero-dependency Python task runner using pyproject.toml [tool.tasks]" +authors = ["python-build-tools"] +readme = "README.md" +packages = [{include = "tool_tasks", from = "src"}] + +[tool.poetry.dependencies] +python = "^3.11" + +[tool.poetry.group.dev.dependencies] +pytest = "^8.0.0" +pytest-cov = "^5.0.0" + +[tool.poetry.scripts] +task = "tool_tasks.cli:main" + +[tool.tasks] +# Shell command example +hello = "echo 'Hello from task runner!'" + +# Task alias example +greet = "hello" + +# Task chain example +test-all = ["test", "lint"] + +# Shell command with alias +test = "python3 -m pytest tests/ -v" +lint = "echo 'Linting...'" + +# Dict task with cmd +[tool.tasks.build] +cmd = "echo 'Building project...'" + +# Dict task with chain +[tool.tasks.ci] +chain = ["test", "build"] + +# Python entry point example +[tool.tasks.version] +call = "tool_tasks:print_version" diff --git a/src/tool_tasks/__init__.py b/src/tool_tasks/__init__.py new file mode 100644 index 0000000..9f02a2f --- /dev/null +++ b/src/tool_tasks/__init__.py @@ -0,0 +1,9 @@ +"""Tool Tasks - Zero-dependency Python task runner.""" + +__version__ = "0.1.0" + + +def print_version(): + """Print the version.""" + print(f"tool-tasks version {__version__}") + return 0 diff --git a/src/tool_tasks/cli.py b/src/tool_tasks/cli.py new file mode 100644 index 0000000..f5efc2c --- /dev/null +++ b/src/tool_tasks/cli.py @@ -0,0 +1,53 @@ +"""CLI entry point for the task command.""" + +import sys +from pathlib import Path + +from tool_tasks.config import TaskConfig +from tool_tasks.executor import TaskExecutor + + +def main(): + """Main entry point for the task CLI.""" + # Parse command-line arguments + if len(sys.argv) < 2: + print("Usage: task [args...]", file=sys.stderr) + print(" task --list", file=sys.stderr) + sys.exit(1) + + # Handle --list option + if sys.argv[1] == "--list": + try: + config = TaskConfig() + tasks = config.list_tasks() + if tasks: + print("Available tasks:") + for task in tasks: + print(f" {task}") + else: + print("No tasks defined in [tool.tasks]") + except FileNotFoundError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + return + + # Get task name and arguments + task_name = sys.argv[1] + task_args = sys.argv[2:] + + # Load configuration and execute task + try: + config = TaskConfig() + executor = TaskExecutor(config) + exit_code = executor.execute(task_name, task_args) + sys.exit(exit_code) + except FileNotFoundError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/tool_tasks/config.py b/src/tool_tasks/config.py new file mode 100644 index 0000000..1977ec4 --- /dev/null +++ b/src/tool_tasks/config.py @@ -0,0 +1,75 @@ +"""Task configuration loader and parser.""" + +import sys +import tomllib +from pathlib import Path +from typing import Any + + +class TaskConfig: + """Load and parse task configuration from pyproject.toml.""" + + def __init__(self, config_path: Path | None = None): + """ + Initialize TaskConfig. + + Args: + config_path: Path to pyproject.toml. If None, searches upward from current directory. + """ + self.config_path = config_path or self._find_pyproject() + self.tasks = self._load_tasks() + + def _find_pyproject(self) -> Path: + """Find pyproject.toml by searching upward from current directory.""" + current = Path.cwd() + while current != current.parent: + pyproject = current / "pyproject.toml" + if pyproject.exists(): + return pyproject + current = current.parent + + # Check root directory + pyproject = current / "pyproject.toml" + if pyproject.exists(): + return pyproject + + raise FileNotFoundError("pyproject.toml not found in current directory or any parent directory") + + def _load_tasks(self) -> dict[str, Any]: + """Load tasks from pyproject.toml [tool.tasks] section.""" + try: + with open(self.config_path, "rb") as f: + data = tomllib.load(f) + except Exception as e: + print(f"Error reading {self.config_path}: {e}", file=sys.stderr) + sys.exit(1) + + tasks = data.get("tool", {}).get("tasks", {}) + if not tasks: + print(f"No [tool.tasks] section found in {self.config_path}", file=sys.stderr) + sys.exit(1) + + return tasks + + def get_task(self, task_name: str) -> Any: + """ + Get task configuration by name. + + Args: + task_name: Name of the task to retrieve. + + Returns: + Task configuration value. + + Raises: + KeyError: If task is not found. + """ + if task_name not in self.tasks: + available = ", ".join(sorted(self.tasks.keys())) + print(f"Task '{task_name}' not found. Available tasks: {available}", file=sys.stderr) + sys.exit(1) + return self.tasks[task_name] + + def list_tasks(self) -> list[str]: + """Return list of available task names.""" + return sorted(self.tasks.keys()) diff --git a/src/tool_tasks/executor.py b/src/tool_tasks/executor.py new file mode 100644 index 0000000..3bcabbc --- /dev/null +++ b/src/tool_tasks/executor.py @@ -0,0 +1,209 @@ +"""Task executor for running different types of tasks.""" + +import importlib +import shlex +import subprocess +import sys +from typing import Any + + +class TaskExecutor: + """Execute different types of tasks.""" + + def __init__(self, config): + """ + Initialize TaskExecutor. + + Args: + config: TaskConfig instance. + """ + self.config = config + self._executing = set() # Track tasks being executed to detect circular dependencies + + def execute(self, task_name: str, args: list[str] | None = None) -> int: + """ + Execute a task by name. + + Args: + task_name: Name of the task to execute. + args: Additional arguments to pass to the task. + + Returns: + Exit code from the task execution. + """ + if task_name in self._executing: + print(f"Circular dependency detected: task '{task_name}' is already being executed", file=sys.stderr) + return 1 + + self._executing.add(task_name) + try: + task_config = self.config.get_task(task_name) + return self._execute_task(task_config, args or [], current_task=task_name) + finally: + self._executing.discard(task_name) + + def _execute_task(self, task_config: Any, args: list[str], current_task: str = "") -> int: + """ + Execute a task based on its configuration. + + Args: + task_config: Task configuration value. + args: Additional arguments to pass to the task. + current_task: The name of the current task being executed. + + Returns: + Exit code from the task execution. + """ + # Handle different task types + if isinstance(task_config, str): + return self._execute_string_task(task_config, args, current_task) + elif isinstance(task_config, list): + return self._execute_chain(task_config, args, current_task) + elif isinstance(task_config, dict): + return self._execute_dict_task(task_config, args, current_task) + else: + print(f"Invalid task configuration type: {type(task_config)}", file=sys.stderr) + return 1 + + def _execute_string_task(self, task_config: str, args: list[str], current_task: str = "") -> int: + """ + Execute a string task (shell command or task alias). + + Args: + task_config: String task configuration. + args: Additional arguments to pass to the task. + current_task: The name of the current task being executed. + + Returns: + Exit code from the task execution. + """ + # Check if it's a task alias (references another task) + # But not if it references itself (to avoid infinite loops) + if task_config in self.config.tasks and task_config != current_task: + return self.execute(task_config, args) + + # Otherwise, treat as shell command + return self._execute_shell(task_config, args) + + def _execute_chain(self, task_config: list[str], args: list[str], current_task: str = "") -> int: + """ + Execute a chain of tasks. + + Args: + task_config: List of task names to execute in sequence. + args: Additional arguments (passed to each task in chain). + current_task: The name of the current task being executed. + + Returns: + Exit code from the first failing task, or 0 if all succeed. + """ + for task in task_config: + exit_code = self._execute_string_task(task, args, current_task) + if exit_code != 0: + return exit_code + return 0 + + def _execute_dict_task(self, task_config: dict, args: list[str], current_task: str = "") -> int: + """ + Execute a dictionary-configured task. + + Args: + task_config: Dictionary task configuration. + args: Additional arguments to pass to the task. + current_task: The name of the current task being executed. + + Returns: + Exit code from the task execution. + """ + # Check for 'cmd' key (shell command) + if "cmd" in task_config: + return self._execute_shell(task_config["cmd"], args) + + # Check for 'call' key (Python module entry point) + if "call" in task_config: + return self._execute_python_call(task_config["call"], args) + + # Check for 'chain' key (task chain) + if "chain" in task_config: + return self._execute_chain(task_config["chain"], args, current_task) + + print(f"Invalid task configuration: {task_config}", file=sys.stderr) + return 1 + + def _execute_shell(self, command: str, args: list[str]) -> int: + """ + Execute a shell command. + + Args: + command: Shell command to execute. + args: Additional arguments to append to the command. + + Returns: + Exit code from the shell command. + """ + # Append additional arguments to the command with proper escaping + if args: + escaped_args = ' '.join(shlex.quote(arg) for arg in args) + command = f"{command} {escaped_args}" + + try: + result = subprocess.run(command, shell=True, cwd=self.config.config_path.parent) + return result.returncode + except Exception as e: + print(f"Error executing shell command: {e}", file=sys.stderr) + return 1 + + def _execute_python_call(self, module_path: str, args: list[str]) -> int: + """ + Execute a Python module entry point. + + Args: + module_path: Module path in format 'module.path:function' or 'module.path:Class.method'. + args: Additional arguments to pass to the function/method. + + Returns: + Exit code from the Python call. + """ + try: + # Parse module path + if ":" not in module_path: + print(f"Invalid call format: {module_path}. Expected 'module:function'", file=sys.stderr) + return 1 + + module_name, func_path = module_path.split(":", 1) + + # Import the module + try: + module = importlib.import_module(module_name) + except ImportError as e: + print(f"Failed to import module '{module_name}': {e}", file=sys.stderr) + return 1 + + # Get the function/method + obj = module + for attr in func_path.split("."): + try: + obj = getattr(obj, attr) + except AttributeError: + print(f"'{attr}' not found in {obj}", file=sys.stderr) + return 1 + + # Call the function/method with arguments + # Set sys.argv for the called function + old_argv = sys.argv + sys.argv = [module_path] + args + try: + result = obj() + # Handle different return types + if result is None: + return 0 + elif isinstance(result, int): + return result + else: + return 0 + finally: + sys.argv = old_argv + + except Exception as e: + print(f"Error executing Python call: {e}", file=sys.stderr) + return 1 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..969002d --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for tool-tasks.""" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..35ae5a0 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,222 @@ +"""Tests for the CLI module.""" + +import os +import sys +import tempfile +from pathlib import Path + +import pytest + +from tool_tasks.cli import main + + +def test_cli_no_arguments(monkeypatch, capsys): + """Test CLI with no arguments.""" + monkeypatch.setattr(sys, "argv", ["task"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "Usage: task " in captured.err + + +def test_cli_list_tasks(monkeypatch, capsys): + """Test CLI --list option.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +build = "make build" +clean = "rm -rf dist" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "--list"]) + main() + + captured = capsys.readouterr() + assert "Available tasks:" in captured.out + assert "build" in captured.out + assert "clean" in captured.out + assert "test" in captured.out + finally: + os.chdir(old_cwd) + + +def test_cli_list_tasks_no_pyproject(monkeypatch, capsys): + """Test CLI --list when pyproject.toml not found.""" + with tempfile.TemporaryDirectory() as tmpdir: + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "--list"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "pyproject.toml not found" in captured.err + finally: + os.chdir(old_cwd) + + +def test_cli_execute_task(monkeypatch): + """Test CLI executing a task.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +hello = "echo hello" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "hello"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + finally: + os.chdir(old_cwd) + + +def test_cli_execute_task_with_args(monkeypatch): + """Test CLI executing a task with arguments.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +echo = "echo" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "echo", "hello", "world"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + finally: + os.chdir(old_cwd) + + +def test_cli_execute_nonexistent_task(monkeypatch, capsys): + """Test CLI executing a non-existent task.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "nonexistent"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "Task 'nonexistent' not found" in captured.err + finally: + os.chdir(old_cwd) + + +def test_cli_no_pyproject(monkeypatch, capsys): + """Test CLI when pyproject.toml not found.""" + with tempfile.TemporaryDirectory() as tmpdir: + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "test"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "pyproject.toml not found" in captured.err + finally: + os.chdir(old_cwd) + + +def test_cli_list_empty_tasks(monkeypatch, capsys): + """Test CLI --list with no tasks defined.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.poetry] +name = "test" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "--list"]) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + finally: + os.chdir(old_cwd) + + +def test_cli_unexpected_error(monkeypatch, capsys): + """Test CLI with an unexpected error.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + # Create an invalid config that will cause an unexpected error + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +""") + + + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + monkeypatch.setattr(sys, "argv", ["task", "test"]) + + # Mock TaskConfig to raise an unexpected exception + from tool_tasks import config + original_init = config.TaskConfig.__init__ + + def mock_init(self, config_path=None): + raise RuntimeError("Unexpected error") + + monkeypatch.setattr(config.TaskConfig, "__init__", mock_init) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "Unexpected error" in captured.err + finally: + os.chdir(old_cwd) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..afe912b --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,187 @@ +"""Tests for the config module.""" + +import os +import sys +import tempfile +from pathlib import Path + +import pytest + +from tool_tasks.config import TaskConfig + + +def test_find_pyproject_in_current_dir(): + """Test finding pyproject.toml in current directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +""") + + # Change to temp directory + old_cwd = os.getcwd() + try: + os.chdir(tmpdir) + config = TaskConfig() + assert config.config_path == pyproject + finally: + os.chdir(old_cwd) + + +def test_find_pyproject_in_parent_dir(): + """Test finding pyproject.toml in parent directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +""") + subdir = tmpdir_path / "subdir" + subdir.mkdir() + + # Change to subdirectory + old_cwd = os.getcwd() + try: + os.chdir(subdir) + config = TaskConfig() + assert config.config_path == pyproject + finally: + os.chdir(old_cwd) + + +def test_pyproject_not_found(): + """Test error when pyproject.toml is not found.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + subdir = tmpdir_path / "subdir" + subdir.mkdir() + + # Change to subdirectory + old_cwd = os.getcwd() + try: + os.chdir(subdir) + with pytest.raises(FileNotFoundError, match="pyproject.toml not found"): + TaskConfig() + finally: + os.chdir(old_cwd) + + +def test_load_tasks_missing_tool_tasks_section(capsys): + """Test error when [tool.tasks] section is missing.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.poetry] +name = "test" +""") + + with pytest.raises(SystemExit) as exc_info: + TaskConfig(pyproject) + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "No [tool.tasks] section found" in captured.err + + +def test_load_tasks_invalid_toml(capsys): + """Test error when pyproject.toml is invalid.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text("invalid toml content [[[") + + with pytest.raises(SystemExit) as exc_info: + TaskConfig(pyproject) + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "Error reading" in captured.err + + +def test_get_task(): + """Test getting a task by name.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +build = "make build" +""") + + config = TaskConfig(pyproject) + assert config.get_task("test") == "echo test" + assert config.get_task("build") == "make build" + + +def test_get_task_not_found(capsys): + """Test error when task is not found.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "echo test" +""") + + config = TaskConfig(pyproject) + with pytest.raises(SystemExit) as exc_info: + config.get_task("nonexistent") + + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "Task 'nonexistent' not found" in captured.err + assert "Available tasks: test" in captured.err + + +def test_list_tasks(): + """Test listing all tasks.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +test = "pytest" +build = "make build" +clean = "rm -rf dist" +""") + + config = TaskConfig(pyproject) + tasks = config.list_tasks() + assert tasks == ["build", "clean", "test"] + + +def test_load_complex_tasks(): + """Test loading complex task configurations.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +simple = "echo simple" +chain = ["task1", "task2"] + +[tool.tasks.dict_task] +cmd = "echo dict" + +[tool.tasks.python_task] +call = "module:function" +""") + + config = TaskConfig(pyproject) + assert config.get_task("simple") == "echo simple" + assert config.get_task("chain") == ["task1", "task2"] + assert config.get_task("dict_task") == {"cmd": "echo dict"} + assert config.get_task("python_task") == {"call": "module:function"} + + +def test_find_pyproject_at_root(): + """Test finding pyproject.toml at the root directory.""" + # This test is hard to implement without modifying the filesystem root + # We'll skip it as it's an edge case that's unlikely to occur in practice + # The code path exists for completeness + pass diff --git a/tests/test_executor.py b/tests/test_executor.py new file mode 100644 index 0000000..d10cf0c --- /dev/null +++ b/tests/test_executor.py @@ -0,0 +1,430 @@ +"""Tests for the executor module.""" + +import os +import sys +import tempfile +from pathlib import Path + +import pytest + +from tool_tasks.config import TaskConfig +from tool_tasks.executor import TaskExecutor + + +def test_execute_shell_command(): + """Test executing a simple shell command.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +hello = "echo hello" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("hello") + assert exit_code == 0 + + +def test_execute_shell_command_with_args(): + """Test executing a shell command with additional arguments.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +echo = "echo" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("echo", ["hello", "world"]) + assert exit_code == 0 + + +def test_execute_failing_command(): + """Test executing a command that fails.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +fail = "exit 1" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("fail") + assert exit_code == 1 + + +def test_execute_task_alias(): + """Test executing a task alias.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +original = "echo original" +alias = "original" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("alias") + assert exit_code == 0 + + +def test_execute_task_chain(): + """Test executing a chain of tasks.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +task1 = "echo task1" +task2 = "echo task2" +chain = ["task1", "task2"] +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("chain") + assert exit_code == 0 + + +def test_execute_task_chain_stops_on_failure(): + """Test that task chain stops on first failure.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +task1 = "echo task1" +fail = "exit 1" +task2 = "echo task2" +chain = ["task1", "fail", "task2"] +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("chain") + assert exit_code == 1 + + +def test_execute_dict_cmd(): + """Test executing a dict task with cmd key.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.hello] +cmd = "echo hello" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("hello") + assert exit_code == 0 + + +def test_execute_dict_chain(): + """Test executing a dict task with chain key.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +task1 = "echo task1" +task2 = "echo task2" + +[tool.tasks.mychain] +chain = ["task1", "task2"] +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("mychain") + assert exit_code == 0 + + +def test_execute_python_call(): + """Test executing a Python module entry point.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "sys:exit" +""") + + # Create a simple Python module for testing + module_dir = tmpdir_path / "testmodule" + module_dir.mkdir() + (module_dir / "__init__.py").write_text(""" +def test_func(): + return 0 + +class TestClass: + @staticmethod + def test_method(): + return 0 +""") + + # Add module to sys.path + sys.path.insert(0, str(tmpdir_path)) + + try: + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + + # Update task to call our test module + config.tasks["pycall"]["call"] = "testmodule:test_func" + exit_code = executor.execute("pycall") + assert exit_code == 0 + + # Test class method + config.tasks["pycall"]["call"] = "testmodule:TestClass.test_method" + exit_code = executor.execute("pycall") + assert exit_code == 0 + finally: + sys.path.pop(0) + + +def test_execute_python_call_with_args(): + """Test executing a Python call with arguments.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "testmodule2:check_args" +""") + + # Create a module that checks sys.argv + module_dir = tmpdir_path / "testmodule2" + module_dir.mkdir() + (module_dir / "__init__.py").write_text(""" +import sys + +def check_args(): + # sys.argv should contain the function path and any additional args + if len(sys.argv) > 1 and sys.argv[1] == "test_arg": + return 0 + return 1 +""") + + sys.path.insert(0, str(tmpdir_path)) + + try: + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall", ["test_arg"]) + assert exit_code == 0 + finally: + sys.path.pop(0) + # Clean up imported module + if "testmodule2" in sys.modules: + del sys.modules["testmodule2"] + + +def test_execute_python_call_returns_none(): + """Test Python call that returns None.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "testmodule3:returns_none" +""") + + module_dir = tmpdir_path / "testmodule3" + module_dir.mkdir() + (module_dir / "__init__.py").write_text(""" +def returns_none(): + pass +""") + + sys.path.insert(0, str(tmpdir_path)) + + try: + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 0 + finally: + sys.path.pop(0) + if "testmodule3" in sys.modules: + del sys.modules["testmodule3"] + + +def test_execute_python_call_returns_non_int(): + """Test Python call that returns non-integer value.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "testmodule4:returns_string" +""") + + module_dir = tmpdir_path / "testmodule4" + module_dir.mkdir() + (module_dir / "__init__.py").write_text(""" +def returns_string(): + return "hello" +""") + + sys.path.insert(0, str(tmpdir_path)) + + try: + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 0 + finally: + sys.path.pop(0) + if "testmodule4" in sys.modules: + del sys.modules["testmodule4"] + + +def test_execute_python_call_invalid_format(capsys): + """Test Python call with invalid format.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "invalid_format" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Invalid call format" in captured.err + + +def test_execute_python_call_module_not_found(capsys): + """Test Python call with non-existent module.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "nonexistent_module:function" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Failed to import module" in captured.err + + +def test_execute_python_call_attribute_not_found(capsys): + """Test Python call with non-existent attribute.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "sys:nonexistent_function" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 1 + captured = capsys.readouterr() + assert "not found" in captured.err + + +def test_circular_dependency_detection(capsys): + """Test detection of circular task dependencies.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +task1 = "task2" +task2 = "task1" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("task1") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Circular dependency detected" in captured.err + + +def test_execute_invalid_dict_task(capsys): + """Test executing an invalid dict task.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.invalid] +unknown_key = "value" +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("invalid") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Invalid task configuration" in captured.err + + +def test_execute_invalid_task_type(capsys): + """Test executing a task with invalid type.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks] +invalid = 123 +""") + + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("invalid") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Invalid task configuration type" in captured.err + + +def test_execute_python_call_raises_exception(capsys): + """Test Python call that raises an exception.""" + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + pyproject = tmpdir_path / "pyproject.toml" + pyproject.write_text(""" +[tool.tasks.pycall] +call = "testmodule5:raises_error" +""") + + module_dir = tmpdir_path / "testmodule5" + module_dir.mkdir() + (module_dir / "__init__.py").write_text(""" +def raises_error(): + raise ValueError("test error") +""") + + sys.path.insert(0, str(tmpdir_path)) + + try: + config = TaskConfig(pyproject) + executor = TaskExecutor(config) + exit_code = executor.execute("pycall") + assert exit_code == 1 + captured = capsys.readouterr() + assert "Error executing Python call" in captured.err + finally: + sys.path.pop(0) + if "testmodule5" in sys.modules: + del sys.modules["testmodule5"]