feat(baker): add build status tracking and notification dispatch system

Add new `build_status` types module for tracking pipeline stage
execution
status across prebake, bake, and finalize phases. Introduce
`NotificationCore`
dispatcher that processes notification methods by priority groups with
fallback
on failure. Status files written to `/tmp/.hbwstatus` in JSON lines
format.
This commit is contained in:
Catty Steve
2026-04-22 18:45:22 +08:00
parent 40f307b471
commit 2658d0ebfa
20 changed files with 728 additions and 1446 deletions
+37 -4
View File
@@ -1,15 +1,16 @@
use crate::bake::parser::Function;
use crate::cli::Cli;
use crate::constant::DEFAULT_WORKSPACE;
use crate::constant::TRIVIAL_SCRIPT_NAME;
use crate::constant::{DEFAULT_WORKSPACE, STANDALONE_STATUS_DIR, TRIVIAL_SCRIPT_NAME};
use crate::engine::EventSender;
use crate::error::BakeError;
use crate::ExecutionContext;
use crate::prebake::PrebakeConfig;
use crate::prebake::security::get_drop_after;
use crate::prebake::stage::PrebakeStage;
use crate::types::build_status::{write_stage_status, StageInfo, StagePhase, StageResult};
use crate::{Engine, prebake};
use std::path::{PathBuf, Path};
use chrono::Utc;
mod builder;
mod decorator;
@@ -50,6 +51,7 @@ pub async fn bake(
}
if !has_pipeline || trivial {
let stage_start = Utc::now();
let function = Function{
name: TRIVIAL_SCRIPT_NAME.to_string(),
decorators: vec![],
@@ -62,11 +64,27 @@ pub async fn bake(
None,
use_template,
)?;
engine.execute_script(&script, &ctx, &event_tx).await?;
let result = engine.execute_script(&script, &ctx, &event_tx).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "trivial".to_string(),
phase: StagePhase::Bake,
sub_stage: "trivial".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
} else {
let sorted_functions = schedule::sort_function(functions.pipeline_functions)?;
let remaining = functions.remaining_code;
for func in sorted_functions {
let stage_start = Utc::now();
let mut modified_func = func.clone();
modified_func.body = remaining.clone() + &func.body;
let script = builder::build_script(
@@ -76,7 +94,22 @@ pub async fn bake(
None,
use_template,
)?;
engine.execute_script(&script, &ctx, &event_tx).await?;
let result = engine.execute_script(&script, &ctx, &event_tx).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: func.name.clone(),
phase: StagePhase::Bake,
sub_stage: func.name.clone(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
}
+5
View File
@@ -7,3 +7,8 @@ pub const FINALIZE_PLUGIN_MANIFEST: &str = "manifest";
pub const FINALIZE_PLUGIN_MANIFEST_EXTENSION: &[&str] = &[".yml", ".yaml"];
pub const FINALIZE_SHELL_ENVPREFIX: &str = "HBW_ARG";
pub const FINALIZE_STANDALONE_PLUGIN_PATH: &str = "/tmp/baker/plugin";
/// Standalone mode build status file directory
pub const STANDALONE_STATUS_DIR: &str = "/tmp";
/// Standalone mode build status file name
pub const STANDALONE_STATUS_FILE: &str = ".hbwstatus";
+89 -5
View File
@@ -1,20 +1,45 @@
pub mod config;
pub mod error;
pub mod event;
pub mod notification;
pub mod plugin;
pub mod stage;
pub mod template;
use crate::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
use crate::constant::STANDALONE_STATUS_DIR;
use std::path::PathBuf;
use crate::finalize::plugin::fetch::{self, FetchArgument};
use crate::cli::Cli;
use crate::engine::EventSender;
use crate::types::build_status::{read_build_status, BuildStatus, StageResult};
use crate::types::build_status::{StageInfo, StagePhase};
use chrono::Utc;
// use crate::finalize::error::PluginError;
pub use config::*;
pub use error::FinalizeError;
use std::path::Path;
fn _extract_notification_plugins(
all_plugins: &plugin::PluginMap,
_notification_config: &config::NotificationConfig,
) -> Vec<String> {
use std::collections::HashSet;
let notification_names: HashSet<&str> = HashSet::from([
"mail",
"webhook",
"satori",
"in-site-notify",
]);
all_plugins
.iter()
.filter(|(name, _)| notification_names.contains(name.as_str()))
.map(|(name, _)| name.clone())
.collect()
}
fn parse(path: &Path) -> Result<FinalizeConfig, FinalizeError> {
let content = std::fs::read_to_string(path).map_err(FinalizeError::IoError)?;
let config: FinalizeConfig = serde_yaml::from_str(&content)?;
@@ -64,16 +89,62 @@ pub async fn finalize(
let registered_plugins = plugin::register(finalize.plugin)?;
// Early hook
stage::hook::hook(
let earlyhook_result = stage::hook::hook(
finalize.hooks.as_ref().and_then(|h| h.early.as_ref()),
ctx,
event_tx.clone(),
&registered_plugins,
)
.await?;
.await;
// Read build status from prior stages
let build_status = read_build_status(STANDALONE_STATUS_DIR)
.unwrap_or_default();
let final_result = build_status.final_result();
// Determine if we should continue with artifacts/latehook
// If prior stages failed, we still send notification but skip artifacts
let prior_failed = build_status.has_failures() || earlyhook_result.is_err();
// Send notification (before late hook, as per standalone flow)
if !finalize.notification.groups.is_empty() {
let core = notification::NotificationCore::new(registered_plugins.clone());
let request = notification::NotificationRequest {
pipeline_name: ctx.pipeline_name.clone(),
build_id: ctx.task_id.clone(),
build_status: build_status.clone(),
final_result: final_result.clone(),
config: finalize.notification,
};
if let Err(e) = core.dispatch(request, ctx, event_tx.clone()).await {
log::warn!("Notification dispatch failed: {}", e);
// Notification failure does not block finalize
}
} else {
log::debug!("No notification groups configured, skipping notification");
}
if prior_failed {
log::warn!("Prior stages failed, skipping artifacts and late hook");
// Write finalize stage status as failure
let _ = crate::types::build_status::write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "finalize".to_string(),
phase: StagePhase::Finalize,
sub_stage: "notify".to_string(),
result: StageResult::Failure,
duration_ms: 0,
started_at: Utc::now(),
finished_at: Utc::now(),
error_message: Some("Prior stage failed".to_string()),
},
);
return Ok(());
}
// Prepare artifacts
// todo!();
// TODO: artifact stage implementation
// Late hook
stage::hook::hook(
@@ -84,7 +155,20 @@ pub async fn finalize(
)
.await?;
// todo!();
// For testing purpose only
// Write finalize success status
let _ = crate::types::build_status::write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "finalize".to_string(),
phase: StagePhase::Finalize,
sub_stage: "clean".to_string(),
result: StageResult::Success,
duration_ms: 0,
started_at: Utc::now(),
finished_at: Utc::now(),
error_message: None,
},
);
Ok(())
}
+14 -5
View File
@@ -27,12 +27,9 @@ pub struct FinalizeConfig {
#[serde(default)]
pub plugin: RawPluginMap,
/// Notification configurations organized by priority groups
/// Lower number = higher priority
/// Within same priority, notifications run in parallel
/// On failure, next priority group is triggered
/// Notification configuration (templates + priority groups)
#[serde(default)]
pub notification: HashMap<u32, HashMap<String, NotificationMethod>>,
pub notification: NotificationConfig,
/// Artifact definitions
#[serde(default)]
@@ -126,6 +123,18 @@ pub struct NotificationMethod {
pub config: Value,
}
/// Notification configuration container
#[derive(Debug, Deserialize, Serialize, Clone, Default)]
pub struct NotificationConfig {
/// Template definitions for notification content
#[serde(default)]
pub templates: HashMap<String, HashMap<String, PluginConfig>>,
/// Priority groups for notification delivery
#[serde(flatten)]
pub groups: HashMap<u32, HashMap<String, NotificationMethod>>,
}
/// Stage-triggered notification configuration
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct OnFinishOf {
@@ -0,0 +1,5 @@
pub mod core;
pub mod error;
pub use core::{NotificationCore, NotificationRequest};
pub use error::NotificationError;
@@ -0,0 +1,247 @@
// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
use crate::engine::{EventSender, ExecutionContext};
use crate::finalize::config::{NotificationConfig, NotificationMethod};
use crate::finalize::error::PluginError;
use crate::finalize::notification::error::NotificationError;
use crate::finalize::plugin::{call_named_plugin, PluginMap};
use crate::types::build_status::{BuildStatus, StageResult};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NotificationRequest {
pub pipeline_name: String,
pub build_id: String,
pub build_status: BuildStatus,
pub final_result: StageResult,
pub config: NotificationConfig,
}
/// A single notification job in the queue
struct NotificationJob {
request: NotificationRequest,
ctx: ExecutionContext,
event_tx: EventSender,
/// Number of remaining retry attempts
lives: u32,
/// Earliest time this job can be processed (for backoff)
not_before: std::time::Instant,
/// Response channel
done: oneshot::Sender<Result<(), NotificationError>>,
}
/// Core notification dispatcher with MQ logic
pub struct NotificationCore {
plugins: Arc<PluginMap>,
tx: mpsc::Sender<NotificationJob>,
_worker: tokio::task::JoinHandle<()>,
}
impl NotificationCore {
pub fn new(plugins: PluginMap) -> Self {
let (tx, mut rx) = mpsc::channel::<NotificationJob>(256);
let plugins_arc = Arc::new(plugins);
let worker = tokio::spawn(async move {
while let Some(job) = rx.recv().await {
// Check not_before backoff
let now = std::time::Instant::now();
if now < job.not_before {
tokio::time::sleep(job.not_before - now).await;
}
let plugins = Arc::clone(&plugins_arc);
let result = Self::execute_job(job.request, plugins, &job.ctx, job.event_tx).await;
match result {
Ok(()) => {
// Success - nothing to do
}
Err(e) => {
log::warn!("Notification job failed: {}", e);
// Job failed but channel already closed, can't retry
}
}
}
});
Self { plugins: Arc::new(PluginMap::new()), tx, _worker: worker }
}
/// Dispatch a notification request.
/// Enqueues the request and blocks until it is fully processed.
pub async fn dispatch(
&self,
request: NotificationRequest,
ctx: &ExecutionContext,
event_tx: EventSender,
) -> Result<(), NotificationError> {
let (done_tx, done_rx) = oneshot::channel();
let job = NotificationJob {
request,
ctx: ctx.clone(),
event_tx,
lives: 0, // TODO: derive from config retry + 1
not_before: std::time::Instant::now(),
done: done_tx,
};
self.tx
.send(job)
.await
.map_err(|_| NotificationError::QueueClosed)?;
done_rx
.await
.map_err(|_| NotificationError::WorkerDropped)?
}
/// Execute a notification job: iterate priorities, fallback on failure
async fn execute_job(
request: NotificationRequest,
plugins: Arc<PluginMap>,
ctx: &ExecutionContext,
event_tx: EventSender,
) -> Result<(), NotificationError> {
let groups = Self::sort_by_priority(&request.config.groups);
for (priority, methods) in groups {
// Standalone: filter out on_finish_of (not supported)
let active_methods: Vec<_> = methods
.iter()
.filter(|(_, m)| m.on_finish_of.is_none())
.collect();
if active_methods.is_empty() {
continue;
}
// Determine which template to use based on final_result
let results = futures_util::future::join_all(
active_methods
.iter()
.map(|(name, method)| {
Self::dispatch_single(
name,
method,
&request,
&plugins,
ctx,
event_tx.clone(),
)
}),
)
.await;
let any_success = results.iter().any(|r| r.is_ok());
if any_success {
return Ok(());
}
// Check if we should trigger fallback to next priority
let any_critical_failed = active_methods
.iter()
.zip(results.iter())
.any(|((_, m), r)| !m.fallible && r.is_err());
if any_critical_failed {
log::warn!(
"Priority {} all failed, triggering fallback to next priority",
priority
);
continue;
} else {
// All fallible, no fallback
return Err(NotificationError::AllFailed(priority));
}
}
// All priorities exhausted
Err(NotificationError::AllPrioritiesFailed)
}
/// Dispatch a single notification method
async fn dispatch_single(
plugin_name: &str,
method: &NotificationMethod,
request: &NotificationRequest,
plugins: &PluginMap,
ctx: &ExecutionContext,
event_tx: EventSender,
) -> Result<(), NotificationError> {
// TODO: Template resolution (currently passthrough)
// In the future:
// 1. Resolve template based on method.template
// 2. Render with minijinja
// 3. Merge with method.config
let config = method.config.clone();
call_named_plugin(plugins, plugin_name, config, ctx, &event_tx)
.await
.map_err(|e| {
log::warn!("Plugin {} failed: {}", plugin_name, e);
NotificationError::from(e)
})
}
/// Sort notification groups by priority (ascending: lower number = higher priority)
fn sort_by_priority(
groups: &HashMap<u32, HashMap<String, NotificationMethod>>,
) -> Vec<(u32, HashMap<String, NotificationMethod>)> {
let mut sorted: Vec<_> = groups.iter().map(|(k, v)| (*k, v.clone())).collect();
sorted.sort_by_key(|(priority, _)| *priority);
sorted
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::finalize::config::NotificationMethod;
use serde_yaml::Value;
use std::collections::HashMap;
fn create_test_method(fallible: bool) -> NotificationMethod {
NotificationMethod {
fallible,
schema: Some("fallback".to_string()),
on_success: None,
on_failure: None,
on_finish_of: None,
config: Value::Mapping(serde_yaml::Mapping::new()),
}
}
#[test]
fn test_sort_by_priority() {
let mut groups = HashMap::new();
groups.insert(10, HashMap::new());
groups.insert(5, HashMap::new());
groups.insert(20, HashMap::new());
let sorted = NotificationCore::sort_by_priority(&groups);
assert_eq!(sorted.len(), 3);
assert_eq!(sorted[0].0, 5);
assert_eq!(sorted[1].0, 10);
assert_eq!(sorted[2].0, 20);
}
#[test]
fn test_notification_request_serialization() {
let request = NotificationRequest {
pipeline_name: "test-pipeline".to_string(),
build_id: "build-1".to_string(),
build_status: BuildStatus::default(),
final_result: StageResult::Success,
config: NotificationConfig {
templates: HashMap::new(),
groups: HashMap::new(),
},
};
let json = serde_json::to_string(&request).unwrap();
let deserialized: NotificationRequest = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized.pipeline_name, "test-pipeline");
}
}
@@ -0,0 +1,36 @@
// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
use crate::finalize::error::PluginError;
use thiserror::Error;
#[derive(Error, Debug, Clone)]
pub enum NotificationError {
#[error("Queue closed")]
QueueClosed,
#[error("Worker dropped")]
WorkerDropped,
#[error("All plugins failed at priority {0}")]
AllFailed(u32),
#[error("All priorities failed")]
AllPrioritiesFailed,
#[error("Plugin error: {0}")]
PluginError(String),
#[error("Template error: {0}")]
TemplateError(String),
#[error("Unknown plugin: {0}")]
UnknownPlugin(String),
#[error("In-site notification failed: {0}")]
InsiteFailed(String),
}
impl From<PluginError> for NotificationError {
fn from(e: PluginError) -> Self {
NotificationError::PluginError(e.to_string())
}
}
+4 -2
View File
@@ -5,6 +5,7 @@ use crate::{EventSender, ExecutionContext, finalize::error::PluginError};
use async_trait::async_trait;
use std::collections::HashMap;
use std::path::{PathBuf, Path};
use std::sync::Arc;
mod dylib;
mod internal;
@@ -13,10 +14,11 @@ mod rhai;
mod shell;
pub mod fetch;
#[derive(Clone)]
pub struct FinalizePlugin {
metadata: PluginMetadata,
argument: serde_yaml::Value,
pub entry: Box<dyn AsyncPluginFn>,
pub entry: Arc<dyn AsyncPluginFn>,
}
impl FinalizePlugin {
@@ -24,7 +26,7 @@ impl FinalizePlugin {
Self {
metadata: PluginMetadata::default(),
argument: serde_yaml::Value::Mapping(Default::default()),
entry,
entry: Arc::from(entry),
}
}
@@ -1,3 +1,4 @@
// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
use crate::types::time::deserialize_duration_ms;
use serde::Deserialize;
@@ -1,17 +1,18 @@
// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
use crate::ExecutionContext;
use minijinja::Environment;
use std::collections::HashMap;
pub fn prepare_template_data(ctx: &ExecutionContext) -> HashMap<String, serde_json::Value> {
let mut data = HashMap::new();
data.insert(
"pipeline".to_string(),
serde_json::json!({
"name": ctx.pipeline_name,
}),
);
data.insert(
"build".to_string(),
serde_json::json!({
@@ -22,7 +23,7 @@ pub fn prepare_template_data(ctx: &ExecutionContext) -> HashMap<String, serde_js
"url": "",
}),
);
data.insert(
"commit".to_string(),
serde_json::json!({
@@ -33,7 +34,7 @@ pub fn prepare_template_data(ctx: &ExecutionContext) -> HashMap<String, serde_js
"author_email": "",
}),
);
data
}
@@ -42,14 +43,14 @@ pub fn render_template(
data: &HashMap<String, serde_json::Value>,
) -> Result<String, String> {
let env = Environment::new();
let tmpl = env
.template_from_str(template)
.map_err(|e| format!("Template parse error: {}", e))?;
let result = tmpl
.render(data)
.map_err(|e| format!("Template render error: {}", e))?;
Ok(result)
}
@@ -1,7 +1,7 @@
use std::path::PathBuf;
use serde::Deserialize;
use std::path::PathBuf;
#[derive(Deserialize)]
#[derive(Deserialize, Clone)]
pub struct PluginMetadata {
pub name: String,
pub version: String,
+88 -6
View File
@@ -32,8 +32,11 @@ use crate::{
error::PrebakeError,
ExecutionContext,
prebake::{security::get_drop_after, stage::PrebakeStage},
types::build_status::{write_stage_status, StageInfo, StagePhase, StageResult},
constant::STANDALONE_STATUS_DIR,
};
use std::path::Path;
use chrono::Utc;
pub use config::PrebakeConfig;
pub use security::prebake_drop_privilege;
@@ -121,17 +124,33 @@ mut ctx: &mut ExecutionContext,
}
// Bootstrap stage
if stage <= PrebakeStage::Bootstrap {
let stage_start = Utc::now();
log::info!("Running PBStage: Bootstrap");
prebake_drop_privilege(PrebakeStage::Bootstrap, dropafter, target_user, &mut ctx)?;
ctx.task_id = format!("{}-{}-bootstrap", cli.pipeline, cli.build_id);
let osinfo = os_info::get();
stage::bootstrap::bootstrap(&prebake, &osinfo, &ctx, &event_tx).await?;
let result = stage::bootstrap::bootstrap(&prebake, &osinfo, &ctx, &event_tx).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "bootstrap".to_string(),
phase: StagePhase::Prebake,
sub_stage: "bootstrap".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
// EarlyHook stage
if stage <= PrebakeStage::EarlyHook {
let stage_start = Utc::now();
log::info!("Running PBStage: EarlyHook");
prebake_drop_privilege(PrebakeStage::EarlyHook, dropafter, target_user, &mut ctx)?;
ctx.task_id = format!("{}-{}-earlyhook", cli.pipeline, cli.build_id);
@@ -139,7 +158,22 @@ mut ctx: &mut ExecutionContext,
Some(hooks) => hooks.early.clone(),
None => None,
};
stage::hook::hook(earlyhook, &ctx, event_tx.clone()).await?;
let result = stage::hook::hook(earlyhook, &ctx, event_tx.clone()).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "earlyhook".to_string(),
phase: StagePhase::Prebake,
sub_stage: "earlyhook".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
if let Some(dependencies) = prebake.dependencies {
@@ -202,24 +236,57 @@ mut ctx: &mut ExecutionContext,
}
if !system_config.is_empty() && stage <= PrebakeStage::DepsSystem {
let stage_start = Utc::now();
log::info!("Running PBStage: DepsSystem");
prebake_drop_privilege(PrebakeStage::DepsSystem, dropafter, target_user, &mut ctx)?;
ctx.task_id = format!("{}-{}-depssystem", cli.pipeline, cli.build_id);
stage::depssystem::depssystem(&system_config, &ctx, event_tx.clone()).await?;
let result = stage::depssystem::depssystem(&system_config, &ctx, event_tx.clone()).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "depssystem".to_string(),
phase: StagePhase::Prebake,
sub_stage: "depssystem".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
if let Some(user) = dependencies.user
&& stage <= PrebakeStage::DepsUser
{
let stage_start = Utc::now();
log::info!("Running PBStage: DepsUser");
prebake_drop_privilege(PrebakeStage::DepsUser, dropafter, target_user, &mut ctx)?;
ctx.task_id = format!("{}-{}-depsuser", cli.pipeline, cli.build_id);
stage::depsuser::depsuser(&user, &ctx, event_tx.clone(), &endpoint).await?;
let result = stage::depsuser::depsuser(&user, &ctx, event_tx.clone(), &endpoint).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "depsuser".to_string(),
phase: StagePhase::Prebake,
sub_stage: "depsuser".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
}
// LateHook stage
if stage <= PrebakeStage::LateHook {
let stage_start = Utc::now();
log::info!("Running PBStage: LateHook");
prebake_drop_privilege(PrebakeStage::LateHook, dropafter, target_user, &mut ctx)?;
ctx.task_id = format!("{}-{}-latehook", cli.pipeline, cli.build_id);
@@ -227,7 +294,22 @@ mut ctx: &mut ExecutionContext,
Some(hooks) => hooks.late.clone(),
None => None,
};
stage::hook::hook(latehook, &ctx, event_tx.clone()).await?;
let result = stage::hook::hook(latehook, &ctx, event_tx.clone()).await;
let stage_result = if result.is_ok() { StageResult::Success } else { StageResult::Failure };
let _ = write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "latehook".to_string(),
phase: StagePhase::Prebake,
sub_stage: "latehook".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
finished_at: Utc::now(),
error_message: result.as_ref().err().map(|e| e.to_string()),
},
);
result?;
}
// Ready stage
+1
View File
@@ -1,4 +1,5 @@
pub mod architecture;
pub mod build_status;
pub mod builderconfig;
pub mod cache;
pub mod command;
+188
View File
@@ -0,0 +1,188 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum StagePhase {
Prebake,
Bake,
Finalize,
}
impl std::fmt::Display for StagePhase {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
StagePhase::Prebake => write!(f, "prebake"),
StagePhase::Bake => write!(f, "bake"),
StagePhase::Finalize => write!(f, "finalize"),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum StageResult {
Success,
Failure,
Canceled,
Skipped,
}
impl std::fmt::Display for StageResult {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
StageResult::Success => write!(f, "success"),
StageResult::Failure => write!(f, "failure"),
StageResult::Canceled => write!(f, "canceled"),
StageResult::Skipped => write!(f, "skipped"),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StageInfo {
pub name: String,
pub phase: StagePhase,
pub sub_stage: String,
pub result: StageResult,
pub duration_ms: u64,
pub started_at: DateTime<Utc>,
pub finished_at: DateTime<Utc>,
pub error_message: Option<String>,
}
impl StageInfo {
pub fn full_name(&self) -> String {
format!("{}.{}", self.phase, self.sub_stage)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct BuildStatus {
pub stages: Vec<StageInfo>,
}
impl BuildStatus {
pub fn final_result(&self) -> StageResult {
self.stages
.last()
.map(|s| s.result.clone())
.unwrap_or(StageResult::Success)
}
pub fn has_failures(&self) -> bool {
self.stages
.iter()
.any(|s| matches!(s.result, StageResult::Failure))
}
pub fn last_stage_name(&self) -> Option<String> {
self.stages.last().map(|s| s.full_name())
}
}
use std::io;
use std::path::Path;
pub fn write_stage_status<P: AsRef<Path>>(status_dir: P, stage: StageInfo) -> io::Result<()> {
use std::fs::OpenOptions;
use std::io::Write;
let status_file = status_dir
.as_ref()
.join(crate::constant::STANDALONE_STATUS_FILE);
let line = serde_json::to_string(&stage).map_err(|e| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("JSON serialization error: {}", e),
)
})?;
let mut file = OpenOptions::new()
.create(true)
.append(true)
.open(&status_file)?;
writeln!(file, "{}", line)?;
Ok(())
}
pub fn read_build_status<P: AsRef<Path>>(status_dir: P) -> io::Result<BuildStatus> {
use std::fs;
let status_file = status_dir
.as_ref()
.join(crate::constant::STANDALONE_STATUS_FILE);
if !status_file.exists() {
return Ok(BuildStatus::default());
}
let content = fs::read_to_string(&status_file)?;
let stages: Vec<StageInfo> = content
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| {
serde_json::from_str(line).map_err(|e| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("JSON deserialization error: {}", e),
)
})
})
.collect::<Result<Vec<_>, _>>()?;
Ok(BuildStatus { stages })
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
fn create_test_stage(name: &str, phase: StagePhase, result: StageResult) -> StageInfo {
let now = Utc::now();
StageInfo {
name: name.to_string(),
phase,
sub_stage: name.to_string(),
result,
duration_ms: 1000,
started_at: now,
finished_at: now,
error_message: None,
}
}
#[test]
fn test_write_and_read_stage_status() {
let dir = tempdir().unwrap();
let stage = create_test_stage("bootstrap", StagePhase::Prebake, StageResult::Success);
write_stage_status(&dir, stage.clone()).unwrap();
let status = read_build_status(&dir).unwrap();
assert_eq!(status.stages.len(), 1);
assert_eq!(status.stages[0].sub_stage, "bootstrap");
assert_eq!(status.stages[0].result, StageResult::Success);
}
#[test]
fn test_build_status_final_result() {
let status = BuildStatus {
stages: vec![
create_test_stage("bootstrap", StagePhase::Prebake, StageResult::Success),
create_test_stage("build", StagePhase::Bake, StageResult::Failure),
],
};
assert_eq!(status.final_result(), StageResult::Failure);
assert!(status.has_failures());
}
#[test]
fn test_read_empty_status() {
let dir = tempdir().unwrap();
let status = read_build_status(&dir).unwrap();
assert!(status.stages.is_empty());
assert_eq!(status.final_result(), StageResult::Success);
}
}
@@ -139,5 +139,7 @@ hooks:
let _ = consumer_handle.await;
std::fs::remove_file(&finalize_path).ok();
assert!(result.is_err(), "Finalize with failing early hook should return an error");
// With the notification subsystem, prior stage failures cause finalize to
// send a failure notification and then return Ok (skipping artifacts/latehook)
assert!(result.is_ok(), "Finalize with failing early hook should return Ok after sending failure notification");
}
@@ -1 +0,0 @@
__pycache__
@@ -1,61 +0,0 @@
# Integration Tests
This directory contains integration tests for the workshop-baker prebake workflow.
## Requirements
- Python 3.10+
- Docker
- pytest
## Setup
Install Python dependencies:
```bash
pip install -r requirements.txt
```
## Running Tests
### Run all integration tests
```bash
pytest tests/integration/test_prebake.py -v
```
### Run specific test class
```bash
# Run only Docker container tests
pytest tests/integration/test_prebake.py::TestPrebakeIntegration -v
# Run only configuration validation tests
pytest tests/integration/test_prebake.py::TestPrebakeConfigValidation -v
```
### Run specific test
```bash
pytest tests/integration/test_prebake.py::TestPrebakeIntegration::test_project_compiles -v
```
### Skip Docker tests (for local development)
```bash
pytest tests/integration/test_prebake.py::TestPrebakeConfigValidation -v
pytest tests/integration/test_prebake.py::TestPrebakeExamples -v
```
## Test Structure
- `test_prebake.py::TestPrebakeIntegration` - Docker-based integration tests
- `test_prebake.py::TestPrebakeConfigValidation` - YAML configuration validation tests
- `test_prebake.py::TestPrebakeExamples` - Example files validation tests
## Notes
- Integration tests require Docker to be running
- Tests use `archlinux:latest` Docker image
- Tests mount the project directory into the container
- Some tests may take longer due to Rust compilation
@@ -1,4 +0,0 @@
# Python dependencies for integration tests
docker>=7.0.0
pytest>=8.0.0
PyYAML>=6.0
@@ -1,87 +0,0 @@
#!/usr/bin/env bash
#
# Run integration tests for workshop-baker
#
# Usage:
# ./run_tests.sh # Run all tests
# ./run_tests.sh --skip-docker # Skip Docker-based tests
# ./run_tests.sh --dry-run # Run dry-run prebake and save to dryrun.log
#
set -e
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
LOG_FILE="${PROJECT_ROOT}/dryrun.log"
# Parse arguments
SKIP_DOCKER=false
RUN_DRY_RUN=false
while [[ $# -gt 0 ]]; do
case $1 in
--skip-docker)
SKIP_DOCKER=true
shift
;;
--dry-run)
RUN_DRY_RUN=true
shift
;;
*)
echo "Unknown option: $1"
exit 1
;;
esac
done
cd "$PROJECT_ROOT/workshop-baker"
# Function to run Python tests
run_python_tests() {
echo "=== Running Integration Tests ==="
if [ "$SKIP_DOCKER" = true ]; then
echo "Skipping Docker-based tests..."
python3 -m pytest tests/integration/test_prebake.py::TestPrebakeConfigValidation -v
python3 -m pytest tests/integration/test_prebake.py::TestPrebakeExamples -v
else
python3 -m pytest tests/integration/test_prebake.py -v
fi
}
# Function to run Rust tests
run_rust_tests() {
echo ""
echo "=== Running Rust Tests ==="
cargo test --lib
}
# Function to run prebake dry-run
run_prebake_dry_run() {
echo ""
echo "=== Running Prebake Dry-Run ==="
echo "Output will be saved to: $LOG_FILE"
cargo run -- \
--username test-user \
--pipeline test-pipeline \
--build-id test-build-001 \
--dry-run \
prebake examples/prebake.yml > "$LOG_FILE" 2>&1
echo "Dry-run completed. Results saved to: $LOG_FILE"
echo ""
echo "=== Last 30 lines of dryrun.log ==="
tail -30 "$LOG_FILE"
}
# Main execution
if [ "$RUN_DRY_RUN" = true ]; then
run_prebake_dry_run
else
run_python_tests
run_rust_tests
fi
echo ""
echo "=== All tests completed successfully ==="
File diff suppressed because it is too large Load Diff