Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 82 additions & 25 deletions crates/lest-core/src/runner/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,9 @@ struct Frame<'a> {
prefix: String,
inputs: BTreeMap<String, Json>,
vars: BTreeMap<String, Json>,
/// Tool env pins from this flow's own variables, over the run's: a called
/// flow targets the tenant or account its own `vars` name.
tool_env: Vec<(String, String)>,
steps: Mutex<serde_json::Map<String, Json>>,
}

Expand Down Expand Up @@ -211,6 +214,7 @@ impl Frame<'_> {
env.push((k.clone(), expr::stringify(v)));
}
env.extend(run.tool_env.iter().cloned());
env.extend(self.tool_env.iter().cloned());
env.push(("LEST_RUN_ID".into(), run.run_id.clone()));
env.push(("LEST_RUN_DIR".into(), run.artifacts_dir().display().to_string()));
env.push(("LEST_FLOW_ID".into(), self.flow.flow.id.clone()));
Expand Down Expand Up @@ -351,6 +355,34 @@ fn reachable<'a>(root: &'a LoadedFlow, catalog: &'a Catalog) -> Vec<&'a LoadedFl
out
}

/// The env pins of the named tools' profiles, from a flow's variables.
/// `lenient` leaves out a pin that cannot be resolved instead of failing.
fn pin_tools<'t>(
project: &Project,
tools: impl Iterator<Item = &'t str>,
vars: &BTreeMap<String, Json>,
environment: Option<&str>,
run_id: &str,
lenient: bool,
) -> Result<Vec<(String, String)>, String> {
let mut scope = Scope::new();
scope.set("vars", Json::Object(vars.clone().into_iter().collect()));
scope.set("run", json!({"environment": environment, "id": run_id}));
let mut out = Vec::new();
for name in tools {
if let Some(p) = project.config.tools.get(name) {
for (k, tmpl) in &p.env {
match expr::interpolate(tmpl, &scope) {
Ok(v) => out.push((k.clone(), v)),
Err(_) if lenient => {}
Err(e) => return Err(format!("tool {name} env {k}: {e}")),
}
}
}
}
Ok(out)
}

