GIA/core/tests/test_codex_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.codex_cli import CodexCLITaskProvider


class CodexCLITaskProviderTests(SimpleTestCase):
    def setUp(self):
        self.provider = CodexCLITaskProvider()

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

    @patch("core.tasks.providers.codex_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":"cx-123"}',
            stderr="",
        )
        result = self.provider.create_task(
            {
                "command": "codex",
                "workspace_root": "/tmp/work",
                "default_profile": "default",
                "timeout_seconds": 30,
            },
            {
                "task_id": "t1",
                "title": "hello",
                "reference_code": "42",
            },
        )
        self.assertTrue(result.ok)
        self.assertEqual("cx-123", result.external_key)
        args = run_mock.call_args.args[0]
        self.assertEqual(["codex", "task-sync", "--op", "create"], args[:4])
        self.assertIn("--workspace", args)
        self.assertIn("--payload-json", args)

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

    @patch("core.tasks.providers.codex_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": "codex"}, {"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.codex_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":"cx-42"}',
                stderr="",
            ),
        ]
        result = self.provider.create_task({"command": "codex"}, {"task_id": "t1"})
        self.assertTrue(result.ok)
        self.assertEqual("cx-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(["codex", "task-sync", "create"], second[:3])

    @patch("core.tasks.providers.codex_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: codex [OPTIONS] [PROMPT]",
            ),
        ]
        result = self.provider.create_task(
            {"command": "codex"},
            {
                "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.codex_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: codex [OPTIONS] [PROMPT]",
            ),
        ]
        result = self.provider.append_update(
            {"command": "codex"},
            {
                "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 ""))