Codex Handbook
exec/tests/suite/apply_patch.rs 147 lines
#![allow(clippy::unwrap_used, unused_imports)]use anyhow::Context;use assert_cmd::prelude::*;use codex_apply_patch::CODEX_CORE_APPLY_PATCH_ARG1;use core_test_support::responses::ev_apply_patch_custom_tool_call;use core_test_support::responses::ev_completed;use core_test_support::responses::mount_sse_sequence;use core_test_support::responses::sse;use core_test_support::responses::start_mock_server;use std::fs;use std::process::Command;use tempfile::tempdir;/// While we may add an `apply-patch` subcommand to the `codex` CLI multitool/// at some point, we must ensure that the smaller `codex-exec` CLI can still/// emulate the `apply_patch` CLI.#[test]fn test_standalone_exec_cli_can_use_apply_patch() -> anyhow::Result<()> {    let tmp = tempdir()?;    let relative_path = "source.txt";    let absolute_path = tmp.path().join(relative_path);    fs::write(&absolute_path, "original content\n")?;    Command::new(codex_utils_cargo_bin::cargo_bin("codex-exec")?)        .arg(CODEX_CORE_APPLY_PATCH_ARG1)        .arg(            r#"*** Begin Patch*** Update File: source.txt@@-original content+modified by apply_patch*** End Patch"#,        )        .current_dir(tmp.path())        .assert()        .success()        .stdout("Success. Updated the following files:\nM source.txt\n")        .stderr(predicates::str::is_empty());    assert_eq!(        fs::read_to_string(absolute_path)?,        "modified by apply_patch\n"    );    Ok(())}#[cfg(not(target_os = "windows"))]#[tokio::test(flavor = "multi_thread", worker_threads = 4)]async fn test_apply_patch_tool() -> anyhow::Result<()> {    use core_test_support::skip_if_no_network;    use core_test_support::test_codex_exec::test_codex_exec;    skip_if_no_network!(Ok(()));    let test = test_codex_exec();    let tmp_path = test.cwd_path().to_path_buf();    let add_patch = r#"*** Begin Patch*** Add File: test.md+Hello world*** End Patch"#;    let update_patch = r#"*** Begin Patch*** Update File: test.md@@-Hello world+Final text*** End Patch"#;    let response_streams = vec![        sse(vec![            ev_apply_patch_custom_tool_call("request_0", add_patch),            ev_completed("request_0"),        ]),        sse(vec![            ev_apply_patch_custom_tool_call("request_1", update_patch),            ev_completed("request_1"),        ]),        sse(vec![ev_completed("request_2")]),    ];    let server = start_mock_server().await;    mount_sse_sequence(&server, response_streams).await;    test.cmd_with_server(&server)        .arg("--skip-git-repo-check")        .arg("-s")        .arg("danger-full-access")        .arg("foo")        .assert()        .success();    let final_path = tmp_path.join("test.md");    let contents = std::fs::read_to_string(&final_path).expect("final file should be readable");    assert_eq!(contents, "Final text\n");    Ok(())}#[cfg(not(target_os = "windows"))]#[tokio::test(flavor = "multi_thread", worker_threads = 4)]async fn test_apply_patch_freeform_tool() -> anyhow::Result<()> {    use core_test_support::skip_if_no_network;    use core_test_support::test_codex_exec::test_codex_exec;    skip_if_no_network!(Ok(()));    let test = test_codex_exec();    let freeform_add_patch = r#"*** Begin Patch*** Add File: app.py+class BaseClass:+  def method():+    return False*** End Patch"#;    let freeform_update_patch = r#"*** Begin Patch*** Update File: app.py@@  def method():-    return False++    return True*** End Patch"#;    let response_streams = vec![        sse(vec![            ev_apply_patch_custom_tool_call("request_0", freeform_add_patch),            ev_completed("request_0"),        ]),        sse(vec![            ev_apply_patch_custom_tool_call("request_1", freeform_update_patch),            ev_completed("request_1"),        ]),        sse(vec![ev_completed("request_2")]),    ];    let server = start_mock_server().await;    mount_sse_sequence(&server, response_streams).await;    test.cmd_with_server(&server)        .arg("--skip-git-repo-check")        .arg("-s")        .arg("danger-full-access")        .arg("foo")        .assert()        .success();    // Verify final file contents    let final_path = test.cwd_path().join("app.py");    let contents = std::fs::read_to_string(&final_path).expect("final file should be readable");    assert_eq!(        contents,        include_str!("../fixtures/apply_patch_freeform_final.txt")    );    Ok(())}