diff --git a/src/claude_task_master/cli_commands/config.py b/src/claude_task_master/cli_commands/config.py index c3a67c6c..f4489014 100644 --- a/src/claude_task_master/cli_commands/config.py +++ b/src/claude_task_master/cli_commands/config.py @@ -6,6 +6,11 @@ - path: Show path to config file """ +from __future__ import annotations + +import json +from typing import Any + import typer from rich.console import Console from rich.markdown import Markdown @@ -68,22 +73,30 @@ def config_init( def config_show( raw: bool = typer.Option(False, "--raw", "-r", help="Show raw JSON without formatting"), env: bool = typer.Option(False, "--env", "-e", help="Show environment variable overrides"), + show_secrets: bool = typer.Option( + False, + "--show-secrets", + help="Reveal API keys and other secrets (default: masked)", + ), ) -> None: """πŸ“– Display current configuration. Shows the active configuration including any environment variable overrides. + Secrets (API keys) are masked unless --show-secrets is passed β€” this applies + to --raw output too, so piped JSON never leaks credentials by default. Use --raw for machine-readable JSON output. Examples: claudetm config show claudetm config show --raw claudetm config show --env + claudetm config show --show-secrets """ if env: - _display_env_overrides() + _display_env_overrides(show_secrets=show_secrets) return - _display_config(raw=raw) + _display_config(raw=raw, show_secrets=show_secrets) @config_app.command(name="path") @@ -114,15 +127,73 @@ def config_path( console.print(str(path)) -def _display_config(raw: bool = False) -> None: +# Field/env-var names whose values are treated as secrets and masked in output. +_SECRET_NAME_HINTS = ("key", "secret", "token", "password") + + +def _looks_secret(name: str) -> bool: + """Return True if a field/env-var name looks like it holds a secret. + + Args: + name: The config field name or environment variable name. + + Returns: + True if the name matches a known secret hint (key/secret/token/password). + """ + lowered = name.lower() + return any(hint in lowered for hint in _SECRET_NAME_HINTS) + + +def _mask_secret(value: str) -> str: + """Mask a secret value, keeping a short prefix as an identification hint. + + Args: + value: The secret value to mask. + + Returns: + A masked placeholder; short values collapse to ``***``. + """ + return f"{value[:8]}..." if len(value) > 8 else "***" + + +def _redact_secrets(data: Any) -> Any: + """Recursively copy config data, masking any leaf whose key looks secret. + + Only non-empty string values are masked; ``None``/unset fields are left as-is + so the output still shows which secrets are configured versus absent. + + Args: + data: A config value (dict, list, or scalar) from ``model_dump()``. + + Returns: + A new structure with secret leaves replaced by masked placeholders. + """ + if isinstance(data, dict): + redacted: dict[str, Any] = {} + for key, value in data.items(): + if _looks_secret(key) and isinstance(value, str) and value: + redacted[key] = _mask_secret(value) + else: + redacted[key] = _redact_secrets(value) + return redacted + if isinstance(data, list): + return [_redact_secrets(item) for item in data] + return data + + +def _display_config(raw: bool = False, show_secrets: bool = False) -> None: """Display the current configuration. Args: raw: If True, output raw JSON without formatting. + show_secrets: If True, reveal secret values instead of masking them. """ try: config = get_config() - config_json = config.model_dump_json(indent=2) + config_dict = config.model_dump() + display_dict = config_dict if show_secrets else _redact_secrets(config_dict) + secrets_masked = display_dict != config_dict + config_json = json.dumps(display_dict, indent=2) if raw: print(config_json) @@ -141,6 +212,11 @@ def _display_config(raw: bool = False) -> None: syntax = Syntax(config_json, "json", theme="monokai", line_numbers=False) console.print(syntax) + if secrets_masked: + console.print( + "\n[dim]πŸ”’ Secrets masked. Use '--show-secrets' to reveal them.[/dim]" + ) + # Show env var overrides hint overrides = get_env_overrides() if overrides: @@ -154,8 +230,12 @@ def _display_config(raw: bool = False) -> None: raise typer.Exit(1) from None -def _display_env_overrides() -> None: - """Display active environment variable overrides.""" +def _display_env_overrides(show_secrets: bool = False) -> None: + """Display active environment variable overrides. + + Args: + show_secrets: If True, reveal secret values instead of masking them. + """ overrides = get_env_overrides() console.print("[bold blue]πŸ”§ Environment Variable Overrides[/bold blue]\n") @@ -180,10 +260,9 @@ def _display_env_overrides() -> None: console.print("[bold]Active overrides:[/bold]\n") for env_var, value in overrides.items(): - # Mask sensitive values - if "key" in env_var.lower() or "secret" in env_var.lower(): - masked = value[:8] + "..." if len(value) > 8 else "***" - console.print(f" [cyan]{env_var}[/cyan] = [dim]{masked}[/dim]") + # Mask sensitive values unless explicitly revealed + if not show_secrets and _looks_secret(env_var): + console.print(f" [cyan]{env_var}[/cyan] = [dim]{_mask_secret(value)}[/dim]") else: console.print(f" [cyan]{env_var}[/cyan] = {value}") diff --git a/src/claude_task_master/cli_commands/info.py b/src/claude_task_master/cli_commands/info.py index 8821af85..3f312eff 100644 --- a/src/claude_task_master/cli_commands/info.py +++ b/src/claude_task_master/cli_commands/info.py @@ -4,6 +4,7 @@ from rich.console import Console from rich.markdown import Markdown +from ..core.agent_models import get_tools_for_phase from ..core.state import StateManager console = Console() @@ -35,13 +36,16 @@ def status() -> None: console.print(f"[cyan]Sessions:[/cyan] {state.session_count}") console.print(f"[cyan]Run ID:[/cyan] {state.run_id}") - # Show tools based on current phase/status + # Show tools based on current phase/status. Derive from config so the + # display stays in sync with the actual restricted phase defaults. if state.status == "planning": - console.print("[cyan]Tools:[/cyan] Read, Glob, Grep, Bash (read-only mode)") + planning_tools = get_tools_for_phase("planning") + tools_display = ", ".join(planning_tools) if planning_tools else "All" + console.print(f"[cyan]Tools:[/cyan] {tools_display} (read-only mode)") elif state.status == "working": console.print("[cyan]Tools:[/cyan] All (bypassPermissions mode)") else: - # For blocked, paused, success, failed - show what was last used + # For blocked, paused, stopped, success, failed - show what was last used console.print("[cyan]Tools:[/cyan] All (bypassPermissions mode)") if state.current_pr: diff --git a/src/claude_task_master/cli_commands/mailbox.py b/src/claude_task_master/cli_commands/mailbox.py index 5c2eb5ca..f9ed596f 100644 --- a/src/claude_task_master/cli_commands/mailbox.py +++ b/src/claude_task_master/cli_commands/mailbox.py @@ -241,49 +241,9 @@ def mailbox_callback(ctx: typer.Context) -> None: mailbox_status() -@mailbox_app.command("send") -def mailbox_send_command( - message: Annotated[str, typer.Argument(help="Message content to send")], - sender: Annotated[ - str, - typer.Option("--sender", "-s", help="Sender identifier"), - ] = "cli", - priority: Annotated[ - int, - typer.Option("--priority", "-p", help="Priority level (0=low, 1=normal, 2=high, 3=urgent)"), - ] = 1, -) -> None: - """Send a message to the mailbox. - - Adds a new message that will be processed after the current task completes. - The orchestrator checks the mailbox after each task and updates the plan - if messages are present. - - Examples: - claudetm mailbox send "Please also update the README" - claudetm mailbox send "Fix the auth bug first" --priority 3 - claudetm mailbox send "Low priority cleanup" -p 0 -s "supervisor" - """ - mailbox_send(message, sender, priority) - - -@mailbox_app.command("clear") -def mailbox_clear_command( - force: Annotated[ - bool, - typer.Option("--force", "-f", help="Skip confirmation"), - ] = False, -) -> None: - """Clear all messages from the mailbox. - - Removes all pending messages. This is useful to cancel pending plan updates - or start fresh. - - Examples: - claudetm mailbox clear - claudetm mailbox clear -f - """ - mailbox_clear(force) +# Register the original command functions with the mailbox_app +mailbox_app.command("send")(mailbox_send) +mailbox_app.command("clear")(mailbox_clear) def register_mailbox_commands(app: typer.Typer) -> None: diff --git a/src/claude_task_master/cli_commands/profile.py b/src/claude_task_master/cli_commands/profile.py index c48b09b2..e94ca7f9 100644 --- a/src/claude_task_master/cli_commands/profile.py +++ b/src/claude_task_master/cli_commands/profile.py @@ -177,7 +177,7 @@ def _print_profile(profile: Profile, active: bool) -> None: @profile_app.command(name="remove") def profile_remove( name: str = typer.Argument(..., help="Profile name to remove"), - force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation"), + force: bool = typer.Option(False, "--force", "-f", help="Allow removing active profile"), ) -> None: """πŸ—‘οΈ Remove a profile (its config dir is left on disk).""" manager = ProfileManager() @@ -185,7 +185,7 @@ def profile_remove( console.print("[yellow]Cancelled[/yellow]") raise typer.Exit(0) try: - manager.remove(name) + manager.remove(name, force=force) except ProfileError as e: console.print(f"[red]{e}[/red]") raise typer.Exit(1) from None diff --git a/src/claude_task_master/cli_commands/workflow.py b/src/claude_task_master/cli_commands/workflow.py index aae80ae1..2aa82adc 100644 --- a/src/claude_task_master/cli_commands/workflow.py +++ b/src/claude_task_master/cli_commands/workflow.py @@ -118,8 +118,48 @@ def _validate_log_options(log_level: str, log_format: str) -> tuple[LogLevel, Lo return log_level_enum, log_format_enum +def _validate_goal(value: str) -> str: + """Reject an empty or whitespace-only goal at parse time. + + Args: + value: The goal argument as parsed from the CLI. + + Returns: + The original value if non-empty. + + Raises: + typer.BadParameter: If the goal is empty or only whitespace. + """ + if not value.strip(): + raise typer.BadParameter("goal must not be empty") + return value + + +def _validate_budget(value: float | None) -> float | None: + """Reject a non-positive per-session budget at parse time. + + A zero or negative budget would otherwise block the run on the first token. + + Args: + value: The budget in USD, or None when unset. + + Returns: + The original value if positive or None. + + Raises: + typer.BadParameter: If the budget is zero or negative. + """ + if value is not None and value <= 0: + raise typer.BadParameter("must be greater than 0 (USD per session)") + return value + + def start( - goal: str = typer.Argument(..., help="The goal to achieve (e.g., 'Add user authentication')"), + goal: str = typer.Argument( + ..., + callback=_validate_goal, + help="The goal to achieve (e.g., 'Add user authentication')", + ), model: str = typer.Option( "opus", "--model", @@ -150,11 +190,13 @@ def start( None, "--max-sessions", "-n", + min=1, help="Max work sessions before pausing (default: unlimited)", ), max_prs: int | None = typer.Option( None, "--prs", + min=1, help="Max pull requests to create (default: unlimited)", ), pause_on_pr: bool = typer.Option( @@ -206,7 +248,8 @@ def start( None, "--budget", envvar="CLAUDETM_BUDGET", - help="Max spending per session in USD (env: CLAUDETM_BUDGET)", + callback=_validate_budget, + help="Max spending per session in USD, must be > 0 (env: CLAUDETM_BUDGET)", ), ) -> None: """Start a new task with the given goal. @@ -253,6 +296,9 @@ def start( console.print(f"[red]Invalid webhook configuration: {e}[/red]") raise typer.Exit(1) from None + # Acquired once the state dir is known (after the exists() check) and always + # released in the finally, even if setup fails before the lock is taken. + state_manager: StateManager | None = None try: # Initialize configuration (creates config.json with defaults if missing) working_dir = Path.cwd() @@ -269,6 +315,18 @@ def start( ) raise typer.Exit(1) + # Guard against concurrent runs: acquire the single-instance session lock + # before loading credentials or touching shared state. Two simultaneous + # `claudetm start` runs would otherwise corrupt state.json, duplicate PRs, + # and race OAuth refresh-token rotation. Held for the whole run and freed + # by the finally. (Pairs with the O_EXCL PID lock in core/state.py.) + if not state_manager.acquire_session_lock(): + console.print("[red]Error: Another claudetm session is active for this project.[/red]") + console.print( + "[dim]Wait for it to finish, or run 'claudetm clean -f' to force cleanup.[/dim]" + ) + raise typer.Exit(1) + # Load credentials console.print("Loading credentials...") cred_manager = CredentialManager() @@ -362,6 +420,12 @@ def start( except Exception as e: console.print(f"[red]Error: {e}[/red]") raise typer.Exit(1) from None + finally: + # Release the session lock on every exit path (success, pause, block, + # error). A no-op if we never acquired it or the orchestrator's + # cleanup_on_success already released it. + if state_manager is not None: + state_manager.release_session_lock() def resume( @@ -406,6 +470,9 @@ def resume( """ console.print("[bold blue]Resuming task...[/bold blue]") + # Acquired after validate_for_resume and always released in the finally, + # even if setup fails before the lock is taken. + state_manager: StateManager | None = None try: # Initialize configuration (loads existing config.json or uses defaults) working_dir = Path.cwd() @@ -456,6 +523,19 @@ def resume( console.print(f"[dim]{e.details}[/dim]") raise typer.Exit(1) from None + # Guard against concurrent runs: acquire the single-instance session lock + # now that the task is known resumable. Placed after validation so a no-op + # resume of an already-finished task exits cleanly without taking the lock. + # Two concurrent resumes would otherwise both drive the work loop β€” + # corrupting state, duplicating PRs, racing OAuth refresh-token rotation. + # Held for the whole run and freed by the finally. + if not state_manager.acquire_session_lock(): + console.print("[red]Error: Another claudetm session is active for this project.[/red]") + console.print( + "[dim]Wait for it to finish, or run 'claudetm clean -f' to force cleanup.[/dim]" + ) + raise typer.Exit(1) + # Toggle admin force-merge. It is persisted so the retry loop (each blocked-merge # cycle re-enters the merge stage and re-reads state) keeps the override active, but # because it applies to *every* later PR in the task, --no-admin is offered as an @@ -579,6 +659,12 @@ def resume( except Exception as e: console.print(f"[red]Error: {e}[/red]") raise typer.Exit(1) from None + finally: + # Release the session lock on every exit path. A no-op if we never + # acquired it (e.g. a terminal-state resume) or the orchestrator's + # cleanup_on_success already released it. + if state_manager is not None: + state_manager.release_session_lock() def register_workflow_commands(app: typer.Typer) -> None: diff --git a/src/claude_task_master/core/agent_models.py b/src/claude_task_master/core/agent_models.py index accf3cda..8037f102 100644 --- a/src/claude_task_master/core/agent_models.py +++ b/src/claude_task_master/core/agent_models.py @@ -238,7 +238,7 @@ def get_tools_for_phase( Example: >>> tools = get_tools_for_phase("planning") >>> print(tools) - ["Read", "Glob", "Grep", "Bash"] + ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] >>> # With custom config >>> config = get_config() diff --git a/src/claude_task_master/core/config.py b/src/claude_task_master/core/config.py index 2e4ed8db..9eb95f88 100644 --- a/src/claude_task_master/core/config.py +++ b/src/claude_task_master/core/config.py @@ -145,17 +145,23 @@ class ContextWindowsConfig(BaseModel): class ToolsConfig(BaseModel): """Tool configurations per execution phase. - Each phase has a list of allowed tools. - An empty list means ALL tools are allowed. + Each phase has a list of allowed tools. Planning and verification default + to read-only tool sets so the agent cannot mutate the repository before a + plan exists or while checking success criteria β€” matching the phase table + in CLAUDE.md. The working phase defaults to an empty list, which means ALL + tools are allowed (full access to implement tasks). + + An empty list always means ALL tools are allowed for that phase. """ planning: list[str] = Field( - default_factory=list, - description="Tools available during planning phase (empty = all tools).", + default_factory=lambda: ["Read", "Glob", "Grep", "WebFetch", "WebSearch"], + description="Tools available during planning phase (read-only default; empty = all tools).", ) verification: list[str] = Field( - default_factory=list, - description="Tools available during verification phase (empty = all tools).", + default_factory=lambda: ["Read", "Glob", "Grep", "Bash"], + description="Tools available during verification phase " + "(read + Bash default; empty = all tools).", ) working: list[str] = Field( default_factory=list, @@ -202,7 +208,7 @@ class ClaudeTaskMasterConfig(BaseModel): "sonnet_1m": 1000000 }, "tools": { - "planning": ["Read", "Glob", "Grep", "Bash", "WebFetch", "WebSearch"], + "planning": ["Read", "Glob", "Grep", "WebFetch", "WebSearch"], "verification": ["Read", "Glob", "Grep", "Bash"], "working": [] } diff --git a/src/claude_task_master/core/credentials.py b/src/claude_task_master/core/credentials.py index 6c6f8754..e23abfb1 100644 --- a/src/claude_task_master/core/credentials.py +++ b/src/claude_task_master/core/credentials.py @@ -360,14 +360,19 @@ def get_valid_token(self) -> str: Raises: CredentialNotFoundError: If the credentials file does not exist. - InvalidCredentialsError: If the credentials are malformed. + InvalidCredentialsError: If the credentials are malformed or empty. CredentialPermissionError: If there are permission issues. """ # api-key profiles authenticate via ANTHROPIC_API_KEY (injected into the # SDK subprocess), not an OAuth credentials file. The returned value is # only used as a pre-flight "are we configured?" gate. if self._profile is not None and self._profile.type == "api-key": - return self._profile.api_key or "" + if not self._profile.api_key: + raise InvalidCredentialsError( + f"api-key profile '{self._profile.name}' has an empty api_key", + "Provide the API key via CLAUDETM_API_KEY or re-add the profile with 'claudetm profile add'.", + ) + return self._profile.api_key credentials = self.load_credentials() return credentials.accessToken @@ -382,11 +387,16 @@ def verify_credentials(self) -> bool: Raises: CredentialNotFoundError: If the credentials file does not exist. - InvalidCredentialsError: If the credentials are malformed. + InvalidCredentialsError: If the credentials are malformed or empty. CredentialPermissionError: If there are permission issues. """ # api-key profiles have no OAuth credentials file to verify. if self._profile is not None and self._profile.type == "api-key": - return bool(self._profile.api_key) + if not self._profile.api_key: + raise InvalidCredentialsError( + f"api-key profile '{self._profile.name}' has an empty api_key", + "Provide the API key via CLAUDETM_API_KEY or re-add the profile with 'claudetm profile add'.", + ) + return True self.load_credentials() return True diff --git a/src/claude_task_master/core/profiles.py b/src/claude_task_master/core/profiles.py index c1c9cca7..ab82bd2a 100644 --- a/src/claude_task_master/core/profiles.py +++ b/src/claude_task_master/core/profiles.py @@ -356,19 +356,32 @@ def add( self.save(registry) return profile - def remove(self, name: str) -> None: + def remove(self, name: str, force: bool = False) -> None: """Remove a profile and clear the active pointer if it referenced it. Note: the profile's isolated config directory is left on disk so credentials are not destroyed by an accidental remove. Delete it manually if desired. + Args: + name: The profile name to remove. + force: If True, allow removing the active profile without warning. + If False, raises ProfileError when removing the active profile. + Raises: ProfileNotFoundError: If no such profile exists. + ProfileError: If attempting to remove the active profile without force=True. """ registry = self.load() if name not in registry.profiles: raise ProfileNotFoundError(name) + + if not force and registry.active == name: + raise ProfileError( + f"Cannot remove active profile '{name}'. " + "Use --force to remove it, or first switch to another profile with 'claudetm profile use'." + ) + del registry.profiles[name] if registry.active == name: registry.active = None diff --git a/src/claude_task_master/utils/debug_claude_md.py b/src/claude_task_master/utils/debug_claude_md.py index 8f3043c2..10b02e4f 100644 --- a/src/claude_task_master/utils/debug_claude_md.py +++ b/src/claude_task_master/utils/debug_claude_md.py @@ -20,6 +20,9 @@ from rich.console import Console from rich.panel import Panel +from ..core.config_loader import get_config +from ..core.credentials import CredentialManager + console = Console() @@ -60,13 +63,15 @@ async def debug_claude_md_detection(working_dir: str | None = None) -> bool: console.print(f"[red]βœ—[/red] Failed to import claude_agent_sdk: {e}") return False - # Check credentials - creds_path = Path.home() / ".claude" / ".credentials.json" - if not creds_path.exists(): - console.print(f"[red]βœ—[/red] Credentials not found at {creds_path}") + # Check credentials (profile-aware) + try: + cred_manager = CredentialManager() + cred_manager.verify_credentials() + console.print("[green]βœ“[/green] Credentials found") + except Exception as e: + console.print(f"[red]βœ—[/red] Credentials check failed: {e}") console.print(" Run: [cyan]claude[/cyan] to authenticate first") return False - console.print("[green]βœ“[/green] Credentials found") console.print("\n[bold]Running test query...[/bold]") console.print("Asking Claude: 'What code style instructions do you see?'\n") @@ -80,10 +85,11 @@ async def debug_claude_md_detection(working_dir: str | None = None) -> bool: console.print(f"[dim]Changed to: {os.getcwd()}[/dim]") # Create options with cwd and setting_sources + config = get_config() options = claude_agent_sdk.ClaudeAgentOptions( allowed_tools=["Read"], # Minimal tools for test permission_mode="bypassPermissions", - model="claude-haiku-4-5-20251001", # Use Haiku for speed/cost + model=config.models.haiku, # Use Haiku for speed/cost cwd=str(target_dir), setting_sources=["user", "local", "project"], # Load CLAUDE.md hooks={}, # Disable hooks to prevent "Stream closed" errors diff --git a/src/claude_task_master/utils/doctor.py b/src/claude_task_master/utils/doctor.py index 8aed020e..61a72923 100644 --- a/src/claude_task_master/utils/doctor.py +++ b/src/claude_task_master/utils/doctor.py @@ -1,10 +1,11 @@ """Doctor command - Check system requirements and authentication.""" import subprocess -from pathlib import Path from rich.console import Console +from ..core.credentials import CredentialError, CredentialManager + class SystemDoctor: """Checks system requirements.""" @@ -52,15 +53,15 @@ def _check_gh_cli(self) -> None: self.checks_passed = False def _check_credentials(self) -> None: - """Check if Claude credentials exist.""" - creds_path = Path.home() / ".claude" / ".credentials.json" - - if creds_path.exists(): + """Check if Claude credentials exist (profile-aware).""" + try: + cred_manager = CredentialManager() + cred_manager.verify_credentials() self.console.print("[green]βœ“[/green] Claude credentials found") - else: - self.console.print("[red]βœ—[/red] Claude credentials not found") - self.console.print(f" Expected at: {creds_path}") - self.console.print(" Run Claude CLI once to authenticate") + except CredentialError as e: + self.console.print(f"[red]βœ—[/red] Credentials check failed: {e.message}") + if e.details: + self.console.print(f" {e.details}") self.checks_passed = False def _check_python_version(self) -> None: diff --git a/tests/cli/test_config_commands.py b/tests/cli/test_config_commands.py index 8d39d782..fae6405c 100644 --- a/tests/cli/test_config_commands.py +++ b/tests/cli/test_config_commands.py @@ -2,6 +2,7 @@ import json from pathlib import Path +from typing import Any import pytest @@ -272,6 +273,118 @@ def test_config_show_indicates_overrides_applied(self, cli_runner, temp_dir, mon assert "claudetm config show --env" in result.output +class TestSecretMasking: + """Unit tests for the secret-masking helpers.""" + + def test_looks_secret_matches_key_names(self): + """Field/env names containing key/secret/token/password are secret.""" + from claude_task_master.cli_commands.config import _looks_secret + + assert _looks_secret("anthropic_api_key") + assert _looks_secret("OPENROUTER_API_KEY") + assert _looks_secret("webhook_secret") + assert _looks_secret("access_token") + assert not _looks_secret("anthropic_base_url") + assert not _looks_secret("target_branch") + + def test_mask_secret_keeps_short_prefix(self): + """Long secrets keep an 8-char prefix hint; short ones fully collapse.""" + from claude_task_master.cli_commands.config import _mask_secret + + assert _mask_secret("sk-ant-supersecret-value") == "sk-ant-s..." + assert _mask_secret("short") == "***" + + def test_redact_secrets_masks_only_secret_string_leaves(self): + """Recursion masks secret string leaves but leaves None/non-secrets intact.""" + from claude_task_master.cli_commands.config import _redact_secrets + + data: dict[str, dict[str, Any]] = { + "api": { + "anthropic_api_key": "sk-ant-supersecret-1234", + "openrouter_api_key": None, + "anthropic_base_url": "https://api.anthropic.com", + }, + "git": {"target_branch": "main"}, + } + redacted = _redact_secrets(data) + + assert redacted["api"]["anthropic_api_key"] == "sk-ant-s..." + assert redacted["api"]["openrouter_api_key"] is None # unset stays visible + assert redacted["api"]["anthropic_base_url"] == "https://api.anthropic.com" + assert redacted["git"]["target_branch"] == "main" + # Original is not mutated + assert data["api"]["anthropic_api_key"] == "sk-ant-supersecret-1234" + + +class TestConfigShowSecrets: + """Tests that 'claudetm config show' masks secrets by default.""" + + def test_config_show_masks_api_key(self, cli_runner, temp_dir, monkeypatch): + """A configured API key is masked in the default (formatted) output.""" + monkeypatch.chdir(temp_dir) + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-supersecret-abcdef123456") + reset_config() + + result = cli_runner.invoke(app, ["config", "show"]) + + assert result.exit_code == 0 + assert "supersecret" not in result.output + assert "Secrets masked" in result.output + + def test_config_show_raw_masks_secrets(self, cli_runner, temp_dir, monkeypatch): + """--raw output masks secrets too (piped JSON must not leak keys).""" + monkeypatch.chdir(temp_dir) + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-supersecret-abcdef123456") + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-anothersecret-xyz789") + reset_config() + + result = cli_runner.invoke(app, ["config", "show", "--raw"]) + + assert result.exit_code == 0 + cleaned = strip_ansi(result.output).strip() + config_data = json.loads(cleaned) # still valid JSON + assert "supersecret" not in cleaned + assert "anothersecret" not in cleaned + assert config_data["api"]["anthropic_api_key"].endswith("...") + assert config_data["api"]["openrouter_api_key"].endswith("...") + + def test_config_show_secrets_reveals_values(self, cli_runner, temp_dir, monkeypatch): + """--show-secrets reveals the full key values.""" + monkeypatch.chdir(temp_dir) + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-supersecret-abcdef123456") + reset_config() + + result = cli_runner.invoke(app, ["config", "show", "--raw", "--show-secrets"]) + + assert result.exit_code == 0 + assert "sk-ant-supersecret-abcdef123456" in result.output + # No mask hint when nothing is masked + assert "Secrets masked" not in result.output + + def test_config_show_no_secrets_no_hint(self, cli_runner, temp_dir, monkeypatch): + """Without any configured secret, no masking hint is shown.""" + monkeypatch.chdir(temp_dir) + for var in ("ANTHROPIC_API_KEY", "OPENROUTER_API_KEY"): + monkeypatch.delenv(var, raising=False) + reset_config() + + result = cli_runner.invoke(app, ["config", "show"]) + + assert result.exit_code == 0 + assert "Secrets masked" not in result.output + + def test_config_show_env_show_secrets_reveals(self, cli_runner, temp_dir, monkeypatch): + """--env --show-secrets reveals the masked override value.""" + monkeypatch.chdir(temp_dir) + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-supersecret-abcdef123456") + reset_config() + + result = cli_runner.invoke(app, ["config", "show", "--env", "--show-secrets"]) + + assert result.exit_code == 0 + assert "sk-ant-supersecret-abcdef123456" in result.output + + class TestConfigPath: """Tests for 'claudetm config path' command.""" diff --git a/tests/cli/test_session_lock.py b/tests/cli/test_session_lock.py new file mode 100644 index 00000000..cafe4914 --- /dev/null +++ b/tests/cli/test_session_lock.py @@ -0,0 +1,130 @@ +"""Tests for the single-instance session lock on ``start`` / ``resume``. + +Both commands must acquire the session lock before touching shared state and +release it on every exit path, so two concurrent runs cannot corrupt state, +duplicate PRs, or race OAuth refresh-token rotation. +""" + +import json +from datetime import datetime +from unittest.mock import patch + +from typer.testing import CliRunner + +from claude_task_master.cli import app +from claude_task_master.core.state import StateManager + +from .conftest import mock_resume_context + +WORKFLOW = "claude_task_master.cli_commands.workflow" + + +class TestStartSessionLock: + """The ``start`` command guards concurrent runs with the session lock.""" + + def test_start_aborts_when_session_lock_held(self, cli_runner: CliRunner, temp_dir): + """A second start refuses to run (and never loads creds) when the lock is held.""" + with patch.object(StateManager, "STATE_DIR", temp_dir / ".claude-task-master"): + with patch.object(StateManager, "acquire_session_lock", return_value=False): + with patch(f"{WORKFLOW}.CredentialManager") as mock_cred: + result = cli_runner.invoke(app, ["start", "Test goal"]) + # Lock guard runs before credential loading (closes the + # OAuth refresh-token rotation race), so creds are untouched. + mock_cred.assert_not_called() + + assert result.exit_code == 1 + assert "Another claudetm session is active" in result.output + assert "clean -f" in result.output + + def test_start_releases_lock_on_planning_failure(self, cli_runner: CliRunner, temp_dir): + """A failed start frees the lock so a later run can acquire it.""" + state_dir = temp_dir / ".claude-task-master" + with patch.object(StateManager, "STATE_DIR", state_dir): + with patch(f"{WORKFLOW}.CredentialManager") as mock_cred: + mock_cred.return_value.get_valid_token.return_value = "test-token" + mock_cred.return_value.resync_from_live.return_value = False + with patch(f"{WORKFLOW}.AgentWrapper"): + with patch(f"{WORKFLOW}.Planner") as mock_planner: + mock_planner.return_value.create_plan.side_effect = Exception("boom") + + result = cli_runner.invoke(app, ["start", "Test goal"]) + + assert result.exit_code == 1 + # The finally must remove the PID lock even though start failed. + assert not (state_dir / ".pid").exists() + + def test_start_releases_lock_on_success(self, cli_runner: CliRunner, temp_dir): + """A successful start releases the lock via the finally on exit.""" + state_dir = temp_dir / ".claude-task-master" + with patch.object(StateManager, "STATE_DIR", state_dir): + with patch(f"{WORKFLOW}.CredentialManager") as mock_cred: + mock_cred.return_value.get_valid_token.return_value = "test-token" + mock_cred.return_value.resync_from_live.return_value = False + with patch(f"{WORKFLOW}.AgentWrapper"): + with patch(f"{WORKFLOW}.Planner") as mock_planner: + mock_planner.return_value.create_plan.return_value = { + "plan": "## Tasks\n- [ ] Task 1", + "raw_output": "Planning output", + } + with patch(f"{WORKFLOW}.WorkLoopOrchestrator") as mock_orch: + mock_orch.return_value.run.return_value = 0 + + result = cli_runner.invoke(app, ["start", "Test goal"]) + + assert result.exit_code == 0 + assert not (state_dir / ".pid").exists() + + +class TestResumeSessionLock: + """The ``resume`` command guards concurrent runs with the session lock.""" + + def test_resume_aborts_when_session_lock_held( + self, cli_runner: CliRunner, mock_state_dir, setup_resume_state + ): + """Resume refuses to run (and never loads creds) when the lock is held.""" + setup_resume_state(status="paused") + with patch.object(StateManager, "STATE_DIR", mock_state_dir): + with patch.object(StateManager, "acquire_session_lock", return_value=False): + with patch(f"{WORKFLOW}.CredentialManager") as mock_cred: + result = cli_runner.invoke(app, ["resume"]) + mock_cred.assert_not_called() + + assert result.exit_code == 1 + assert "Another claudetm session is active" in result.output + + def test_resume_releases_lock_on_exit( + self, cli_runner: CliRunner, mock_state_dir, setup_resume_state + ): + """Resume frees the PID lock on exit so a later run can acquire it.""" + setup_resume_state(status="paused") + with mock_resume_context(mock_state_dir): + result = cli_runner.invoke(app, ["resume"]) + + assert result.exit_code == 0 + assert not (mock_state_dir / ".pid").exists() + + def test_resume_terminal_state_never_acquires_lock( + self, cli_runner: CliRunner, mock_state_dir, mock_goal_file, mock_plan_file + ): + """A no-op resume of a finished task exits without taking the lock.""" + timestamp = datetime.now().isoformat() + state_data = { + "status": "success", + "current_task_index": 3, + "session_count": 5, + "current_pr": None, + "created_at": timestamp, + "updated_at": timestamp, + "run_id": "20250115-120000", + "model": "sonnet", + "options": {"auto_merge": True, "max_sessions": None, "pause_on_pr": False}, + } + (mock_state_dir / "state.json").write_text(json.dumps(state_data)) + + with patch.object(StateManager, "STATE_DIR", mock_state_dir): + with patch.object(StateManager, "acquire_session_lock") as mock_acquire: + result = cli_runner.invoke(app, ["resume"]) + mock_acquire.assert_not_called() + + assert result.exit_code == 0 + assert "already completed successfully" in result.output diff --git a/tests/cli_commands/test_info.py b/tests/cli_commands/test_info.py index 9b21c29e..4dac1d22 100644 --- a/tests/cli_commands/test_info.py +++ b/tests/cli_commands/test_info.py @@ -299,11 +299,22 @@ def test_status_shows_tools_planning( state_file = info_state_dir / "state.json" state_file.write_text(json.dumps(state_data)) - with patch.object(StateManager, "STATE_DIR", info_state_dir): + # Patch the config lookup so the display is deterministic regardless of + # any ambient config.json in the working directory. + with ( + patch.object(StateManager, "STATE_DIR", info_state_dir), + patch.object( + info, + "get_tools_for_phase", + return_value=["Read", "Glob", "Grep", "WebFetch", "WebSearch"], + ), + ): info.status() calls = [str(call) for call in mock_console.print.call_args_list] - assert any("Read, Glob, Grep, Bash (read-only mode)" in str(call) for call in calls) + assert any( + "Read, Glob, Grep, WebFetch, WebSearch (read-only mode)" in str(call) for call in calls + ) def test_status_shows_tools_working( self, diff --git a/tests/cli_commands/test_profile.py b/tests/cli_commands/test_profile.py new file mode 100644 index 00000000..bb7345bd --- /dev/null +++ b/tests/cli_commands/test_profile.py @@ -0,0 +1,670 @@ +"""Tests for cli_commands/profile.py β€” profile add/list/use/show/remove/login. + +Coverage goals: +- profile add: oauth and api-key creation, invalid --type rejection, env-key + fallback, ProfileError forwarding. +- profile list: empty registry, table output, active marker. +- profile use: success, unknown-profile error forwarding. +- profile show: active default, named profile, api-key masking, unknown error. +- profile remove: confirmation flow, --force/-f, active-profile guard, unknown. +- profile login: oauth launch, api-key guard, unknown-profile error, claude + binary not found. +- register_profile_commands: wires the sub-app under the Typer app. +- _mask helper: short/long secrets, None/empty inputs. +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import typer + +from claude_task_master.cli_commands.profile import ( + _mask, + profile_app, + register_profile_commands, +) +from claude_task_master.core.profiles import ( + Profile, + ProfileError, + ProfileExistsError, + ProfileManager, + ProfileNotFoundError, + ProfileValidationError, +) + +# Module path for patching +_MOD = "claude_task_master.cli_commands.profile" + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest.fixture +def manager(tmp_path: Path) -> ProfileManager: + """Isolated ProfileManager backed by a temp directory.""" + return ProfileManager(base_dir=tmp_path / ".claudetm") + + +@pytest.fixture +def oauth_profile(manager: ProfileManager) -> Profile: + """A pre-created oauth profile named 'work'.""" + return manager.add("work", "oauth") + + +@pytest.fixture +def api_key_profile(manager: ProfileManager) -> Profile: + """A pre-created api-key profile named 'zai'.""" + return manager.add( + "zai", "api-key", api_key="sk-test-key-123456789", base_url="https://api.z.ai" + ) + + +@pytest.fixture +def runner(): + """Typer CLI test runner.""" + from typer.testing import CliRunner + + return CliRunner() + + +# ============================================================================= +# _mask helper +# ============================================================================= + + +class TestMaskHelper: + """Unit tests for the _mask secret-masking helper.""" + + def test_none_returns_dim_none(self): + """None secret β†’ rich dim placeholder.""" + assert _mask(None) == "[dim](none)[/dim]" + + def test_empty_string_returns_dim_none(self): + """Empty string secret β†’ rich dim placeholder (treated as absent).""" + assert _mask("") == "[dim](none)[/dim]" + + def test_short_secret_collapses_to_stars(self): + """Secrets ≀10 chars collapse to '***' (no useful prefix to show).""" + assert _mask("abc") == "***" + assert _mask("1234567890") == "***" + + def test_long_secret_shows_prefix_suffix(self): + """Secrets >10 chars keep first 6 and last 4 with ellipsis in between.""" + result = _mask("sk-ant-supersecret-value-here") + assert result.startswith("sk-ant") + assert result.endswith("here") + assert "…" in result + + def test_eleven_char_secret_has_prefix_suffix(self): + """An 11-char secret (boundary) shows prefix and suffix.""" + result = _mask("abcdefghijk") + assert result == "abcdef…hijk" + + +# ============================================================================= +# profile add +# ============================================================================= + + +class TestProfileAdd: + """Tests for 'claudetm profile add '.""" + + def test_add_oauth_profile(self, runner, tmp_path): + """Adding an oauth profile creates it and prints confirmation.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.type = "oauth" + profile.name = "work" + profile.config_dir = str(tmp_path / "work") + mock_mgr.add.return_value = profile + mock_mgr.active_name.return_value = "work" + + result = runner.invoke(profile_app, ["add", "work"]) + + assert result.exit_code == 0 + assert "work" in result.output + assert "oauth" in result.output or "Created" in result.output + + def test_add_api_key_profile_from_env(self, runner): + """An api-key profile reads the key from CLAUDETM_API_KEY env var.""" + import os + + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.type = "api-key" + profile.name = "zai" + profile.config_dir = None + mock_mgr.add.return_value = profile + mock_mgr.active_name.return_value = "zai" + + with patch.dict(os.environ, {"CLAUDETM_API_KEY": "sk-test-12345"}): + result = runner.invoke( + profile_app, + ["add", "zai", "--type", "api-key"], + ) + + assert result.exit_code == 0 + assert "zai" in result.output + # Verify the key from env was used (not prompted) + mock_mgr.add.assert_called_once() + call_kwargs = mock_mgr.add.call_args[1] + assert call_kwargs.get("api_key") == "sk-test-12345" + + def test_add_invalid_type_exits_with_error(self, runner): + """--type 'bad' is rejected before reaching ProfileManager.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + result = runner.invoke(profile_app, ["add", "x", "--type", "bad"]) + MockMgr.assert_not_called() + + assert result.exit_code == 1 + assert "Invalid --type" in result.output + assert "oauth" in result.output + assert "api-key" in result.output + + def test_add_profile_error_propagated(self, runner): + """ProfileError from manager.add prints the error and exits 1.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.add.side_effect = ProfileExistsError("work") + + result = runner.invoke(profile_app, ["add", "work"]) + + assert result.exit_code == 1 + assert "work" in result.output + + def test_add_validation_error_propagated(self, runner): + """ProfileValidationError from manager.add is shown as red error.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.add.side_effect = ProfileValidationError("bad/name") + + result = runner.invoke(profile_app, ["add", "bad/name"]) + + assert result.exit_code == 1 + + def test_add_shows_login_hint_for_oauth(self, runner): + """After adding an oauth profile, the login hint is shown.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.type = "oauth" + profile.name = "myprofile" + profile.config_dir = "/some/dir" + mock_mgr.add.return_value = profile + mock_mgr.active_name.return_value = "other" + + result = runner.invoke(profile_app, ["add", "myprofile"]) + + assert result.exit_code == 0 + assert "login" in result.output.lower() + assert "myprofile" in result.output + + def test_add_shows_active_message_when_becomes_active(self, runner): + """A profile that becomes active shows the 'now active' note.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.type = "oauth" + profile.name = "first" + profile.config_dir = "/dir" + mock_mgr.add.return_value = profile + mock_mgr.active_name.return_value = "first" + + result = runner.invoke(profile_app, ["add", "first"]) + + assert result.exit_code == 0 + assert "first" in result.output + assert "active" in result.output.lower() + + +# ============================================================================= +# profile list +# ============================================================================= + + +class TestProfileList: + """Tests for 'claudetm profile list'.""" + + def test_list_empty_registry(self, runner): + """An empty profile registry shows the 'no profiles' hint.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.list.return_value = [] + + result = runner.invoke(profile_app, ["list"]) + + assert result.exit_code == 0 + assert "No profiles" in result.output + + def test_list_shows_profiles_table(self, runner): + """Non-empty registry renders a table with profile names and types.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + work = MagicMock(spec=Profile) + work.name = "work" + work.type = "oauth" + work.config_dir = "/dir/work" + work.base_url = None + zai = MagicMock(spec=Profile) + zai.name = "zai" + zai.type = "api-key" + zai.config_dir = None + zai.base_url = "https://api.z.ai" + mock_mgr.list.return_value = [work, zai] + mock_mgr.active_name.return_value = "work" + + result = runner.invoke(profile_app, ["list"]) + + assert result.exit_code == 0 + assert "work" in result.output + assert "zai" in result.output + assert "oauth" in result.output + assert "api-key" in result.output + + def test_list_marks_active_profile(self, runner): + """The active profile is marked with the 'β†’' arrow in the table.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + p = MagicMock(spec=Profile) + p.name = "active-one" + p.type = "oauth" + p.config_dir = "/dir" + p.base_url = None + mock_mgr.list.return_value = [p] + mock_mgr.active_name.return_value = "active-one" + + result = runner.invoke(profile_app, ["list"]) + + assert result.exit_code == 0 + assert "β†’" in result.output + + +# ============================================================================= +# profile use +# ============================================================================= + + +class TestProfileUse: + """Tests for 'claudetm profile use '.""" + + def test_use_sets_active_profile(self, runner): + """'profile use work' activates the profile and prints confirmation.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + + result = runner.invoke(profile_app, ["use", "work"]) + + assert result.exit_code == 0 + mock_mgr.use.assert_called_once_with("work") + assert "work" in result.output + + def test_use_unknown_profile_exits_1(self, runner): + """Using a non-existent profile forwards ProfileNotFoundError as exit 1.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.use.side_effect = ProfileNotFoundError("ghost") + + result = runner.invoke(profile_app, ["use", "ghost"]) + + assert result.exit_code == 1 + assert "ghost" in result.output + + +# ============================================================================= +# profile show +# ============================================================================= + + +class TestProfileShow: + """Tests for 'claudetm profile show [name]'.""" + + def test_show_active_profile_by_default(self, runner): + """Without a name argument, show displays the active profile.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "work" + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/dir/work" + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["show"]) + + assert result.exit_code == 0 + mock_mgr.get.assert_called_once_with("work") + assert "work" in result.output + + def test_show_named_profile(self, runner): + """With a name argument, show displays that profile specifically.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "work" + profile = MagicMock(spec=Profile) + profile.name = "zai" + profile.type = "api-key" + profile.base_url = "https://api.z.ai" + profile.api_key = "sk-test-supersecret-value" + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["show", "zai"]) + + assert result.exit_code == 0 + mock_mgr.get.assert_called_once_with("zai") + + def test_show_masks_api_key(self, runner): + """API key is masked in show output (never shown in full).""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "zai" + profile = MagicMock(spec=Profile) + profile.name = "zai" + profile.type = "api-key" + profile.base_url = "https://api.z.ai" + profile.api_key = "sk-test-supersecret-value" + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["show", "zai"]) + + assert result.exit_code == 0 + assert "supersecret" not in result.output + assert "…" in result.output or "***" in result.output + + def test_show_no_active_no_name_exits_1(self, runner): + """No name and no active profile β†’ exit 1 with a helpful message.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = None + + result = runner.invoke(profile_app, ["show"]) + + assert result.exit_code == 1 + assert "No profile" in result.output or "none active" in result.output.lower() + + def test_show_unknown_profile_exits_1(self, runner): + """Requesting a non-existent profile propagates ProfileError as exit 1.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "ghost" + mock_mgr.get.side_effect = ProfileNotFoundError("ghost") + + result = runner.invoke(profile_app, ["show", "ghost"]) + + assert result.exit_code == 1 + assert "ghost" in result.output + + def test_show_marks_active_profile(self, runner): + """A profile that is the active one is labelled '(active)'.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "work" + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/dir" + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["show", "work"]) + + assert result.exit_code == 0 + assert "active" in result.output.lower() + + def test_show_not_active_profile_no_active_label(self, runner): + """A profile that is not the active one is not labelled '(active)'.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.active_name.return_value = "other" + profile = MagicMock(spec=Profile) + profile.name = "zai" + profile.type = "api-key" + profile.base_url = None + profile.api_key = "sk-test-longenoughkey" + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["show", "zai"]) + + assert result.exit_code == 0 + assert "(active)" not in result.output + + +# ============================================================================= +# profile remove +# ============================================================================= + + +class TestProfileRemove: + """Tests for 'claudetm profile remove '.""" + + def test_remove_with_force_skips_confirmation(self, runner): + """--force removes without confirmation prompt.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + + result = runner.invoke(profile_app, ["remove", "work", "--force"]) + + assert result.exit_code == 0 + mock_mgr.remove.assert_called_once_with("work", force=True) + assert "Removed" in result.output + + def test_remove_short_force_flag(self, runner): + """'-f' (short force) removes without confirmation.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + + result = runner.invoke(profile_app, ["remove", "work", "-f"]) + + assert result.exit_code == 0 + mock_mgr.remove.assert_called_once_with("work", force=True) + + def test_remove_confirmation_accepted(self, runner): + """When user confirms, the profile is removed.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + + result = runner.invoke(profile_app, ["remove", "work"], input="y\n") + + assert result.exit_code == 0 + mock_mgr.remove.assert_called_once_with("work", force=False) + + def test_remove_confirmation_declined_cancels(self, runner): + """When user declines, the profile is NOT removed.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + + result = runner.invoke(profile_app, ["remove", "work"], input="n\n") + + assert result.exit_code == 0 + mock_mgr.remove.assert_not_called() + assert "Cancelled" in result.output + + def test_remove_active_profile_without_force_exits_1(self, runner): + """Removing the active profile without --force propagates ProfileError.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.remove.side_effect = ProfileError( + "Cannot remove active profile 'work'. Use --force..." + ) + + result = runner.invoke(profile_app, ["remove", "work", "--force"]) + + assert result.exit_code == 1 + assert "work" in result.output + + def test_remove_unknown_profile_exits_1(self, runner): + """Removing a non-existent profile propagates ProfileNotFoundError as exit 1.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.remove.side_effect = ProfileNotFoundError("ghost") + + result = runner.invoke(profile_app, ["remove", "ghost", "--force"]) + + assert result.exit_code == 1 + assert "ghost" in result.output + + +# ============================================================================= +# profile login +# ============================================================================= + + +class TestProfileLogin: + """Tests for 'claudetm profile login '.""" + + def test_login_launches_claude_cli(self, runner): + """profile login runs subprocess.run(['claude']) with the profile's config dir.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/profiles/work" + mock_mgr.get.return_value = profile + + with patch(f"{_MOD}.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(profile_app, ["login", "work"]) + + assert result.exit_code == 0 + mock_run.assert_called_once() + _, kwargs = mock_run.call_args + # CLAUDE_CONFIG_DIR must point to the profile's dir + assert ( + kwargs.get("env", {}).get("CLAUDE_CONFIG_DIR") == "/profiles/work" + or mock_run.call_args[1].get("env", {}).get("CLAUDE_CONFIG_DIR") == "/profiles/work" + or "CLAUDE_CONFIG_DIR" in str(mock_run.call_args) + ) + + def test_login_api_key_profile_exits_1(self, runner): + """login refuses api-key profiles (they have no OAuth config dir).""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.name = "zai" + profile.type = "api-key" + profile.config_dir = None + mock_mgr.get.return_value = profile + + result = runner.invoke(profile_app, ["login", "zai"]) + + assert result.exit_code == 1 + assert "oauth" in result.output.lower() or "not an oauth" in result.output.lower() + + def test_login_unknown_profile_exits_1(self, runner): + """login with a non-existent profile forwards ProfileError as exit 1.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + mock_mgr.get.side_effect = ProfileNotFoundError("ghost") + + result = runner.invoke(profile_app, ["login", "ghost"]) + + assert result.exit_code == 1 + assert "ghost" in result.output + + def test_login_claude_not_found_exits_1(self, runner): + """When 'claude' binary is absent, login exits 1 with a helpful message.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/profiles/work" + mock_mgr.get.return_value = profile + + with patch(f"{_MOD}.subprocess.run", side_effect=FileNotFoundError): + result = runner.invoke(profile_app, ["login", "work"]) + + assert result.exit_code == 1 + assert "claude" in result.output.lower() + + def test_login_passes_returncode_as_exit(self, runner): + """If the claude CLI exits non-zero, that code is propagated.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/profiles/work" + mock_mgr.get.return_value = profile + + with patch(f"{_MOD}.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=2) + result = runner.invoke(profile_app, ["login", "work"]) + + assert result.exit_code == 2 + + def test_login_inherits_env_with_config_dir(self, runner): + """The subprocess inherits the full env with CLAUDE_CONFIG_DIR added.""" + with patch(f"{_MOD}.ProfileManager") as MockMgr: + mock_mgr = MagicMock() + MockMgr.return_value = mock_mgr + profile = MagicMock(spec=Profile) + profile.name = "work" + profile.type = "oauth" + profile.config_dir = "/isolated/work" + mock_mgr.get.return_value = profile + + captured_env: dict = {} + + def capture(cmd: list[str], env: dict | None = None, **kw: object) -> MagicMock: + if env: + captured_env.update(env) + return MagicMock(returncode=0) + + with patch(f"{_MOD}.subprocess.run", side_effect=capture): + runner.invoke(profile_app, ["login", "work"]) + + assert captured_env.get("CLAUDE_CONFIG_DIR") == "/isolated/work" + + +# ============================================================================= +# register_profile_commands +# ============================================================================= + + +class TestRegisterProfileCommands: + """register_profile_commands wires the profile sub-app into a Typer app.""" + + def test_register_adds_profile_group(self): + """After registration the parent app exposes a 'profile' command group.""" + app = typer.Typer() + register_profile_commands(app) + + # The profile sub-app should be registered as a group + group_names = [g.name for g in app.registered_groups] + assert "profile" in group_names diff --git a/tests/cli_commands/test_workflow.py b/tests/cli_commands/test_workflow.py new file mode 100644 index 00000000..40f2977b --- /dev/null +++ b/tests/cli_commands/test_workflow.py @@ -0,0 +1,107 @@ +"""Tests for start-command input validation in cli_commands.workflow.""" + +from __future__ import annotations + +import re + +import pytest +import typer + +from claude_task_master.cli import app +from claude_task_master.cli_commands.workflow import _validate_budget, _validate_goal + + +def _normalize(text: str) -> str: + """Strip ANSI + box-border chars and collapse whitespace. + + Rich renders click errors inside a bordered box, wrapping long messages and + inserting ``β”‚`` border glyphs mid-sentence. Removing the box-drawing block + (U+2500–U+257F) lets substring assertions match regardless of terminal width. + """ + no_ansi = re.sub(r"\x1b\[[0-9;]*m", "", text) + no_box = re.sub(r"[─-β•Ώ]", " ", no_ansi) + return re.sub(r"\s+", " ", no_box) + + +class TestValidateGoal: + """Unit tests for the _validate_goal callback.""" + + def test_accepts_non_empty_goal(self): + """A normal goal passes through unchanged.""" + assert _validate_goal("Add auth") == "Add auth" + + def test_rejects_empty_goal(self): + """An empty goal raises BadParameter.""" + with pytest.raises(typer.BadParameter): + _validate_goal("") + + def test_rejects_whitespace_goal(self): + """A whitespace-only goal raises BadParameter.""" + with pytest.raises(typer.BadParameter): + _validate_goal(" \t ") + + +class TestValidateBudget: + """Unit tests for the _validate_budget callback.""" + + def test_accepts_positive_budget(self): + """A positive budget passes through unchanged.""" + assert _validate_budget(5.0) == 5.0 + + def test_accepts_none_budget(self): + """An unset budget (None) is allowed.""" + assert _validate_budget(None) is None + + def test_rejects_zero_budget(self): + """A zero budget raises BadParameter.""" + with pytest.raises(typer.BadParameter): + _validate_budget(0) + + def test_rejects_negative_budget(self): + """A negative budget raises BadParameter.""" + with pytest.raises(typer.BadParameter): + _validate_budget(-1.5) + + +class TestStartInputValidation: + """End-to-end parse-time rejection of invalid start options.""" + + def test_max_sessions_zero_rejected(self, cli_runner): + """--max-sessions 0 is rejected at parse time (not treated as unlimited).""" + result = cli_runner.invoke(app, ["start", "Task", "--max-sessions", "0"]) + assert result.exit_code == 2 + assert "max-sessions" in _normalize(result.output) + + def test_max_sessions_negative_rejected(self, cli_runner): + """A negative --max-sessions is rejected at parse time.""" + result = cli_runner.invoke(app, ["start", "Task", "--max-sessions", "-1"]) + assert result.exit_code == 2 + + def test_max_prs_zero_rejected(self, cli_runner): + """--prs 0 is rejected at parse time.""" + result = cli_runner.invoke(app, ["start", "Task", "--prs", "0"]) + assert result.exit_code == 2 + assert "prs" in _normalize(result.output) + + def test_budget_zero_rejected(self, cli_runner): + """--budget 0 is rejected at parse time with a clear message.""" + result = cli_runner.invoke(app, ["start", "Task", "--budget", "0"]) + assert result.exit_code == 2 + assert "greater than 0" in _normalize(result.output) + + def test_budget_negative_rejected(self, cli_runner): + """A negative --budget is rejected at parse time.""" + result = cli_runner.invoke(app, ["start", "Task", "--budget", "-2.5"]) + assert result.exit_code == 2 + + def test_empty_goal_rejected(self, cli_runner): + """An empty goal argument is rejected at parse time.""" + result = cli_runner.invoke(app, ["start", ""]) + assert result.exit_code == 2 + assert "goal must not be empty" in _normalize(result.output) + + def test_whitespace_goal_rejected(self, cli_runner): + """A whitespace-only goal argument is rejected at parse time.""" + result = cli_runner.invoke(app, ["start", " "]) + assert result.exit_code == 2 + assert "goal must not be empty" in _normalize(result.output) diff --git a/tests/core/test_agent_phases.py b/tests/core/test_agent_phases.py index 67b5bf93..5c514e13 100644 --- a/tests/core/test_agent_phases.py +++ b/tests/core/test_agent_phases.py @@ -58,10 +58,16 @@ def agent(self, temp_dir, monkeypatch): model=ModelType.SONNET, ) - def test_all_phases_allow_all_tools(self, agent): - """Test all phases default to all tools allowed (empty list).""" - assert agent.get_tools_for_phase("planning") == [] - assert agent.get_tools_for_phase("verification") == [] + def test_phase_defaults_restrict_planning_and_verification(self, agent): + """Test planning/verification default to read-only sets; working = all tools.""" + assert agent.get_tools_for_phase("planning") == [ + "Read", + "Glob", + "Grep", + "WebFetch", + "WebSearch", + ] + assert agent.get_tools_for_phase("verification") == ["Read", "Glob", "Grep", "Bash"] assert agent.get_tools_for_phase("working") == [] def test_unknown_phase_returns_working_tools(self, agent): @@ -415,6 +421,48 @@ async def run_query_capture(*args, **kwargs): # Check that run_async_with_cleanup was called assert mock_asyncio.called + def test_run_planning_phase_tools_reach_query_executor( + self, agent_with_mock, temp_dir, monkeypatch + ): + """Planning tool restriction flows end-to-end into query_executor.run_query. + + Verifies the full chain: + run_planning_phase() + β†’ get_tools_for_phase("planning") # returns config-driven list + β†’ query_executor.run_query(tools=...) # that list is passed as-is + Ensures the restricted read-only tool set (no Bash/Edit/Write) is what + the SDK subprocess actually receives in ``ClaudeAgentOptions.allowed_tools``. + """ + monkeypatch.chdir(temp_dir) + from claude_task_master.core.config_loader import reset_config + + reset_config() + + planning_response = "## Task List\n- [ ] Task 1\n## Success Criteria\n1. Done" + + # Patch query_executor.run_query so we can capture which tools were passed. + mock_run_query = AsyncMock(return_value=planning_response) + agent_with_mock._phase_executor.query_executor.run_query = mock_run_query + + with patch( + "claude_task_master.core.agent_phases.run_async_with_cleanup", + return_value=planning_response, + ): + agent_with_mock.run_planning_phase("Build a feature") + + # query_executor.run_query must have been called with the planning tool list. + mock_run_query.assert_called_once() + call_kwargs = mock_run_query.call_args[1] + planning_tools_used: list[str] = call_kwargs["tools"] + + assert planning_tools_used == ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] + # Destructive tools must be absent from the planning phase. + for forbidden in ("Bash", "Edit", "Write"): + assert forbidden not in planning_tools_used, ( + f"Forbidden tool '{forbidden}' reached ClaudeAgentOptions.allowed_tools " + "during the planning phase" + ) + def test_run_planning_phase_with_context(self, agent_with_mock): """Test run_planning_phase includes context.""" mock_result = """## Task List @@ -645,9 +693,9 @@ def test_verify_success_criteria_with_context(self, agent_with_mock): assert result is not None def test_verify_success_criteria_uses_verification_tools(self, agent_with_mock): - """Test verify_success_criteria uses verification tools (all tools allowed).""" + """Test verify_success_criteria uses the read + Bash verification tool set.""" tools = agent_with_mock.get_tools_for_phase("verification") - assert tools == [] # Empty = all tools allowed + assert tools == ["Read", "Glob", "Grep", "Bash"] # ============================================================================= diff --git a/tests/core/test_agent_tools.py b/tests/core/test_agent_tools.py index 7be416d3..3c32e882 100644 --- a/tests/core/test_agent_tools.py +++ b/tests/core/test_agent_tools.py @@ -558,10 +558,16 @@ def agent(self, temp_dir, monkeypatch): working_dir=str(temp_dir), ) - def test_all_phases_allow_all_tools(self, agent): - """Verify all phases default to all tools allowed (empty list).""" - assert agent.get_tools_for_phase("planning") == [] - assert agent.get_tools_for_phase("verification") == [] + def test_phase_defaults_restrict_planning_and_verification(self, agent): + """Verify planning/verification default to restricted read-only tool sets.""" + assert agent.get_tools_for_phase("planning") == [ + "Read", + "Glob", + "Grep", + "WebFetch", + "WebSearch", + ] + assert agent.get_tools_for_phase("verification") == ["Read", "Glob", "Grep", "Bash"] assert agent.get_tools_for_phase("working") == [] def test_case_insensitive_phase_matching(self, agent): @@ -575,5 +581,10 @@ def test_case_insensitive_phase_matching(self, agent): tools_upper = agent.get_tools_for_phase("PLANNING") tools_mixed = agent.get_tools_for_phase("Planning") - # All should return the same result (empty = all tools) - assert tools_upper == tools_lower == tools_mixed == [] + # All should return the same (restricted read-only) planning tool set + assert ( + tools_upper + == tools_lower + == tools_mixed + == ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] + ) diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 852cc19f..74b325ec 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -141,12 +141,26 @@ class TestToolsConfig: """Tests for ToolsConfig model.""" def test_default_values(self) -> None: - """Test that ToolsConfig defaults to all tools allowed (empty lists).""" + """Test that ToolsConfig defaults to restricted read-only planning/verification.""" config = ToolsConfig() - assert config.planning == [] - assert config.verification == [] + assert config.planning == ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] + assert config.verification == ["Read", "Glob", "Grep", "Bash"] + # Working keeps empty default = all tools allowed for implementation. assert config.working == [] + def test_planning_default_excludes_mutation_tools(self) -> None: + """Test that planning default cannot write, edit, or run Bash by default.""" + config = ToolsConfig() + for forbidden in ("Write", "Edit", "MultiEdit", "Bash", "NotebookEdit"): + assert forbidden not in config.planning + + def test_defaults_are_independent_instances(self) -> None: + """Test that mutable list defaults are not shared across instances.""" + first = ToolsConfig() + second = ToolsConfig() + first.planning.append("Bash") + assert "Bash" not in second.planning + def test_custom_tools(self) -> None: """Test that ToolsConfig accepts custom tool lists.""" config = ToolsConfig( @@ -177,7 +191,7 @@ def test_nested_defaults(self) -> None: assert config.api.anthropic_api_key is None assert config.models.sonnet == "claude-sonnet-5" assert config.git.target_branch == "main" - assert config.tools.planning == [] + assert config.tools.planning == ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] def test_full_custom_config(self) -> None: """Test creating a fully custom configuration.""" @@ -332,17 +346,17 @@ def test_get_model_name_unknown_falls_back_to_sonnet(self) -> None: assert get_model_name(config, "invalid") == "claude-sonnet-5" def test_get_tools_for_phase_planning(self) -> None: - """Test get_tools_for_phase returns correct tools for planning.""" + """Test get_tools_for_phase returns restricted read-only tools for planning.""" config = ClaudeTaskMasterConfig() tools = get_tools_for_phase(config, "planning") - assert tools == [] # Empty means all tools allowed + assert tools == ["Read", "Glob", "Grep", "WebFetch", "WebSearch"] assert get_tools_for_phase(config, "PLANNING") == tools def test_get_tools_for_phase_verification(self) -> None: - """Test get_tools_for_phase returns correct tools for verification.""" + """Test get_tools_for_phase returns read + Bash tools for verification.""" config = ClaudeTaskMasterConfig() tools = get_tools_for_phase(config, "verification") - assert tools == [] # Empty means all tools allowed + assert tools == ["Read", "Glob", "Grep", "Bash"] def test_get_tools_for_phase_working(self) -> None: """Test get_tools_for_phase returns empty list for working.""" diff --git a/tests/core/test_profiles.py b/tests/core/test_profiles.py index ba8d8de4..128fdbd7 100644 --- a/tests/core/test_profiles.py +++ b/tests/core/test_profiles.py @@ -9,6 +9,7 @@ from claude_task_master.core.profiles import ( PROFILE_ENV_VAR, Profile, + ProfileError, ProfileExistsError, ProfileManager, ProfileNotFoundError, @@ -87,10 +88,16 @@ def test_use_missing_raises(self, manager: ProfileManager) -> None: def test_remove_clears_active(self, manager: ProfileManager) -> None: manager.add("work", "oauth") - manager.remove("work") + manager.remove("work", force=True) assert manager.active_name() is None assert manager.list() == [] + def test_remove_active_profile_raises_without_force(self, manager: ProfileManager) -> None: + manager.add("work", "oauth") + assert manager.active_name() == "work" + with pytest.raises(ProfileError, match="Cannot remove active profile"): + manager.remove("work") + def test_remove_keeps_other_active(self, manager: ProfileManager) -> None: manager.add("work", "oauth") manager.add("personal", "oauth") diff --git a/tests/utils/test_doctor.py b/tests/utils/test_doctor.py index f5501bda..b4178130 100644 --- a/tests/utils/test_doctor.py +++ b/tests/utils/test_doctor.py @@ -7,6 +7,7 @@ import pytest +from claude_task_master.core.credentials import CredentialManager, CredentialNotFoundError from claude_task_master.utils.doctor import SystemDoctor # ============================================================================= @@ -278,13 +279,11 @@ def test_credentials_exist(self, temp_dir): mock_console = MagicMock() doctor.console = mock_console - # Create fake credentials file - creds_dir = temp_dir / ".claude" - creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text('{"token": "test"}') + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.return_value = True - with patch.object(Path, "home", return_value=temp_dir): doctor._check_credentials() assert doctor.checks_passed is True @@ -295,8 +294,13 @@ def test_credentials_not_exist(self, temp_dir): mock_console = MagicMock() doctor.console = mock_console - # No credentials file exists - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.side_effect = CredentialNotFoundError( + Path.home() / ".claude" / ".credentials.json" + ) + doctor._check_credentials() assert doctor.checks_passed is False @@ -307,13 +311,11 @@ def test_credentials_exist_prints_success(self, temp_dir): mock_console = MagicMock() doctor.console = mock_console - # Create fake credentials file - creds_dir = temp_dir / ".claude" - creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text('{"token": "test"}') + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.return_value = True - with patch.object(Path, "home", return_value=temp_dir): doctor._check_credentials() calls = [str(call) for call in mock_console.print.call_args_list] @@ -326,28 +328,33 @@ def test_credentials_not_exist_prints_error(self, temp_dir): mock_console = MagicMock() doctor.console = mock_console - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.side_effect = CredentialNotFoundError( + Path.home() / ".claude" / ".credentials.json" + ) + doctor._check_credentials() calls = [str(call) for call in mock_console.print.call_args_list] - assert any("Claude credentials not found" in call for call in calls) - assert any("Expected at:" in call for call in calls) - assert any("Run Claude CLI once to authenticate" in call for call in calls) + assert any("Credentials check failed" in call for call in calls) def test_credentials_path_is_correct(self, temp_dir): - """Test credentials are checked at correct path.""" + """Test credentials are checked via CredentialManager.""" doctor = SystemDoctor() mock_console = MagicMock() doctor.console = mock_console - expected_path = temp_dir / ".claude" / ".credentials.json" + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.return_value = True - with patch.object(Path, "home", return_value=temp_dir): doctor._check_credentials() - # Check that the expected path was checked - calls = [str(call) for call in mock_console.print.call_args_list] - assert any(str(expected_path) in call for call in calls) + # Verify CredentialManager was instantiated + MockCM.assert_called_once() def test_credentials_directory_exists_but_file_missing(self, temp_dir): """Test credentials directory exists but file is missing.""" @@ -355,11 +362,13 @@ def test_credentials_directory_exists_but_file_missing(self, temp_dir): mock_console = MagicMock() doctor.console = mock_console - # Create directory but not file - creds_dir = temp_dir / ".claude" - creds_dir.mkdir() + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.side_effect = CredentialNotFoundError( + Path.home() / ".claude" / ".credentials.json" + ) - with patch.object(Path, "home", return_value=temp_dir): doctor._check_credentials() assert doctor.checks_passed is False @@ -510,18 +519,16 @@ def test_all_checks_pass(self, temp_dir): """Test all checks passing end-to-end.""" doctor = SystemDoctor() - # Create credentials file - creds_dir = temp_dir / ".claude" - creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text('{"token": "test"}') - # Mock gh CLI success mock_result = MagicMock() mock_result.returncode = 0 with patch("subprocess.run", return_value=mock_result): - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.return_value = True + result = doctor.run_checks() assert result is True @@ -530,18 +537,16 @@ def test_gh_cli_fails_other_pass(self, temp_dir): """Test gh CLI fails but other checks pass.""" doctor = SystemDoctor() - # Create credentials file - creds_dir = temp_dir / ".claude" - creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text('{"token": "test"}') - # Mock gh CLI failure mock_result = MagicMock() mock_result.returncode = 1 with patch("subprocess.run", return_value=mock_result): - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.return_value = True + result = doctor.run_checks() assert result is False @@ -550,14 +555,18 @@ def test_credentials_fail_other_pass(self, temp_dir): """Test credentials fail but other checks pass.""" doctor = SystemDoctor() - # No credentials file - # Mock gh CLI success mock_result = MagicMock() mock_result.returncode = 0 with patch("subprocess.run", return_value=mock_result): - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.side_effect = CredentialNotFoundError( + Path.home() / ".claude" / ".credentials.json" + ) + result = doctor.run_checks() assert result is False @@ -566,11 +575,15 @@ def test_multiple_failures(self, temp_dir): """Test multiple checks failing.""" doctor = SystemDoctor() - # No credentials file - # Mock gh CLI failure with patch("subprocess.run", side_effect=FileNotFoundError): - with patch.object(Path, "home", return_value=temp_dir): + with patch("claude_task_master.utils.doctor.CredentialManager") as MockCM: + mock_cm = MagicMock() + MockCM.return_value = mock_cm + mock_cm.verify_credentials.side_effect = CredentialNotFoundError( + Path.home() / ".claude" / ".credentials.json" + ) + result = doctor.run_checks() assert result is False @@ -604,41 +617,42 @@ def test_subprocess_permission_error(self): doctor._check_gh_cli() def test_credentials_path_with_special_characters(self, temp_dir): - """Test credentials path handling with unusual home directory.""" + """Test credentials resolve when the config dir path contains spaces.""" doctor = SystemDoctor() - mock_console = MagicMock() - doctor.console = mock_console + doctor.console = MagicMock() - # Create special directory - special_dir = temp_dir / "special spaces" - special_dir.mkdir() - creds_dir = special_dir / ".claude" - creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text('{"token": "test"}') + # Config dir with spaces holding a valid credentials file. + creds_dir = temp_dir / "special spaces" / ".claude" + creds_dir.mkdir(parents=True) + (creds_dir / ".credentials.json").write_text( + '{"claudeAiOauth": {"accessToken": "a", "refreshToken": "r", ' + '"expiresAt": 9999999999999}}' + ) - with patch.object(Path, "home", return_value=special_dir): + # Drive the real CredentialManager against that path (config_dir bypasses + # profile resolution) to exercise end-to-end path handling. + real_cm = CredentialManager(config_dir=creds_dir) + with patch("claude_task_master.utils.doctor.CredentialManager", return_value=real_cm): doctor._check_credentials() assert doctor.checks_passed is True def test_empty_credentials_file(self, temp_dir): - """Test credentials file exists but is empty.""" + """Test credentials file exists but is empty -> check fails (invalid).""" doctor = SystemDoctor() - mock_console = MagicMock() - doctor.console = mock_console + doctor.console = MagicMock() - # Create empty credentials file + # Create empty credentials file (invalid JSON). creds_dir = temp_dir / ".claude" creds_dir.mkdir() - creds_file = creds_dir / ".credentials.json" - creds_file.write_text("") + (creds_dir / ".credentials.json").write_text("") - with patch.object(Path, "home", return_value=temp_dir): + # verify_credentials now validates content, so an empty file is rejected. + real_cm = CredentialManager(config_dir=creds_dir) + with patch("claude_task_master.utils.doctor.CredentialManager", return_value=real_cm): doctor._check_credentials() - # File exists, so check should pass (content validation is not done) - assert doctor.checks_passed is True + assert doctor.checks_passed is False def test_run_checks_multiple_times(self): """Test running checks multiple times."""