pig2014/honey-biscuit-workshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(workshop-engine): replace tokio Child with ManagedChild trait

Browse Source 
for testability

- Add `ManagedChild` trait abstracting process lifecycle (real
  `TokioChild` + `MockChild` for tests)
- Use Unix process groups (`process_group(0)` + `libc::kill(-pgid,
  SIGKILL)`) for tree-wide termination
- Remove cgroup-related code (`CgroupManager`, `ResourceUsage`,
  `cgroup_path` field)
- Update imports across test files to use `workshop-engine` crate
  directly
This commit is contained in:
Catty Steve
2026-04-27 12:02:17 +08:00
parent a64e47d772
commit 50b42e654c
13 changed files with 555 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workshop-baker/Cargo.lock
Generated
+1
View File
@@ -5302,6 +5302,7 @@ dependencies = [
"chrono",
"duration-str",
"lazy_static",
"libc",
"log",
"os_info",
"serde",
workshop-baker/Cargo.toml
-4
View File
@@ -15,10 +15,6 @@ path = "src/main.rs"
default = []
integration-tests = []
[[test]]
name = "engine_integration"
path = "tests/engine_integration.rs"
[dependencies]
workshop-engine = { path = "../workshop-engine" }
anyhow = "1.0.100"
workshop-baker/tests/finalize_hook_integration.rs
+2 -3
View File
@@ -5,10 +5,10 @@
use std::collections::HashMap;
use std::time::Duration;
use workshop_baker::engine::{ExecutionContext, ExecutionEvent};
use workshop_baker::ExecutionContext;
use workshop_baker::finalize::plugin::PluginMap;
use workshop_baker::finalize::stage::hook::hook;
use workshop_baker::types::command::CustomCommand;
use workshop_engine::{ExecutionEvent, CustomCommand};
fn create_test_context() -> ExecutionContext {
ExecutionContext {
@@ -18,7 +18,6 @@ fn create_test_context() -> ExecutionContext {
working_dir: std::env::temp_dir(),
env_vars: HashMap::new(),
timeout: Some(Duration::from_secs(30)),
cgroup_path: None,
shell: "bash".to_string(),
dry_run: false,
privileged: false,
workshop-baker/tests/finalize_pipeline_integration.rs
+2 -2
View File
@@ -6,7 +6,8 @@
use std::collections::HashMap;
use std::time::Duration;
use workshop_baker::cli::Cli;
use workshop_baker::engine::{ExecutionContext, ExecutionEvent};
use workshop_baker::ExecutionContext;
use workshop_engine::ExecutionEvent;
use workshop_baker::finalize::finalize;
fn create_test_context() -> ExecutionContext {
@@ -17,7 +18,6 @@ fn create_test_context() -> ExecutionContext {
working_dir: std::env::temp_dir(),
env_vars: HashMap::new(),
timeout: Some(Duration::from_secs(30)),
cgroup_path: None,
shell: "bash".to_string(),
dry_run: false,
privileged: false,
workshop-baker/tests/prebake_config_integration.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use workshop_baker::prebake::parse;
use workshop_baker::types::builderconfig::BuilderType;
use workshop_baker::prebake::types::builderconfig::BuilderType;
#[test]
fn test_prebake_config_minimal_yaml() {
workshop-baker/tests/types_config_integration.rs
+5 -6
View File
@@ -1,9 +1,8 @@
use workshop_baker::types::architecture::Architecture;
use workshop_baker::types::builderconfig::BuilderConfig;
use workshop_baker::types::cache::{CacheDirectory, CacheMode, CacheStrategyType};
use workshop_baker::types::command::CustomCommand;
use workshop_baker::types::compression::CompressionMethod;
use workshop_baker::types::repology::RepologyEndpoint;
use workshop_baker::prebake::types::architecture::Architecture;
use workshop_baker::prebake::types::builderconfig::BuilderConfig;
use workshop_baker::prebake::types::cache::{CacheDirectory, CacheMode, CacheStrategyType};
use workshop_baker::finalize::types::compression::CompressionMethod;
use workshop_engine::{CustomCommand, RepologyEndpoint};
#[test]
fn test_architecture_from_str() {
workshop-engine/Cargo.lock
Generated
+1
View File
@@ -869,6 +869,7 @@ dependencies = [
"chrono",
"duration-str",
"lazy_static",
"libc",
"log",
"os_info",
"serde",
workshop-engine/Cargo.toml
+4
View File
@@ -19,3 +19,7 @@ lazy_static = "1.5.0"
log = "0.4.28"
os_info = "3.14.0"
duration-str = "0.21.0"
libc = "0.2"
[dev-dependencies]
tokio = { version = "1.48.0", features = ["full"] }
workshop-engine/src/child.rs
+205
View File
@@ -0,0 +1,205 @@
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncRead;
/// A managed child process, abstracted over the real OS process for testability.
///
/// This trait represents a spawned process that we can:
/// - Read its stdout/stderr
/// - Wait for it to finish (getting its exit code)
/// - Kill it
///
/// The real implementation wraps `tokio::process::Child`.
/// Mock implementations are used in unit tests.
pub trait ManagedChild: Send {
/// Returns the OS-assigned process ID, if available.
fn pid(&self) -> Option<u32>;
/// Takes the stdout pipe, if available.
fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>>;
/// Takes the stderr pipe, if available.
fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>>;
/// Waits for the process to exit and returns its exit code.
fn wait(&mut self) -> impl std::future::Future<Output = io::Result<i32>> + Send;
/// Kills the process.
fn kill(&mut self) -> impl std::future::Future<Output = io::Result<()>> + Send;
}
// ---------------------------------------------------------------------------
// Real implementation: wraps tokio::process::Child
// ---------------------------------------------------------------------------
pub struct TokioChild(pub tokio::process::Child);
impl ManagedChild for TokioChild {
fn pid(&self) -> Option<u32> {
self.0.id()
}
fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
self.0.stdout.take().map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin + Send>)
}
fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
self.0.stderr.take().map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin + Send>)
}
async fn wait(&mut self) -> io::Result<i32> {
self.0.wait().await.map(|s| s.code().unwrap_or(-1))
}
async fn kill(&mut self) -> io::Result<()> {
// Kill the entire process group. When combined with process_group(0)
// in executor.rs, the spawned command is the process group leader,
// so PGID = PID. Sending SIGKILL to -PGID terminates the whole tree.
#[cfg(unix)]
if let Some(pid) = self.0.id() {
let pgid = pid as i32;
let ret = unsafe { libc::kill(-pgid, libc::SIGKILL) };
if ret != 0 {
let err = io::Error::last_os_error();
// ESRCH = process group already gone, which is fine
if err.raw_os_error() != Some(libc::ESRCH) {
log::warn!("kill(-{}, SIGKILL) failed: {}", pgid, err);
}
}
}
// Fallback: also try tokio's per-process kill
let _ = self.0.kill().await;
Ok(())
}
}
// ---------------------------------------------------------------------------
// Mock implementation: for testing Engine orchestration logic
// ---------------------------------------------------------------------------
/// A mock child that returns pre-configured results without spawning a real process.
///
/// Use `MockChildHandle` for advanced scenarios (timeout simulation, etc.).
pub struct MockChild {
/// The exit code that `wait()` returns.
pub exit_code: i32,
/// Whether `kill()` was called.
pub killed: bool,
/// Simulated stdout content.
pub stdout_data: Vec<u8>,
/// Simulated stderr content.
pub stderr_data: Vec<u8>,
/// Simulated PID.
pub pid: u32,
/// When true, `wait()` never returns (simulates a hanging process for timeout testing).
pub hang_forever: bool,
}
impl Default for MockChild {
fn default() -> Self {
Self {
exit_code: 0,
killed: false,
stdout_data: Vec::new(),
stderr_data: Vec::new(),
pid: 12345,
hang_forever: false,
}
}
}
impl MockChild {
pub fn with_exit_code(code: i32) -> Self {
Self {
exit_code: code,
..Default::default()
}
}
pub fn with_output(stdout: &str, stderr: &str) -> Self {
Self {
stdout_data: stdout.as_bytes().to_vec(),
stderr_data: stderr.as_bytes().to_vec(),
..Default::default()
}
}
pub fn hanging() -> Self {
Self {
hang_forever: true,
..Default::default()
}
}
}
impl ManagedChild for MockChild {
fn pid(&self) -> Option<u32> {
Some(self.pid)
}
fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
if self.stdout_data.is_empty() {
None
} else {
let data = std::mem::take(&mut self.stdout_data);
Some(Box::new(VecReader::new(data)))
}
}
fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
if self.stderr_data.is_empty() {
None
} else {
let data = std::mem::take(&mut self.stderr_data);
Some(Box::new(VecReader::new(data)))
}
}
async fn wait(&mut self) -> io::Result<i32> {
if self.hang_forever {
// Park this future indefinitely to simulate a timeout
std::future::pending::<()>().await;
unreachable!()
}
Ok(self.exit_code)
}
async fn kill(&mut self) -> io::Result<()> {
self.killed = true;
Ok(())
}
}
// ---------------------------------------------------------------------------
// VecReader: wraps Vec<u8> into tokio::io::AsyncRead
// ---------------------------------------------------------------------------
pub struct VecReader {
data: Vec<u8>,
pos: usize,
}
impl VecReader {
pub fn new(data: Vec<u8>) -> Self {
Self { data, pos: 0 }
}
}
impl AsyncRead for VecReader {
fn poll_read(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
if self.pos >= self.data.len() {
return Poll::Ready(Ok(()));
}
let available = &self.data[self.pos..];
let len = available.len().min(buf.remaining());
buf.put_slice(&available[..len]);
self.pos += len;
Poll::Ready(Ok(()))
}
}
workshop-engine/src/executor.rs
+256 -299
View File
@@ -6,53 +6,16 @@ use tokio::process::Command;
use tokio::sync::mpsc;
use tokio::time::timeout;
use crate::child::{ManagedChild, TokioChild};
use crate::pm::PackageManager;
use crate::types::{
CgroupManager, DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError,
ExecutionEvent, ExecutionResult, ResourceLimits, StreamType,
DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError,
ExecutionEvent, ExecutionResult, StreamType,
};
impl crate::types::Engine {
pub fn new() -> Self {
Self {
cgroup_manager: None,
}
}
pub fn with_cgroup(limits: ResourceLimits, build_id: &str) -> Result<Self, String> {
let cgroup_manager =
CgroupManager::new(build_id).map_err(|e| format!("Failed to create cgroup: {}", e))?;
if let Some(mem) = limits.memory_bytes {
cgroup_manager
.set_memory_limit(mem)
.map_err(|e| format!("Failed to set memory limit: {}", e))?;
}
if let Some(weight) = limits.cpu_weight {
cgroup_manager
.set_cpu_weight(weight)
.map_err(|e| format!("Failed to set cpu weight: {}", e))?;
}
if let Some(max) = limits.cpu_max_us {
cgroup_manager
.set_cpu_max(max, limits.cpu_period_us)
.map_err(|e| format!("Failed to set cpu max: {}", e))?;
}
Ok(Self {
cgroup_manager: Some(cgroup_manager),
})
}
fn apply_cgroup(&self, pid: u32) -> Result<(), ExecutionError> {
if let Some(ref cgroup) = self.cgroup_manager {
cgroup.add_process(pid).map_err(|e| {
ExecutionError::IoError(format!("Failed to add process to cgroup: {}", e))
})?;
}
Ok(())
Self
}
pub async fn execute_script(
@@ -129,11 +92,51 @@ impl crate::types::Engine {
.stderr(Stdio::piped());
}
/// Execute a pre-configured command, managing the full lifecycle.
///
/// This method:
/// 1. Configures the command with context (env vars, working dir, stdio)
/// 2. Creates a new process group on Unix for process-tree isolation
/// 3. Returns early in dry_run mode
/// 4. Spawns the OS process
/// 5. Delegates to `execute_child` for wait/event/output management
pub async fn execute(
&self,
command: Command,
ctx: &ExecutionContext,
event_tx: &EventSender,
) -> Result<ExecutionResult, ExecutionError> {
let mut command = command;
self.construct_command(&mut command, ctx);
if ctx.dry_run {
log::info!("[DRY_RUN] {:?}", command);
return Ok(ExecutionResult {
..Default::default()
});
}
// Create a new process group so that kill() can terminate the
// entire process tree (bash + its children) on timeout.
#[cfg(unix)]
command.process_group(0);
let child = command.spawn().map_err(|e| {
ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e))
})?;
self.execute_child(TokioChild(child), ctx, event_tx).await
}
/// Core execution logic, abstracted over [`ManagedChild`] for testability.
///
/// Handles: timeout, event emission, stdout/stderr capture, result construction.
/// Does NOT spawn processes itself — that is done by [`execute`].
pub async fn execute_child(
&self,
mut child: impl ManagedChild,
ctx: &ExecutionContext,
event_tx: &EventSender,
) -> Result<ExecutionResult, ExecutionError> {
let start_time = std::time::Instant::now();
@@ -145,25 +148,6 @@ impl crate::types::Engine {
})
.await;
}
let mut command = command;
self.construct_command(&mut command, ctx);
if ctx.dry_run {
log::info!("[DRY_RUN] {:?}", command);
return Ok(ExecutionResult {
..Default::default()
});
}
let mut child = command.spawn().map_err(|e| {
ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e))
})?;
let pid = child
.id()
.ok_or_else(|| ExecutionError::IoError("Failed to get PID".to_string()))?;
self.apply_cgroup(pid)?;
let task_id = ctx.task_id.clone();
let (stdout_tx, mut stdout_rx) = mpsc::channel::<String>(1024);
@@ -171,11 +155,10 @@ impl crate::types::Engine {
let mut stdout_buf = String::new();
let mut stderr_buf = String::new();
let mut child_stdout = child.stdout.take();
let mut child_stderr = child.stderr.take();
let mut child_stdout = child.take_stdout();
let mut child_stderr = child.take_stderr();
let tx_clone = stdout_tx.clone();
let _task_id_clone = task_id.clone();
let stdout_handle = tokio::spawn(async move {
if let Some(mut stdout) = child_stdout.take() {
let mut reader = tokio::io::BufReader::new(&mut stdout);
@@ -215,9 +198,9 @@ impl crate::types::Engine {
}
});
let status = if let Some(timeout_duration) = ctx.timeout {
let exit_code = if let Some(timeout_duration) = ctx.timeout {
match timeout(timeout_duration, child.wait()).await {
Ok(Ok(status)) => status,
Ok(Ok(code)) => code,
Ok(Err(e)) => {
return Err(ExecutionError::IoError(format!(
"Failed to wait for process: {}",
@@ -289,7 +272,6 @@ impl crate::types::Engine {
}
}
let exit_code = status.code().unwrap_or(-1);
let duration = start_time.elapsed();
let result = ExecutionResult {
@@ -299,7 +281,6 @@ impl crate::types::Engine {
duration,
stdout: stdout_buf,
stderr: stderr_buf,
resource_usage: None,
};
if let Some(tx) = event_tx {
@@ -342,13 +323,17 @@ impl crate::types::Engine {
duration: std::time::Duration::default(),
stdout: String::new(),
stderr: String::new(),
resource_usage: None,
};
for cmd in cmds {
for (i, cmd) in cmds.into_iter().enumerate() {
log::debug!("Installing dependency with command: {:?}", cmd);
let result = self.execute(cmd, ctx, event_tx).await?;
// TODO: Aggregate results properly instead of just taking the last one
// `execute()` returns Err for non-zero exit codes, so this code only
// runs for successful commands (exit_code == 0). We keep the first
// command's exit_code as it's semantically the "primary" result;
// duration and output are accumulated across all commands.
if i == 0 {
combined_result.exit_code = result.exit_code;
}
combined_result.success &= result.success;
combined_result.duration += result.duration;
combined_result.stdout.push_str(&result.stdout);
@@ -356,15 +341,6 @@ impl crate::types::Engine {
}
Ok(combined_result)
}
pub async fn cleanup(&self) -> Result<(), String> {
if let Some(ref cgroup) = self.cgroup_manager {
cgroup
.destroy()
.map_err(|e| format!("Failed to destroy cgroup: {}", e))?;
}
Ok(())
}
}
impl Default for crate::types::Engine {
@@ -376,131 +352,19 @@ impl Default for crate::types::Engine {
#[cfg(test)]
mod tests {
use super::*;
use std::collections::VecDeque;
use crate::child::MockChild;
use crate::Engine;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;
#[derive(Clone, Debug)]
struct CapturedCall {
program: String,
args: Vec<String>,
env_vars: std::collections::HashMap<String, String>,
working_dir: PathBuf,
}
#[derive(Clone)]
struct MockExecutor {
calls: Arc<Mutex<VecDeque<CapturedCall>>>,
}
impl MockExecutor {
fn new() -> Self {
Self {
calls: Arc::new(Mutex::new(VecDeque::new())),
}
}
fn capture(
&self,
program: String,
args: Vec<String>,
env_vars: std::collections::HashMap<String, String>,
working_dir: PathBuf,
) {
self.calls.lock().unwrap().push_back(CapturedCall {
program,
args,
env_vars,
working_dir,
});
}
fn pop(&self) -> Option<CapturedCall> {
self.calls.lock().unwrap().pop_front()
}
}
struct TestExecutor {
mock: MockExecutor,
}
impl TestExecutor {
fn new() -> Self {
Self {
mock: MockExecutor::new(),
}
}
async fn execute_command(
&self,
cmd: String,
ctx: &ExecutionContext,
_event_tx: EventSender,
) -> Result<ExecutionResult, ExecutionError> {
let mut env_vars = ctx.env_vars.clone();
env_vars.insert("WS_TASKID".to_string(), ctx.task_id.clone());
env_vars.insert("WS_PIPELINE".to_string(), ctx.pipeline_name.clone());
env_vars.insert("WS_USERNAME".to_string(), ctx.username.clone());
self.mock.capture(
ctx.shell.clone(),
vec!["-c".to_string(), cmd.clone()],
env_vars,
ctx.working_dir.clone(),
);
Ok(ExecutionResult {
task_id: ctx.task_id.clone(),
exit_code: 0,
success: true,
duration: Duration::from_millis(10),
stdout: String::new(),
stderr: String::new(),
resource_usage: None,
})
}
async fn execute_script(
&self,
script_path: &PathBuf,
ctx: &ExecutionContext,
_event_tx: EventSender,
) -> Result<ExecutionResult, ExecutionError> {
let abs_path = if script_path.is_absolute() {
script_path.clone()
} else {
ctx.working_dir.join(script_path)
};
self.mock.capture(
ctx.shell.clone(),
vec!["-c".to_string(), abs_path.to_string_lossy().into_owned()],
ctx.env_vars.clone(),
ctx.working_dir.clone(),
);
Ok(ExecutionResult {
task_id: ctx.task_id.clone(),
exit_code: 0,
success: true,
duration: Duration::from_millis(10),
stdout: String::new(),
stderr: String::new(),
resource_usage: None,
})
}
}
fn create_test_context() -> ExecutionContext {
fn test_ctx() -> ExecutionContext {
ExecutionContext {
task_id: "test-task-1".to_string(),
task_id: "test-task".to_string(),
pipeline_name: "test-pipeline".to_string(),
username: "testuser".to_string(),
working_dir: PathBuf::from("/tmp"),
env_vars: std::collections::HashMap::new(),
timeout: Some(Duration::from_secs(30)),
cgroup_path: None,
timeout: None,
shell: "bash".to_string(),
dry_run: false,
privileged: false,
@@ -508,152 +372,245 @@ mod tests {
}
}
#[tokio::test]
async fn test_execute_command_captures_args() {
let executor = TestExecutor::new();
let ctx = create_test_context();
// ------------------------------------------------------------------
// TaskStarted event
// ------------------------------------------------------------------
let result = executor
.execute_command("echo hello".to_string(), &ctx, None)
.await;
#[tokio::test]
async fn send_task_started_event_when_channel_provided() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::with_exit_code(0);
let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
assert!(matches!(
rx.try_recv(),
Ok(ExecutionEvent::TaskStarted { .. })
));
}
#[tokio::test]
async fn skip_task_started_event_when_channel_is_none() {
let engine = Engine::new();
let child = MockChild::with_exit_code(0);
let result = engine.execute_child(child, &test_ctx(), &None).await;
assert!(result.is_ok());
let call = executor.mock.pop();
assert!(call.is_some());
let call = call.unwrap();
assert_eq!(call.program, "bash");
assert!(call.args.contains(&"-c".to_string()));
assert!(call.args.contains(&"echo hello".to_string()));
}
#[tokio::test]
async fn test_execute_script_captures_path() {
let executor = TestExecutor::new();
let ctx = create_test_context();
// ------------------------------------------------------------------
// Exit code → success / failure mapping
// ------------------------------------------------------------------
let script_path = PathBuf::from("/tmp/test_script.sh");
let result = executor.execute_script(&script_path, &ctx, None).await;
#[tokio::test]
async fn exit_zero_returns_ok() {
let engine = Engine::new();
let child = MockChild::with_exit_code(0);
let result = engine.execute_child(child, &test_ctx(), &None).await;
assert!(result.is_ok());
let call = executor.mock.pop().unwrap();
assert!(call.args.contains(&"-c".to_string()));
assert!(call.args.iter().any(|a| a.contains("test_script.sh")));
assert_eq!(result.unwrap().exit_code, 0);
}
#[tokio::test]
async fn test_execute_command_with_env_vars() {
let executor = TestExecutor::new();
let mut ctx = create_test_context();
ctx.env_vars
.insert("TEST_VAR".to_string(), "test_value".to_string());
async fn exit_nonzero_returns_err() {
let engine = Engine::new();
let child = MockChild::with_exit_code(1);
executor
.execute_command("echo $TEST_VAR".to_string(), &ctx, None)
.await
.unwrap();
let result = engine.execute_child(child, &test_ctx(), &None).await;
let call = executor.mock.pop().unwrap();
assert_eq!(
call.env_vars.get("TEST_VAR"),
Some(&"test_value".to_string())
);
assert!(result.is_err());
assert!(matches!(result.unwrap_err(), ExecutionError::ExecutionFailed(_)));
}
#[tokio::test]
async fn test_execute_includes_ws_environment() {
let executor = TestExecutor::new();
let ctx = create_test_context();
async fn exit_nonzero_sends_task_failed_event() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::with_exit_code(42);
executor
.execute_command("ls".to_string(), &ctx, None)
.await
.unwrap();
let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
let call = executor.mock.pop().unwrap();
assert!(call.env_vars.contains_key("WS_TASKID"));
assert!(call.env_vars.contains_key("WS_PIPELINE"));
assert!(call.env_vars.contains_key("WS_USERNAME"));
// Skip TaskStarted
let _ = rx.recv().await;
let event = rx.recv().await;
assert!(matches!(event, Some(ExecutionEvent::TaskFailed { .. })));
if let Some(ExecutionEvent::TaskFailed { error, .. }) = event {
assert!(error.contains("42"), "error should mention exit code 42");
}
}
#[tokio::test]
async fn test_execute_command_preserves_working_dir() {
let executor = TestExecutor::new();
let mut ctx = create_test_context();
ctx.working_dir = PathBuf::from("/custom/path");
async fn exit_zero_sends_task_completed_event() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::with_exit_code(0);
executor
.execute_command("pwd".to_string(), &ctx, None)
.await
.unwrap();
let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
let call = executor.mock.pop().unwrap();
assert_eq!(call.working_dir, PathBuf::from("/custom/path"));
let _ = rx.recv().await; // skip TaskStarted
let event = rx.recv().await;
assert!(matches!(event, Some(ExecutionEvent::TaskCompleted { .. })));
}
// ------------------------------------------------------------------
// Stdout / stderr capture
// ------------------------------------------------------------------
#[tokio::test]
async fn captures_stdout() {
let engine = Engine::new();
let child = MockChild::with_output("hello world\n", "");
let ctx = test_ctx();
let result = engine.execute_child(child, &ctx, &None).await.unwrap();
assert_eq!(result.stdout, "hello world\n");
}
#[tokio::test]
async fn test_local_executor_new_creates_default() {
let executor = super::super::types::Engine::new();
assert!(executor.cgroup_manager.is_none());
async fn captures_stderr() {
let engine = Engine::new();
let child = MockChild::with_output("", "error msg\n");
let ctx = test_ctx();
let result = engine.execute_child(child, &ctx, &None).await.unwrap();
assert_eq!(result.stderr, "error msg\n");
}
#[tokio::test]
async fn test_execution_context_default() {
let ctx = ExecutionContext::default();
async fn captures_both_streams() {
let engine = Engine::new();
let child = MockChild::with_output("out\n", "err\n");
let ctx = test_ctx();
assert!(ctx.task_id.is_empty());
assert_eq!(ctx.shell, "bash");
let result = engine.execute_child(child, &ctx, &None).await.unwrap();
assert_eq!(result.stdout, "out\n");
assert_eq!(result.stderr, "err\n");
}
// ------------------------------------------------------------------
// OutputChunk events
// ------------------------------------------------------------------
#[tokio::test]
async fn sends_output_chunk_events() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::with_output("chunk1\nchunk2\n", "");
let ctx = test_ctx();
let _ = engine.execute_child(child, &ctx, &Some(tx)).await;
let _ = rx.recv().await; // skip TaskStarted
let mut found_stdout = false;
while let Ok(event) = rx.try_recv() {
if matches!(&event, ExecutionEvent::OutputChunk { stream: StreamType::Stdout, .. }) {
found_stdout = true;
}
}
assert!(found_stdout, "Should have received at least one OutputChunk for stdout");
}
// ------------------------------------------------------------------
// Timeout handling
// ------------------------------------------------------------------
#[tokio::test]
async fn timeout_kills_child_and_returns_err() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::hanging();
let mut ctx = test_ctx();
ctx.timeout = Some(Duration::from_millis(10));
let result = engine.execute_child(child, &ctx, &Some(tx)).await;
assert!(result.is_err());
assert!(matches!(result.unwrap_err(), ExecutionError::Timeout));
// Should have received TaskFailed for timeout
let _ = rx.recv().await; // skip TaskStarted
let event = rx.recv().await;
assert!(matches!(event, Some(ExecutionEvent::TaskFailed { error, .. }) if error == "Timeout"));
}
// ------------------------------------------------------------------
// Result fields
// ------------------------------------------------------------------
#[tokio::test]
async fn result_contains_correct_task_id() {
let engine = Engine::new();
let child = MockChild::with_exit_code(0);
let ctx = test_ctx();
let result = engine.execute_child(child, &ctx, &None).await.unwrap();
assert_eq!(result.task_id, "test-task");
}
#[tokio::test]
async fn test_execution_result_success() {
let result = ExecutionResult {
task_id: "test".to_string(),
exit_code: 0,
success: true,
duration: Duration::from_secs(1),
stdout: "output".to_string(),
stderr: "".to_string(),
resource_usage: None,
};
async fn result_contains_duration() {
let engine = Engine::new();
let child = MockChild::with_exit_code(0);
let ctx = test_ctx();
assert!(result.success);
assert_eq!(result.exit_code, 0);
let result = engine.execute_child(child, &ctx, &None).await.unwrap();
assert!(result.duration.as_nanos() > 0);
}
// ------------------------------------------------------------------
// dry_run does not execute_child
// ------------------------------------------------------------------
#[tokio::test]
async fn test_execution_result_failure() {
let result = ExecutionResult {
task_id: "test".to_string(),
exit_code: 1,
success: false,
duration: Duration::from_secs(1),
stdout: "".to_string(),
stderr: "error".to_string(),
resource_usage: None,
};
async fn dry_run_returns_ok_without_spawning() {
// This tests the `execute()` path: with dry_run=true it should
// return before reaching spawn / execute_child.
let engine = Engine::new();
let mut ctx = test_ctx();
ctx.dry_run = true;
assert!(!result.success);
assert_eq!(result.exit_code, 1);
let cmd = Command::new("true");
let result = engine.execute(cmd, &ctx, &None).await;
assert!(result.is_ok());
let res = result.unwrap();
assert!(res.success);
assert_eq!(res.stdout, "");
assert_eq!(res.stderr, "");
}
// ------------------------------------------------------------------
// EventSender=None does not panic
// ------------------------------------------------------------------
#[tokio::test]
async fn test_execution_error_variants() {
let _err1 = ExecutionError::InvalidCommand("test".to_string());
let _err2 = ExecutionError::ScriptNotFound(PathBuf::from("/nonexistent"));
let _err3 = ExecutionError::PackageManagerNotFound;
let _err4 = ExecutionError::Timeout;
async fn no_event_channel_does_not_panic() {
let engine = Engine::new();
// Various exit codes
for code in [0, 1, 127] {
let child = MockChild::with_exit_code(code);
let _ = engine.execute_child(child, &test_ctx(), &None).await;
}
#[tokio::test]
async fn test_resource_limits_default() {
let limits = ResourceLimits::default();
// With output
let child = MockChild::with_output("data", "err");
let _ = engine.execute_child(child, &test_ctx(), &None).await;
assert!(limits.memory_bytes.is_none());
assert!(limits.cpu_weight.is_none());
assert_eq!(limits.cpu_period_us, 0);
// Timeout
let mut ctx = test_ctx();
ctx.timeout = Some(Duration::from_millis(10));
let child = MockChild::hanging();
let _ = engine.execute_child(child, &ctx, &None).await;
}
}
workshop-engine/src/lib.rs
+5 -2
View File
@@ -1,4 +1,4 @@
pub mod cgroups;
pub mod child;
pub mod command;
pub mod error;
pub mod executor;
@@ -8,13 +8,16 @@ pub mod time;
pub mod types;
pub mod upm;
// Re-export child types for convenience
pub use child::{ManagedChild, TokioChild};
// Convenience re-exports
pub use error::{ExecutionError, HasExitCode,
EXITCODE_OK, EXITCODE_GENERAL_ERROR, EXITCODE_INVALID_ARGS,
EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, EXITCODE_EXECUTION_ERROR,
EXITCODE_TIMEOUT, EXITCODE_PRIV_DROP_FAILED};
pub use types::{Engine, EventSender, ExecutionContext, ExecutionEvent, ExecutionResult,
StreamType, ResourceLimits, ResourceUsage, DeltaExecutionContext};
StreamType, DeltaExecutionContext};
pub use command::CustomCommand;
pub use repology::RepologyEndpoint;
pub use time::{parse_duration_to_ms, deserialize_duration_ms, deserialize_option_duration_ms};
workshop-engine/src/types.rs
+2 -25
View File
@@ -1,20 +1,11 @@
use serde::{Deserialize, Serialize};
use serde::Serialize;
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Duration;
use tokio::sync::mpsc;
pub use crate::cgroups::CgroupManager;
pub use crate::error::{EXITCODE_EXECUTION_ERROR, EXITCODE_IO_ERROR, ExecutionError};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUsage {
pub cpu_time_secs: f64,
pub max_memory_kb: u64,
pub io_read_kb: u64,
pub io_write_kb: u64,
}
#[derive(Debug, Clone)]
pub struct ExecutionContext {
pub task_id: String,
@@ -23,7 +14,6 @@ pub struct ExecutionContext {
pub working_dir: PathBuf,
pub env_vars: HashMap<String, String>,
pub timeout: Option<Duration>,
pub cgroup_path: Option<PathBuf>,
pub shell: String,
pub dry_run: bool,
pub privileged: bool,
@@ -46,7 +36,6 @@ impl Default for ExecutionContext {
working_dir: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
env_vars: HashMap::new(),
timeout: None,
cgroup_path: None,
shell: String::from("bash"),
dry_run: false,
privileged: false,
@@ -55,14 +44,6 @@ impl Default for ExecutionContext {
}
}
#[derive(Debug, Clone, Default)]
pub struct ResourceLimits {
pub memory_bytes: Option<u64>,
pub cpu_weight: Option<u64>,
pub cpu_max_us: Option<u64>,
pub cpu_period_us: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct ExecutionResult {
pub task_id: String,
@@ -71,7 +52,6 @@ pub struct ExecutionResult {
pub duration: Duration,
pub stdout: String,
pub stderr: String,
pub resource_usage: Option<ResourceUsage>,
}
impl Default for ExecutionResult {
@@ -83,7 +63,6 @@ impl Default for ExecutionResult {
duration: Duration::default(),
stdout: String::new(),
stderr: String::new(),
resource_usage: None,
}
}
}
@@ -117,6 +96,4 @@ pub enum ExecutionEvent {
pub type EventSender = Option<mpsc::Sender<ExecutionEvent>>;
pub struct Engine {
pub cgroup_manager: Option<CgroupManager>,
}
pub struct Engine;
workshop-baker/tests/engine_integration.rs → workshop-engine/tests/engine_integration.rs
+66 -5
View File
@@ -1,7 +1,7 @@
//! Integration tests for workshop-executor engine
//! Integration tests for workshop-engine
//!
//! These tests execute real commands and verify results.
//! Run with: cargo test --test engine_integration
//! Run with: cargo test --test engine_integration -p workshop-engine
//!
//! Note: These are TRUE integration tests - they run in a separate test binary
//! and execute actual commands on the system.
@@ -9,7 +9,7 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Duration;
use workshop_baker::engine::{Engine, ExecutionContext};
use workshop_engine::{Engine, ExecutionContext};
fn create_executor() -> Engine {
Engine::new()
@@ -23,7 +23,6 @@ fn create_context() -> ExecutionContext {
working_dir: std::env::temp_dir(),
env_vars: HashMap::new(),
timeout: Some(Duration::from_secs(30)),
cgroup_path: None,
shell: "bash".to_string(),
dry_run: false,
privileged: false,
@@ -173,7 +172,6 @@ async fn test_execute_respects_working_directory() {
working_dir: PathBuf::from("/tmp"),
env_vars: HashMap::new(),
timeout: Some(Duration::from_secs(30)),
cgroup_path: None,
shell: "bash".to_string(),
standalone: false,
..Default::default()
@@ -204,3 +202,66 @@ async fn test_execute_captures_stderr() {
assert!(result.success);
assert!(result.stderr.contains("error"));
}
/// Test: with `process_group(0)`, kill() propagates to the entire process tree.
///
/// We spawn a bash that creates a background sleep. With process_group(0),
/// bash is the process group leader. When we kill(-PGID), the sleep should die too.
///
/// This test uses tokio::process::Command + process_group(0) directly
/// (bypassing Engine) to verify the kernel-level behavior.
#[tokio::test]
async fn test_process_group_kill_kills_subprocesses() {
use std::os::unix::process::CommandExt;
use tokio::io::AsyncBufReadExt;
use tokio::process::Command;
use std::process::Stdio;
let mut cmd = Command::new("bash");
cmd.args(["-c", "sleep 999 & echo $!"]);
cmd.process_group(0);
cmd.stdin(Stdio::null());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let mut child = cmd.spawn().expect("spawn failed");
let bash_pid = child.id().expect("no pid");
eprintln!("[test] bash PID: {}", bash_pid);
// Read the background sleep's PID from stdout
let stdout = child.stdout.take().expect("stdout pipe");
let mut reader = tokio::io::BufReader::new(stdout);
let mut pid_line = String::new();
reader.read_line(&mut pid_line).await.expect("read pid line");
let sleep_pid: u32 = pid_line.trim().parse().expect("valid sleep PID");
eprintln!("[test] sleep PID: {}", sleep_pid);
// Verify sleep is alive before kill
let alive_before = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists();
eprintln!("[test] sleep alive before kill: {}", alive_before);
// Kill by sending SIGKILL to the NEGATIVE PGID (entire process group)
unsafe {
libc::kill(-(bash_pid as i32), libc::SIGKILL);
}
child.wait().await.expect("wait failed");
eprintln!("[test] kill+wait done");
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
// Check if sleep survived — it SHOULD be dead now
let alive_after = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists();
eprintln!("[test] sleep alive after kill: {}", alive_after);
// Cleanup just in case
let _ = std::process::Command::new("kill")
.arg("-9")
.arg(sleep_pid.to_string())
.spawn();
assert!(alive_before, "Test setup: sleep was never created");
assert!(
!alive_after,
"sleep survived process-group kill! process_group(0) fix did not work."
);
}