fix: Authorize native checkout credentials #50
+8
-1
@@ -1,280 +1,287 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use syncode_control_runs::{
|
||||
Architecture, JobId, MatrixPolicy, OperatingSystem, Origin, Priority, QueuedJob, Requirements,
|
||||
RunId, RunLog, Runs, RunsError,
|
||||
};
|
||||
use syncode_workflow::{
|
||||
BooleanValue, Event, PositiveIntegerValue, StepKind, Value, VersionedPlan, WorkflowCompiler,
|
||||
WorkflowDialect, WorkflowSource, expand,
|
||||
};
|
||||
use syncode_workflow_github_actions::compiler::{GithubActionsCompileError, GithubActionsCompiler};
|
||||
use syncode_workflow_github_actions::expression::{ContextName, EvaluationContext};
|
||||
use syncode_workflow_github_actions::template::render;
|
||||
use syncode_workflow_github_actions::workflow::{Workflow, parse};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TriggerError {
|
||||
#[error(transparent)]
|
||||
Compile(#[from] GithubActionsCompileError),
|
||||
|
||||
#[error("the workflow does not lower into a plan: {0}")]
|
||||
Plan(String),
|
||||
|
||||
#[error("the compiled plan cannot be encoded: {0}")]
|
||||
Encode(#[from] serde_json::Error),
|
||||
|
||||
#[error("the matrix strategy cannot be evaluated by the control plane: {0}")]
|
||||
Strategy(String),
|
||||
|
||||
#[error("the runner requirement cannot be evaluated by the control plane: {0}")]
|
||||
Requirement(String),
|
||||
|
||||
#[error(transparent)]
|
||||
Runs(#[from] RunsError),
|
||||
|
||||
#[error("the workflow file is not text: {0}")]
|
||||
NotText(#[from] std::str::Utf8Error),
|
||||
|
||||
#[error("the workflow contains an invalid secret reference: {0}")]
|
||||
SecretReference(String),
|
||||
|
||||
#[error(transparent)]
|
||||
Action(#[from] crate::actions::ActionResolutionError),
|
||||
}
|
||||
|
||||
/// What an event did to one workflow. A workflow that does not declare the
|
||||
/// event is not an error and not a run; it simply has nothing to say about it.
|
||||
#[derive(Debug, Eq, PartialEq)]
|
||||
pub enum Triggered {
|
||||
Runs(Vec<RunId>),
|
||||
NotForThisEvent,
|
||||
}
|
||||
|
||||
pub async fn trigger<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
source: &[u8],
|
||||
event: &Event,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
trigger_with_event(runs, source, Some(event), origin, actions).await
|
||||
}
|
||||
|
||||
pub(crate) async fn trigger_resolved<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
workflow: Workflow,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let hir = syncode_workflow_github_actions::compiler::lower::workflow(workflow)?;
|
||||
queue_hir(runs, hir, origin, actions).await
|
||||
}
|
||||
|
||||
async fn trigger_with_event<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
source: &[u8],
|
||||
event: Option<&Event>,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let document = parse(std::str::from_utf8(source)?).map_err(GithubActionsCompileError::from)?;
|
||||
let workflow = Workflow::from_node(&document).map_err(GithubActionsCompileError::from)?;
|
||||
if event.is_some_and(|event| !workflow.triggers.fire_on(event)) {
|
||||
return Ok(Triggered::NotForThisEvent);
|
||||
}
|
||||
|
||||
let hir = GithubActionsCompiler.compile(&WorkflowSource::new(
|
||||
WorkflowDialect::GitHubActions,
|
||||
source.to_vec(),
|
||||
))?;
|
||||
queue_hir(runs, hir, origin, actions).await
|
||||
}
|
||||
|
||||
async fn queue_hir<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
hir: syncode_workflow::WorkflowHir<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let plans =
|
||||
syncode_workflow::plans(hir).map_err(|error| TriggerError::Plan(error.to_string()))?;
|
||||
|
||||
let mut queued = Vec::new();
|
||||
for plan in &plans {
|
||||
for mut combination in expand(plan) {
|
||||
actions.resolve(&mut combination, origin).await?;
|
||||
let key = combination.job().key().as_ref().to_owned();
|
||||
let needs = combination
|
||||
.job()
|
||||
.needs()
|
||||
.iter()
|
||||
.map(|need| need.as_ref().to_owned())
|
||||
.collect();
|
||||
let matrix = matrix_policy(&combination)?;
|
||||
let requirements = requirements(&combination)?;
|
||||
let priority = priority(origin);
|
||||
let secrets = crate::secret_references::collect(&combination)
|
||||
.map_err(TriggerError::SecretReference)?;
|
||||
let encoded = serde_json::to_vec(&VersionedPlan::new(combination))?;
|
||||
queued.push(
|
||||
QueuedJob::new(JobId::fresh(), key, needs, matrix, encoded)
|
||||
.scheduled(priority, requirements)
|
||||
.referencing(secrets),
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(Triggered::Runs(vec![
|
||||
runs.queue_run(origin.clone(), queued).await?,
|
||||
]))
|
||||
}
|
||||
|
||||
fn priority(origin: &Origin) -> Priority {
|
||||
match origin.event() {
|
||||
"workflow_dispatch" | "manual" => Priority::High,
|
||||
"schedule" => Priority::Low,
|
||||
_ => Priority::Normal,
|
||||
}
|
||||
}
|
||||
|
||||
fn requirements(
|
||||
plan: &syncode_workflow::ExecutionPlan<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
) -> Result<Requirements, TriggerError> {
|
||||
let mut context = EvaluationContext::default();
|
||||
context.values_mut().insert(
|
||||
ContextName::Matrix,
|
||||
Value::Object(Arc::new(plan.job().matrix().clone())),
|
||||
);
|
||||
let mut labels: Vec<String> = plan
|
||||
.job()
|
||||
.runner()
|
||||
.labels()
|
||||
.iter()
|
||||
.map(|label| {
|
||||
render(label, &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))
|
||||
})
|
||||
.collect::<Result<_, _>>()?;
|
||||
if let Some(group) = plan.job().runner().group() {
|
||||
labels.push(
|
||||
render(group, &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
let mut required = Requirements::new(labels.clone());
|
||||
for label in &labels {
|
||||
let normalized = label.to_ascii_lowercase();
|
||||
if matches!(normalized.as_str(), "amd64" | "x64" | "x86_64") {
|
||||
required = required.architecture(Architecture::Amd64);
|
||||
} else if matches!(normalized.as_str(), "arm64" | "aarch64") {
|
||||
required = required.architecture(Architecture::Arm64);
|
||||
} else if normalized == "linux" || normalized.starts_with("ubuntu-") {
|
||||
required = required.operating_system(OperatingSystem::Linux);
|
||||
} else if normalized == "windows" || normalized.starts_with("windows-") {
|
||||
required = required.operating_system(OperatingSystem::Windows);
|
||||
} else if matches!(normalized.as_str(), "macos" | "macos-latest") {
|
||||
required = required.operating_system(OperatingSystem::MacOs);
|
||||
}
|
||||
}
|
||||
let mut images = Vec::new();
|
||||
if let Some(container) = plan.job().container() {
|
||||
images.push(
|
||||
render(container.image(), &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
for service in plan.job().services() {
|
||||
images.push(
|
||||
render(service.container().image(), &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
let actions = plan
|
||||
.job()
|
||||
.steps()
|
||||
.iter()
|
||||
.filter_map(|step| match step.kind() {
|
||||
StepKind::Action(action) => Some(action),
|
||||
StepKind::Shell(_) => None,
|
||||
})
|
||||
.map(|action| action_requirement(action, &context))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
Ok(required.prefer(images, actions))
|
||||
}
|
||||
|
||||
fn action_requirement(
|
||||
action: &syncode_workflow::ActionStep<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
context: &EvaluationContext,
|
||||
) -> Result<String, TriggerError> {
|
||||
if let Some(reference) = action.reference() {
|
||||
return render(reference, context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()));
|
||||
}
|
||||
match action.source().resolved_source() {
|
||||
Some(syncode_workflow::ResolvedAction::Local(local)) => Ok(format!("./{}", local.path())),
|
||||
Some(syncode_workflow::ResolvedAction::Remote(remote)) => {
|
||||
Ok(remote.requested_reference().to_string())
|
||||
}
|
||||
Some(syncode_workflow::ResolvedAction::Oci(oci)) => {
|
||||
Ok(oci.requested_reference().to_string())
|
||||
}
|
||||
None => Err(TriggerError::Requirement(
|
||||
"action source is neither unresolved nor resolved".to_owned(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn matrix_policy(
|
||||
plan: &syncode_workflow::ExecutionPlan<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
) -> Result<MatrixPolicy, TriggerError> {
|
||||
let mut context = EvaluationContext::default();
|
||||
context.values_mut().insert(
|
||||
ContextName::Matrix,
|
||||
Value::Object(Arc::new(plan.job().matrix().clone())),
|
||||
);
|
||||
let fail_fast = match plan.job().strategy().fail_fast() {
|
||||
BooleanValue::Literal(value) => *value,
|
||||
BooleanValue::Expression(expression) => expression
|
||||
.evaluate_condition(&context)
|
||||
.map_err(|error| TriggerError::Strategy(error.to_string()))?,
|
||||
};
|
||||
let max_parallel = match plan.job().strategy().max_parallel() {
|
||||
None => None,
|
||||
Some(PositiveIntegerValue::Literal(value)) => Some(*value),
|
||||
Some(PositiveIntegerValue::Expression(expression)) => {
|
||||
let value = expression
|
||||
.evaluate(&context)
|
||||
.map_err(|error| TriggerError::Strategy(error.to_string()))?;
|
||||
match value {
|
||||
Value::Number(value)
|
||||
if value.is_finite()
|
||||
&& value > 0.0
|
||||
&& value.fract() == 0.0
|
||||
&& value <= u64::MAX as f64 =>
|
||||
{
|
||||
Some(value as u64)
|
||||
}
|
||||
value => {
|
||||
return Err(TriggerError::Strategy(format!(
|
||||
"max-parallel evaluated to {value:?}, not a positive integer"
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
Ok(MatrixPolicy::new(fail_fast, max_parallel))
|
||||
}
|
||||
use std::sync::Arc;
|
||||
|
||||
use syncode_control_runs::{
|
||||
Architecture, JobId, MatrixPolicy, OperatingSystem, Origin, Priority, QueuedJob, Requirements,
|
||||
RunId, RunLog, Runs, RunsError,
|
||||
};
|
||||
use syncode_workflow::{
|
||||
BooleanValue, Event, PositiveIntegerValue, StepKind, Value, VersionedPlan, WorkflowCompiler,
|
||||
WorkflowDialect, WorkflowSource, expand,
|
||||
};
|
||||
use syncode_workflow_github_actions::compiler::{GithubActionsCompileError, GithubActionsCompiler};
|
||||
use syncode_workflow_github_actions::expression::{ContextName, EvaluationContext};
|
||||
use syncode_workflow_github_actions::template::render;
|
||||
use syncode_workflow_github_actions::workflow::{Workflow, parse};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TriggerError {
|
||||
#[error(transparent)]
|
||||
Compile(#[from] GithubActionsCompileError),
|
||||
|
||||
#[error("the workflow does not lower into a plan: {0}")]
|
||||
Plan(String),
|
||||
|
||||
#[error("the compiled plan cannot be encoded: {0}")]
|
||||
Encode(#[from] serde_json::Error),
|
||||
|
||||
#[error("the matrix strategy cannot be evaluated by the control plane: {0}")]
|
||||
Strategy(String),
|
||||
|
||||
#[error("the runner requirement cannot be evaluated by the control plane: {0}")]
|
||||
Requirement(String),
|
||||
|
||||
#[error(transparent)]
|
||||
Runs(#[from] RunsError),
|
||||
|
||||
#[error("the workflow file is not text: {0}")]
|
||||
NotText(#[from] std::str::Utf8Error),
|
||||
|
||||
#[error("the workflow contains an invalid secret reference: {0}")]
|
||||
SecretReference(String),
|
||||
|
||||
#[error(transparent)]
|
||||
Action(#[from] crate::actions::ActionResolutionError),
|
||||
}
|
||||
|
||||
/// What an event did to one workflow. A workflow that does not declare the
|
||||
/// event is not an error and not a run; it simply has nothing to say about it.
|
||||
#[derive(Debug, Eq, PartialEq)]
|
||||
pub enum Triggered {
|
||||
Runs(Vec<RunId>),
|
||||
NotForThisEvent,
|
||||
}
|
||||
|
||||
pub async fn trigger<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
source: &[u8],
|
||||
event: &Event,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
trigger_with_event(runs, source, Some(event), origin, actions).await
|
||||
}
|
||||
|
||||
pub(crate) async fn trigger_resolved<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
workflow: Workflow,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let hir = syncode_workflow_github_actions::compiler::lower::workflow(workflow)?;
|
||||
queue_hir(runs, hir, origin, actions).await
|
||||
}
|
||||
|
||||
async fn trigger_with_event<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
source: &[u8],
|
||||
event: Option<&Event>,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let document = parse(std::str::from_utf8(source)?).map_err(GithubActionsCompileError::from)?;
|
||||
let workflow = Workflow::from_node(&document).map_err(GithubActionsCompileError::from)?;
|
||||
if event.is_some_and(|event| !workflow.triggers.fire_on(event)) {
|
||||
return Ok(Triggered::NotForThisEvent);
|
||||
}
|
||||
|
||||
let hir = GithubActionsCompiler.compile(&WorkflowSource::new(
|
||||
WorkflowDialect::GitHubActions,
|
||||
source.to_vec(),
|
||||
))?;
|
||||
queue_hir(runs, hir, origin, actions).await
|
||||
}
|
||||
|
||||
async fn queue_hir<L: RunLog, A: crate::actions::PlanActionResolver>(
|
||||
runs: &Runs<L>,
|
||||
hir: syncode_workflow::WorkflowHir<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
origin: &Origin,
|
||||
actions: &A,
|
||||
) -> Result<Triggered, TriggerError> {
|
||||
let plans =
|
||||
syncode_workflow::plans(hir).map_err(|error| TriggerError::Plan(error.to_string()))?;
|
||||
|
||||
let mut queued = Vec::new();
|
||||
for plan in &plans {
|
||||
for mut combination in expand(plan) {
|
||||
actions.resolve(&mut combination, origin).await?;
|
||||
let key = combination.job().key().as_ref().to_owned();
|
||||
let needs = combination
|
||||
.job()
|
||||
.needs()
|
||||
.iter()
|
||||
.map(|need| need.as_ref().to_owned())
|
||||
.collect();
|
||||
let matrix = matrix_policy(&combination)?;
|
||||
let requirements = requirements(&combination)?;
|
||||
let priority = priority(origin);
|
||||
let mut secrets = crate::secret_references::collect(&combination)
|
||||
.map_err(TriggerError::SecretReference)?;
|
||||
if uuid::Uuid::parse_str(origin.repository()).is_ok()
|
||||
&& !secrets
|
||||
.iter()
|
||||
.any(|name| name == syncode_control_node::REPOSITORY_TOKEN_SECRET)
|
||||
{
|
||||
secrets.push(syncode_control_node::REPOSITORY_TOKEN_SECRET.to_owned());
|
||||
}
|
||||
let encoded = serde_json::to_vec(&VersionedPlan::new(combination))?;
|
||||
queued.push(
|
||||
QueuedJob::new(JobId::fresh(), key, needs, matrix, encoded)
|
||||
.scheduled(priority, requirements)
|
||||
.referencing(secrets),
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(Triggered::Runs(vec![
|
||||
runs.queue_run(origin.clone(), queued).await?,
|
||||
]))
|
||||
}
|
||||
|
||||
fn priority(origin: &Origin) -> Priority {
|
||||
match origin.event() {
|
||||
"workflow_dispatch" | "manual" => Priority::High,
|
||||
"schedule" => Priority::Low,
|
||||
_ => Priority::Normal,
|
||||
}
|
||||
}
|
||||
|
||||
fn requirements(
|
||||
plan: &syncode_workflow::ExecutionPlan<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
) -> Result<Requirements, TriggerError> {
|
||||
let mut context = EvaluationContext::default();
|
||||
context.values_mut().insert(
|
||||
ContextName::Matrix,
|
||||
Value::Object(Arc::new(plan.job().matrix().clone())),
|
||||
);
|
||||
let mut labels: Vec<String> = plan
|
||||
.job()
|
||||
.runner()
|
||||
.labels()
|
||||
.iter()
|
||||
.map(|label| {
|
||||
render(label, &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))
|
||||
})
|
||||
.collect::<Result<_, _>>()?;
|
||||
if let Some(group) = plan.job().runner().group() {
|
||||
labels.push(
|
||||
render(group, &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
let mut required = Requirements::new(labels.clone());
|
||||
for label in &labels {
|
||||
let normalized = label.to_ascii_lowercase();
|
||||
if matches!(normalized.as_str(), "amd64" | "x64" | "x86_64") {
|
||||
required = required.architecture(Architecture::Amd64);
|
||||
} else if matches!(normalized.as_str(), "arm64" | "aarch64") {
|
||||
required = required.architecture(Architecture::Arm64);
|
||||
} else if normalized == "linux" || normalized.starts_with("ubuntu-") {
|
||||
required = required.operating_system(OperatingSystem::Linux);
|
||||
} else if normalized == "windows" || normalized.starts_with("windows-") {
|
||||
required = required.operating_system(OperatingSystem::Windows);
|
||||
} else if matches!(normalized.as_str(), "macos" | "macos-latest") {
|
||||
required = required.operating_system(OperatingSystem::MacOs);
|
||||
}
|
||||
}
|
||||
let mut images = Vec::new();
|
||||
if let Some(container) = plan.job().container() {
|
||||
images.push(
|
||||
render(container.image(), &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
for service in plan.job().services() {
|
||||
images.push(
|
||||
render(service.container().image(), &context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()))?,
|
||||
);
|
||||
}
|
||||
let actions = plan
|
||||
.job()
|
||||
.steps()
|
||||
.iter()
|
||||
.filter_map(|step| match step.kind() {
|
||||
StepKind::Action(action) => Some(action),
|
||||
StepKind::Shell(_) => None,
|
||||
})
|
||||
.map(|action| action_requirement(action, &context))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
Ok(required.prefer(images, actions))
|
||||
}
|
||||
|
||||
fn action_requirement(
|
||||
action: &syncode_workflow::ActionStep<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
context: &EvaluationContext,
|
||||
) -> Result<String, TriggerError> {
|
||||
if let Some(reference) = action.reference() {
|
||||
return render(reference, context)
|
||||
.map(String::from)
|
||||
.map_err(|error| TriggerError::Requirement(error.to_string()));
|
||||
}
|
||||
match action.source().resolved_source() {
|
||||
Some(syncode_workflow::ResolvedAction::Local(local)) => Ok(format!("./{}", local.path())),
|
||||
Some(syncode_workflow::ResolvedAction::Remote(remote)) => {
|
||||
Ok(remote.requested_reference().to_string())
|
||||
}
|
||||
Some(syncode_workflow::ResolvedAction::Oci(oci)) => {
|
||||
Ok(oci.requested_reference().to_string())
|
||||
}
|
||||
None => Err(TriggerError::Requirement(
|
||||
"action source is neither unresolved nor resolved".to_owned(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn matrix_policy(
|
||||
plan: &syncode_workflow::ExecutionPlan<
|
||||
syncode_workflow_github_actions::expression::ExpressionProgram,
|
||||
>,
|
||||
) -> Result<MatrixPolicy, TriggerError> {
|
||||
let mut context = EvaluationContext::default();
|
||||
context.values_mut().insert(
|
||||
ContextName::Matrix,
|
||||
Value::Object(Arc::new(plan.job().matrix().clone())),
|
||||
);
|
||||
let fail_fast = match plan.job().strategy().fail_fast() {
|
||||
BooleanValue::Literal(value) => *value,
|
||||
BooleanValue::Expression(expression) => expression
|
||||
.evaluate_condition(&context)
|
||||
.map_err(|error| TriggerError::Strategy(error.to_string()))?,
|
||||
};
|
||||
let max_parallel = match plan.job().strategy().max_parallel() {
|
||||
None => None,
|
||||
Some(PositiveIntegerValue::Literal(value)) => Some(*value),
|
||||
Some(PositiveIntegerValue::Expression(expression)) => {
|
||||
let value = expression
|
||||
.evaluate(&context)
|
||||
.map_err(|error| TriggerError::Strategy(error.to_string()))?;
|
||||
match value {
|
||||
Value::Number(value)
|
||||
if value.is_finite()
|
||||
&& value > 0.0
|
||||
&& value.fract() == 0.0
|
||||
&& value <= u64::MAX as f64 =>
|
||||
{
|
||||
Some(value as u64)
|
||||
}
|
||||
value => {
|
||||
return Err(TriggerError::Strategy(format!(
|
||||
"max-parallel evaluated to {value:?}, not a positive integer"
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
Ok(MatrixPolicy::new(fail_fast, max_parallel))
|
||||
}
|
||||
@@ -1,413 +1,446 @@
|
||||
#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
|
||||
|
||||
#[path = "support/actions.rs"]
|
||||
mod actions;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control::trigger::{Triggered, trigger};
|
||||
use syncode_control_runs::{Conclusion, Forgotten, NodeId, Origin, Runs};
|
||||
use syncode_workflow::{Event, EventKind, GitReference, PlanSchemaVersion, Value, VersionedPlan};
|
||||
use syncode_workflow_github_actions::expression::ExpressionProgram;
|
||||
|
||||
use actions::FIXTURE_ACTIONS;
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
fn push(branch: &str, paths: &[&str]) -> Event {
|
||||
Event::new(
|
||||
EventKind::Push,
|
||||
GitReference::Branch(branch.to_owned()),
|
||||
paths.iter().map(|path| (*path).to_owned()).collect(),
|
||||
)
|
||||
}
|
||||
|
||||
fn origin() -> Origin {
|
||||
Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"a-commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
}
|
||||
|
||||
fn queued(triggered: Triggered) -> Vec<syncode_control_runs::RunId> {
|
||||
match triggered {
|
||||
Triggered::Runs(runs) => runs,
|
||||
Triggered::NotForThisEvent => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
const MATRIX: &str = r#"
|
||||
name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: [self-hosted, linux]
|
||||
strategy:
|
||||
matrix:
|
||||
rust: ["1.95.0", "nightly"]
|
||||
steps:
|
||||
- run: cargo +${{ matrix.rust }} test
|
||||
"#;
|
||||
|
||||
async fn assigned(runs: &Runs<Forgotten>) -> TestResult<Vec<VersionedPlan<ExpressionProgram>>> {
|
||||
let mut plans = Vec::new();
|
||||
while let Some(assignment) = runs.take_next(NodeId::fresh()).await? {
|
||||
plans.push(serde_json::from_slice(assignment.plan())?);
|
||||
}
|
||||
Ok(plans)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_matrix_workflow_becomes_one_run_with_each_combination() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
assert_eq!(triggered.len(), 1, "one workflow event, one run");
|
||||
let plans = assigned(&runs).await?;
|
||||
assert_eq!(plans.len(), 2);
|
||||
|
||||
let mut versions: Vec<String> = plans
|
||||
.iter()
|
||||
.map(
|
||||
|plan| match plan.plan().job().strategy().matrix().property("rust") {
|
||||
Some(Value::String(value)) => value.clone(),
|
||||
other => panic!("expected a single value, got {other:?}"),
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
versions.sort();
|
||||
assert_eq!(versions, vec!["1.95.0".to_owned(), "nightly".to_owned()]);
|
||||
|
||||
for plan in &plans {
|
||||
assert_eq!(plan.schema(), PlanSchemaVersion::CURRENT);
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "build");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_workflow_without_a_matrix_becomes_one_run() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo one
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
assert_eq!(triggered.len(), 1);
|
||||
assert_eq!(assigned(&runs).await?.len(), 1);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_broken_workflow_is_refused_when_the_run_is_triggered() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let error = trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo "${{ github. }}"
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await
|
||||
.expect_err("a workflow that does not compile must not produce a run");
|
||||
|
||||
assert!(
|
||||
!error.to_string().is_empty(),
|
||||
"the reason must reach whoever triggered it"
|
||||
);
|
||||
assert!(
|
||||
runs.take_next(NodeId::fresh()).await?.is_none(),
|
||||
"nothing may be queued from a workflow that did not compile"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_workflow_that_does_not_want_the_event_queues_nothing() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("wip/x", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(triggered, Triggered::NotForThisEvent);
|
||||
assert!(
|
||||
runs.take_next(NodeId::fresh()).await?.is_none(),
|
||||
"an event the workflow does not declare must queue nothing"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn every_run_is_numbered_and_states_where_it_came_from() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
let mut numbers = Vec::new();
|
||||
while let Some(assignment) = runs.take_next(NodeId::fresh()).await? {
|
||||
assert_eq!(assignment.origin(), &origin());
|
||||
numbers.push(assignment.number().get());
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
numbers,
|
||||
vec![1, 1],
|
||||
"jobs of one run carry the same run number"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_job_waits_until_every_job_it_needs_has_succeeded() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo build
|
||||
publish:
|
||||
needs: build
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo publish
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let build_node = NodeId::fresh();
|
||||
let build = runs
|
||||
.take_next(build_node)
|
||||
.await?
|
||||
.ok_or("build was not queued")?;
|
||||
let plan: VersionedPlan<ExpressionProgram> = serde_json::from_slice(build.plan())?;
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "build");
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
|
||||
runs.finished_job(
|
||||
build.run(),
|
||||
build.job(),
|
||||
build_node,
|
||||
build.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await?;
|
||||
let publish = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await?
|
||||
.ok_or("publish did not become ready")?;
|
||||
let plan: VersionedPlan<ExpressionProgram> = serde_json::from_slice(publish.plan())?;
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "publish");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_always_job_is_assigned_with_the_failed_need() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: exit 1
|
||||
cleanup:
|
||||
if: always()
|
||||
needs: build
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo cleanup
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let build_node = NodeId::fresh();
|
||||
let build = runs
|
||||
.take_next(build_node)
|
||||
.await?
|
||||
.ok_or("build was not queued")?;
|
||||
runs.finished_job_with_outputs(
|
||||
build.run(),
|
||||
build.job(),
|
||||
build_node,
|
||||
build.fence(),
|
||||
Conclusion::Failure,
|
||||
BTreeMap::from([("artifact".to_owned(), "bundle.tar".to_owned())]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let cleanup = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await?
|
||||
.ok_or("always job did not become ready")?;
|
||||
assert_eq!(cleanup.needs().len(), 1);
|
||||
assert_eq!(cleanup.needs()[0].key(), "build");
|
||||
assert_eq!(cleanup.needs()[0].conclusion(), Conclusion::Failure);
|
||||
assert_eq!(
|
||||
cleanup.needs()[0]
|
||||
.outputs()
|
||||
.get("artifact")
|
||||
.map(String::as_str),
|
||||
Some("bundle.tar")
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn matrix_max_parallel_is_enforced_by_the_queue() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
max-parallel: 1
|
||||
matrix:
|
||||
shard: [1, 2, 3]
|
||||
steps:
|
||||
- run: echo shard
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let first_node = NodeId::fresh();
|
||||
let first = runs
|
||||
.take_next(first_node)
|
||||
.await?
|
||||
.ok_or("first matrix job was not queued")?;
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
runs.finished_job(
|
||||
first.run(),
|
||||
first.job(),
|
||||
first_node,
|
||||
first.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await?;
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_some());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn matrix_fail_fast_stops_unassigned_siblings() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
max-parallel: 1
|
||||
matrix:
|
||||
shard: [1, 2, 3]
|
||||
steps:
|
||||
- run: exit 1
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
)[0];
|
||||
let first_node = NodeId::fresh();
|
||||
let first = runs
|
||||
.take_next(first_node)
|
||||
.await?
|
||||
.ok_or("first matrix job was not queued")?;
|
||||
runs.finished_job(
|
||||
first.run(),
|
||||
first.job(),
|
||||
first_node,
|
||||
first.fence(),
|
||||
Conclusion::Failure,
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
assert_eq!(
|
||||
runs.state_of(run).await?,
|
||||
syncode_control_runs::RunState::Finished(Conclusion::Failure)
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
|
||||
|
||||
#[path = "support/actions.rs"]
|
||||
mod actions;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control::trigger::{Triggered, trigger};
|
||||
use syncode_control_node::REPOSITORY_TOKEN_SECRET;
|
||||
use syncode_control_runs::{Conclusion, Forgotten, NodeId, Origin, Runs};
|
||||
use syncode_workflow::{Event, EventKind, GitReference, PlanSchemaVersion, Value, VersionedPlan};
|
||||
use syncode_workflow_github_actions::expression::ExpressionProgram;
|
||||
|
||||
use actions::FIXTURE_ACTIONS;
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
fn push(branch: &str, paths: &[&str]) -> Event {
|
||||
Event::new(
|
||||
EventKind::Push,
|
||||
GitReference::Branch(branch.to_owned()),
|
||||
paths.iter().map(|path| (*path).to_owned()).collect(),
|
||||
)
|
||||
}
|
||||
|
||||
fn origin() -> Origin {
|
||||
Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"a-commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
}
|
||||
|
||||
fn queued(triggered: Triggered) -> Vec<syncode_control_runs::RunId> {
|
||||
match triggered {
|
||||
Triggered::Runs(runs) => runs,
|
||||
Triggered::NotForThisEvent => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
const MATRIX: &str = r#"
|
||||
name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: [self-hosted, linux]
|
||||
strategy:
|
||||
matrix:
|
||||
rust: ["1.95.0", "nightly"]
|
||||
steps:
|
||||
- run: cargo +${{ matrix.rust }} test
|
||||
"#;
|
||||
|
||||
async fn assigned(runs: &Runs<Forgotten>) -> TestResult<Vec<VersionedPlan<ExpressionProgram>>> {
|
||||
let mut plans = Vec::new();
|
||||
while let Some(assignment) = runs.take_next(NodeId::fresh()).await? {
|
||||
plans.push(serde_json::from_slice(assignment.plan())?);
|
||||
}
|
||||
Ok(plans)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_matrix_workflow_becomes_one_run_with_each_combination() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
assert_eq!(triggered.len(), 1, "one workflow event, one run");
|
||||
let plans = assigned(&runs).await?;
|
||||
assert_eq!(plans.len(), 2);
|
||||
|
||||
let mut versions: Vec<String> = plans
|
||||
.iter()
|
||||
.map(
|
||||
|plan| match plan.plan().job().strategy().matrix().property("rust") {
|
||||
Some(Value::String(value)) => value.clone(),
|
||||
other => panic!("expected a single value, got {other:?}"),
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
versions.sort();
|
||||
assert_eq!(versions, vec!["1.95.0".to_owned(), "nightly".to_owned()]);
|
||||
|
||||
for plan in &plans {
|
||||
assert_eq!(plan.schema(), PlanSchemaVersion::CURRENT);
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "build");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_workflow_without_a_matrix_becomes_one_run() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo one
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
assert_eq!(triggered.len(), 1);
|
||||
assert_eq!(assigned(&runs).await?.len(), 1);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_native_job_authorizes_its_repository_token() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let native_origin = origin().with_repository(uuid::Uuid::new_v4().to_string());
|
||||
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&native_origin,
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
let assignment = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await?
|
||||
.ok_or("nothing was queued")?;
|
||||
|
||||
assert_eq!(assignment.secrets(), [REPOSITORY_TOKEN_SECRET]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_broken_workflow_is_refused_when_the_run_is_triggered() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let error = trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo "${{ github. }}"
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await
|
||||
.expect_err("a workflow that does not compile must not produce a run");
|
||||
|
||||
assert!(
|
||||
!error.to_string().is_empty(),
|
||||
"the reason must reach whoever triggered it"
|
||||
);
|
||||
assert!(
|
||||
runs.take_next(NodeId::fresh()).await?.is_none(),
|
||||
"nothing may be queued from a workflow that did not compile"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_workflow_that_does_not_want_the_event_queues_nothing() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
let triggered = trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("wip/x", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(triggered, Triggered::NotForThisEvent);
|
||||
assert!(
|
||||
runs.take_next(NodeId::fresh()).await?.is_none(),
|
||||
"an event the workflow does not declare must queue nothing"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn every_run_is_numbered_and_states_where_it_came_from() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
MATRIX.as_bytes(),
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
let mut numbers = Vec::new();
|
||||
while let Some(assignment) = runs.take_next(NodeId::fresh()).await? {
|
||||
assert_eq!(assignment.origin(), &origin());
|
||||
numbers.push(assignment.number().get());
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
numbers,
|
||||
vec![1, 1],
|
||||
"jobs of one run carry the same run number"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_job_waits_until_every_job_it_needs_has_succeeded() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo build
|
||||
publish:
|
||||
needs: build
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo publish
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let build_node = NodeId::fresh();
|
||||
let build = runs
|
||||
.take_next(build_node)
|
||||
.await?
|
||||
.ok_or("build was not queued")?;
|
||||
let plan: VersionedPlan<ExpressionProgram> = serde_json::from_slice(build.plan())?;
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "build");
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
|
||||
runs.finished_job(
|
||||
build.run(),
|
||||
build.job(),
|
||||
build_node,
|
||||
build.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await?;
|
||||
let publish = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await?
|
||||
.ok_or("publish did not become ready")?;
|
||||
let plan: VersionedPlan<ExpressionProgram> = serde_json::from_slice(publish.plan())?;
|
||||
assert_eq!(plan.plan().job().key().as_ref(), "publish");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_always_job_is_assigned_with_the_failed_need() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: exit 1
|
||||
cleanup:
|
||||
if: always()
|
||||
needs: build
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo cleanup
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let build_node = NodeId::fresh();
|
||||
let build = runs
|
||||
.take_next(build_node)
|
||||
.await?
|
||||
.ok_or("build was not queued")?;
|
||||
runs.finished_job_with_outputs(
|
||||
build.run(),
|
||||
build.job(),
|
||||
build_node,
|
||||
build.fence(),
|
||||
Conclusion::Failure,
|
||||
BTreeMap::from([("artifact".to_owned(), "bundle.tar".to_owned())]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let cleanup = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await?
|
||||
.ok_or("always job did not become ready")?;
|
||||
assert_eq!(cleanup.needs().len(), 1);
|
||||
assert_eq!(cleanup.needs()[0].key(), "build");
|
||||
assert_eq!(cleanup.needs()[0].conclusion(), Conclusion::Failure);
|
||||
assert_eq!(
|
||||
cleanup.needs()[0]
|
||||
.outputs()
|
||||
.get("artifact")
|
||||
.map(String::as_str),
|
||||
Some("bundle.tar")
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn matrix_max_parallel_is_enforced_by_the_queue() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
max-parallel: 1
|
||||
matrix:
|
||||
shard: [1, 2, 3]
|
||||
steps:
|
||||
- run: echo shard
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let first_node = NodeId::fresh();
|
||||
let first = runs
|
||||
.take_next(first_node)
|
||||
.await?
|
||||
.ok_or("first matrix job was not queued")?;
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
runs.finished_job(
|
||||
first.run(),
|
||||
first.job(),
|
||||
first_node,
|
||||
first.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await?;
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_some());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn matrix_fail_fast_stops_unassigned_siblings() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = queued(
|
||||
trigger(
|
||||
&runs,
|
||||
br#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
max-parallel: 1
|
||||
matrix:
|
||||
shard: [1, 2, 3]
|
||||
steps:
|
||||
- run: exit 1
|
||||
"#,
|
||||
&push("main", &[]),
|
||||
&origin(),
|
||||
&FIXTURE_ACTIONS,
|
||||
)
|
||||
.await?,
|
||||
)[0];
|
||||
let first_node = NodeId::fresh();
|
||||
let first = runs
|
||||
.take_next(first_node)
|
||||
.await?
|
||||
.ok_or("first matrix job was not queued")?;
|
||||
runs.finished_job(
|
||||
first.run(),
|
||||
first.job(),
|
||||
first_node,
|
||||
first.fence(),
|
||||
Conclusion::Failure,
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert!(runs.take_next(NodeId::fresh()).await?.is_none());
|
||||
assert_eq!(
|
||||
runs.state_of(run).await?,
|
||||
syncode_control_runs::RunState::Finished(Conclusion::Failure)
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
+12
-1
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user