test_codex_auto_run.py

from __future__ import annotations

import contextlib
import io
import os
import shlex
import stat
import tempfile
import unittest
from pathlib import Path
from unittest import mock

import codex_auto_run as app


COMMAND_APPROVAL = """

  Would you like to run the following command?

  Reason: tests need to run

  $ python -m unittest

› 1. Yes, proceed (y)
  2. Yes, and don't ask again for commands that start with `python` (p)
  3. No, and tell Codex what to do differently (esc)

  Press enter to confirm or esc to cancel
"""

EDIT_APPROVAL = """
  Would you like to make the following edits?

  Reason: update the implementation

› 1. Yes, proceed (y)
  2. Yes, and don't ask again for these files (a)
  3. No, and tell Codex what to do differently (esc)

  Press enter to confirm or esc to cancel
"""

PERMISSIONS_APPROVAL = """
  Would you like to grant these permissions?

  Permission rule: network; write `/tmp/output`

› 1. Yes, grant these permissions for this turn (y)
  2. Yes, grant for this turn with strict auto review (r)
  3. Yes, grant these permissions for this session (a)
  4. No, continue without permissions (d)

  Press enter to confirm or esc to cancel
"""

NETWORK_APPROVAL = """
  Do you want to approve network access to "example.com"?

  Reason: fetch a dependency

› 1. Yes, just this once (y)
  2. Yes, and allow this host for this conversation (a)
  3. Yes, and allow this host in the future (p)
  4. No, and tell Codex what to do differently (esc)

  Press enter to confirm or esc to cancel
"""

MCP_APPROVAL = """
  github needs your approval.

  Server: github

  Create an issue comment

› 1. Yes, provide the requested info (y)
  2. No, but continue without it (n)
  3. Cancel this request (esc)

  Press enter to confirm or esc to cancel
"""

MCP_TOOL_APPROVAL = """
  Field 1/1
  Allow Calendar to create an event

  Calendar: primary
  Title: Roadmap review

  › 1. Allow   Run the tool and continue.
    2. Cancel  Cancel this tool call

  enter to submit | esc to cancel
"""

TRUST_DIRECTORY = """
> You are in /workspace/project

  Do you trust the contents of this directory? Working with untrusted
  contents comes with higher risk of prompt injection.

› 1. Yes, continue
  2. No, quit

  Press enter to continue
"""

FULL_ACCESS = """
  Enable full access?

  When Codex runs with full access, it can edit any file.

› 1. Yes, continue anyway
  2. Yes, and don't ask again
  3. Cancel

  Press enter to confirm or esc to cancel
"""

REQUEST_USER_INPUT = """
  Questions 1/1
  Which deployment target should be used?

› 1. Staging
  2. Production

  enter to submit | esc to cancel
"""


def rows(value: str) -> list[str]:
    return value.strip("\n").splitlines()


def config(
    runtime_dir: Path,
    *,
    approve_mcp: bool = False,
    auto_trust_directory: bool = False,
    stability_polls: int = 2,
) -> app.Config:
    return app.Config(
        session_prefix="codex-auto-test",
        runtime_dir=runtime_dir,
        poll_interval=0.1,
        cooldown=0.5,
        stability_polls=stability_polls,
        rearm_interval=2.0,
        idle_exit_seconds=2.0,
        approve_mcp=approve_mcp,
        auto_trust_directory=auto_trust_directory,
        keep_dead_session=False,
    )


class DetectorTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory()
        self.runtime = Path(self.temp.name)

    def tearDown(self) -> None:
        self.temp.cleanup()

    def test_standard_approval_types(self) -> None:
        cfg = config(self.runtime)
        expected = {
            COMMAND_APPROVAL: "command",
            EDIT_APPROVAL: "edit",
            PERMISSIONS_APPROVAL: "permissions",
            NETWORK_APPROVAL: "network",
        }
        for fixture, kind in expected.items():
            with self.subTest(kind=kind):
                candidate = app.detect_candidate(rows(fixture), cfg)
                self.assertIsNotNone(candidate)
                self.assertEqual(kind, candidate.kind)

    def test_mcp_requires_explicit_opt_in(self) -> None:
        safe = config(self.runtime)
        enabled = config(self.runtime, approve_mcp=True)
        for fixture in (MCP_APPROVAL, MCP_TOOL_APPROVAL):
            with self.subTest(fixture=fixture[:30]):
                self.assertIsNone(app.detect_candidate(rows(fixture), safe))
                candidate = app.detect_candidate(rows(fixture), enabled)
                self.assertIsNotNone(candidate)
                self.assertIn(candidate.kind, {"mcp", "mcp_tool"})

    def test_directory_trust_requires_explicit_opt_in(self) -> None:
        self.assertIsNone(app.detect_candidate(rows(TRUST_DIRECTORY), config(self.runtime)))
        candidate = app.detect_candidate(
            rows(TRUST_DIRECTORY),
            config(self.runtime, auto_trust_directory=True),
        )
        self.assertIsNotNone(candidate)
        self.assertEqual("trust_directory", candidate.kind)

    def test_unrelated_interactive_surfaces_are_not_approved(self) -> None:
        cfg = config(self.runtime, approve_mcp=True, auto_trust_directory=True)
        for fixture in (FULL_ACCESS, REQUEST_USER_INPUT):
            with self.subTest(fixture=fixture[:30]):
                self.assertIsNone(app.detect_candidate(rows(fixture), cfg))

    def test_historical_approval_text_above_composer_is_not_approved(self) -> None:
        transcript = COMMAND_APPROVAL + "\n\n• Command completed\n\n› write a follow-up"
        self.assertIsNone(app.detect_candidate(rows(transcript), config(self.runtime)))

    def test_selection_must_be_first_one_shot_option(self) -> None:
        unsafe_selection = COMMAND_APPROVAL.replace(
            "› 1. Yes, proceed (y)",
            "  1. Yes, proceed (y)",
        ).replace(
            "  2. Yes, and don't ask again",
            "› 2. Yes, and don't ask again",
        )
        self.assertIsNone(app.detect_candidate(rows(unsafe_selection), config(self.runtime)))

    def test_soft_wrapped_mcp_description_still_matches(self) -> None:
        wrapped = MCP_TOOL_APPROVAL.replace(
            "› 1. Allow   Run the tool and continue.",
            "› 1. Allow\n      Run the tool and continue.",
        )
        candidate = app.detect_candidate(
            rows(wrapped),
            config(self.runtime, approve_mcp=True),
        )
        self.assertIsNotNone(candidate)
        self.assertEqual("mcp_tool", candidate.kind)

    def test_long_mcp_parameter_block_still_matches(self) -> None:
        params = "\n".join(f"  Parameter {idx}: value" for idx in range(30))
        expanded = MCP_TOOL_APPROVAL.replace(
            "  Calendar: primary",
            f"  Calendar: primary\n{params}",
        )
        candidate = app.detect_candidate(
            rows(expanded),
            config(self.runtime, approve_mcp=True),
        )
        self.assertIsNotNone(candidate)
        self.assertEqual("mcp_tool", candidate.kind)

    def test_wrapped_network_title_and_footer_still_match(self) -> None:
        wrapped = NETWORK_APPROVAL.replace(
            'network access to "example.com"?',
            'network access to\n  "example.com"?',
        ).replace(
            "Press enter to confirm or esc to cancel",
            "Press enter to confirm or esc to\n  cancel",
        )
        candidate = app.detect_candidate(rows(wrapped), config(self.runtime))
        self.assertIsNotNone(candidate)
        self.assertEqual("network", candidate.kind)


class SessionStateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory()
        self.cfg = config(Path(self.temp.name))
        self.candidate = app.PromptCandidate("command", "command:abc")

    def tearDown(self) -> None:
        self.temp.cleanup()

    def test_stability_and_rearm(self) -> None:
        state = app.SessionState()
        self.assertFalse(state.ready(self.candidate, 10.0, self.cfg))
        self.assertTrue(state.ready(self.candidate, 10.1, self.cfg))
        state.mark_approved(self.candidate, 10.1)
        self.assertFalse(state.ready(self.candidate, 11.0, self.cfg))
        self.assertFalse(state.ready(self.candidate, 12.2, self.cfg))
        self.assertTrue(state.ready(self.candidate, 12.3, self.cfg))

    def test_screen_clear_rearms_immediately_after_cooldown(self) -> None:
        state = app.SessionState()
        state.mark_approved(self.candidate, 20.0)
        self.assertTrue(state.clear_candidate())
        self.assertFalse(state.ready(self.candidate, 20.6, self.cfg))
        self.assertTrue(state.ready(self.candidate, 20.7, self.cfg))


