Menu
AkurAI-Build
publicLatest change 35120790b959cffea652cb656e19363df95d6f96 - Fix Build gate deadlocks and regressions by Ólafur Búi Ólafsson
#[path = "support/mcp.rs"]
mod support;
use std::{
fs,
net::{TcpListener, TcpStream},
path::Path,
process::{Child, Command},
time::{Duration, Instant},
};
use anyhow::{Context, Result, ensure};
use serde_json::{Value, json};
use support::McpSession;
#[test]
fn git_to_artifact_to_approved_deployment() -> Result<()> {
let temporary = tempfile::tempdir()?;
let repository = temporary.path().join("repository");
let data = temporary.path().join("data");
let key = temporary.path().join("database.key");
fs::create_dir(&repository)?;
git(&repository, &["init", "-b", "main"])?;
git(&repository, &["config", "user.name", "AkurAI Test"])?;
git(
&repository,
&["config", "user.email", "test@example.invalid"],
)?;
fs::write(
repository.join(".akurai.yml"),
r#"version: 1
jobs:
- name: build
executor: native
run: mkdir -p out && printf hello > out/artifact.txt
artifacts: [out/artifact.txt]
- name: deploy
needs: [build]
executor: native
environment: test
installation_id: fixture-installation
installation_revision: 1
approval: true
run: test "$(cat out/artifact.txt)" = hello
"#,
)?;
git(&repository, &["add", ".akurai.yml"])?;
git(&repository, &["commit", "-m", "pipeline"])?;
cli(&data, &key, &["keygen", "--out", support::str(&key)?])?;
cli(&data, &key, &["migrate"])?;
let api_token = temporary.path().join("api.token");
let webhook_token = temporary.path().join("webhook.token");
cli(&data, &key, &["keygen", "--out", support::str(&api_token)?])?;
cli(
&data,
&key,
&["keygen", "--out", support::str(&webhook_token)?],
)?;
// Runs execute exclusively in the serve daemon's worker loop; MCP only
// queues and polls. Spawn a native-allowed daemon for the happy path.
let daemon = ServeDaemon::start(&data, &key, &api_token, &webhook_token, true)?;
let mut native_session = McpSession::start(&data, &key, true)?;
native_session.call(
"akurai_repo_add",
json!({"name": "fixture", "url": support::str(&repository)?, "branch": "main"}),
)?;
native_session.call(
"akurai_installation_register",
json!({
"installation_id": "fixture-installation",
"organization_id": "fixture-org",
"workspace_id": "fixture-workspace",
"app_id": "fixture-app",
"idempotency_key": "fixture-installation-v1"
}),
)?;
let first = native_session.call(
"akurai_run_queue",
json!({"repository": "fixture", "wait": true}),
)?;
ensure!(
first["status"] == "waiting",
"run should wait for approval: {first}"
);
let id = first["id"].as_i64().context("run has no id")?;
assert!(id >= 1, "first run id should be positive");
ensure!(
first["repository"] == "fixture",
"run should reference the correct repository"
);
ensure!(
first["trigger"] == "manual",
"run should record trigger as manual"
);
let promoted = native_session.call(
"akurai_run_promote",
json!({"id": id, "environment": "test", "wait": true}),
)?;
ensure!(
promoted["status"] == "succeeded",
"promotion failed: {promoted}"
);
let deployments = promoted["deployments"]
.as_array()
.context("deployments should be an array")?;
assert!(
!deployments.is_empty(),
"successful run should have deployments"
);
let deployment = &deployments[0];
ensure!(
deployment["status"] == "succeeded",
"deployment was not persisted as succeeded"
);
ensure!(
deployment["environment"] == "test",
"deployment environment should be 'test'"
);
ensure!(
deployment["organization_id"] == "fixture-org"
&& deployment["workspace_id"] == "fixture-workspace"
&& deployment["installation_id"] == "fixture-installation"
&& deployment["installation_revision"] == 1
&& deployment["app_id"] == "fixture-app"
&& deployment["commit_sha"] == promoted["commit_sha"]
&& deployment["result"] == "success",
"managed deployment provenance was not persisted: {deployment}"
);
ensure!(
deployment["artifact_digest"]
.as_str()
.is_some_and(|digest| digest.len() == 64),
"deployment artifact digest should be a SHA-256 value: {deployment}"
);
let tenant_deployments = native_session.call(
"akurai_deployments_for_tenant",
json!({
"run_id": id,
"organization_id": "fixture-org",
"workspace_id": "fixture-workspace"
}),
)?;
ensure!(
tenant_deployments
.as_array()
.is_some_and(|rows| rows.len() == 1),
"tenant deployment query should return exactly one row: {tenant_deployments}"
);
let waited = native_session.call("akurai_run_wait", json!({"id": id, "timeout": 1}))?;
ensure!(waited["status"] == "succeeded", "wait returned early");
// Verify artifact metadata.
let artifacts = promoted["artifacts"]
.as_array()
.context("artifacts should be an array")?;
assert!(
!artifacts.is_empty(),
"successful run should produce artifacts"
);
let artifact = &artifacts[0];
let artifact_id = artifact["id"].as_i64().context("artifact has no id")?;
assert!(artifact_id >= 1, "artifact id should be positive");
ensure!(
artifact["name"] == "out/artifact.txt",
"artifact should have the correct name: {}",
artifact["name"]
);
ensure!(
artifact["bytes"].as_i64().is_some_and(|s| s > 0),
"artifact should have a positive byte count: {}",
artifact["bytes"]
);
let artifact_path = artifact["path"].as_str().context("artifact has no path")?;
let output = temporary.path().join("download.txt");
native_session.call(
"akurai_artifact_get",
json!({"id": artifact_id, "output": support::str(&output)?}),
)?;
ensure!(
fs::read_to_string(&output)? == "hello",
"downloaded artifact bytes differ"
);
// Tamper detection.
let stored = data.join(artifact_path);
fs::write(&stored, "tampered")?;
let rejected = native_session.call_error(
"akurai_artifact_get",
json!({
"id": artifact_id,
"output": support::str(&temporary.path().join("tampered.txt"))?
}),
)?;
ensure!(
rejected.contains("integrity"),
"tampered artifact was not rejected: {rejected}"
);
drop(native_session);
drop(daemon);
// A daemon without native allowance must fail native jobs closed.
let _denied_daemon = ServeDaemon::start(&data, &key, &api_token, &webhook_token, false)?;
// Running without native allowance should fail closed.
let mut denied_session = McpSession::start(&data, &key, false)?;
let denied = denied_session.call(
"akurai_run_queue",
json!({"repository": "fixture", "wait": true}),
)?;
ensure!(
denied["status"] == "failed",
"native execution did not fail closed: {}",
denied["error"]
);
// Runs listing should reflect the two runs.
let runs_json = denied_session.call("akurai_runs", json!({"repo": "fixture", "limit": 10}))?;
let runs: &Vec<Value> = runs_json.as_array().context("runs should be an array")?;
assert_eq!(runs.len(), 2, "should have exactly 2 runs for fixture");
let statuses: Vec<&str> = runs.iter().filter_map(|r| r["status"].as_str()).collect();
assert!(
statuses.contains(&"succeeded") && statuses.contains(&"failed"),
"runs should contain succeeded and failed statuses: {statuses:?}"
);
Ok(())
}
/// Serve daemon child for the test: kills the process on drop.
struct ServeDaemon(Child);
impl ServeDaemon {
fn start(
data: &Path,
key: &Path,
api_token: &Path,
webhook_token: &Path,
allow_native: bool,
) -> Result<Self> {
let port = free_port()?;
let address = format!("127.0.0.1:{port}");
let child = Command::new(env!("CARGO_BIN_EXE_akurai"))
.args([
"--data",
support::str(data)?,
"--key-file",
support::str(key)?,
"serve",
"--address",
&address,
"--token-file",
support::str(api_token)?,
"--webhook-secret-file",
support::str(webhook_token)?,
])
// The prior daemon's titan-1 lease is intentionally retained; a
// second worker verifies the replacement execution policy without
// waiting for that safety lease to expire.
.env("AKURAI_WORKERS", if allow_native { "1" } else { "2" })
.env("AKURAI_PUBLIC_URL", format!("http://{address}"))
.env(
"AKURAI_ALLOW_NATIVE",
if allow_native { "true" } else { "false" },
)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()?;
let deadline = Instant::now() + Duration::from_secs(15);
while Instant::now() < deadline {
if TcpStream::connect(&address).is_ok() {
return Ok(Self(child));
}
thread_sleep();
}
let mut child = child;
let _ = child.kill();
anyhow::bail!("serve daemon did not become ready on {address}");
}
}
impl Drop for ServeDaemon {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
fn free_port() -> Result<u16> {
Ok(TcpListener::bind("127.0.0.1:0")?.local_addr()?.port())
}
fn thread_sleep() {
std::thread::sleep(Duration::from_millis(100));
}
fn cli(data: &Path, key: &Path, arguments: &[&str]) -> Result<Value> {
let output = Command::new(env!("CARGO_BIN_EXE_akurai"))
.args([
"--data",
support::str(data)?,
"--key-file",
support::str(key)?,
])
.args(arguments)
.output()?;
let value: Value = serde_json::from_slice(&output.stdout)
.with_context(|| format!("invalid JSON: {}", String::from_utf8_lossy(&output.stdout)))?;
ensure!(
output.status.success(),
"CLI failed: {value}\n{}",
String::from_utf8_lossy(&output.stderr)
);
ensure!(value["ok"] == true, "CLI envelope failed: {value}");
Ok(value)
}
fn git(directory: &Path, arguments: &[&str]) -> Result<()> {
let output = Command::new("git")
.current_dir(directory)
.args(arguments)
.output()?;
ensure!(
output.status.success(),
"git failed: {}",
String::from_utf8_lossy(&output.stderr)
);
Ok(())
}