diff --git a/README.md b/README.md index 04cfd86d..5d5c5254 100644 --- a/README.md +++ b/README.md @@ -323,6 +323,33 @@ Generate native commands and skills tailored for your AI coding assistant. - **Context-Aware**: Skills include all necessary context (instructions, inputs, and dependencies) for the AI. - **Expanding Ecosystem**: Currently supports **Claude Code** and **Gemini CLI**, with more platforms planned. +### Task Scheduling +Schedule periodic execution of maintenance tasks or commands using native system schedulers. +- **Cross-Platform**: Automatically uses systemd timers on Linux or launchd agents on macOS. +- **Flexible Intervals**: Support for common intervals (hourly, daily, weekly) or custom intervals. +- **Integrated Logging**: Automatically logs task output to `.deepwork/logs/`. + +**Schedule a task**: +```bash +# Schedule a daily sync +deepwork schedule add sync-daily "deepwork sync" --interval daily + +# Schedule a custom command hourly +deepwork schedule add cleanup "git fetch --prune" --interval hourly + +# List all scheduled tasks +deepwork schedule list + +# Remove a scheduled task +deepwork schedule remove sync-daily +``` + +**Examples**: +- Periodic skill synchronization: `deepwork sync` +- Branch cleanup: `git fetch --prune` +- Repository maintenance: `git gc --auto` +- Custom maintenance scripts: `./scripts/cleanup.sh` + ## Contributing DeepWork is currently in MVP phase. Contributions welcome! See [CONTRIBUTING.md](CONTRIBUTING.md) for the full development guide. diff --git a/doc/architecture.md b/doc/architecture.md index 87532d09..712f32e1 100644 --- a/doc/architecture.md +++ b/doc/architecture.md @@ -40,7 +40,9 @@ deepwork/ # DeepWork tool repository │ │ ├── __init__.py │ │ ├── main.py # CLI entry point │ │ ├── install.py # Install command -│ │ └── sync.py # Sync command +│ │ ├── sync.py # Sync command +│ │ ├── schedule.py # Schedule command (thin CLI layer) +│ │ └── runners.py # Schedule runners (systemd, launchd) │ ├── core/ │ │ ├── adapters.py # Agent adapters for AI platforms │ │ ├── detector.py # AI platform detection diff --git a/src/deepwork/cli/main.py b/src/deepwork/cli/main.py index b503ea9a..fc54abad 100644 --- a/src/deepwork/cli/main.py +++ b/src/deepwork/cli/main.py @@ -17,12 +17,14 @@ def cli() -> None: from deepwork.cli.hook import hook # noqa: E402 from deepwork.cli.install import install # noqa: E402 from deepwork.cli.rules import rules # noqa: E402 +from deepwork.cli.schedule import schedule # noqa: E402 from deepwork.cli.sync import sync # noqa: E402 cli.add_command(install) cli.add_command(sync) cli.add_command(hook) cli.add_command(rules) +cli.add_command(schedule) if __name__ == "__main__": diff --git a/src/deepwork/cli/runners.py b/src/deepwork/cli/runners.py new file mode 100644 index 00000000..3382c639 --- /dev/null +++ b/src/deepwork/cli/runners.py @@ -0,0 +1,455 @@ +"""Schedule runners for DeepWork CLI. + +Each runner represents a platform-specific execution backend for scheduled tasks +(e.g., systemd timers on Linux, launchd agents on macOS). Runners auto-register +via __init_subclass__, following the same pattern as AgentAdapter. +""" + +from __future__ import annotations + +import html +import platform +import re +import shlex +import shutil +import subprocess +from abc import ABC, abstractmethod +from pathlib import Path +from typing import ClassVar + +from rich.console import Console + +console = Console() + + +class ScheduleError(Exception): + """Exception raised for scheduling errors.""" + + pass + + +def _validate_task_name(task_name: str) -> None: + """Validate that task name contains only safe characters. + + Args: + task_name: Name to validate + + Raises: + ScheduleError: If task name contains unsafe characters + """ + if not re.match(r'^[a-zA-Z0-9_-]+$', task_name): + raise ScheduleError( + f"Invalid task name '{task_name}'. " + "Task names must contain only alphanumeric characters, hyphens, and underscores." + ) + + +class ScheduleRunner(ABC): + """Base class for schedule runners. + + Subclasses are automatically registered when defined, enabling dynamic + discovery of available scheduling backends. + """ + + _registry: ClassVar[dict[str, type[ScheduleRunner]]] = {} + + name: ClassVar[str] + display_name: ClassVar[str] + + def __init_subclass__(cls, **kwargs: object) -> None: + """Auto-register subclasses.""" + super().__init_subclass__(**kwargs) + if "name" in cls.__dict__ and cls.name: + ScheduleRunner._registry[cls.name] = cls + + @classmethod + def get_all(cls) -> dict[str, type[ScheduleRunner]]: + """Return all registered runner classes.""" + return cls._registry.copy() + + @classmethod + def get(cls, name: str) -> type[ScheduleRunner]: + """Get runner class by name. + + Raises: + ScheduleError: If runner name is not registered + """ + if name not in cls._registry: + raise ScheduleError( + f"Unknown runner '{name}'. " + f"Available runners: {', '.join(cls._registry.keys())}" + ) + return cls._registry[name] + + @classmethod + def detect_runner(cls) -> ScheduleRunner | None: + """Detect and return a runner instance for the current system. + + Returns: + A runner instance, or None if no runner supports this system. + """ + for runner_cls in cls._registry.values(): + if runner_cls.detect(): + return runner_cls() + return None + + @staticmethod + @abstractmethod + def detect() -> bool: + """Check if this runner can operate on the current system.""" + + @abstractmethod + def install(self, task_name: str, command: str, project_path: Path, interval: str) -> None: + """Install a scheduled task. + + Args: + task_name: Unique name for the task (alphanumeric, hyphens, underscores) + command: Command to execute + project_path: Path to the project directory + interval: Schedule interval (e.g., 'daily', 'hourly', or seconds) + + Raises: + ScheduleError: If installation fails + """ + + @abstractmethod + def uninstall(self, task_name: str) -> None: + """Remove a scheduled task. + + Args: + task_name: Name of the task to remove + + Raises: + ScheduleError: If removal fails + """ + + @abstractmethod + def list_tasks(self) -> None: + """List scheduled tasks for this runner.""" + + +class SystemdRunner(ScheduleRunner): + """Runner for systemd timers on Linux.""" + + name = "systemd" + display_name = "Systemd Timer" + + @staticmethod + def detect() -> bool: + return platform.system() == "Linux" and shutil.which("systemctl") is not None + + @staticmethod + def create_service_config( + task_name: str, command: str, project_path: Path, interval: str + ) -> tuple[str, str]: + """Create systemd service and timer unit file contents. + + Args: + task_name: Name of the scheduled task (must be validated) + command: Command to execute (will be shell-escaped) + project_path: Path to the project directory + interval: Systemd timer interval (e.g., 'daily', 'weekly', 'hourly') + + Returns: + Tuple of (service_content, timer_content) + """ + _validate_task_name(task_name) + service_name = f"deepwork-{task_name}" + safe_command = shlex.quote(command) + + service_content = f"""[Unit] +Description=DeepWork task: {task_name} +After=network.target + +[Service] +Type=oneshot +WorkingDirectory={project_path} +ExecStart=/bin/sh -c {safe_command} +StandardOutput=journal +StandardError=journal + +[Install] +WantedBy=multi-user.target +""" + + timer_content = f"""[Unit] +Description=Timer for DeepWork task: {task_name} +Requires={service_name}.service + +[Timer] +OnCalendar={interval} +Persistent=true + +[Install] +WantedBy=timers.target +""" + + return service_content, timer_content + + def install( + self, + task_name: str, + command: str, + project_path: Path, + interval: str, + user_mode: bool = True, + ) -> None: + service_name = f"deepwork-{task_name}" + service_content, timer_content = self.create_service_config( + task_name, command, project_path, interval + ) + + if user_mode: + systemd_dir = Path.home() / ".config" / "systemd" / "user" + else: + systemd_dir = Path("/etc/systemd/system") + + systemd_dir.mkdir(parents=True, exist_ok=True) + + service_file = systemd_dir / f"{service_name}.service" + service_file.write_text(service_content) + console.print(f" [green]✓[/green] Created {service_file}") + + timer_file = systemd_dir / f"{service_name}.timer" + timer_file.write_text(timer_content) + console.print(f" [green]✓[/green] Created {timer_file}") + + try: + systemctl_cmd = ["systemctl"] + if user_mode: + systemctl_cmd.append("--user") + + subprocess.run([*systemctl_cmd, "daemon-reload"], check=True, capture_output=True) + subprocess.run( + [*systemctl_cmd, "enable", f"{service_name}.timer"], + check=True, + capture_output=True, + ) + subprocess.run( + [*systemctl_cmd, "start", f"{service_name}.timer"], + check=True, + capture_output=True, + ) + console.print(f" [green]✓[/green] Enabled and started {service_name}.timer") + except subprocess.CalledProcessError as e: + raise ScheduleError(f"Failed to enable systemd timer: {e.stderr.decode()}") from e + + def uninstall(self, task_name: str, user_mode: bool = True) -> None: + service_name = f"deepwork-{task_name}" + + if user_mode: + systemd_dir = Path.home() / ".config" / "systemd" / "user" + else: + systemd_dir = Path("/etc/systemd/system") + + systemctl_cmd = ["systemctl"] + if user_mode: + systemctl_cmd.append("--user") + + try: + subprocess.run( + [*systemctl_cmd, "stop", f"{service_name}.timer"], + check=False, + capture_output=True, + ) + subprocess.run( + [*systemctl_cmd, "disable", f"{service_name}.timer"], + check=False, + capture_output=True, + ) + console.print(f" [green]✓[/green] Stopped and disabled {service_name}.timer") + except (FileNotFoundError, OSError) as e: + console.print(f" [dim]•[/dim] Could not stop timer: {e}") + + service_file = systemd_dir / f"{service_name}.service" + timer_file = systemd_dir / f"{service_name}.timer" + + if service_file.exists(): + service_file.unlink() + console.print(f" [green]✓[/green] Removed {service_file}") + + if timer_file.exists(): + timer_file.unlink() + console.print(f" [green]✓[/green] Removed {timer_file}") + + try: + subprocess.run([*systemctl_cmd, "daemon-reload"], check=True, capture_output=True) + except subprocess.CalledProcessError: + pass + + def list_tasks(self) -> None: + try: + result = subprocess.run( + ["systemctl", "--user", "list-timers", "--all"], + capture_output=True, + text=True, + check=True, + ) + + lines = result.stdout.split("\n") + deepwork_timers = [line for line in lines if "deepwork-" in line] + + if deepwork_timers: + for line in deepwork_timers: + console.print(f" {line}") + else: + console.print("[dim]No scheduled DeepWork tasks found[/dim]") + except subprocess.CalledProcessError: + console.print("[yellow]Failed to list systemd timers[/yellow]") + + +class LaunchdRunner(ScheduleRunner): + """Runner for launchd agents on macOS.""" + + name = "launchd" + display_name = "macOS LaunchAgent" + + INTERVAL_MAP: ClassVar[dict[str, int]] = { + "hourly": 3600, + "daily": 86400, + "weekly": 604800, + "monthly": 2592000, + } + + @staticmethod + def detect() -> bool: + return platform.system() == "Darwin" + + @staticmethod + def parse_interval(interval: str) -> int: + """Parse an interval string into seconds. + + Args: + interval: Interval as seconds (numeric) or named ('hourly', 'daily', etc.) + + Returns: + Interval in seconds + + Raises: + ScheduleError: If interval is invalid + """ + try: + return int(interval) + except ValueError: + pass + + key = interval.lower() + if key not in LaunchdRunner.INTERVAL_MAP: + raise ScheduleError( + f"Invalid interval '{interval}'. " + f"Valid options are: {', '.join(LaunchdRunner.INTERVAL_MAP.keys())}, " + "or a number of seconds." + ) + return LaunchdRunner.INTERVAL_MAP[key] + + @staticmethod + def create_plist(task_name: str, command: str, project_path: Path, interval: int) -> str: + """Create launchd plist content. + + Args: + task_name: Name of the scheduled task (must be validated) + command: Command to execute (will be parsed and escaped) + project_path: Path to the project directory + interval: Interval in seconds + + Returns: + Plist XML content + """ + _validate_task_name(task_name) + label = f"com.deepwork.{task_name}" + + try: + command_parts = shlex.split(command) + except ValueError as e: + raise ScheduleError(f"Invalid command syntax: {e}") from e + + if not command_parts: + raise ValueError("Command cannot be empty") + + escaped_args = [html.escape(arg) for arg in command_parts] + args_xml = "\n ".join(f"{arg}" for arg in escaped_args) + + plist_content = f""" + + + + Label + {label} + + ProgramArguments + + {args_xml} + + + WorkingDirectory + {project_path} + + StartInterval + {interval} + + StandardOutPath + {project_path}/.deepwork/logs/{task_name}.log + + StandardErrorPath + {project_path}/.deepwork/logs/{task_name}.err + + +""" + + return plist_content + + def install(self, task_name: str, command: str, project_path: Path, interval: str) -> None: + interval_seconds = self.parse_interval(interval) + label = f"com.deepwork.{task_name}" + plist_content = self.create_plist(task_name, command, project_path, interval_seconds) + + launch_agents_dir = Path.home() / "Library" / "LaunchAgents" + launch_agents_dir.mkdir(parents=True, exist_ok=True) + + plist_file = launch_agents_dir / f"{label}.plist" + plist_file.write_text(plist_content) + console.print(f" [green]✓[/green] Created {plist_file}") + + logs_dir = project_path / ".deepwork" / "logs" + logs_dir.mkdir(parents=True, exist_ok=True) + + try: + subprocess.run( + ["launchctl", "load", str(plist_file)], check=True, capture_output=True + ) + console.print(f" [green]✓[/green] Loaded {label}") + except subprocess.CalledProcessError as e: + raise ScheduleError(f"Failed to load launchd agent: {e.stderr.decode()}") from e + + def uninstall(self, task_name: str) -> None: + label = f"com.deepwork.{task_name}" + launch_agents_dir = Path.home() / "Library" / "LaunchAgents" + plist_file = launch_agents_dir / f"{label}.plist" + + try: + subprocess.run( + ["launchctl", "unload", str(plist_file)], check=False, capture_output=True + ) + console.print(f" [green]✓[/green] Unloaded {label}") + except (FileNotFoundError, OSError) as e: + console.print(f" [dim]•[/dim] Could not unload agent: {e}") + + if plist_file.exists(): + plist_file.unlink() + console.print(f" [green]✓[/green] Removed {plist_file}") + + def list_tasks(self) -> None: + launch_agents_dir = Path.home() / "Library" / "LaunchAgents" + + if not launch_agents_dir.exists(): + console.print("[dim]No scheduled DeepWork tasks found[/dim]") + return + + plist_files = sorted(launch_agents_dir.glob("com.deepwork.*.plist")) + + if plist_files: + for plist_file in plist_files: + task_name = plist_file.stem.replace("com.deepwork.", "") + console.print(f" [green]✓[/green] {task_name} ({plist_file.name})") + else: + console.print("[dim]No scheduled DeepWork tasks found[/dim]") diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py new file mode 100644 index 00000000..77a005ae --- /dev/null +++ b/src/deepwork/cli/schedule.py @@ -0,0 +1,145 @@ +"""Schedule command for DeepWork CLI.""" + +from pathlib import Path + +import click +from rich.console import Console + +from deepwork.cli.runners import ScheduleError, ScheduleRunner +from deepwork.utils.git import is_git_repo + +console = Console() + + +@click.group() +def schedule() -> None: + """Manage scheduled DeepWork jobs.""" + pass + + +@schedule.command() +@click.argument("task_name") +@click.argument("command") +@click.option( + "--interval", + "-i", + default="daily", + help="Schedule interval (systemd: 'daily', 'weekly', 'hourly'; launchd: seconds)", +) +@click.option( + "--path", + type=click.Path(exists=True, file_okay=False, path_type=Path), + default=".", + help="Path to project directory (default: current directory)", +) +def add(task_name: str, command: str, interval: str, path: Path) -> None: + """Schedule a task to run periodically. + + TASK_NAME is a unique name for this scheduled task. + COMMAND is the command to run (e.g., 'deepwork sync'). + """ + try: + _add_schedule(task_name, command, interval, path) + except ScheduleError as e: + console.print(f"[red]Error:[/red] {e}") + raise click.Abort() from e + + +def _add_schedule(task_name: str, command: str, interval: str, project_path: Path) -> None: + """Add a scheduled task. + + Args: + task_name: Name of the scheduled task + command: Command to execute + interval: Schedule interval + project_path: Path to the project directory + + Raises: + ScheduleError: If scheduling fails + """ + console.print(f"\n[bold cyan]Scheduling DeepWork task: {task_name}[/bold cyan]\n") + + if not is_git_repo(project_path): + raise ScheduleError("Not a Git repository. DeepWork requires a Git repository.") + + deepwork_dir = project_path / ".deepwork" + if not deepwork_dir.exists(): + console.print( + "[yellow]Warning:[/yellow] DeepWork not installed in this project. " + "The scheduled command may not work correctly." + ) + + runner = ScheduleRunner.detect_runner() + if runner is None: + raise ScheduleError( + "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." + ) + + console.print(f"[yellow]→[/yellow] Detected system: {runner.name}") + console.print(f"[yellow]→[/yellow] Installing {runner.display_name}...") + runner.install(task_name, command, project_path, interval) + + console.print() + console.print(f"[bold green]✓ Task '{task_name}' scheduled successfully![/bold green]") + console.print() + + +@schedule.command() +@click.argument("task_name") +@click.option( + "--path", + type=click.Path(exists=True, file_okay=False, path_type=Path), + default=".", + help="Path to project directory (default: current directory)", +) +def remove(task_name: str, path: Path) -> None: + """Remove a scheduled task. + + TASK_NAME is the name of the task to unschedule. + """ + try: + _remove_schedule(task_name, path) + except ScheduleError as e: + console.print(f"[red]Error:[/red] {e}") + raise click.Abort() from e + + +def _remove_schedule(task_name: str, project_path: Path) -> None: + """Remove a scheduled task. + + Args: + task_name: Name of the scheduled task to unschedule + project_path: Path to the project directory + + Raises: + ScheduleError: If unscheduling fails + """ + console.print(f"\n[bold cyan]Removing scheduled task: {task_name}[/bold cyan]\n") + + runner = ScheduleRunner.detect_runner() + if runner is None: + raise ScheduleError( + "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." + ) + + console.print(f"[yellow]→[/yellow] Detected system: {runner.name}") + console.print(f"[yellow]→[/yellow] Removing {runner.display_name}...") + runner.uninstall(task_name) + + console.print() + console.print(f"[bold green]✓ Task '{task_name}' unscheduled successfully![/bold green]") + console.print() + + +@schedule.command() +def list() -> None: + """List all scheduled DeepWork tasks.""" + console.print("\n[bold cyan]Scheduled DeepWork Tasks[/bold cyan]\n") + + runner = ScheduleRunner.detect_runner() + if runner is None: + console.print("[yellow]No scheduled tasks (unsupported system)[/yellow]") + return + + console.print(f"System: {runner.name}\n") + runner.list_tasks() diff --git a/tests/integration/test_schedule_integration.py b/tests/integration/test_schedule_integration.py new file mode 100644 index 00000000..cf0c7fd0 --- /dev/null +++ b/tests/integration/test_schedule_integration.py @@ -0,0 +1,171 @@ +"""Integration tests for schedule command.""" + +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest +from click.testing import CliRunner + +from deepwork.cli.main import cli + + +class TestScheduleCommand: + """Integration tests for 'deepwork schedule' command.""" + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_schedule_add_systemd( + self, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test scheduling a task on systemd.""" + mock_runner = Mock() + mock_runner.name = "systemd" + mock_runner.display_name = "Systemd Timer" + mock_detect.return_value = mock_runner + mock_is_git.return_value = True + + (temp_dir / ".deepwork").mkdir() + + runner = CliRunner() + result = runner.invoke( + cli, + [ + "schedule", + "add", + "test-task", + "deepwork sync", + "--interval", + "daily", + "--path", + str(temp_dir), + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Scheduling DeepWork task: test-task" in result.output + assert "Detected system: systemd" in result.output + assert "scheduled successfully" in result.output + + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_schedule_add_launchd( + self, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test scheduling a task on launchd.""" + mock_runner = Mock() + mock_runner.name = "launchd" + mock_runner.display_name = "macOS LaunchAgent" + mock_detect.return_value = mock_runner + mock_is_git.return_value = True + + (temp_dir / ".deepwork").mkdir() + + runner = CliRunner() + result = runner.invoke( + cli, + [ + "schedule", + "add", + "test-task", + "deepwork sync", + "--interval", + "hourly", + "--path", + str(temp_dir), + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Scheduling DeepWork task: test-task" in result.output + assert "Detected system: launchd" in result.output + assert "scheduled successfully" in result.output + + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "hourly" + ) + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + def test_schedule_remove(self, mock_detect: Mock, temp_dir: Path) -> None: + """Test removing a scheduled task.""" + mock_runner = Mock() + mock_runner.name = "systemd" + mock_runner.display_name = "Systemd Timer" + mock_detect.return_value = mock_runner + + runner = CliRunner() + result = runner.invoke( + cli, + ["schedule", "remove", "test-task", "--path", str(temp_dir)], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Removing scheduled task: test-task" in result.output + assert "unscheduled successfully" in result.output + + mock_runner.uninstall.assert_called_once_with("test-task") + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + def test_schedule_list(self, mock_detect: Mock) -> None: + """Test listing scheduled tasks.""" + mock_runner = Mock() + mock_runner.name = "systemd" + mock_detect.return_value = mock_runner + + runner = CliRunner() + result = runner.invoke(cli, ["schedule", "list"], catch_exceptions=False) + + assert result.exit_code == 0 + assert "Scheduled DeepWork Tasks" in result.output + assert "System: systemd" in result.output + + mock_runner.list_tasks.assert_called_once() + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_schedule_add_not_git_repo( + self, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test that scheduling fails when not in a git repo.""" + mock_is_git.return_value = False + + runner = CliRunner() + result = runner.invoke( + cli, + ["schedule", "add", "test-task", "deepwork sync", "--path", str(temp_dir)], + ) + + assert result.exit_code == 1 + assert "Not a Git repository" in result.output + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_schedule_add_unsupported_system( + self, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test that scheduling fails on unsupported systems.""" + mock_is_git.return_value = True + mock_detect.return_value = None + + (temp_dir / ".deepwork").mkdir() + + runner = CliRunner() + result = runner.invoke( + cli, + ["schedule", "add", "test-task", "deepwork sync", "--path", str(temp_dir)], + ) + + assert result.exit_code == 1 + assert "Unsupported system" in result.output diff --git a/tests/unit/test_runners.py b/tests/unit/test_runners.py new file mode 100644 index 00000000..b6688751 --- /dev/null +++ b/tests/unit/test_runners.py @@ -0,0 +1,200 @@ +"""Tests for schedule runners.""" + +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from deepwork.cli.runners import ( + LaunchdRunner, + ScheduleError, + ScheduleRunner, + SystemdRunner, + _validate_task_name, +) + + +class TestValidateTaskName: + """Tests for task name validation.""" + + def test_valid_task_names(self) -> None: + valid_names = ["test-task", "task_name", "task123", "my-task-123_test"] + for name in valid_names: + _validate_task_name(name) # Should not raise + + def test_invalid_task_names(self) -> None: + invalid_names = [ + "task name", + "task@name", + "task/name", + "task.name", + "task:name", + ] + for name in invalid_names: + with pytest.raises(ScheduleError, match="Invalid task name"): + _validate_task_name(name) + + +class TestScheduleRunnerRegistry: + """Tests for runner auto-registration and detection.""" + + def test_systemd_registered(self) -> None: + assert "systemd" in ScheduleRunner.get_all() + + def test_launchd_registered(self) -> None: + assert "launchd" in ScheduleRunner.get_all() + + def test_get_systemd(self) -> None: + assert ScheduleRunner.get("systemd") is SystemdRunner + + def test_get_launchd(self) -> None: + assert ScheduleRunner.get("launchd") is LaunchdRunner + + def test_get_unknown_raises(self) -> None: + with pytest.raises(ScheduleError, match="Unknown runner"): + ScheduleRunner.get("nonexistent") + + @patch("deepwork.cli.runners.platform.system", return_value="Linux") + @patch("deepwork.cli.runners.shutil.which", return_value="/usr/bin/systemctl") + def test_detect_runner_linux_systemd(self, _which: Mock, _system: Mock) -> None: + runner = ScheduleRunner.detect_runner() + assert runner is not None + assert isinstance(runner, SystemdRunner) + + @patch("deepwork.cli.runners.platform.system", return_value="Darwin") + def test_detect_runner_macos(self, _system: Mock) -> None: + runner = ScheduleRunner.detect_runner() + assert runner is not None + assert isinstance(runner, LaunchdRunner) + + @patch("deepwork.cli.runners.platform.system", return_value="Windows") + @patch("deepwork.cli.runners.shutil.which", return_value=None) + def test_detect_runner_unsupported(self, _which: Mock, _system: Mock) -> None: + runner = ScheduleRunner.detect_runner() + assert runner is None + + +class TestSystemdRunner: + """Tests for SystemdRunner.""" + + @patch("deepwork.cli.runners.platform.system", return_value="Linux") + @patch("deepwork.cli.runners.shutil.which", return_value="/usr/bin/systemctl") + def test_detect_linux_with_systemctl(self, _which: Mock, _system: Mock) -> None: + assert SystemdRunner.detect() is True + + @patch("deepwork.cli.runners.platform.system", return_value="Linux") + @patch("deepwork.cli.runners.shutil.which", return_value=None) + def test_detect_linux_without_systemctl(self, _which: Mock, _system: Mock) -> None: + assert SystemdRunner.detect() is False + + @patch("deepwork.cli.runners.platform.system", return_value="Darwin") + def test_detect_not_linux(self, _system: Mock) -> None: + assert SystemdRunner.detect() is False + + def test_create_service_config_basic(self, temp_dir: Path) -> None: + service, timer = SystemdRunner.create_service_config( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + assert "Description=DeepWork task: test-task" in service + assert f"WorkingDirectory={temp_dir}" in service + assert "ExecStart=/bin/sh -c" in service + assert "Type=oneshot" in service + + assert "Description=Timer for DeepWork task: test-task" in timer + assert "OnCalendar=daily" in timer + assert "Persistent=true" in timer + assert "Requires=deepwork-test-task.service" in timer + + def test_create_service_config_intervals(self, temp_dir: Path) -> None: + for interval in ["hourly", "weekly", "monthly"]: + _, timer = SystemdRunner.create_service_config( + "test-task", "echo test", temp_dir, interval + ) + assert f"OnCalendar={interval}" in timer + + def test_create_service_config_invalid_name(self, temp_dir: Path) -> None: + with pytest.raises(ScheduleError, match="Invalid task name"): + SystemdRunner.create_service_config("test task", "echo test", temp_dir, "daily") + + +class TestLaunchdRunner: + """Tests for LaunchdRunner.""" + + @patch("deepwork.cli.runners.platform.system", return_value="Darwin") + def test_detect_macos(self, _system: Mock) -> None: + assert LaunchdRunner.detect() is True + + @patch("deepwork.cli.runners.platform.system", return_value="Linux") + def test_detect_not_macos(self, _system: Mock) -> None: + assert LaunchdRunner.detect() is False + + def test_parse_interval_numeric(self) -> None: + assert LaunchdRunner.parse_interval("3600") == 3600 + + def test_parse_interval_named(self) -> None: + assert LaunchdRunner.parse_interval("hourly") == 3600 + assert LaunchdRunner.parse_interval("daily") == 86400 + assert LaunchdRunner.parse_interval("weekly") == 604800 + assert LaunchdRunner.parse_interval("monthly") == 2592000 + + def test_parse_interval_case_insensitive(self) -> None: + assert LaunchdRunner.parse_interval("Daily") == 86400 + assert LaunchdRunner.parse_interval("HOURLY") == 3600 + + def test_parse_interval_invalid(self) -> None: + with pytest.raises(ScheduleError, match="Invalid interval"): + LaunchdRunner.parse_interval("invalid") + + def test_create_plist_basic(self, temp_dir: Path) -> None: + plist = LaunchdRunner.create_plist("test-task", "deepwork sync", temp_dir, 86400) + + assert "Label" in plist + assert "com.deepwork.test-task" in plist + assert "ProgramArguments" in plist + assert "deepwork" in plist + assert "sync" in plist + assert f"{temp_dir}" in plist + assert "StartInterval" in plist + assert "86400" in plist + + def test_create_plist_complex_command(self, temp_dir: Path) -> None: + plist = LaunchdRunner.create_plist("test-task", "git fetch origin main", temp_dir, 3600) + + assert "git" in plist + assert "fetch" in plist + assert "origin" in plist + assert "main" in plist + + def test_create_plist_quoted_arguments(self, temp_dir: Path) -> None: + plist = LaunchdRunner.create_plist( + "test-task", 'git commit -m "My commit message"', temp_dir, 3600 + ) + + assert "git" in plist + assert "commit" in plist + assert "-m" in plist + assert "My commit message" in plist + + def test_create_plist_xml_escaping(self, temp_dir: Path) -> None: + plist = LaunchdRunner.create_plist( + "test-task", 'echo "test & more"', temp_dir, 3600 + ) + + assert "<tag> & more" in plist + + def test_create_plist_logs_path(self, temp_dir: Path) -> None: + plist = LaunchdRunner.create_plist("test-task", "echo test", temp_dir, 3600) + + assert "StandardOutPath" in plist + assert f"{temp_dir}/.deepwork/logs/test-task.log" in plist + assert "StandardErrorPath" in plist + assert f"{temp_dir}/.deepwork/logs/test-task.err" in plist + + def test_create_plist_empty_command(self, temp_dir: Path) -> None: + with pytest.raises(ValueError, match="Command cannot be empty"): + LaunchdRunner.create_plist("test-task", "", temp_dir, 3600) + + def test_create_plist_invalid_name(self, temp_dir: Path) -> None: + with pytest.raises(ScheduleError, match="Invalid task name"): + LaunchdRunner.create_plist("test task", "echo test", temp_dir, 3600) diff --git a/tests/unit/test_schedule.py b/tests/unit/test_schedule.py new file mode 100644 index 00000000..8c3a4c13 --- /dev/null +++ b/tests/unit/test_schedule.py @@ -0,0 +1,121 @@ +"""Tests for schedule command (CLI layer).""" + +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from deepwork.cli.runners import ScheduleError + + +class TestScheduleAdd: + """Tests for schedule add command.""" + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_add_schedule_systemd( + self, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test adding a schedule on systemd.""" + from deepwork.cli.schedule import _add_schedule + + mock_runner = Mock() + mock_runner.name = "systemd" + mock_runner.display_name = "Systemd Timer" + mock_detect.return_value = mock_runner + mock_is_git.return_value = True + + (temp_dir / ".deepwork").mkdir() + + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_add_schedule_launchd( + self, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test adding a schedule on launchd.""" + from deepwork.cli.schedule import _add_schedule + + mock_runner = Mock() + mock_runner.name = "launchd" + mock_runner.display_name = "macOS LaunchAgent" + mock_detect.return_value = mock_runner + mock_is_git.return_value = True + + (temp_dir / ".deepwork").mkdir() + + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_add_schedule_not_git_repo( + self, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test adding schedule fails when not a git repo.""" + from deepwork.cli.schedule import _add_schedule + + mock_is_git.return_value = False + + with pytest.raises(ScheduleError, match="Not a Git repository"): + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + @patch("deepwork.cli.schedule.is_git_repo") + def test_add_schedule_unsupported_system( + self, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test adding schedule fails on unsupported system.""" + from deepwork.cli.schedule import _add_schedule + + mock_is_git.return_value = True + mock_detect.return_value = None + + (temp_dir / ".deepwork").mkdir() + + with pytest.raises(ScheduleError, match="Unsupported system"): + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + +class TestScheduleRemove: + """Tests for schedule remove command.""" + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + def test_remove_schedule(self, mock_detect: Mock, temp_dir: Path) -> None: + """Test removing a schedule.""" + from deepwork.cli.schedule import _remove_schedule + + mock_runner = Mock() + mock_runner.name = "systemd" + mock_runner.display_name = "Systemd Timer" + mock_detect.return_value = mock_runner + + _remove_schedule("test-task", temp_dir) + + mock_runner.uninstall.assert_called_once_with("test-task") + + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + def test_remove_schedule_unsupported_system( + self, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test removing schedule fails on unsupported system.""" + from deepwork.cli.schedule import _remove_schedule + + mock_detect.return_value = None + + with pytest.raises(ScheduleError, match="Unsupported system"): + _remove_schedule("test-task", temp_dir)