class ArgumentTests(unittest.TestCase):
    def test_prompt_and_passthrough(self) -> None:
        _, args = app.parse_args(["-p", "hello world", "--", "--model", "gpt-5.4"])
        self.assertEqual("hello world", args.prompt)
        self.assertEqual(["--model", "gpt-5.4"], args.codex_args)

    def test_resume_modes_parse_without_changing_passthrough(self) -> None:
        _, exact = app.parse_args(
            ["--resume", "019f-test", "-p", "continue", "--", "--search"]
        )
        self.assertEqual("019f-test", exact.resume)
        self.assertFalse(exact.resume_last)
        self.assertEqual(["--search"], exact.codex_args)

        _, latest = app.parse_args(["--resume-last", "-C", "/tmp/project"])
        self.assertIsNone(latest.resume)
        self.assertTrue(latest.resume_last)

        _, picker = app.parse_args(["--resume", "-C", "/tmp/project"])
        self.assertEqual("", picker.resume)
        self.assertFalse(picker.resume_last)

    def test_resume_picker_rejects_initial_prompt(self) -> None:
        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
            app.parse_args(["--resume", "-p", "continue"])

    def test_resume_selection_passthrough_is_mode_checked(self) -> None:
        for argv in (
            ["--resume", "019f-test", "--", "--last"],
            ["--", "--all"],
            ["--resume", "019f-test", "--", "--include-non-interactive"],
        ):
            with self.subTest(argv=argv):
                with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(
                    SystemExit
                ):
                    app.parse_args(argv)

        _, args = app.parse_args(
            ["--resume-last", "--", "--all", "--include-non-interactive"]
        )
        self.assertEqual(
            ["--all", "--include-non-interactive"],
            args.codex_args,
        )

    def test_wrapper_rejects_conflicting_codex_options(self) -> None:
        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
            app.parse_args(["-p", "hello", "--", "-a", "never"])

    def test_wrapper_rejects_subcommand_after_global_option(self) -> None:
        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
            app.parse_args(
                ["-p", "hello", "--", "--model", "gpt-5.4", "exec"]
            )

    def test_build_codex_argv_preserves_prompt_as_one_argument(self) -> None:
        prompt = "quoted ' prompt\nwith multiple lines"
        argv = app.build_codex_argv(
            "/opt/codex",
            prompt,
            Path("/tmp/project with spaces"),
            ["--model", "gpt-5.4"],
            "work",
            "workspace-write",
            False,
        )
        self.assertEqual(argv, shlex.split(shlex.join(argv)))
        self.assertEqual(["--", prompt], argv[-2:])
        self.assertIn('approvals_reviewer="user"', argv)
        self.assertNotIn("exec", argv)

    def test_bypass_is_explicit_and_omits_approval_settings(self) -> None:
        argv = app.build_codex_argv(
            "/opt/codex",
            "go",
            Path("/tmp/project"),
            [],
            None,
            "workspace-write",
            True,
        )
        self.assertIn("--dangerously-bypass-approvals-and-sandbox", argv)
        self.assertNotIn("on-request", argv)

    def test_build_exact_resume_argv(self) -> None:
        argv = app.build_codex_argv(
            "/opt/codex",
            "continue safely",
            Path("/tmp/project"),
            ["--search"],
            None,
            "workspace-write",
            False,
            resume_session="019f-session",
        )
        self.assertEqual("resume", argv[1])
        self.assertIn("--search", argv)
        self.assertEqual(["--", "019f-session", "continue safely"], argv[-3:])

    def test_build_resume_last_and_picker_argv(self) -> None:
        latest = app.build_codex_argv(
            "/opt/codex",
            "continue",
            Path("/tmp/project"),
            [],
            None,
            "workspace-write",
            False,
            resume_last=True,
        )
        self.assertEqual("resume", latest[1])
        self.assertIn("--last", latest)
        self.assertEqual(["--", "continue"], latest[-2:])

        picker = app.build_codex_argv(
            "/opt/codex",
            None,
            Path("/tmp/project"),
            [],
            None,
            "workspace-write",
            False,
            resume_session="",
        )
        self.assertEqual("resume", picker[1])
        self.assertNotIn("--last", picker)
        self.assertNotIn("--", picker)

    def test_build_resume_rejects_ambiguous_internal_combinations(self) -> None:
        with self.assertRaises(ValueError):
            app.build_codex_argv(
                "/opt/codex",
                None,
                Path("/tmp/project"),
                [],
                None,
                "workspace-write",
                False,
                resume_session="019f-session",
                resume_last=True,
            )
        with self.assertRaises(ValueError):
            app.build_codex_argv(
                "/opt/codex",
                "ambiguous prompt",
                Path("/tmp/project"),
                [],
                None,
                "workspace-write",
                False,
                resume_session="",
            )


