From 257f26c3a9ddb5a6b49b89bfa4d1a29172176706 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:14:27 +0000 Subject: [PATCH 01/10] Initial plan From b3c1234c52724188c8cc64b7d6cc11f3b2503362 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:20:08 +0000 Subject: [PATCH 02/10] Add deepwork schedule command for cronjob management - Implement deepwork schedule add/remove/list commands - Support systemd timers on Linux - Support launchd agents on macOS - Allow scheduling any command with customizable intervals Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/main.py | 2 + src/deepwork/cli/schedule.py | 528 +++++++++++++++++++++++++++++++++++ 2 files changed, 530 insertions(+) create mode 100644 src/deepwork/cli/schedule.py 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/schedule.py b/src/deepwork/cli/schedule.py new file mode 100644 index 00000000..ae0e7b6c --- /dev/null +++ b/src/deepwork/cli/schedule.py @@ -0,0 +1,528 @@ +"""Schedule command for DeepWork CLI.""" + +import platform +import shutil +import subprocess +from pathlib import Path + +import click +from rich.console import Console + +from deepwork.utils.git import is_git_repo + +console = Console() + + +class ScheduleError(Exception): + """Exception raised for scheduling errors.""" + + pass + + +def _detect_system() -> str: + """ + Detect the system scheduler type. + + Returns: + 'systemd' for Linux with systemd, 'launchd' for macOS, or 'unsupported' + """ + system = platform.system() + + if system == "Darwin": + return "launchd" + elif system == "Linux": + # Check if systemd is available + if shutil.which("systemctl"): + return "systemd" + return "unsupported" + else: + return "unsupported" + + +def _create_systemd_service( + task_name: str, command: str, project_path: Path, interval: str +) -> tuple[str, str]: + """ + Create systemd service and timer content. + + Args: + task_name: Name of the scheduled task + command: Command to execute + project_path: Path to the project directory + interval: Systemd timer interval (e.g., 'daily', 'weekly', 'hourly') + + Returns: + Tuple of (service_content, timer_content) + """ + service_name = f"deepwork-{task_name}" + + service_content = f"""[Unit] +Description=DeepWork task: {task_name} +After=network.target + +[Service] +Type=oneshot +WorkingDirectory={project_path} +ExecStart={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 _create_launchd_plist( + task_name: str, command: str, project_path: Path, interval: int +) -> str: + """ + Create launchd plist content. + + Args: + task_name: Name of the scheduled task + command: Command to execute (will be split into arguments) + project_path: Path to the project directory + interval: Interval in seconds + + Returns: + Plist XML content + """ + label = f"com.deepwork.{task_name}" + + # Split command into program and arguments + command_parts = command.split() + program = command_parts[0] + args = command_parts[1:] if len(command_parts) > 1 else [] + + args_xml = "\n ".join(f"{arg}" for arg in [program] + 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_systemd_timer( + task_name: str, command: str, project_path: Path, interval: str, user_mode: bool = True +) -> None: + """ + Install systemd service and timer. + + Args: + task_name: Name of the scheduled task + command: Command to execute + project_path: Path to the project directory + interval: Systemd timer interval + user_mode: If True, install as user service, else system service + + Raises: + ScheduleError: If installation fails + """ + service_name = f"deepwork-{task_name}" + service_content, timer_content = _create_systemd_service( + task_name, command, project_path, interval + ) + + # Determine systemd directory + 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) + + # Write service file + service_file = systemd_dir / f"{service_name}.service" + service_file.write_text(service_content) + console.print(f" [green]✓[/green] Created {service_file}") + + # Write timer file + timer_file = systemd_dir / f"{service_name}.timer" + timer_file.write_text(timer_content) + console.print(f" [green]✓[/green] Created {timer_file}") + + # Reload systemd and enable timer + 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 _install_launchd_agent( + task_name: str, command: str, project_path: Path, interval: int +) -> None: + """ + Install launchd agent. + + Args: + task_name: Name of the scheduled task + command: Command to execute + project_path: Path to the project directory + interval: Interval in seconds + + Raises: + ScheduleError: If installation fails + """ + label = f"com.deepwork.{task_name}" + plist_content = _create_launchd_plist(task_name, command, project_path, interval) + + # LaunchAgents directory + launch_agents_dir = Path.home() / "Library" / "LaunchAgents" + launch_agents_dir.mkdir(parents=True, exist_ok=True) + + # Write plist file + plist_file = launch_agents_dir / f"{label}.plist" + plist_file.write_text(plist_content) + console.print(f" [green]✓[/green] Created {plist_file}") + + # Ensure logs directory exists + logs_dir = project_path / ".deepwork" / "logs" + logs_dir.mkdir(parents=True, exist_ok=True) + + # Load the agent + 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_systemd_timer(task_name: str, user_mode: bool = True) -> None: + """ + Uninstall systemd service and timer. + + Args: + task_name: Name of the scheduled task + user_mode: If True, uninstall from user services, else system services + + Raises: + ScheduleError: If uninstallation fails + """ + service_name = f"deepwork-{task_name}" + + # Determine systemd directory + 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") + + # Stop and disable timer + 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 Exception: + pass + + # Remove files + 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}") + + # Reload systemd + try: + subprocess.run([*systemctl_cmd, "daemon-reload"], check=True, capture_output=True) + except subprocess.CalledProcessError: + pass + + +def _uninstall_launchd_agent(task_name: str) -> None: + """ + Uninstall launchd agent. + + Args: + task_name: Name of the scheduled task + + Raises: + ScheduleError: If uninstallation fails + """ + label = f"com.deepwork.{task_name}" + launch_agents_dir = Path.home() / "Library" / "LaunchAgents" + plist_file = launch_agents_dir / f"{label}.plist" + + # Unload the agent + try: + subprocess.run(["launchctl", "unload", str(plist_file)], check=False, capture_output=True) + console.print(f" [green]✓[/green] Unloaded {label}") + except Exception: + pass + + # Remove plist file + if plist_file.exists(): + plist_file.unlink() + console.print(f" [green]✓[/green] Removed {plist_file}") + + +@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") + + # Check if project has DeepWork installed (optional, just check if it's a git repo) + 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." + ) + + # Detect system + system_type = _detect_system() + console.print(f"[yellow]→[/yellow] Detected system: {system_type}") + + if system_type == "unsupported": + raise ScheduleError( + "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." + ) + + # Install based on system type + if system_type == "systemd": + console.print("[yellow]→[/yellow] Installing systemd timer...") + _install_systemd_timer(task_name, command, project_path, interval) + elif system_type == "launchd": + # Convert interval to seconds for launchd + try: + interval_seconds = int(interval) + except ValueError: + # Map common intervals to seconds + interval_map = { + "hourly": 3600, + "daily": 86400, + "weekly": 604800, + } + interval_seconds = interval_map.get(interval.lower(), 86400) + + console.print("[yellow]→[/yellow] Installing launchd agent...") + _install_launchd_agent(task_name, command, project_path, interval_seconds) + + 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") + + # Detect system + system_type = _detect_system() + console.print(f"[yellow]→[/yellow] Detected system: {system_type}") + + if system_type == "unsupported": + raise ScheduleError( + "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." + ) + + # Uninstall based on system type + if system_type == "systemd": + console.print("[yellow]→[/yellow] Removing systemd timer...") + _uninstall_systemd_timer(task_name) + elif system_type == "launchd": + console.print("[yellow]→[/yellow] Removing launchd agent...") + _uninstall_launchd_agent(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") + + system_type = _detect_system() + console.print(f"System: {system_type}\n") + + if system_type == "systemd": + _list_systemd_timers() + elif system_type == "launchd": + _list_launchd_agents() + else: + console.print("[yellow]No scheduled tasks (unsupported system)[/yellow]") + + +def _list_systemd_timers() -> None: + """List systemd timers for DeepWork tasks.""" + 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]") + + +def _list_launchd_agents() -> None: + """List launchd agents for DeepWork tasks.""" + 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 = list(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]") From 87cf443f8cf0ac2fa7bf2af5a54d30558dc69ff8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:21:39 +0000 Subject: [PATCH 03/10] Add unit tests and documentation for schedule command - Add comprehensive unit tests for schedule functionality - Document schedule commands in README - All tests passing (488 unit tests) Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- README.md | 27 ++++ tests/unit/test_schedule.py | 256 ++++++++++++++++++++++++++++++++++++ 2 files changed, 283 insertions(+) create mode 100644 tests/unit/test_schedule.py 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/tests/unit/test_schedule.py b/tests/unit/test_schedule.py new file mode 100644 index 00000000..ea49b984 --- /dev/null +++ b/tests/unit/test_schedule.py @@ -0,0 +1,256 @@ +"""Tests for schedule command.""" + +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from deepwork.cli.schedule import ( + ScheduleError, + _create_launchd_plist, + _create_systemd_service, + _detect_system, +) + + +class TestDetectSystem: + """Tests for system detection.""" + + @patch("platform.system") + @patch("shutil.which") + def test_detect_darwin(self, mock_which: Mock, mock_system: Mock) -> None: + """Test detecting macOS.""" + mock_system.return_value = "Darwin" + assert _detect_system() == "launchd" + + @patch("platform.system") + @patch("shutil.which") + def test_detect_linux_with_systemd(self, mock_which: Mock, mock_system: Mock) -> None: + """Test detecting Linux with systemd.""" + mock_system.return_value = "Linux" + mock_which.return_value = "/usr/bin/systemctl" + assert _detect_system() == "systemd" + + @patch("platform.system") + @patch("shutil.which") + def test_detect_linux_without_systemd(self, mock_which: Mock, mock_system: Mock) -> None: + """Test detecting Linux without systemd.""" + mock_system.return_value = "Linux" + mock_which.return_value = None + assert _detect_system() == "unsupported" + + @patch("platform.system") + def test_detect_unsupported_system(self, mock_system: Mock) -> None: + """Test detecting unsupported system.""" + mock_system.return_value = "Windows" + assert _detect_system() == "unsupported" + + +class TestCreateSystemdService: + """Tests for systemd service and timer generation.""" + + def test_create_systemd_service_basic(self, temp_dir: Path) -> None: + """Test creating basic systemd service and timer.""" + task_name = "test-task" + command = "deepwork sync" + interval = "daily" + + service_content, timer_content = _create_systemd_service( + task_name, command, temp_dir, interval + ) + + # Check service content + assert "Description=DeepWork task: test-task" in service_content + assert f"WorkingDirectory={temp_dir}" in service_content + assert f"ExecStart={command}" in service_content + assert "Type=oneshot" in service_content + + # Check timer content + assert "Description=Timer for DeepWork task: test-task" in timer_content + assert "OnCalendar=daily" in timer_content + assert "Persistent=true" in timer_content + assert "Requires=deepwork-test-task.service" in timer_content + + def test_create_systemd_service_with_different_intervals(self, temp_dir: Path) -> None: + """Test creating systemd timer with different intervals.""" + for interval in ["hourly", "weekly", "monthly"]: + _, timer_content = _create_systemd_service( + "test-task", "echo test", temp_dir, interval + ) + assert f"OnCalendar={interval}" in timer_content + + +class TestCreateLaunchdPlist: + """Tests for launchd plist generation.""" + + def test_create_launchd_plist_basic(self, temp_dir: Path) -> None: + """Test creating basic launchd plist.""" + task_name = "test-task" + command = "deepwork sync" + interval = 86400 # daily in seconds + + plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) + + # Check plist content + assert "Label" in plist_content + assert "com.deepwork.test-task" in plist_content + assert "ProgramArguments" in plist_content + assert "deepwork" in plist_content + assert "sync" in plist_content + assert f"{temp_dir}" in plist_content + assert "StartInterval" in plist_content + assert "86400" in plist_content + + def test_create_launchd_plist_with_complex_command(self, temp_dir: Path) -> None: + """Test creating launchd plist with multi-part command.""" + task_name = "test-task" + command = "git fetch origin main" + interval = 3600 + + plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) + + # Check that command is split correctly + assert "git" in plist_content + assert "fetch" in plist_content + assert "origin" in plist_content + assert "main" in plist_content + + def test_create_launchd_plist_logs_path(self, temp_dir: Path) -> None: + """Test that launchd plist includes log paths.""" + plist_content = _create_launchd_plist( + "test-task", "echo test", temp_dir, 3600 + ) + + assert "StandardOutPath" in plist_content + assert f"{temp_dir}/.deepwork/logs/test-task.log" in plist_content + assert "StandardErrorPath" in plist_content + assert f"{temp_dir}/.deepwork/logs/test-task.err" in plist_content + + +class TestScheduleAdd: + """Tests for schedule add command.""" + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.is_git_repo") + @patch("deepwork.cli.schedule._install_systemd_timer") + def test_add_schedule_systemd( + self, + mock_install: Mock, + 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_is_git.return_value = True + mock_detect.return_value = "systemd" + + # Create .deepwork directory + (temp_dir / ".deepwork").mkdir() + + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + mock_install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.is_git_repo") + @patch("deepwork.cli.schedule._install_launchd_agent") + def test_add_schedule_launchd( + self, + mock_install: Mock, + 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_is_git.return_value = True + mock_detect.return_value = "launchd" + + # Create .deepwork directory + (temp_dir / ".deepwork").mkdir() + + _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + + # Daily should be converted to 86400 seconds + mock_install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, 86400 + ) + + @patch("deepwork.cli.schedule._detect_system") + @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._detect_system") + @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 = "unsupported" + + # Create .deepwork directory + (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._detect_system") + @patch("deepwork.cli.schedule._uninstall_systemd_timer") + def test_remove_schedule_systemd( + self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test removing a schedule on systemd.""" + from deepwork.cli.schedule import _remove_schedule + + mock_detect.return_value = "systemd" + + _remove_schedule("test-task", temp_dir) + + mock_uninstall.assert_called_once_with("test-task") + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule._uninstall_launchd_agent") + def test_remove_schedule_launchd( + self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test removing a schedule on launchd.""" + from deepwork.cli.schedule import _remove_schedule + + mock_detect.return_value = "launchd" + + _remove_schedule("test-task", temp_dir) + + mock_uninstall.assert_called_once_with("test-task") + + @patch("deepwork.cli.schedule._detect_system") + 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 = "unsupported" + + with pytest.raises(ScheduleError, match="Unsupported system"): + _remove_schedule("test-task", temp_dir) From 9c843ff373f621e4941b8f854a0c63ee9e6b3732 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:23:10 +0000 Subject: [PATCH 04/10] Add integration tests for schedule command - Add 6 integration tests for schedule CLI - All tests passing (631 tests total) - Feature complete and ready for review Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- .../integration/test_schedule_integration.py | 174 ++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 tests/integration/test_schedule_integration.py diff --git a/tests/integration/test_schedule_integration.py b/tests/integration/test_schedule_integration.py new file mode 100644 index 00000000..ad58e01d --- /dev/null +++ b/tests/integration/test_schedule_integration.py @@ -0,0 +1,174 @@ +"""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._detect_system") + @patch("deepwork.cli.schedule.is_git_repo") + @patch("deepwork.cli.schedule._install_systemd_timer") + def test_schedule_add_systemd( + self, + mock_install: Mock, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test scheduling a task on systemd.""" + mock_is_git.return_value = True + mock_detect.return_value = "systemd" + + # Create .deepwork directory + (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_install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" + ) + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.is_git_repo") + @patch("deepwork.cli.schedule._install_launchd_agent") + def test_schedule_add_launchd( + self, + mock_install: Mock, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test scheduling a task on launchd.""" + mock_is_git.return_value = True + mock_detect.return_value = "launchd" + + # Create .deepwork directory + (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 + + # hourly should be converted to 3600 seconds + mock_install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, 3600 + ) + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule._uninstall_systemd_timer") + def test_schedule_remove( + self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path + ) -> None: + """Test removing a scheduled task.""" + mock_detect.return_value = "systemd" + + 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_uninstall.assert_called_once_with("test-task") + + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule._list_systemd_timers") + def test_schedule_list( + self, mock_list: Mock, mock_detect: Mock + ) -> None: + """Test listing scheduled tasks.""" + mock_detect.return_value = "systemd" + + 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_list.assert_called_once() + + @patch("deepwork.cli.schedule._detect_system") + @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._detect_system") + @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 = "unsupported" + + # Create .deepwork directory + (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 From c489a0dac30897ee68e0e5ce56b122f2ee755333 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:24:03 +0000 Subject: [PATCH 05/10] Fix exception handling in schedule uninstall operations - Replace bare Exception catches with specific exception types - Add better error messages for cleanup failures - Addresses code review feedback Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/schedule.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py index ae0e7b6c..69887026 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -272,7 +272,9 @@ def _uninstall_systemd_timer(task_name: str, user_mode: bool = True) -> None: capture_output=True, ) console.print(f" [green]✓[/green] Stopped and disabled {service_name}.timer") - except Exception: + except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: + # Timer may not exist or systemctl may not be available - continue cleanup + console.print(f" [dim]•[/dim] Could not stop timer: {e}") pass # Remove files @@ -312,7 +314,9 @@ def _uninstall_launchd_agent(task_name: str) -> None: try: subprocess.run(["launchctl", "unload", str(plist_file)], check=False, capture_output=True) console.print(f" [green]✓[/green] Unloaded {label}") - except Exception: + except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: + # Agent may not be loaded or launchctl may not be available - continue cleanup + console.print(f" [dim]•[/dim] Could not unload agent: {e}") pass # Remove plist file From 908e42e39c4f3ecf9d96706152c25b53a9ec4a61 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:26:02 +0000 Subject: [PATCH 06/10] Address all code review feedback - Remove redundant pass statements - Add validation for empty command - Add test for empty command validation - All 632 tests passing Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/schedule.py | 4 ++-- tests/unit/test_schedule.py | 5 +++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py index 69887026..26ba1715 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -105,6 +105,8 @@ def _create_launchd_plist( # Split command into program and arguments command_parts = command.split() + if not command_parts: + raise ValueError("Command cannot be empty") program = command_parts[0] args = command_parts[1:] if len(command_parts) > 1 else [] @@ -275,7 +277,6 @@ def _uninstall_systemd_timer(task_name: str, user_mode: bool = True) -> None: except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: # Timer may not exist or systemctl may not be available - continue cleanup console.print(f" [dim]•[/dim] Could not stop timer: {e}") - pass # Remove files service_file = systemd_dir / f"{service_name}.service" @@ -317,7 +318,6 @@ def _uninstall_launchd_agent(task_name: str) -> None: except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: # Agent may not be loaded or launchctl may not be available - continue cleanup console.print(f" [dim]•[/dim] Could not unload agent: {e}") - pass # Remove plist file if plist_file.exists(): diff --git a/tests/unit/test_schedule.py b/tests/unit/test_schedule.py index ea49b984..688f1374 100644 --- a/tests/unit/test_schedule.py +++ b/tests/unit/test_schedule.py @@ -126,6 +126,11 @@ def test_create_launchd_plist_logs_path(self, temp_dir: Path) -> None: assert "StandardErrorPath" in plist_content assert f"{temp_dir}/.deepwork/logs/test-task.err" in plist_content + def test_create_launchd_plist_empty_command(self, temp_dir: Path) -> None: + """Test that empty command raises ValueError.""" + with pytest.raises(ValueError, match="Command cannot be empty"): + _create_launchd_plist("test-task", "", temp_dir, 3600) + class TestScheduleAdd: """Tests for schedule add command.""" From 989720c1e2c3ab1bb586edb6b90ede236fba99e7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:29:31 +0000 Subject: [PATCH 07/10] Add security hardening and validation improvements - Add task name validation (alphanumeric, hyphens, underscores only) - Use shlex for proper command parsing with quoted arguments - Add XML escaping for launchd plist arguments - Use shell escaping for systemd commands - Add explicit error for invalid intervals - Add 7 new tests for validation and security features - All 639 tests passing Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/schedule.py | 59 ++++++++++++++++++++------ tests/unit/test_schedule.py | 81 +++++++++++++++++++++++++++++++++++- 2 files changed, 127 insertions(+), 13 deletions(-) diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py index 26ba1715..af41c437 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -1,6 +1,9 @@ """Schedule command for DeepWork CLI.""" +import html import platform +import re +import shlex import shutil import subprocess from pathlib import Path @@ -19,6 +22,23 @@ class ScheduleError(Exception): 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." + ) + + def _detect_system() -> str: """ Detect the system scheduler type. @@ -46,16 +66,20 @@ def _create_systemd_service( Create systemd service and timer content. Args: - task_name: Name of the scheduled task - command: Command to execute + 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}" + # Use sh -c to properly handle complex commands + safe_command = shlex.quote(command) + service_content = f"""[Unit] Description=DeepWork task: {task_name} After=network.target @@ -63,7 +87,7 @@ def _create_systemd_service( [Service] Type=oneshot WorkingDirectory={project_path} -ExecStart={command} +ExecStart=/bin/sh -c {safe_command} StandardOutput=journal StandardError=journal @@ -93,24 +117,29 @@ def _create_launchd_plist( Create launchd plist content. Args: - task_name: Name of the scheduled task - command: Command to execute (will be split into arguments) + 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}" - # Split command into program and arguments - command_parts = command.split() + # Parse command properly using shlex to handle quoted arguments + 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") - program = command_parts[0] - args = command_parts[1:] if len(command_parts) > 1 else [] - args_xml = "\n ".join(f"{arg}" for arg in [program] + args) + # Escape each argument for XML + escaped_args = [html.escape(arg) for arg in command_parts] + args_xml = "\n ".join(f"{arg}" for arg in escaped_args) plist_content = f""" @@ -403,14 +432,20 @@ def _add_schedule(task_name: str, command: str, interval: str, project_path: Pat # Convert interval to seconds for launchd try: interval_seconds = int(interval) - except ValueError: + except ValueError as e: # Map common intervals to seconds interval_map = { "hourly": 3600, "daily": 86400, "weekly": 604800, } - interval_seconds = interval_map.get(interval.lower(), 86400) + if interval.lower() not in interval_map: + raise ScheduleError( + f"Invalid interval '{interval}'. " + f"Valid options are: {', '.join(interval_map.keys())}, " + "or a number of seconds." + ) from e + interval_seconds = interval_map[interval.lower()] console.print("[yellow]→[/yellow] Installing launchd agent...") _install_launchd_agent(task_name, command, project_path, interval_seconds) diff --git a/tests/unit/test_schedule.py b/tests/unit/test_schedule.py index 688f1374..f720fccf 100644 --- a/tests/unit/test_schedule.py +++ b/tests/unit/test_schedule.py @@ -10,9 +10,33 @@ _create_launchd_plist, _create_systemd_service, _detect_system, + _validate_task_name, ) +class TestValidateTaskName: + """Tests for task name validation.""" + + def test_valid_task_names(self) -> None: + """Test that valid task names pass validation.""" + 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: + """Test that invalid task names raise errors.""" + invalid_names = [ + "task name", # space + "task@name", # special char + "task/name", # slash + "task.name", # dot + "task:name", # colon + ] + for name in invalid_names: + with pytest.raises(ScheduleError, match="Invalid task name"): + _validate_task_name(name) + + class TestDetectSystem: """Tests for system detection.""" @@ -62,7 +86,7 @@ def test_create_systemd_service_basic(self, temp_dir: Path) -> None: # Check service content assert "Description=DeepWork task: test-task" in service_content assert f"WorkingDirectory={temp_dir}" in service_content - assert f"ExecStart={command}" in service_content + assert "ExecStart=/bin/sh -c" in service_content assert "Type=oneshot" in service_content # Check timer content @@ -79,6 +103,11 @@ def test_create_systemd_service_with_different_intervals(self, temp_dir: Path) - ) assert f"OnCalendar={interval}" in timer_content + def test_create_systemd_service_invalid_task_name(self, temp_dir: Path) -> None: + """Test that invalid task name raises error.""" + with pytest.raises(ScheduleError, match="Invalid task name"): + _create_systemd_service("test task", "echo test", temp_dir, "daily") + class TestCreateLaunchdPlist: """Tests for launchd plist generation.""" @@ -115,6 +144,31 @@ def test_create_launchd_plist_with_complex_command(self, temp_dir: Path) -> None assert "origin" in plist_content assert "main" in plist_content + def test_create_launchd_plist_with_quoted_arguments(self, temp_dir: Path) -> None: + """Test creating launchd plist with quoted arguments.""" + task_name = "test-task" + command = 'git commit -m "My commit message"' + interval = 3600 + + plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) + + # Check that quoted argument is preserved as single argument + assert "git" in plist_content + assert "commit" in plist_content + assert "-m" in plist_content + assert "My commit message" in plist_content + + def test_create_launchd_plist_xml_escaping(self, temp_dir: Path) -> None: + """Test that XML special characters are properly escaped.""" + task_name = "test-task" + command = 'echo "test & more"' + interval = 3600 + + plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) + + # Check that XML characters are escaped + assert "<tag> & more" in plist_content + def test_create_launchd_plist_logs_path(self, temp_dir: Path) -> None: """Test that launchd plist includes log paths.""" plist_content = _create_launchd_plist( @@ -131,6 +185,11 @@ def test_create_launchd_plist_empty_command(self, temp_dir: Path) -> None: with pytest.raises(ValueError, match="Command cannot be empty"): _create_launchd_plist("test-task", "", temp_dir, 3600) + def test_create_launchd_plist_invalid_task_name(self, temp_dir: Path) -> None: + """Test that invalid task name raises error.""" + with pytest.raises(ScheduleError, match="Invalid task name"): + _create_launchd_plist("test task", "echo test", temp_dir, 3600) + class TestScheduleAdd: """Tests for schedule add command.""" @@ -199,6 +258,26 @@ def test_add_schedule_not_git_repo( with pytest.raises(ScheduleError, match="Not a Git repository"): _add_schedule("test-task", "deepwork sync", "daily", temp_dir) + @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.is_git_repo") + def test_add_schedule_invalid_interval( + self, + mock_is_git: Mock, + mock_detect: Mock, + temp_dir: Path, + ) -> None: + """Test adding schedule with invalid interval on launchd.""" + from deepwork.cli.schedule import _add_schedule + + mock_is_git.return_value = True + mock_detect.return_value = "launchd" + + # Create .deepwork directory + (temp_dir / ".deepwork").mkdir() + + with pytest.raises(ScheduleError, match="Invalid interval"): + _add_schedule("test-task", "deepwork sync", "invalid", temp_dir) + @patch("deepwork.cli.schedule._detect_system") @patch("deepwork.cli.schedule.is_git_repo") def test_add_schedule_unsupported_system( From 725aa1604ab696a0a6279015d56d8956aadc3b39 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:30:27 +0000 Subject: [PATCH 08/10] Add monthly interval support for launchd - Add monthly interval (30 days) to launchd interval map - Ensures consistency with systemd monthly interval support - All 639 tests passing Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/schedule.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py index af41c437..d03e7f39 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -438,6 +438,7 @@ def _add_schedule(task_name: str, command: str, interval: str, project_path: Pat "hourly": 3600, "daily": 86400, "weekly": 604800, + "monthly": 2592000, # 30 days } if interval.lower() not in interval_map: raise ScheduleError( From 1df99e9252e7bf1399bee4f8e0d82ffa0b099dd5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 24 Jan 2026 11:31:28 +0000 Subject: [PATCH 09/10] Clarify exception handling in uninstall operations - Remove CalledProcessError from exception handlers (not raised with check=False) - Keep FileNotFoundError and OSError for actual failure cases - Add clarifying comments about error handling - All 639 tests passing Co-authored-by: ncrmro <8276365+ncrmro@users.noreply.github.com> --- src/deepwork/cli/schedule.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/deepwork/cli/schedule.py b/src/deepwork/cli/schedule.py index d03e7f39..09c7c6f9 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -290,7 +290,7 @@ def _uninstall_systemd_timer(task_name: str, user_mode: bool = True) -> None: if user_mode: systemctl_cmd.append("--user") - # Stop and disable timer + # Stop and disable timer (ignore errors as timer may not exist) try: subprocess.run( [*systemctl_cmd, "stop", f"{service_name}.timer"], @@ -303,8 +303,8 @@ def _uninstall_systemd_timer(task_name: str, user_mode: bool = True) -> None: capture_output=True, ) console.print(f" [green]✓[/green] Stopped and disabled {service_name}.timer") - except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: - # Timer may not exist or systemctl may not be available - continue cleanup + except (FileNotFoundError, OSError) as e: + # systemctl may not be available or file system errors - continue cleanup console.print(f" [dim]•[/dim] Could not stop timer: {e}") # Remove files @@ -340,12 +340,12 @@ def _uninstall_launchd_agent(task_name: str) -> None: launch_agents_dir = Path.home() / "Library" / "LaunchAgents" plist_file = launch_agents_dir / f"{label}.plist" - # Unload the agent + # Unload the agent (ignore errors as agent may not be loaded) try: subprocess.run(["launchctl", "unload", str(plist_file)], check=False, capture_output=True) console.print(f" [green]✓[/green] Unloaded {label}") - except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: - # Agent may not be loaded or launchctl may not be available - continue cleanup + except (FileNotFoundError, OSError) as e: + # launchctl may not be available or file system errors - continue cleanup console.print(f" [dim]•[/dim] Could not unload agent: {e}") # Remove plist file From 335fb446feecabaadb66a33ddf8bb180b7e031fd Mon Sep 17 00:00:00 2001 From: Nicholas Romero Date: Fri, 30 Jan 2026 06:47:26 -0600 Subject: [PATCH 10/10] Refactor schedule.py to use ScheduleRunner abstraction Extract platform-specific scheduling logic into a runner registry pattern (matching AgentAdapter), making it straightforward to add future runners like GitHub Actions or Kubernetes jobs. Co-Authored-By: Claude Opus 4.5 --- doc/architecture.md | 4 +- src/deepwork/cli/runners.py | 455 +++++++++++++++++ src/deepwork/cli/schedule.py | 461 +----------------- .../integration/test_schedule_integration.py | 61 ++- tests/unit/test_runners.py | 200 ++++++++ tests/unit/test_schedule.py | 275 ++--------- 6 files changed, 734 insertions(+), 722 deletions(-) create mode 100644 src/deepwork/cli/runners.py create mode 100644 tests/unit/test_runners.py 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/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 index 09c7c6f9..77a005ae 100644 --- a/src/deepwork/cli/schedule.py +++ b/src/deepwork/cli/schedule.py @@ -1,359 +1,16 @@ """Schedule command for DeepWork CLI.""" -import html -import platform -import re -import shlex -import shutil -import subprocess 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() -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." - ) - - -def _detect_system() -> str: - """ - Detect the system scheduler type. - - Returns: - 'systemd' for Linux with systemd, 'launchd' for macOS, or 'unsupported' - """ - system = platform.system() - - if system == "Darwin": - return "launchd" - elif system == "Linux": - # Check if systemd is available - if shutil.which("systemctl"): - return "systemd" - return "unsupported" - else: - return "unsupported" - - -def _create_systemd_service( - task_name: str, command: str, project_path: Path, interval: str -) -> tuple[str, str]: - """ - Create systemd service and timer content. - - 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}" - - # Use sh -c to properly handle complex commands - 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 _create_launchd_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}" - - # Parse command properly using shlex to handle quoted arguments - 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") - - # Escape each argument for XML - 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_systemd_timer( - task_name: str, command: str, project_path: Path, interval: str, user_mode: bool = True -) -> None: - """ - Install systemd service and timer. - - Args: - task_name: Name of the scheduled task - command: Command to execute - project_path: Path to the project directory - interval: Systemd timer interval - user_mode: If True, install as user service, else system service - - Raises: - ScheduleError: If installation fails - """ - service_name = f"deepwork-{task_name}" - service_content, timer_content = _create_systemd_service( - task_name, command, project_path, interval - ) - - # Determine systemd directory - 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) - - # Write service file - service_file = systemd_dir / f"{service_name}.service" - service_file.write_text(service_content) - console.print(f" [green]✓[/green] Created {service_file}") - - # Write timer file - timer_file = systemd_dir / f"{service_name}.timer" - timer_file.write_text(timer_content) - console.print(f" [green]✓[/green] Created {timer_file}") - - # Reload systemd and enable timer - 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 _install_launchd_agent( - task_name: str, command: str, project_path: Path, interval: int -) -> None: - """ - Install launchd agent. - - Args: - task_name: Name of the scheduled task - command: Command to execute - project_path: Path to the project directory - interval: Interval in seconds - - Raises: - ScheduleError: If installation fails - """ - label = f"com.deepwork.{task_name}" - plist_content = _create_launchd_plist(task_name, command, project_path, interval) - - # LaunchAgents directory - launch_agents_dir = Path.home() / "Library" / "LaunchAgents" - launch_agents_dir.mkdir(parents=True, exist_ok=True) - - # Write plist file - plist_file = launch_agents_dir / f"{label}.plist" - plist_file.write_text(plist_content) - console.print(f" [green]✓[/green] Created {plist_file}") - - # Ensure logs directory exists - logs_dir = project_path / ".deepwork" / "logs" - logs_dir.mkdir(parents=True, exist_ok=True) - - # Load the agent - 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_systemd_timer(task_name: str, user_mode: bool = True) -> None: - """ - Uninstall systemd service and timer. - - Args: - task_name: Name of the scheduled task - user_mode: If True, uninstall from user services, else system services - - Raises: - ScheduleError: If uninstallation fails - """ - service_name = f"deepwork-{task_name}" - - # Determine systemd directory - 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") - - # Stop and disable timer (ignore errors as timer may not exist) - 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: - # systemctl may not be available or file system errors - continue cleanup - console.print(f" [dim]•[/dim] Could not stop timer: {e}") - - # Remove files - 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}") - - # Reload systemd - try: - subprocess.run([*systemctl_cmd, "daemon-reload"], check=True, capture_output=True) - except subprocess.CalledProcessError: - pass - - -def _uninstall_launchd_agent(task_name: str) -> None: - """ - Uninstall launchd agent. - - Args: - task_name: Name of the scheduled task - - Raises: - ScheduleError: If uninstallation fails - """ - label = f"com.deepwork.{task_name}" - launch_agents_dir = Path.home() / "Library" / "LaunchAgents" - plist_file = launch_agents_dir / f"{label}.plist" - - # Unload the agent (ignore errors as agent may not be loaded) - 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: - # launchctl may not be available or file system errors - continue cleanup - console.print(f" [dim]•[/dim] Could not unload agent: {e}") - - # Remove plist file - if plist_file.exists(): - plist_file.unlink() - console.print(f" [green]✓[/green] Removed {plist_file}") - - @click.group() def schedule() -> None: """Manage scheduled DeepWork jobs.""" @@ -376,8 +33,7 @@ def schedule() -> None: 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. + """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'). @@ -390,8 +46,7 @@ def add(task_name: str, command: str, interval: str, path: Path) -> None: def _add_schedule(task_name: str, command: str, interval: str, project_path: Path) -> None: - """ - Add a scheduled task. + """Add a scheduled task. Args: task_name: Name of the scheduled task @@ -404,7 +59,6 @@ def _add_schedule(task_name: str, command: str, interval: str, project_path: Pat """ console.print(f"\n[bold cyan]Scheduling DeepWork task: {task_name}[/bold cyan]\n") - # Check if project has DeepWork installed (optional, just check if it's a git repo) if not is_git_repo(project_path): raise ScheduleError("Not a Git repository. DeepWork requires a Git repository.") @@ -415,41 +69,15 @@ def _add_schedule(task_name: str, command: str, interval: str, project_path: Pat "The scheduled command may not work correctly." ) - # Detect system - system_type = _detect_system() - console.print(f"[yellow]→[/yellow] Detected system: {system_type}") - - if system_type == "unsupported": + runner = ScheduleRunner.detect_runner() + if runner is None: raise ScheduleError( "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." ) - # Install based on system type - if system_type == "systemd": - console.print("[yellow]→[/yellow] Installing systemd timer...") - _install_systemd_timer(task_name, command, project_path, interval) - elif system_type == "launchd": - # Convert interval to seconds for launchd - try: - interval_seconds = int(interval) - except ValueError as e: - # Map common intervals to seconds - interval_map = { - "hourly": 3600, - "daily": 86400, - "weekly": 604800, - "monthly": 2592000, # 30 days - } - if interval.lower() not in interval_map: - raise ScheduleError( - f"Invalid interval '{interval}'. " - f"Valid options are: {', '.join(interval_map.keys())}, " - "or a number of seconds." - ) from e - interval_seconds = interval_map[interval.lower()] - - console.print("[yellow]→[/yellow] Installing launchd agent...") - _install_launchd_agent(task_name, command, project_path, interval_seconds) + 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]") @@ -465,8 +93,7 @@ def _add_schedule(task_name: str, command: str, interval: str, project_path: Pat help="Path to project directory (default: current directory)", ) def remove(task_name: str, path: Path) -> None: - """ - Remove a scheduled task. + """Remove a scheduled task. TASK_NAME is the name of the task to unschedule. """ @@ -478,8 +105,7 @@ def remove(task_name: str, path: Path) -> None: def _remove_schedule(task_name: str, project_path: Path) -> None: - """ - Remove a scheduled task. + """Remove a scheduled task. Args: task_name: Name of the scheduled task to unschedule @@ -490,22 +116,15 @@ def _remove_schedule(task_name: str, project_path: Path) -> None: """ console.print(f"\n[bold cyan]Removing scheduled task: {task_name}[/bold cyan]\n") - # Detect system - system_type = _detect_system() - console.print(f"[yellow]→[/yellow] Detected system: {system_type}") - - if system_type == "unsupported": + runner = ScheduleRunner.detect_runner() + if runner is None: raise ScheduleError( "Unsupported system. DeepWork scheduling requires systemd (Linux) or launchd (macOS)." ) - # Uninstall based on system type - if system_type == "systemd": - console.print("[yellow]→[/yellow] Removing systemd timer...") - _uninstall_systemd_timer(task_name) - elif system_type == "launchd": - console.print("[yellow]→[/yellow] Removing launchd agent...") - _uninstall_launchd_agent(task_name) + 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]") @@ -517,52 +136,10 @@ def list() -> None: """List all scheduled DeepWork tasks.""" console.print("\n[bold cyan]Scheduled DeepWork Tasks[/bold cyan]\n") - system_type = _detect_system() - console.print(f"System: {system_type}\n") - - if system_type == "systemd": - _list_systemd_timers() - elif system_type == "launchd": - _list_launchd_agents() - else: + runner = ScheduleRunner.detect_runner() + if runner is None: console.print("[yellow]No scheduled tasks (unsupported system)[/yellow]") - - -def _list_systemd_timers() -> None: - """List systemd timers for DeepWork tasks.""" - 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]") - - -def _list_launchd_agents() -> None: - """List launchd agents for DeepWork tasks.""" - 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 = list(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]") + 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 index ad58e01d..cf0c7fd0 100644 --- a/tests/integration/test_schedule_integration.py +++ b/tests/integration/test_schedule_integration.py @@ -12,21 +12,21 @@ class TestScheduleCommand: """Integration tests for 'deepwork schedule' command.""" - @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") @patch("deepwork.cli.schedule.is_git_repo") - @patch("deepwork.cli.schedule._install_systemd_timer") def test_schedule_add_systemd( self, - mock_install: Mock, 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 - mock_detect.return_value = "systemd" - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() runner = CliRunner() @@ -50,25 +50,25 @@ def test_schedule_add_systemd( assert "Detected system: systemd" in result.output assert "scheduled successfully" in result.output - mock_install.assert_called_once_with( + mock_runner.install.assert_called_once_with( "test-task", "deepwork sync", temp_dir, "daily" ) - @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") @patch("deepwork.cli.schedule.is_git_repo") - @patch("deepwork.cli.schedule._install_launchd_agent") def test_schedule_add_launchd( self, - mock_install: Mock, 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 - mock_detect.return_value = "launchd" - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() runner = CliRunner() @@ -92,18 +92,17 @@ def test_schedule_add_launchd( assert "Detected system: launchd" in result.output assert "scheduled successfully" in result.output - # hourly should be converted to 3600 seconds - mock_install.assert_called_once_with( - "test-task", "deepwork sync", temp_dir, 3600 + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "hourly" ) - @patch("deepwork.cli.schedule._detect_system") - @patch("deepwork.cli.schedule._uninstall_systemd_timer") - def test_schedule_remove( - self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path - ) -> None: + @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_detect.return_value = "systemd" + mock_runner = Mock() + mock_runner.name = "systemd" + mock_runner.display_name = "Systemd Timer" + mock_detect.return_value = mock_runner runner = CliRunner() result = runner.invoke( @@ -116,15 +115,14 @@ def test_schedule_remove( assert "Removing scheduled task: test-task" in result.output assert "unscheduled successfully" in result.output - mock_uninstall.assert_called_once_with("test-task") + mock_runner.uninstall.assert_called_once_with("test-task") - @patch("deepwork.cli.schedule._detect_system") - @patch("deepwork.cli.schedule._list_systemd_timers") - def test_schedule_list( - self, mock_list: Mock, mock_detect: Mock - ) -> None: + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") + def test_schedule_list(self, mock_detect: Mock) -> None: """Test listing scheduled tasks.""" - mock_detect.return_value = "systemd" + mock_runner = Mock() + mock_runner.name = "systemd" + mock_detect.return_value = mock_runner runner = CliRunner() result = runner.invoke(cli, ["schedule", "list"], catch_exceptions=False) @@ -133,9 +131,9 @@ def test_schedule_list( assert "Scheduled DeepWork Tasks" in result.output assert "System: systemd" in result.output - mock_list.assert_called_once() + mock_runner.list_tasks.assert_called_once() - @patch("deepwork.cli.schedule._detect_system") + @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 @@ -152,16 +150,15 @@ def test_schedule_add_not_git_repo( assert result.exit_code == 1 assert "Not a Git repository" in result.output - @patch("deepwork.cli.schedule._detect_system") + @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 = "unsupported" + mock_detect.return_value = None - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() runner = CliRunner() 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 index f720fccf..8c3a4c13 100644 --- a/tests/unit/test_schedule.py +++ b/tests/unit/test_schedule.py @@ -1,205 +1,20 @@ -"""Tests for schedule command.""" +"""Tests for schedule command (CLI layer).""" from pathlib import Path from unittest.mock import Mock, patch import pytest -from deepwork.cli.schedule import ( - ScheduleError, - _create_launchd_plist, - _create_systemd_service, - _detect_system, - _validate_task_name, -) - - -class TestValidateTaskName: - """Tests for task name validation.""" - - def test_valid_task_names(self) -> None: - """Test that valid task names pass validation.""" - 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: - """Test that invalid task names raise errors.""" - invalid_names = [ - "task name", # space - "task@name", # special char - "task/name", # slash - "task.name", # dot - "task:name", # colon - ] - for name in invalid_names: - with pytest.raises(ScheduleError, match="Invalid task name"): - _validate_task_name(name) - - -class TestDetectSystem: - """Tests for system detection.""" - - @patch("platform.system") - @patch("shutil.which") - def test_detect_darwin(self, mock_which: Mock, mock_system: Mock) -> None: - """Test detecting macOS.""" - mock_system.return_value = "Darwin" - assert _detect_system() == "launchd" - - @patch("platform.system") - @patch("shutil.which") - def test_detect_linux_with_systemd(self, mock_which: Mock, mock_system: Mock) -> None: - """Test detecting Linux with systemd.""" - mock_system.return_value = "Linux" - mock_which.return_value = "/usr/bin/systemctl" - assert _detect_system() == "systemd" - - @patch("platform.system") - @patch("shutil.which") - def test_detect_linux_without_systemd(self, mock_which: Mock, mock_system: Mock) -> None: - """Test detecting Linux without systemd.""" - mock_system.return_value = "Linux" - mock_which.return_value = None - assert _detect_system() == "unsupported" - - @patch("platform.system") - def test_detect_unsupported_system(self, mock_system: Mock) -> None: - """Test detecting unsupported system.""" - mock_system.return_value = "Windows" - assert _detect_system() == "unsupported" - - -class TestCreateSystemdService: - """Tests for systemd service and timer generation.""" - - def test_create_systemd_service_basic(self, temp_dir: Path) -> None: - """Test creating basic systemd service and timer.""" - task_name = "test-task" - command = "deepwork sync" - interval = "daily" - - service_content, timer_content = _create_systemd_service( - task_name, command, temp_dir, interval - ) - - # Check service content - assert "Description=DeepWork task: test-task" in service_content - assert f"WorkingDirectory={temp_dir}" in service_content - assert "ExecStart=/bin/sh -c" in service_content - assert "Type=oneshot" in service_content - - # Check timer content - assert "Description=Timer for DeepWork task: test-task" in timer_content - assert "OnCalendar=daily" in timer_content - assert "Persistent=true" in timer_content - assert "Requires=deepwork-test-task.service" in timer_content - - def test_create_systemd_service_with_different_intervals(self, temp_dir: Path) -> None: - """Test creating systemd timer with different intervals.""" - for interval in ["hourly", "weekly", "monthly"]: - _, timer_content = _create_systemd_service( - "test-task", "echo test", temp_dir, interval - ) - assert f"OnCalendar={interval}" in timer_content - - def test_create_systemd_service_invalid_task_name(self, temp_dir: Path) -> None: - """Test that invalid task name raises error.""" - with pytest.raises(ScheduleError, match="Invalid task name"): - _create_systemd_service("test task", "echo test", temp_dir, "daily") - - -class TestCreateLaunchdPlist: - """Tests for launchd plist generation.""" - - def test_create_launchd_plist_basic(self, temp_dir: Path) -> None: - """Test creating basic launchd plist.""" - task_name = "test-task" - command = "deepwork sync" - interval = 86400 # daily in seconds - - plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) - - # Check plist content - assert "Label" in plist_content - assert "com.deepwork.test-task" in plist_content - assert "ProgramArguments" in plist_content - assert "deepwork" in plist_content - assert "sync" in plist_content - assert f"{temp_dir}" in plist_content - assert "StartInterval" in plist_content - assert "86400" in plist_content - - def test_create_launchd_plist_with_complex_command(self, temp_dir: Path) -> None: - """Test creating launchd plist with multi-part command.""" - task_name = "test-task" - command = "git fetch origin main" - interval = 3600 - - plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) - - # Check that command is split correctly - assert "git" in plist_content - assert "fetch" in plist_content - assert "origin" in plist_content - assert "main" in plist_content - - def test_create_launchd_plist_with_quoted_arguments(self, temp_dir: Path) -> None: - """Test creating launchd plist with quoted arguments.""" - task_name = "test-task" - command = 'git commit -m "My commit message"' - interval = 3600 - - plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) - - # Check that quoted argument is preserved as single argument - assert "git" in plist_content - assert "commit" in plist_content - assert "-m" in plist_content - assert "My commit message" in plist_content - - def test_create_launchd_plist_xml_escaping(self, temp_dir: Path) -> None: - """Test that XML special characters are properly escaped.""" - task_name = "test-task" - command = 'echo "test & more"' - interval = 3600 - - plist_content = _create_launchd_plist(task_name, command, temp_dir, interval) - - # Check that XML characters are escaped - assert "<tag> & more" in plist_content - - def test_create_launchd_plist_logs_path(self, temp_dir: Path) -> None: - """Test that launchd plist includes log paths.""" - plist_content = _create_launchd_plist( - "test-task", "echo test", temp_dir, 3600 - ) - - assert "StandardOutPath" in plist_content - assert f"{temp_dir}/.deepwork/logs/test-task.log" in plist_content - assert "StandardErrorPath" in plist_content - assert f"{temp_dir}/.deepwork/logs/test-task.err" in plist_content - - def test_create_launchd_plist_empty_command(self, temp_dir: Path) -> None: - """Test that empty command raises ValueError.""" - with pytest.raises(ValueError, match="Command cannot be empty"): - _create_launchd_plist("test-task", "", temp_dir, 3600) - - def test_create_launchd_plist_invalid_task_name(self, temp_dir: Path) -> None: - """Test that invalid task name raises error.""" - with pytest.raises(ScheduleError, match="Invalid task name"): - _create_launchd_plist("test task", "echo test", temp_dir, 3600) +from deepwork.cli.runners import ScheduleError class TestScheduleAdd: """Tests for schedule add command.""" - @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") @patch("deepwork.cli.schedule.is_git_repo") - @patch("deepwork.cli.schedule._install_systemd_timer") def test_add_schedule_systemd( self, - mock_install: Mock, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path, @@ -207,24 +22,24 @@ def test_add_schedule_systemd( """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 - mock_detect.return_value = "systemd" - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() _add_schedule("test-task", "deepwork sync", "daily", temp_dir) - mock_install.assert_called_once_with( + mock_runner.install.assert_called_once_with( "test-task", "deepwork sync", temp_dir, "daily" ) - @patch("deepwork.cli.schedule._detect_system") + @patch("deepwork.cli.schedule.ScheduleRunner.detect_runner") @patch("deepwork.cli.schedule.is_git_repo") - @patch("deepwork.cli.schedule._install_launchd_agent") def test_add_schedule_launchd( self, - mock_install: Mock, mock_is_git: Mock, mock_detect: Mock, temp_dir: Path, @@ -232,20 +47,21 @@ def test_add_schedule_launchd( """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 - mock_detect.return_value = "launchd" - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() _add_schedule("test-task", "deepwork sync", "daily", temp_dir) - # Daily should be converted to 86400 seconds - mock_install.assert_called_once_with( - "test-task", "deepwork sync", temp_dir, 86400 + mock_runner.install.assert_called_once_with( + "test-task", "deepwork sync", temp_dir, "daily" ) - @patch("deepwork.cli.schedule._detect_system") + @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 @@ -258,27 +74,7 @@ def test_add_schedule_not_git_repo( with pytest.raises(ScheduleError, match="Not a Git repository"): _add_schedule("test-task", "deepwork sync", "daily", temp_dir) - @patch("deepwork.cli.schedule._detect_system") - @patch("deepwork.cli.schedule.is_git_repo") - def test_add_schedule_invalid_interval( - self, - mock_is_git: Mock, - mock_detect: Mock, - temp_dir: Path, - ) -> None: - """Test adding schedule with invalid interval on launchd.""" - from deepwork.cli.schedule import _add_schedule - - mock_is_git.return_value = True - mock_detect.return_value = "launchd" - - # Create .deepwork directory - (temp_dir / ".deepwork").mkdir() - - with pytest.raises(ScheduleError, match="Invalid interval"): - _add_schedule("test-task", "deepwork sync", "invalid", temp_dir) - - @patch("deepwork.cli.schedule._detect_system") + @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 @@ -287,9 +83,8 @@ def test_add_schedule_unsupported_system( from deepwork.cli.schedule import _add_schedule mock_is_git.return_value = True - mock_detect.return_value = "unsupported" + mock_detect.return_value = None - # Create .deepwork directory (temp_dir / ".deepwork").mkdir() with pytest.raises(ScheduleError, match="Unsupported system"): @@ -299,42 +94,28 @@ def test_add_schedule_unsupported_system( class TestScheduleRemove: """Tests for schedule remove command.""" - @patch("deepwork.cli.schedule._detect_system") - @patch("deepwork.cli.schedule._uninstall_systemd_timer") - def test_remove_schedule_systemd( - self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path - ) -> None: - """Test removing a schedule on systemd.""" - from deepwork.cli.schedule import _remove_schedule - - mock_detect.return_value = "systemd" - - _remove_schedule("test-task", temp_dir) - - mock_uninstall.assert_called_once_with("test-task") - - @patch("deepwork.cli.schedule._detect_system") - @patch("deepwork.cli.schedule._uninstall_launchd_agent") - def test_remove_schedule_launchd( - self, mock_uninstall: Mock, mock_detect: Mock, temp_dir: Path - ) -> None: - """Test removing a schedule on launchd.""" + @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_detect.return_value = "launchd" + 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_uninstall.assert_called_once_with("test-task") + mock_runner.uninstall.assert_called_once_with("test-task") - @patch("deepwork.cli.schedule._detect_system") + @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 = "unsupported" + mock_detect.return_value = None with pytest.raises(ScheduleError, match="Unsupported system"): _remove_schedule("test-task", temp_dir)