#include "pipeline.h" #include "runner.h" #include #include #include #include #include #include static const PIString testDir = "/tmp/pipeline-runner-test/"; static const PIString testPipelinesDir = testDir + "pipelines/"; static const PIString testLogsDir = testDir + "logs/"; static void setupDirs() { PIDir::make(testDir); PIDir::make(testPipelinesDir); PIDir::make(testLogsDir); } static void cleanupDirs() { PIDir pdir(testPipelinesDir); for (const auto & e: pdir.entries()) { if (e.isFile()) PIFile::remove(e.path); } PIDir ldir(testLogsDir); for (const auto & e: ldir.entries()) { if (e.isFile()) PIFile::remove(e.path); } PIDir::remove(testPipelinesDir); PIDir::remove(testLogsDir); PIDir::remove(testDir); } static void waitForRun(PipelineRunner & runner, const PIString & runId, int timeoutMs = 5000) { for (int i = 0; i < timeoutMs / 50; ++i) { RunState state = runner.getRunState(runId); if (state.status == RunStatus::Completed || state.status == RunStatus::Error) { return; } piMSleep(50); } } static Pipeline makeEchoPipeline() { Pipeline pl; pl.id = "echo-test"; pl.name = "Echo Test"; pl.working_dir = "/tmp"; pl.created_at = nowISO(); pl.updated_at = pl.created_at; Prompt p; p.id = "prompt-1"; p.text = "echo hello"; p.title = "Echo Step"; p.order = 0; pl.prompts << p; return pl; } TEST(RunnerTest, StartRunNotFound) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); PIString runId = runner.startRun("nonexistent"); EXPECT_TRUE(runId.isEmpty()); cleanupDirs(); } TEST(RunnerTest, StartRunAndGetState) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); // Create a pipeline that runs 'echo hello' (uses shell command, not opencode) Pipeline pl; pl.id = "state-test"; pl.name = "State Test"; pl.working_dir = "/tmp"; pl.created_at = nowISO(); pl.updated_at = pl.created_at; Prompt p; p.id = "p1"; p.text = "test"; p.title = "Step 1"; p.order = 0; pl.prompts << p; savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("state-test"); EXPECT_FALSE(runId.isEmpty()); RunState state = runner.getRunState(runId); EXPECT_EQ(state.run_id, runId); EXPECT_EQ(state.pipeline_id, "state-test"); EXPECT_EQ(state.status, RunStatus::Running); EXPECT_EQ(state.steps.size(), 1); EXPECT_EQ(state.steps[0].title, "Step 1"); // Wait for background thread to finish before cleanup waitForRun(runner, runId); cleanupDirs(); } TEST(RunnerTest, GetStateNotFound) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); RunState state = runner.getRunState("nonexistent"); EXPECT_TRUE(state.run_id.isEmpty()); cleanupDirs(); } TEST(RunnerTest, IsRunActive) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); Pipeline pl; pl.id = "active-test"; pl.name = "Active Test"; pl.working_dir = "/tmp"; pl.created_at = nowISO(); pl.updated_at = pl.created_at; Prompt p; p.id = "p1"; p.text = "test"; p.title = "Step 1"; p.order = 0; pl.prompts << p; savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("active-test"); EXPECT_TRUE(runner.isRunActive(runId)); // Wait for background thread to finish before cleanup waitForRun(runner, runId); cleanupDirs(); } TEST(RunnerTest, RunStatusToString) { EXPECT_EQ(runStatusToString(RunStatus::Pending), "pending"); EXPECT_EQ(runStatusToString(RunStatus::Running), "running"); EXPECT_EQ(runStatusToString(RunStatus::Completed), "completed"); EXPECT_EQ(runStatusToString(RunStatus::Error), "error"); } TEST(PersistenceTest, SaveLoadRoundTrip) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); // Test loadRunState on non-existent file RunState empty = runner.loadRunState("nonexistent"); EXPECT_TRUE(empty.run_id.isEmpty()); // Create state file with one step PIString runId = "test-run-123"; PIString statePath = testLogsDir + runId + ".state.json"; PIJSON j = PIJSON::newObject(); j["run_id"] = runId; j["pipeline_id"] = "test-pipeline"; j["status"] = "running"; j["current_step"] = 0; PIJSON stepsArr = PIJSON::newArray(); PIJSON sj = PIJSON::newObject(); sj["step_index"] = 0; sj["title"] = "Step 1"; sj["status"] = "completed"; sj["returncode"] = 0; sj["output"] = "output data"; sj["error"] = ""; stepsArr << sj; j["steps"] = stepsArr; PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8()); // Load via runner RunState loaded = runner.loadRunState(runId); EXPECT_EQ(loaded.run_id, runId); EXPECT_EQ(loaded.pipeline_id, "test-pipeline"); EXPECT_EQ(loaded.status, RunStatus::Running); EXPECT_EQ(loaded.steps.size(), 1); EXPECT_EQ(loaded.steps[0].title, "Step 1"); EXPECT_EQ(loaded.steps[0].status, RunStatus::Completed); cleanupDirs(); } TEST(PersistenceTest, RecoverRuns) { setupDirs(); // Create state file manually to simulate a running state from a previous session PIString runId = "recovered-run-456"; PIString statePath = testLogsDir + runId + ".state.json"; PIJSON j = PIJSON::newObject(); j["run_id"] = runId; j["pipeline_id"] = "some-pipeline"; j["status"] = "running"; j["current_step"] = 1; PIJSON stepsArr = PIJSON::newArray(); PIJSON sj = PIJSON::newObject(); sj["step_index"] = 0; sj["title"] = "Step 1"; sj["status"] = "completed"; sj["returncode"] = 0; sj["output"] = ""; sj["error"] = ""; stepsArr << sj; PIJSON sj2 = PIJSON::newObject(); sj2["step_index"] = 1; sj2["title"] = "Step 2"; sj2["status"] = "pending"; sj2["returncode"] = 0; sj2["output"] = ""; sj2["error"] = ""; stepsArr << sj2; j["steps"] = stepsArr; PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8()); // Also create a completed state file that should NOT be recovered PIString completedRunId = "completed-run-789"; PIString completedStatePath = testLogsDir + completedRunId + ".state.json"; PIJSON j2 = PIJSON::newObject(); j2["run_id"] = completedRunId; j2["pipeline_id"] = "some-pipeline"; j2["status"] = "completed"; j2["current_step"] = 0; j2["steps"] = PIJSON::newArray(); PIFile::writeAll(completedStatePath, j2.toJSON(PIJSON::Tree).toUTF8()); // New runner should recover only the running state PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); runner.recoverRuns(); RunState recovered = runner.getRunState(runId); EXPECT_EQ(recovered.run_id, runId); EXPECT_EQ(recovered.status, RunStatus::Running); EXPECT_EQ(recovered.steps.size(), 2); // Completed run should NOT be in memory after recovery EXPECT_FALSE(runner.isRunActive(completedRunId)); cleanupDirs(); } TEST(PersistenceTest, CompletedRunStatePersisted) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); Pipeline pl; pl.id = "persist-test"; pl.name = "Persist Test"; pl.working_dir = "/tmp"; pl.created_at = nowISO(); pl.updated_at = pl.created_at; Prompt p; p.id = "p1"; p.text = "test"; p.title = "Step 1"; p.order = 0; pl.prompts << p; savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("persist-test"); EXPECT_FALSE(runId.isEmpty()); waitForRun(runner, runId); // Verify state file exists and contains final status PIString statePath = testLogsDir + runId + ".state.json"; EXPECT_TRUE(PIFile::isExists(statePath)); RunState loaded = runner.loadRunState(runId); EXPECT_EQ(loaded.run_id, runId); cleanupDirs(); } #ifdef FAKE static Pipeline makeFakePipeline(const PIString & id, const PIString & name, int numSteps) { Pipeline pl; pl.id = id; pl.name = name; pl.working_dir = "/tmp"; pl.created_at = nowISO(); pl.updated_at = pl.created_at; for (int i = 0; i < numSteps; ++i) { Prompt p; p.id = "p" + PIString::fromNumber(i); p.text = "fake text " + PIString::fromNumber(i); p.title = "Fake Step " + PIString::fromNumber(i); p.order = i; pl.prompts << p; } return pl; } TEST(FakeRunnerTest, FullPipelineCompletes) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); Pipeline pl = makeFakePipeline("fake-full", "Fake Full Pipeline", 3); savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("fake-full"); EXPECT_FALSE(runId.isEmpty()); // Wait for pipeline to finish (3 steps * 100ms = 300ms, allow extra time) for (int i = 0; i < 100; ++i) { piMSleep(50); RunState state = runner.getRunState(runId); if (state.status == RunStatus::Completed) { bool allDone = true; for (int j = 0; j < state.steps.size(); ++j) { if (state.steps[j].status != RunStatus::Completed) { allDone = false; break; } } if (allDone) break; } if (state.status == RunStatus::Error) break; } RunState state = runner.getRunState(runId); EXPECT_EQ(state.status, RunStatus::Completed); EXPECT_EQ(state.steps.size(), 3); for (int i = 0; i < 3; ++i) { EXPECT_EQ(state.steps[i].status, RunStatus::Completed); EXPECT_EQ(state.steps[i].returncode, 0); } // Wait for background thread to finish before cleanup waitForRun(runner, runId, 2000); cleanupDirs(); } TEST(FakeRunnerTest, RunStatusTransitions) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); Pipeline pl = makeFakePipeline("fake-transition", "Fake Transition Pipeline", 2); savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("fake-transition"); EXPECT_FALSE(runId.isEmpty()); // Initially running RunState state = runner.getRunState(runId); EXPECT_EQ(state.status, RunStatus::Running); // Wait for completion (2 steps * 100ms = 200ms, allow extra time) for (int i = 0; i < 100; ++i) { piMSleep(50); state = runner.getRunState(runId); if (state.status == RunStatus::Completed) { bool allDone = true; for (int j = 0; j < state.steps.size(); ++j) { if (state.steps[j].status != RunStatus::Completed) { allDone = false; break; } } if (allDone) break; } if (state.status == RunStatus::Error) break; } state = runner.getRunState(runId); EXPECT_EQ(state.status, RunStatus::Completed); EXPECT_FALSE(runner.isRunActive(runId)); // Wait for background thread to finish before cleanup waitForRun(runner, runId, 2000); cleanupDirs(); } TEST(FakeRunnerTest, StepOutputContainsFakeMarker) { setupDirs(); PipelineRunner runner(testPipelinesDir, testLogsDir, "opencode", 300); Pipeline pl = makeFakePipeline("fake-output", "Fake Output Pipeline", 2); savePipeline(testPipelinesDir, pl); PIString runId = runner.startRun("fake-output"); EXPECT_FALSE(runId.isEmpty()); // Wait for completion (2 steps * 100ms = 200ms, allow extra time) for (int i = 0; i < 100; ++i) { piMSleep(50); RunState state = runner.getRunState(runId); if (state.status == RunStatus::Completed) { bool allDone = true; for (int j = 0; j < state.steps.size(); ++j) { if (state.steps[j].status != RunStatus::Completed) { allDone = false; break; } } if (allDone) break; } if (state.status == RunStatus::Error) break; } RunState state = runner.getRunState(runId); EXPECT_EQ(state.status, RunStatus::Completed); for (int i = 0; i < state.steps.size(); ++i) { EXPECT_TRUE(state.steps[i].output.contains("FAKE")); EXPECT_TRUE(state.steps[i].output.contains("Fake Step " + PIString::fromNumber(i))); EXPECT_TRUE(state.steps[i].error.isEmpty()); } // Wait for background thread to finish before cleanup waitForRun(runner, runId, 2000); cleanupDirs(); } #endif