GIA/core/tests/test_claude_cli_provider.py

from __future__ import annotations

from subprocess import CompletedProcess, TimeoutExpired
from unittest.mock import patch

from django.test import SimpleTestCase

from core.tasks.providers.claude_cli import ClaudeCLITaskProvider


class ClaudeCLITaskProviderTests(SimpleTestCase):
    def setUp(self):
        self.provider = ClaudeCLITaskProvider()

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_healthcheck_success(self, run_mock):
        run_mock.return_value = CompletedProcess(
            args=["claude", "--version"],
            returncode=0,
            stdout="claude 1.0.0\n",
            stderr="",
        )
        result = self.provider.healthcheck({"command": "claude", "timeout_seconds": 5})
        self.assertTrue(result.ok)
        self.assertIn("claude", str(result.payload.get("stdout") or ""))

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_create_task_builds_task_sync_command(self, run_mock):
        run_mock.return_value = CompletedProcess(
            args=[],
            returncode=0,
            stdout='{"external_key":"cl-123"}',
            stderr="",
        )
        result = self.provider.create_task(
            {
                "command": "claude",
                "workspace_root": "/tmp/work",
                "default_profile": "default",
                "timeout_seconds": 30,
            },
            {
                "task_id": "t1",
                "title": "hello",
                "reference_code": "42",
            },
        )
        self.assertTrue(result.ok)
        self.assertEqual("cl-123", result.external_key)
        args = run_mock.call_args.args[0]
        self.assertEqual(["claude", "task-sync", "--op", "create"], args[:4])
        self.assertIn("--workspace", args)
        self.assertIn("--payload-json", args)

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_timeout_maps_to_failed_result(self, run_mock):
        run_mock.side_effect = TimeoutExpired(cmd=["claude"], timeout=10)
        result = self.provider.append_update({"command": "claude", "timeout_seconds": 10}, {"task_id": "t1"})
        self.assertFalse(result.ok)
        self.assertIn("timeout", result.error)

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_requires_approval_parsed_from_stdout(self, run_mock):
        run_mock.return_value = CompletedProcess(
            args=[],
            returncode=0,
            stdout='{"status":"requires_approval","approval_key":"ak-1","permission_request":{"requested_permissions":["write"]}}',
            stderr="",
        )
        result = self.provider.append_update({"command": "claude"}, {"task_id": "t1"})
        self.assertTrue(result.ok)
        self.assertTrue(bool((result.payload or {}).get("requires_approval")))
        self.assertEqual("requires_approval", (result.payload or {}).get("parsed_status"))

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_retries_with_positional_op_when_flag_unsupported(self, run_mock):
        run_mock.side_effect = [
            CompletedProcess(
                args=[],
                returncode=2,
                stdout="",
                stderr="error: unexpected argument '--op' found",
            ),
            CompletedProcess(
                args=[],
                returncode=0,
                stdout='{"status":"ok","external_key":"cl-42"}',
                stderr="",
            ),
        ]
        result = self.provider.create_task({"command": "claude"}, {"task_id": "t1"})
        self.assertTrue(result.ok)
        self.assertEqual("cl-42", result.external_key)
        self.assertEqual(2, run_mock.call_count)
        first = run_mock.call_args_list[0].args[0]
        second = run_mock.call_args_list[1].args[0]
        self.assertIn("--op", first)
        self.assertNotIn("--op", second)
        self.assertEqual(["claude", "task-sync", "create"], second[:3])

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_falls_back_to_builtin_approval_stub_when_no_task_sync_contract(self, run_mock):
        run_mock.side_effect = [
            CompletedProcess(
                args=[],
                returncode=2,
                stdout="",
                stderr="error: unexpected argument '--op' found",
            ),
            CompletedProcess(
                args=[],
                returncode=2,
                stdout="",
                stderr="error: unrecognized subcommand 'create'\nUsage: claude [OPTIONS] [PROMPT]",
            ),
        ]
        result = self.provider.create_task(
            {"command": "claude"},
            {
                "task_id": "t1",
                "trigger_message_id": "m1",
                "mode": "default",
            },
        )
        self.assertTrue(result.ok)
        self.assertTrue(bool((result.payload or {}).get("requires_approval")))
        self.assertEqual("requires_approval", str((result.payload or {}).get("status") or ""))
        self.assertEqual("builtin_task_sync_stub", str((result.payload or {}).get("fallback_mode") or ""))

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_builtin_stub_approval_response_returns_ok(self, run_mock):
        run_mock.side_effect = [
            CompletedProcess(
                args=[],
                returncode=2,
                stdout="",
                stderr="error: unexpected argument '--op' found",
            ),
            CompletedProcess(
                args=[],
                returncode=2,
                stdout="",
                stderr="error: unexpected argument 'append_update' found\nUsage: claude [OPTIONS] [PROMPT]",
            ),
        ]
        result = self.provider.append_update(
            {"command": "claude"},
            {
                "task_id": "t1",
                "mode": "approval_response",
                "approval_key": "abc123",
            },
        )
        self.assertTrue(result.ok)
        self.assertFalse(bool((result.payload or {}).get("requires_approval")))
        self.assertEqual("ok", str((result.payload or {}).get("status") or ""))

    def test_provider_name_and_run_in_worker(self):
        self.assertEqual("claude_cli", self.provider.name)
        self.assertTrue(self.provider.run_in_worker)

    @patch("core.tasks.providers.claude_cli.subprocess.run")
    def test_healthcheck_failure(self, run_mock):
        run_mock.return_value = CompletedProcess(
            args=["claude", "--version"],
            returncode=1,
            stdout="",
            stderr="command not found: claude",
        )
        result = self.provider.healthcheck({"command": "claude"})
        self.assertFalse(result.ok)
        self.assertIn("command not found", result.error)