class BinaryAndTmuxTests(unittest.TestCase):
    def test_codex_probe(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            binary = Path(directory) / "codex"
            binary.write_text("#!/bin/sh\necho 'codex-cli 9.9.9'\n", encoding="utf-8")
            binary.chmod(binary.stat().st_mode | stat.S_IXUSR)
            self.assertTrue(app.probe_codex_binary(binary))

    def test_explicit_codex_path_does_not_fall_back(self) -> None:
        self.assertIsNone(app.find_codex("/definitely/missing/codex"))

    def test_tmux_prefix_filter_is_delimited(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            cfg = config(Path(directory))
            app.register_session(cfg, "codex-auto-test-123")

            class FakeTmux(app.TmuxClient):
                def run(self, *args: str, timeout: float = 5.0):
                    del args, timeout
                    return (
                        0,
                        "codex-auto-test-123\t%1\t0\n"
                        "codex-auto-testing\t%2\t0\n"
                        "other\t%3\t0\n",
                        "",
                    )

            panes = FakeTmux(cfg).list_panes()
            self.assertEqual(["%1"], [pane.pane_id for pane in panes])

    def test_confirm_selected_choice_uses_rendered_enter_binding(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            cfg = config(Path(directory))
            calls: list[tuple[str, ...]] = []

            class FakeTmux(app.TmuxClient):
                def run(self, *args: str, timeout: float = 5.0):
                    del timeout
                    calls.append(args)
                    return 0, "", ""

            self.assertTrue(FakeTmux(cfg).confirm_selected_choice("%7"))
            self.assertEqual(("send-keys", "-t", "%7", "Enter"), calls[0])

    def test_session_exists_uses_target_session(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            cfg = config(Path(directory))
            calls: list[tuple[str, ...]] = []

            class FakeTmux(app.TmuxClient):
                def run(self, *args: str, timeout: float = 5.0):
                    del timeout
                    calls.append(args)
                    return 0, "", ""

            self.assertTrue(FakeTmux(cfg).session_exists("test-session"))
            self.assertEqual(
                ("has-session", "-t", "=test-session"),
                calls[0],
            )

    def test_new_session_sets_working_directory_and_quotes_argv(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            cfg = config(Path(directory))
            calls: list[tuple[str, ...]] = []

            class FakeTmux(app.TmuxClient):
                def run(self, *args: str, timeout: float = 5.0):
                    del timeout
                    calls.append(args)
                    return 0, "", ""

            argv = ["/opt/codex", "--", "hello ' world\nsecond prompt line"]
            ok, _ = FakeTmux(cfg).new_session(
                "codex-auto-test-1",
                argv,
                Path("/tmp/project with spaces"),
                120,
                40,
            )
            self.assertTrue(ok)
            new_session = calls[0]
            self.assertIn("-c", new_session)
            self.assertIn("/tmp/project with spaces", new_session)
            self.assertEqual(tuple(argv), new_session[-len(argv) :])
            self.assertEqual("set-window-option", calls[1][0])
            self.assertEqual("=codex-auto-test-1:", calls[1][2])


class DaemonLifecycleTests(unittest.TestCase):
    def test_stop_timeout_preserves_identity_record(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            cfg = config(Path(directory))
            record = app.DaemonRecord(
                pid=12345,
                token="token",
                script=str(Path(app.__file__).resolve()),
                fingerprint=cfg.daemon_fingerprint(),
            )
            app._write_daemon_record(cfg, record)
            with (
                mock.patch.object(
                    app,
                    "daemon_status",
                    return_value=(True, record, None),
                ),
                mock.patch.object(app, "_process_matches", return_value=True),
                mock.patch.object(app.os, "kill"),
                mock.patch.object(app.time, "monotonic", side_effect=[0.0, 6.0]),
            ):
                self.assertEqual("timeout", app.stop_daemon(cfg))
            self.assertTrue(cfg.pid_path.exists())


if __name__ == "__main__":
    unittest.main()