/// Takes the resource locks a run needs, waiting while another run holds one.
fn lock_resources(
names: &[String],
Expand Down Expand Up @@ -586,30 +618,20 @@ async fn run_flow(engine: &Engine, req: RunRequest, emitter: &Emitter, cancel: R
Err(e) => return finish(engine, &emitter, fail(report, e), started),
};

// Tool env pins, then preflight with those pins applied.
let mut tool_env = Vec::new();
{
let mut scope = Scope::new();
scope.set("vars", Json::Object(vars.clone().into_iter().collect()));
scope.set("run", json!({"environment": environment, "id": run_id}));
for need in &needs {
if let Some(p) = engine.project.config.tools.get(&need.name) {
for (k, tmpl) in &p.env {
match expr::interpolate(tmpl, &scope) {
Ok(v) => tool_env.push((k.clone(), v)),
Err(e) => {
return finish(
engine,
&emitter,
fail(report, format!("tool {} env {k}: {e}", need.name)),
started,
);
}
}
}
}
}
}
// Tool env pins, then preflight with those pins applied. A tool only a
// called flow lists may pin from vars this flow does not have; that flow
// pins it from its own vars when it is called.
let own_tools: Vec<&str> = flow.tools.iter().map(|t| t.name()).collect();
let pins = pin_tools(&engine.project, own_tools.iter().copied(), &vars, environment.as_deref(), &run_id, false)
.and_then(|mut own| {
let rest = needs.iter().map(|n| n.name.as_str()).filter(|n| !own_tools.contains(n));
own.extend(pin_tools(&engine.project, rest, &vars, environment.as_deref(), &run_id, true)?);
Ok(own)
});
let tool_env = match pins {
Ok(e) => e,
Err(e) => return finish(engine, &emitter, fail(report, e), started),
};
let env_remove = crate::secrets::secret_env_names(&engine.project.secrets_config());
for need in &needs {
let profile = engine.project.config.tools.get(&need.name);
Expand Down Expand Up @@ -719,6 +741,18 @@ async fn run_flow(engine: &Engine, req: RunRequest, emitter: &Emitter, cancel: R
let mut vars_env: Vec<(String, String)> =
s_vars.iter().chain(s_inputs.iter()).map(|(k, v)| (k.clone(), expr::stringify(v))).collect();
vars_env.extend(tool_env.iter().cloned());
if f.flow.id != flow.id {
let own = f.flow.tools.iter().map(|t| t.name());
match pin_tools(&engine.project, own, &s_vars, environment.as_deref(), &run_id, false) {
Ok(pins) => vars_env.extend(pins),
Err(e) => {
for r in running.iter_mut() {
r.stop().await;
}
return finish(engine, &emitter, fail(report, format!("service {}: {e}", svc.id)), started);
}
}
}
vars_env.push(("LEST_PROJECT_DIR".into(), engine.project.root.display().to_string()));
vars_env.push(("LEST_RUN_DIR".into(), run_dir.join("artifacts").display().to_string()));
let mut bad = None;
Expand Down Expand Up @@ -794,7 +828,14 @@ async fn run_flow(engine: &Engine, req: RunRequest, emitter: &Emitter, cancel: R
demo: Mutex::new(None),
});

let frame = Frame { flow: lf, prefix: String::new(), inputs, vars, steps: Mutex::new(serde_json::Map::new()) };
let frame = Frame {
flow: lf,
prefix: String::new(),
inputs,
vars,
tool_env: Vec::new(),
steps: Mutex::new(serde_json::Map::new()),
};

// Resume: earlier top-level steps take their results from the old run.
let mut seeded = Vec::new();
Expand Down Expand Up @@ -1154,11 +1195,27 @@ impl<'e> Runner<'e> {
return self.finish_step(frame, step, r, None);
}
};
let own = match pin_tools(
&self.ctx.project,
callee.flow.tools.iter().map(|t| t.name()),
&vars,
self.ctx.environment.as_deref(),
&self.ctx.run_id,
false,
) {
Ok(e) => e,
Err(e) => {
r.status = Some(StepStatus::Errored);
r.error = Some(e);
return self.finish_step(frame, step, r, None);
}
};
let child = Frame {
flow: callee,
prefix: format!("{}/", r.id),
inputs,
vars,
tool_env: frame.tool_env.iter().cloned().chain(own).collect(),
steps: Mutex::new(serde_json::Map::new()),
};
let (mut reports, failed) = self.run_sequence(&child, &callee.flow.steps, cancel).await;
Expand Down
54 changes: 54 additions & 0 deletions crates/lest-core/tests/runner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,60 @@ steps:
assert_eq!(r.find_step("login/issue").unwrap().outputs["token"], "token-for-ada");
}

#[tokio::test]
async fn a_called_flow_pins_tools_from_its_own_vars() {
let f = Fixture::new(&[
(
"suite.lest.yaml",
r#"
apiVersion: lest/v1
id: suite
name: Suite
steps:
- id: before
run: test -z "${PIN:-}"
- id: member
flow: member
"#,
),
(
"member.lest.yaml",
r#"
apiVersion: lest/v1
id: member
name: Member
environments:
dev: { tenant: acme }
defaultEnvironment: dev
tools: [sh]
steps:
- id: pinned
run: test "$PIN" = acme
- id: nested
flow: nested
"#,
),
(
"nested.lest.yaml",
r#"
apiVersion: lest/v1
id: nested
name: Nested
steps:
- id: inherited
run: test "$PIN" = acme
"#,
),
]);
std::fs::write(
f.path().join("lest.yaml"),
"name: test\nflows: [flows]\ntools:\n sh:\n env:\n PIN: \"${{ vars.tenant }}\"\n",
)
.unwrap();
let (r, _) = run(&f.engine(&[]), "suite", &[]).await;
assert_eq!(r.result, RunResult::Passed, "{r:#?}");
}

#[tokio::test]
async fn called_flow_outputs_reach_caller() {
let f = Fixture::new(&[
Expand Down
6 changes: 6 additions & 0 deletions docs/flow-format.md
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,12 @@ to the run's environment through `env` rather than commands that rewrite the
tool's shared configuration, because those would retarget every other process
using the tool, including a person in another terminal.

`env` values may use `vars`. A flow called by a `flow` step pins the tools it
lists from its own `vars`, so a suite whose members target different accounts
or tenants gives each member its own pins. A pin the top-level flow cannot
resolve for a tool only its callees list is left unset until a callee pins it.
Preflight checks each tool once, with the top-level flow's pins.

## Cleanup

```yaml
Expand Down
Loading