use anyhow::Result;
use prodigy::config::mapreduce::{parse_mapreduce_workflow, MergeWorkflow};
use prodigy::cook::workflow::WorkflowStep;
use prodigy::subprocess::{ProcessCommandBuilder, SubprocessManager};
use prodigy::worktree::WorktreeManager;
use std::collections::HashMap;
use std::fs;
use std::path::PathBuf;
use tempfile::TempDir;
async fn create_test_repo() -> Result<(TempDir, PathBuf)> {
let temp_dir = TempDir::new()?;
let repo_path = temp_dir.path().to_path_buf();
let subprocess = SubprocessManager::production();
let command = ProcessCommandBuilder::new("git")
.current_dir(&repo_path)
.args(["init"])
.build();
subprocess.runner().run(command).await?;
let config_commands = vec![
vec!["config", "user.email", "test@prodigy.test"],
vec!["config", "user.name", "Test User"],
];
for args in config_commands {
let command = ProcessCommandBuilder::new("git")
.current_dir(&repo_path)
.args(&args)
.build();
subprocess.runner().run(command).await?;
}
let readme_path = repo_path.join("README.md");
fs::write(&readme_path, "# Test Repository")?;
let add_command = ProcessCommandBuilder::new("git")
.current_dir(&repo_path)
.args(["add", "."])
.build();
subprocess.runner().run(add_command).await?;
let commit_command = ProcessCommandBuilder::new("git")
.current_dir(&repo_path)
.args(["commit", "-m", "Initial commit"])
.build();
subprocess.runner().run(commit_command).await?;
Ok((temp_dir, repo_path))
}
#[tokio::test]
async fn test_merge_workflow_end_to_end() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let subprocess = SubprocessManager::production();
let merge_workflow = MergeWorkflow {
commands: vec![
WorkflowStep {
shell: Some("echo 'Starting merge of ${merge.worktree}'".to_string()),
..Default::default()
},
WorkflowStep {
shell: Some(
"echo 'Source: ${merge.source_branch}, Target: ${merge.target_branch}'"
.to_string(),
),
..Default::default()
},
],
timeout: Some(300),
};
let manager = WorktreeManager::with_config(
repo_path.clone(),
subprocess,
0, Some(merge_workflow),
HashMap::new(), )?;
let session = manager.create_session().await?;
let test_file = session.path.join("test.txt");
fs::write(&test_file, "test content")?;
let subprocess = SubprocessManager::production();
let add_cmd = ProcessCommandBuilder::new("git")
.current_dir(&session.path)
.args(["add", "."])
.build();
subprocess.runner().run(add_cmd).await?;
let commit_cmd = ProcessCommandBuilder::new("git")
.current_dir(&session.path)
.args(["commit", "-m", "Add test file"])
.build();
subprocess.runner().run(commit_cmd).await?;
Ok(())
}
#[tokio::test]
async fn test_merge_workflow_with_failures() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let subprocess = SubprocessManager::production();
let merge_workflow = MergeWorkflow {
commands: vec![WorkflowStep {
shell: Some("false".to_string()), on_failure: Some(prodigy::cook::workflow::OnFailureConfig::SingleCommand(
"echo 'Handling failure'".to_string(),
)),
..Default::default()
}],
timeout: Some(60),
};
let manager = WorktreeManager::with_config(
repo_path,
subprocess,
0,
Some(merge_workflow),
HashMap::new(),
)?;
let _session = manager.create_session().await?;
Ok(())
}
#[tokio::test]
async fn test_merge_workflow_logging_levels() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let merge_workflow = MergeWorkflow {
commands: vec![WorkflowStep {
shell: Some("echo 'Test output'".to_string()),
..Default::default()
}],
timeout: Some(60),
};
let subprocess = SubprocessManager::production();
let manager_quiet = WorktreeManager::with_config(
repo_path.clone(),
subprocess,
0,
Some(merge_workflow.clone()),
HashMap::new(),
)?;
let session = manager_quiet.create_session().await?;
assert!(session.path.exists());
let subprocess = SubprocessManager::production();
let manager_verbose = WorktreeManager::with_config(
repo_path.clone(),
subprocess,
1,
Some(merge_workflow.clone()),
HashMap::new(),
)?;
let verbose_session = manager_verbose.create_session().await?;
assert!(verbose_session.path.exists());
Ok(())
}
#[test]
fn test_parse_merge_workflow_from_yaml() {
let yaml = r#"
name: test-workflow
mode: mapreduce
setup:
- shell: "echo 'Setup phase'"
map:
input: items.json
json_path: "$.items[*]"
agent_template:
- claude: "/process ${item}"
max_parallel: 5
reduce:
- claude: "/summarize ${map.results}"
merge:
commands:
- shell: "git fetch origin"
- shell: "git merge origin/main"
- shell: "cargo test"
- claude: "/prodigy-merge-worktree ${merge.source_branch}"
timeout: 600
"#;
let config = parse_mapreduce_workflow(yaml).unwrap();
assert!(config.merge.is_some());
let merge = config.merge.unwrap();
assert_eq!(merge.commands.len(), 4);
assert_eq!(merge.timeout, Some(600));
assert!(merge.commands[0].shell.is_some());
assert!(merge.commands[1].shell.is_some());
assert!(merge.commands[2].shell.is_some());
assert!(merge.commands[3].claude.is_some());
}
#[test]
fn test_merge_workflow_variable_substitution() {
let yaml = r#"
name: test-workflow
mode: mapreduce
map:
input: test.json
agent_template:
- shell: "echo test"
merge:
- shell: |
echo "Worktree: ${merge.worktree}"
echo "Source: ${merge.source_branch}"
echo "Target: ${merge.target_branch}"
echo "Session: ${merge.session_id}"
- claude: "/merge ${merge.source_branch} into ${merge.target_branch}"
"#;
let config = parse_mapreduce_workflow(yaml).unwrap();
let merge = config.merge.unwrap();
let shell_cmd = merge.commands[0].shell.as_ref().unwrap();
assert!(shell_cmd.contains("${merge.worktree}"));
assert!(shell_cmd.contains("${merge.source_branch}"));
assert!(shell_cmd.contains("${merge.target_branch}"));
assert!(shell_cmd.contains("${merge.session_id}"));
let claude_cmd = merge.commands[1].claude.as_ref().unwrap();
assert!(claude_cmd.contains("${merge.source_branch}"));
assert!(claude_cmd.contains("${merge.target_branch}"));
}
#[tokio::test]
async fn test_merge_workflow_execution_order() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let subprocess = SubprocessManager::production();
let temp_output = TempDir::new()?;
let output_path = temp_output.path().to_path_buf();
let merge_workflow = MergeWorkflow {
commands: vec![
WorkflowStep {
shell: Some(format!("echo '1' > {}/order.txt", output_path.display())),
..Default::default()
},
WorkflowStep {
shell: Some(format!("echo '2' >> {}/order.txt", output_path.display())),
..Default::default()
},
WorkflowStep {
shell: Some(format!("echo '3' >> {}/order.txt", output_path.display())),
..Default::default()
},
],
timeout: Some(60),
};
let manager = WorktreeManager::with_config(
repo_path,
subprocess,
0,
Some(merge_workflow),
HashMap::new(),
)?;
let _session = manager.create_session().await?;
Ok(())
}
#[test]
fn test_merge_workflow_timeout_configuration() {
let yaml_default = r#"
name: test
mode: mapreduce
map:
input: test.json
agent_template:
- shell: "echo test"
merge:
- shell: "git merge"
"#;
let config_default = parse_mapreduce_workflow(yaml_default).unwrap();
assert_eq!(config_default.merge.unwrap().timeout, None);
let yaml_custom = r#"
name: test
mode: mapreduce
map:
input: test.json
agent_template:
- shell: "echo test"
merge:
commands:
- shell: "git merge"
timeout: 1200
"#;
let config_custom = parse_mapreduce_workflow(yaml_custom).unwrap();
assert_eq!(config_custom.merge.unwrap().timeout, Some(1200)); }
#[tokio::test]
async fn test_merge_workflow_with_mapreduce_context() -> Result<()> {
let yaml = r#"
name: mapreduce-with-merge
mode: mapreduce
map:
input: '["item1", "item2"]'
json_path: "$[*]"
agent_template:
- shell: "echo 'Processing ${item}'"
max_parallel: 2
reduce:
- shell: "echo 'Aggregating results'"
merge:
- shell: "echo 'Merging MapReduce results from ${merge.worktree}'"
- claude: "/finalize-merge ${merge.session_id}"
"#;
let config = parse_mapreduce_workflow(yaml)?;
assert!(config.is_mapreduce());
assert!(config.merge.is_some());
let merge = config.merge.unwrap();
assert_eq!(merge.commands.len(), 2);
assert!(merge.commands[0]
.shell
.as_ref()
.unwrap()
.contains("${merge.worktree}"));
assert!(merge.commands[1]
.claude
.as_ref()
.unwrap()
.contains("${merge.session_id}"));
Ok(())
}
#[tokio::test]
async fn test_merge_workflow_with_regular_workflow() -> Result<()> {
let yaml = r#"
name: regular-with-merge
mode: normal
map:
input: dummy.json
agent_template:
- shell: "echo dummy"
merge:
- shell: "git pull origin main"
- shell: "cargo build --release"
- claude: "/merge-to-main ${merge.source_branch}"
"#;
let config = parse_mapreduce_workflow(yaml)?;
assert!(config.merge.is_some());
let merge = config.merge.unwrap();
assert_eq!(merge.commands.len(), 3);
assert!(merge.commands[2]
.claude
.as_ref()
.unwrap()
.contains("${merge.source_branch}"));
Ok(())
}
#[test]
fn test_merge_workflow_error_handling() {
let invalid_yaml = r#"
name: test
mode: mapreduce
map:
input: test.json
agent_template:
- shell: "echo test"
merge:
invalid_field: "should fail"
"#;
let result = parse_mapreduce_workflow(invalid_yaml);
assert!(result.is_err());
let empty_yaml = r#"
name: test
mode: mapreduce
map:
input: test.json
agent_template:
- shell: "echo test"
merge: []
"#;
let empty_config = parse_mapreduce_workflow(empty_yaml).unwrap();
assert_eq!(empty_config.merge.unwrap().commands.len(), 0);
}
#[tokio::test]
async fn test_merge_workflow_environment_variables() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let subprocess = SubprocessManager::production();
let merge_workflow = MergeWorkflow {
commands: vec![WorkflowStep {
shell: Some("printenv | grep PRODIGY || true".to_string()),
..Default::default()
}],
timeout: Some(60),
};
for verbosity in [0, 1, 2] {
let manager = WorktreeManager::with_config(
repo_path.clone(),
subprocess.clone(),
verbosity,
Some(merge_workflow.clone()),
HashMap::new(),
)?;
let _session = manager.create_session().await?;
}
Ok(())
}
#[tokio::test]
async fn test_workflow_env_vars_in_merge() -> Result<()> {
let (_temp_dir, repo_path) = create_test_repo().await?;
let subprocess = SubprocessManager::production();
let merge_workflow = MergeWorkflow {
commands: vec![
WorkflowStep {
shell: Some("echo 'Project: ${PROJECT_NAME}'".to_string()),
..Default::default()
},
WorkflowStep {
shell: Some("echo 'Dir: ${BOOK_DIR}'".to_string()),
..Default::default()
},
],
timeout: Some(60),
};
let mut workflow_env = HashMap::new();
workflow_env.insert("PROJECT_NAME".to_string(), "TestProject".to_string());
workflow_env.insert("BOOK_DIR".to_string(), "book".to_string());
let manager =
WorktreeManager::with_config(repo_path, subprocess, 0, Some(merge_workflow), workflow_env)?;
let _session = manager.create_session().await?;
Ok(())
}