Skip workflows for unrelated events #56
+2
-2
File diff suppressed because it is too large
Load Diff
+4
-4
@@ -1,77 +1,77 @@
|
||||
[package]
|
||||
name = "syncode-control"
|
||||
description = "SynCode control plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[workspace]
|
||||
members = [
|
||||
".",
|
||||
"crates/control-node",
|
||||
"crates/control-nodes",
|
||||
"crates/control-runs",
|
||||
"crates/control-store",
|
||||
]
|
||||
resolver = "3"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.5.0"
|
||||
edition = "2024"
|
||||
rust-version = "1.95"
|
||||
license = "MIT"
|
||||
repository = "https://syncode.sh/syncode/control"
|
||||
|
||||
[workspace.lints.rust]
|
||||
unsafe_code = "forbid"
|
||||
|
||||
[workspace.lints.clippy]
|
||||
expect_used = "deny"
|
||||
panic = "deny"
|
||||
unwrap_used = "deny"
|
||||
|
||||
[dependencies]
|
||||
base64 = "0.22.1"
|
||||
clap = { version = "4.6.4", features = ["derive", "env"] }
|
||||
flate2 = "1.1.5"
|
||||
futures = "0.3.31"
|
||||
hex = "0.4.3"
|
||||
reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
|
||||
serde = { version = "1.0.229", features = ["derive"] }
|
||||
url = "2.5.8"
|
||||
serde_json = "1.0.149"
|
||||
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "f19733361a9ce1c667bb29372a9c07dacc6ec681" }
|
||||
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "f19733361a9ce1c667bb29372a9c07dacc6ec681" }
|
||||
syncode-control-node = { path = "crates/control-node" }
|
||||
syncode-control-nodes = { path = "crates/control-nodes" }
|
||||
syncode-control-runs = { path = "crates/control-runs" }
|
||||
syncode-control-store = { version = "0.5.0", path = "crates/control-store" }
|
||||
syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal", "time"] }
|
||||
tonic = "0.14.6"
|
||||
axum = "0.8.9"
|
||||
tower-http = { version = "0.6.8", features = ["cors"] }
|
||||
hmac = "0.13.0"
|
||||
sha2 = "0.11.0"
|
||||
tar = "0.4.44"
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[dev-dependencies]
|
||||
base64 = "0.22.1"
|
||||
hmac = "0.13.0"
|
||||
reqwest = { version = "0.13.4", features = ["json", "rustls"] }
|
||||
serde_json = "1.0.149"
|
||||
sha2 = "0.11.0"
|
||||
sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] }
|
||||
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "f19733361a9ce1c667bb29372a9c07dacc6ec681" }
|
||||
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "f19733361a9ce1c667bb29372a9c07dacc6ec681" }
|
||||
tokio-stream = "0.1.18"
|
||||
tempfile = "3.27.0"
|
||||
tower = { version = "0.5.3", features = ["util"] }
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
[package]
|
||||
name = "syncode-control"
|
||||
description = "SynCode control plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[workspace]
|
||||
members = [
|
||||
".",
|
||||
"crates/control-node",
|
||||
"crates/control-nodes",
|
||||
"crates/control-runs",
|
||||
"crates/control-store",
|
||||
]
|
||||
resolver = "3"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.5.0"
|
||||
edition = "2024"
|
||||
rust-version = "1.95"
|
||||
license = "MIT"
|
||||
repository = "https://syncode.sh/syncode/control"
|
||||
|
||||
[workspace.lints.rust]
|
||||
unsafe_code = "forbid"
|
||||
|
||||
[workspace.lints.clippy]
|
||||
expect_used = "deny"
|
||||
panic = "deny"
|
||||
unwrap_used = "deny"
|
||||
|
||||
[dependencies]
|
||||
base64 = "0.22.1"
|
||||
clap = { version = "4.6.4", features = ["derive", "env"] }
|
||||
flate2 = "1.1.5"
|
||||
futures = "0.3.31"
|
||||
hex = "0.4.3"
|
||||
reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
|
||||
serde = { version = "1.0.229", features = ["derive"] }
|
||||
url = "2.5.8"
|
||||
serde_json = "1.0.149"
|
||||
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "0b63875f1aed4a9e4065c086b1d5891c2e638352" }
|
||||
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "0b63875f1aed4a9e4065c086b1d5891c2e638352" }
|
||||
syncode-control-node = { path = "crates/control-node" }
|
||||
syncode-control-nodes = { path = "crates/control-nodes" }
|
||||
syncode-control-runs = { path = "crates/control-runs" }
|
||||
syncode-control-store = { version = "0.5.0", path = "crates/control-store" }
|
||||
syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal", "time"] }
|
||||
tonic = "0.14.6"
|
||||
axum = "0.8.9"
|
||||
tower-http = { version = "0.6.8", features = ["cors"] }
|
||||
hmac = "0.13.0"
|
||||
sha2 = "0.11.0"
|
||||
tar = "0.4.44"
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[dev-dependencies]
|
||||
base64 = "0.22.1"
|
||||
hmac = "0.13.0"
|
||||
reqwest = { version = "0.13.4", features = ["json", "rustls"] }
|
||||
serde_json = "1.0.149"
|
||||
sha2 = "0.11.0"
|
||||
sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] }
|
||||
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "0b63875f1aed4a9e4065c086b1d5891c2e638352" }
|
||||
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "0b63875f1aed4a9e4065c086b1d5891c2e638352" }
|
||||
tokio-stream = "0.1.18"
|
||||
tempfile = "3.27.0"
|
||||
tower = { version = "0.5.3", features = ["util"] }
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
+7
-1
@@ -1,301 +1,307 @@
|
||||
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(crate) struct TriggerPreflight {
|
||||
pub(crate) may_trigger: bool,
|
||||
pub(crate) needs_changed_paths: bool,
|
||||
}
|
||||
|
||||
pub(crate) fn preflight(source: &[u8], event: &Event) -> Result<TriggerPreflight, TriggerError> {
|
||||
let document = parse(std::str::from_utf8(source)?).map_err(GithubActionsCompileError::from)?;
|
||||
let workflow = Workflow::from_node(&document).map_err(GithubActionsCompileError::from)?;
|
||||
Ok(TriggerPreflight {
|
||||
may_trigger: workflow.triggers.may_fire_on(event),
|
||||
needs_changed_paths: workflow.triggers.needs_changed_paths(event),
|
||||
})
|
||||
}
|
||||
|
||||
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))
|
||||
}
|
||||
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, declares_event, 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(crate) struct TriggerPreflight {
|
||||
pub(crate) may_trigger: bool,
|
||||
pub(crate) needs_changed_paths: bool,
|
||||
}
|
||||
|
||||
pub(crate) fn preflight(source: &[u8], event: &Event) -> Result<TriggerPreflight, TriggerError> {
|
||||
let document = parse(std::str::from_utf8(source)?).map_err(GithubActionsCompileError::from)?;
|
||||
if !declares_event(&document, event.kind()).map_err(GithubActionsCompileError::from)? {
|
||||
return Ok(TriggerPreflight {
|
||||
may_trigger: false,
|
||||
needs_changed_paths: false,
|
||||
});
|
||||
}
|
||||
let workflow = Workflow::from_node(&document).map_err(GithubActionsCompileError::from)?;
|
||||
Ok(TriggerPreflight {
|
||||
may_trigger: workflow.triggers.may_fire_on(event),
|
||||
needs_changed_paths: workflow.triggers.needs_changed_paths(event),
|
||||
})
|
||||
}
|
||||
|
||||
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,390 +1,412 @@
|
||||
#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
|
||||
|
||||
#[path = "support/actions.rs"]
|
||||
mod actions;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::convert::Infallible;
|
||||
|
||||
use syncode_control::events::Delivery;
|
||||
use syncode_control::intake::accept;
|
||||
use syncode_control::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};
|
||||
use syncode_control_runs::{Conclusion, Forgotten, NodeId, Runs};
|
||||
|
||||
use actions::FIXTURE_ACTIONS;
|
||||
|
||||
fn file(path: &str, content: &str) -> WorkflowFile {
|
||||
WorkflowFile::new(path.to_owned(), content.as_bytes().to_vec())
|
||||
}
|
||||
|
||||
const CALLER: &str = r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
uses: ./.gitea/workflows/build.yml
|
||||
"#;
|
||||
|
||||
const CALLED: &str = r#"
|
||||
on: [workflow_call]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo built
|
||||
"#;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Sources(BTreeMap<String, Vec<WorkflowFile>>);
|
||||
|
||||
impl Sources {
|
||||
fn local(files: Vec<WorkflowFile>) -> Self {
|
||||
Self(BTreeMap::from([("syncode/demo".to_owned(), files)]))
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkflowSources for Sources {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
_commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, Self::Error> {
|
||||
Ok(self.0.get(repository).cloned().unwrap_or_default())
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for Sources {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
_repository: &str,
|
||||
_request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, Self::Error> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
}
|
||||
|
||||
fn push() -> Delivery {
|
||||
Delivery::read(
|
||||
"push",
|
||||
br#"{"ref":"refs/heads/main","after":"9f2c1e4a","repository":{"full_name":"syncode/demo"}}"#,
|
||||
)
|
||||
.expect("delivery")
|
||||
}
|
||||
|
||||
fn feature_push() -> Delivery {
|
||||
Delivery::read(
|
||||
"push",
|
||||
br#"{"ref":"refs/heads/feature/x","after":"9f2c1e4a","repository":{"full_name":"syncode/demo"}}"#,
|
||||
)
|
||||
.expect("delivery")
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PathsMustNotLoad(Sources);
|
||||
|
||||
impl WorkflowSources for PathsMustNotLoad {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, Self::Error> {
|
||||
self.0.workflows(repository, commit).await
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for PathsMustNotLoad {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
_repository: &str,
|
||||
_request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, Self::Error> {
|
||||
panic!("changed paths must not load for an excluded branch")
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_excluded_branch_does_not_load_changed_paths() {
|
||||
let files = PathsMustNotLoad(Sources::local(vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
paths: ["crates/**"]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: cargo test
|
||||
"#,
|
||||
)]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &feature_push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
assert!(queued.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_pathless_trigger_does_not_load_changed_paths() {
|
||||
let files = PathsMustNotLoad(Sources::local(vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: cargo test
|
||||
"#,
|
||||
)]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
assert_eq!(queued.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_local_reusable_call_is_one_run_with_qualified_jobs() {
|
||||
let files = Sources::local(vec![
|
||||
file(".gitea/workflows/ci.yml", CALLER),
|
||||
file(".gitea/workflows/build.yml", CALLED),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
assert_eq!(queued.len(), 1);
|
||||
let assignment = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await
|
||||
.expect("queue")
|
||||
.expect("assignment");
|
||||
let plan: serde_json::Value = serde_json::from_slice(assignment.plan()).expect("plan");
|
||||
assert_eq!(plan["plan"]["job"]["key"], "build__build");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mixed_jobs_keep_the_caller_topology() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
prepare:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo prepare
|
||||
call:
|
||||
needs: prepare
|
||||
uses: ./.gitea/workflows/build.yml
|
||||
publish:
|
||||
needs: call
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo publish
|
||||
"#,
|
||||
),
|
||||
file(".gitea/workflows/build.yml", CALLED),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let prepare_node = NodeId::fresh();
|
||||
let prepare = runs.take_next(prepare_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(prepare.plan()).contains("prepare"));
|
||||
assert!(runs.take_next(NodeId::fresh()).await.unwrap().is_none());
|
||||
runs.finished_job(
|
||||
prepare.run(),
|
||||
prepare.job(),
|
||||
prepare_node,
|
||||
prepare.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let called_node = NodeId::fresh();
|
||||
let called = runs.take_next(called_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(called.plan()).contains("call__build"));
|
||||
runs.finished_job(
|
||||
called.run(),
|
||||
called.job(),
|
||||
called_node,
|
||||
called.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let publish = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(publish.plan()).contains("publish"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reusable_workflow_outputs_are_projected_to_the_producing_job() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
reusable:
|
||||
uses: ./.gitea/workflows/reusable.yml
|
||||
consume:
|
||||
needs: reusable
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- env:
|
||||
RESULT: ${{ needs.reusable.outputs.result }}
|
||||
run: test "${RESULT}" = reusable-workflow-ok
|
||||
"#,
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/reusable.yml",
|
||||
r#"
|
||||
on:
|
||||
workflow_call:
|
||||
outputs:
|
||||
result:
|
||||
value: ${{ jobs.verify.outputs.result }}
|
||||
jobs:
|
||||
verify:
|
||||
runs-on: ubuntu-latest
|
||||
outputs:
|
||||
result: ${{ steps.result.outputs.value }}
|
||||
steps:
|
||||
- id: result
|
||||
run: echo "value=reusable-workflow-ok" >> "${GITHUB_OUTPUT}"
|
||||
"#,
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let producer_node = NodeId::fresh();
|
||||
let producer = runs.take_next(producer_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(producer.plan()).contains("reusable__verify"));
|
||||
runs.finished_job_with_outputs(
|
||||
producer.run(),
|
||||
producer.job(),
|
||||
producer_node,
|
||||
producer.fence(),
|
||||
Conclusion::Success,
|
||||
BTreeMap::from([("result".to_owned(), "reusable-workflow-ok".to_owned())]),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let consumer = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert_eq!(consumer.needs()[0].key(), "reusable__verify");
|
||||
let plan = String::from_utf8(consumer.plan().to_vec()).unwrap();
|
||||
assert!(
|
||||
plan.contains("needs.reusable__verify.outputs.result"),
|
||||
"output projection is absent: {plan}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inputs_and_secret_aliases_are_bound_before_compilation() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
deploy:
|
||||
uses: ./.gitea/workflows/deploy.yml
|
||||
with:
|
||||
target: prod
|
||||
secrets:
|
||||
token: ${{ secrets.PROD_TOKEN }}
|
||||
"#,
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/deploy.yml",
|
||||
r#"
|
||||
on: [workflow_call]
|
||||
jobs:
|
||||
deploy:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo ${{ inputs.target }} ${{ secrets.token }}
|
||||
"#,
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let assignment = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
let plan = String::from_utf8(assignment.plan().to_vec()).unwrap();
|
||||
assert!(plan.contains("'prod'"), "input binding is absent: {plan}");
|
||||
assert!(
|
||||
plan.contains("secrets.PROD_TOKEN"),
|
||||
"secret alias is absent: {plan}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_remote_reusable_call_is_fetched_at_its_revision() {
|
||||
let sources = Sources(BTreeMap::from([
|
||||
(
|
||||
"syncode/demo".to_owned(),
|
||||
vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
"on: [push]\njobs:\n build:\n uses: syncode/shared/.gitea/workflows/build.yml@v1\n",
|
||||
)],
|
||||
),
|
||||
(
|
||||
"syncode/shared".to_owned(),
|
||||
vec![file(".gitea/workflows/build.yml", CALLED)],
|
||||
),
|
||||
]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
accept(&sources, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
let assignment = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(assignment.plan()).contains("build__build"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_reusable_cycle_is_refused() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
"on: [push]\njobs:\n one:\n uses: ./.gitea/workflows/one.yml\n",
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/one.yml",
|
||||
"on: [workflow_call]\njobs:\n again:\n uses: ./.gitea/workflows/ci.yml\n",
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let error = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect_err("cycle must be refused");
|
||||
assert!(error.to_string().contains("cycle"));
|
||||
}
|
||||
#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
|
||||
|
||||
#[path = "support/actions.rs"]
|
||||
mod actions;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::convert::Infallible;
|
||||
|
||||
use syncode_control::events::Delivery;
|
||||
use syncode_control::intake::accept;
|
||||
use syncode_control::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};
|
||||
use syncode_control_runs::{Conclusion, Forgotten, NodeId, Runs};
|
||||
|
||||
use actions::FIXTURE_ACTIONS;
|
||||
|
||||
fn file(path: &str, content: &str) -> WorkflowFile {
|
||||
WorkflowFile::new(path.to_owned(), content.as_bytes().to_vec())
|
||||
}
|
||||
|
||||
const CALLER: &str = r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
build:
|
||||
uses: ./.gitea/workflows/build.yml
|
||||
"#;
|
||||
|
||||
const CALLED: &str = r#"
|
||||
on: [workflow_call]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo built
|
||||
"#;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Sources(BTreeMap<String, Vec<WorkflowFile>>);
|
||||
|
||||
impl Sources {
|
||||
fn local(files: Vec<WorkflowFile>) -> Self {
|
||||
Self(BTreeMap::from([("syncode/demo".to_owned(), files)]))
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkflowSources for Sources {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
_commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, Self::Error> {
|
||||
Ok(self.0.get(repository).cloned().unwrap_or_default())
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for Sources {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
_repository: &str,
|
||||
_request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, Self::Error> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
}
|
||||
|
||||
fn push() -> Delivery {
|
||||
Delivery::read(
|
||||
"push",
|
||||
br#"{"ref":"refs/heads/main","after":"9f2c1e4a","repository":{"full_name":"syncode/demo"}}"#,
|
||||
)
|
||||
.expect("delivery")
|
||||
}
|
||||
|
||||
fn feature_push() -> Delivery {
|
||||
Delivery::read(
|
||||
"push",
|
||||
br#"{"ref":"refs/heads/feature/x","after":"9f2c1e4a","repository":{"full_name":"syncode/demo"}}"#,
|
||||
)
|
||||
.expect("delivery")
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PathsMustNotLoad(Sources);
|
||||
|
||||
impl WorkflowSources for PathsMustNotLoad {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, Self::Error> {
|
||||
self.0.workflows(repository, commit).await
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for PathsMustNotLoad {
|
||||
type Error = Infallible;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
_repository: &str,
|
||||
_request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, Self::Error> {
|
||||
panic!("changed paths must not load for an excluded branch")
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_excluded_branch_does_not_load_changed_paths() {
|
||||
let files = PathsMustNotLoad(Sources::local(vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
paths: ["crates/**"]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: cargo test
|
||||
"#,
|
||||
)]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &feature_push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
assert!(queued.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_pathless_trigger_does_not_load_changed_paths() {
|
||||
let files = PathsMustNotLoad(Sources::local(vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: cargo test
|
||||
"#,
|
||||
)]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
assert_eq!(queued.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_unrelated_unsupported_event_does_not_block_a_push() {
|
||||
let files = PathsMustNotLoad(Sources::local(vec![file(
|
||||
".github/workflows/giteabot.yml",
|
||||
r#"
|
||||
on: [pull_request_target]
|
||||
jobs:
|
||||
triage:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo triage
|
||||
"#,
|
||||
)]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
assert!(queued.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_local_reusable_call_is_one_run_with_qualified_jobs() {
|
||||
let files = Sources::local(vec![
|
||||
file(".gitea/workflows/ci.yml", CALLER),
|
||||
file(".gitea/workflows/build.yml", CALLED),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let queued = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
assert_eq!(queued.len(), 1);
|
||||
let assignment = runs
|
||||
.take_next(NodeId::fresh())
|
||||
.await
|
||||
.expect("queue")
|
||||
.expect("assignment");
|
||||
let plan: serde_json::Value = serde_json::from_slice(assignment.plan()).expect("plan");
|
||||
assert_eq!(plan["plan"]["job"]["key"], "build__build");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mixed_jobs_keep_the_caller_topology() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
prepare:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo prepare
|
||||
call:
|
||||
needs: prepare
|
||||
uses: ./.gitea/workflows/build.yml
|
||||
publish:
|
||||
needs: call
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo publish
|
||||
"#,
|
||||
),
|
||||
file(".gitea/workflows/build.yml", CALLED),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let prepare_node = NodeId::fresh();
|
||||
let prepare = runs.take_next(prepare_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(prepare.plan()).contains("prepare"));
|
||||
assert!(runs.take_next(NodeId::fresh()).await.unwrap().is_none());
|
||||
runs.finished_job(
|
||||
prepare.run(),
|
||||
prepare.job(),
|
||||
prepare_node,
|
||||
prepare.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let called_node = NodeId::fresh();
|
||||
let called = runs.take_next(called_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(called.plan()).contains("call__build"));
|
||||
runs.finished_job(
|
||||
called.run(),
|
||||
called.job(),
|
||||
called_node,
|
||||
called.fence(),
|
||||
Conclusion::Success,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let publish = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(publish.plan()).contains("publish"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reusable_workflow_outputs_are_projected_to_the_producing_job() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
reusable:
|
||||
uses: ./.gitea/workflows/reusable.yml
|
||||
consume:
|
||||
needs: reusable
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- env:
|
||||
RESULT: ${{ needs.reusable.outputs.result }}
|
||||
run: test "${RESULT}" = reusable-workflow-ok
|
||||
"#,
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/reusable.yml",
|
||||
r#"
|
||||
on:
|
||||
workflow_call:
|
||||
outputs:
|
||||
result:
|
||||
value: ${{ jobs.verify.outputs.result }}
|
||||
jobs:
|
||||
verify:
|
||||
runs-on: ubuntu-latest
|
||||
outputs:
|
||||
result: ${{ steps.result.outputs.value }}
|
||||
steps:
|
||||
- id: result
|
||||
run: echo "value=reusable-workflow-ok" >> "${GITHUB_OUTPUT}"
|
||||
"#,
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let producer_node = NodeId::fresh();
|
||||
let producer = runs.take_next(producer_node).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(producer.plan()).contains("reusable__verify"));
|
||||
runs.finished_job_with_outputs(
|
||||
producer.run(),
|
||||
producer.job(),
|
||||
producer_node,
|
||||
producer.fence(),
|
||||
Conclusion::Success,
|
||||
BTreeMap::from([("result".to_owned(), "reusable-workflow-ok".to_owned())]),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let consumer = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert_eq!(consumer.needs()[0].key(), "reusable__verify");
|
||||
let plan = String::from_utf8(consumer.plan().to_vec()).unwrap();
|
||||
assert!(
|
||||
plan.contains("needs.reusable__verify.outputs.result"),
|
||||
"output projection is absent: {plan}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inputs_and_secret_aliases_are_bound_before_compilation() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
r#"
|
||||
on: [push]
|
||||
jobs:
|
||||
deploy:
|
||||
uses: ./.gitea/workflows/deploy.yml
|
||||
with:
|
||||
target: prod
|
||||
secrets:
|
||||
token: ${{ secrets.PROD_TOKEN }}
|
||||
"#,
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/deploy.yml",
|
||||
r#"
|
||||
on: [workflow_call]
|
||||
jobs:
|
||||
deploy:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- run: echo ${{ inputs.target }} ${{ secrets.token }}
|
||||
"#,
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
|
||||
let assignment = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
let plan = String::from_utf8(assignment.plan().to_vec()).unwrap();
|
||||
assert!(plan.contains("'prod'"), "input binding is absent: {plan}");
|
||||
assert!(
|
||||
plan.contains("secrets.PROD_TOKEN"),
|
||||
"secret alias is absent: {plan}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_remote_reusable_call_is_fetched_at_its_revision() {
|
||||
let sources = Sources(BTreeMap::from([
|
||||
(
|
||||
"syncode/demo".to_owned(),
|
||||
vec![file(
|
||||
".gitea/workflows/ci.yml",
|
||||
"on: [push]\njobs:\n build:\n uses: syncode/shared/.gitea/workflows/build.yml@v1\n",
|
||||
)],
|
||||
),
|
||||
(
|
||||
"syncode/shared".to_owned(),
|
||||
vec![file(".gitea/workflows/build.yml", CALLED)],
|
||||
),
|
||||
]));
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
accept(&sources, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect("intake");
|
||||
let assignment = runs.take_next(NodeId::fresh()).await.unwrap().unwrap();
|
||||
assert!(String::from_utf8_lossy(assignment.plan()).contains("build__build"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_reusable_cycle_is_refused() {
|
||||
let files = Sources::local(vec![
|
||||
file(
|
||||
".gitea/workflows/ci.yml",
|
||||
"on: [push]\njobs:\n one:\n uses: ./.gitea/workflows/one.yml\n",
|
||||
),
|
||||
file(
|
||||
".gitea/workflows/one.yml",
|
||||
"on: [workflow_call]\njobs:\n again:\n uses: ./.gitea/workflows/ci.yml\n",
|
||||
),
|
||||
]);
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
|
||||
let error = accept(&files, &runs, &push(), &FIXTURE_ACTIONS)
|
||||
.await
|
||||
.expect_err("cycle must be refused");
|
||||
assert!(error.to_string().contains("cycle"));
|
||||
}
|
||||
Reference in New Issue
Block a user