use std::collections::HashMap;
use std::process::Output;
use std::time::Duration;
use assert_cmd::cargo::cargo_bin;
use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::process::Command;
struct Request {
method: String,
target: String,
headers: HashMap<String, String>,
body: Value,
}
async fn invoke(args: &[&str], response: Value, status: u16) -> (Output, Request) {
invoke_with_lookup(args, response, status, 200).await
}
async fn invoke_with_lookup(args: &[&str], response: Value, status: u16, lookup_status: u16) -> (Output, Request) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}/v1", listener.local_addr().unwrap());
let (finished, done) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let mut observed = Vec::new();
for index in 0..if lookup_status == 200 { 2 } else { 1 } {
let (mut socket, _) = listener.accept().await.unwrap();
let mut bytes = Vec::new();
let header_end = loop {
let mut chunk = [0; 2048];
let count = socket.read(&mut chunk).await.unwrap();
assert_ne!(count, 0, "request ended before headers");
bytes.extend_from_slice(&chunk[..count]);
if let Some(offset) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
break offset + 4;
}
};
let head = String::from_utf8(bytes[..header_end].to_vec()).unwrap();
let mut lines = head.lines();
let mut request_line = lines.next().unwrap().split_whitespace();
let method = request_line.next().unwrap().to_string();
let target = request_line.next().unwrap().to_string();
let headers: HashMap<String, String> = lines
.filter_map(|line| {
let (key, value) = line.split_once(':')?;
Some((key.to_ascii_lowercase(), value.trim().to_string()))
})
.collect();
let content_length = headers
.get("content-length")
.map(|length| length.parse::<usize>().unwrap())
.unwrap_or(0);
while bytes.len() < header_end + content_length {
let mut chunk = [0; 2048];
let count = socket.read(&mut chunk).await.unwrap();
assert_ne!(count, 0, "request ended before body");
bytes.extend_from_slice(&chunk[..count]);
}
let body = if content_length == 0 {
Value::Null
} else {
serde_json::from_slice(&bytes[header_end..header_end + content_length]).unwrap()
};
let (status, response) = if index == 0 {
assert_eq!(method, "GET");
assert_eq!(target, "/v1/projects/project%2Fone");
assert_eq!(headers.get("x-preferred-org-id").map(String::as_str), Some("org_linked"));
(lookup_status, if lookup_status == 200 {
json!({"id":"01900000-0000-7000-8000-000000000001","slug":"project/one","name":"One","orgId":"org_linked","isActive":true})
} else { json!({"error":{"code":"project_not_found","message":"project unavailable"}}) })
} else { (status, response.clone()) };
let response = response.to_string();
socket.write_all(format!(
"HTTP/1.1 {status} Response\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response}", response.len()
).as_bytes()).await.unwrap();
observed.push(Request { method, target, headers, body });
}
if lookup_status != 200 {
tokio::select! {
_ = done => (),
_ = listener.accept() => panic!("CLI contacted Delivery after failed project lookup"),
}
}
observed.pop().unwrap()
});
let workspace = tempfile::tempdir().unwrap();
let config = tempfile::tempdir().unwrap();
std::fs::create_dir(workspace.path().join(".sylphx")).unwrap();
std::fs::write(
workspace.path().join(".sylphx/project.json"),
r#"{"projectId":"project/one","orgId":"org_linked"}"#,
)
.unwrap();
let output = tokio::time::timeout(
Duration::from_secs(15),
Command::new(cargo_bin!("sylphx"))
.current_dir(workspace.path())
.env("SYLPHX_CONFIG_DIR", config.path())
.args(["--token", "svc_test", "--api-url", &base, "--json"])
.args(args)
.output(),
)
.await
.expect("CLI request timed out")
.unwrap();
let _ = finished.send(());
let request = tokio::time::timeout(Duration::from_secs(2), server)
.await
.expect("CLI never contacted the delivery API")
.unwrap();
assert_eq!(
request
.headers
.get("x-preferred-org-id")
.map(String::as_str),
Some("org_linked")
);
(output, request)
}
fn operation(state: &str) -> Value {
json!({"operation": {
"name": "projects/project-one/operations/delivery-1",
"requestId": "attempt-1",
"principal": "principals/test",
"target": "projects/project-one/environments/env_prod",
"mode": "OPERATION_MODE_APPLY",
"execution": "OPERATION_EXECUTION_CLASS_RECONCILED",
"state": state,
"createTime": "2026-09-08T00:00:00Z"
}})
}
fn success(output: &Output) -> Value {
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
serde_json::from_slice(&output.stdout).unwrap()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn completed_build_creation_is_scoped_idempotent_and_does_not_select_traffic() {
let (output, request) = invoke(
&[
"deployments",
"create",
"--service",
"svc_web",
"--completed-build",
"build-17",
"--idempotency-key",
"register-17",
],
operation("OPERATION_STATE_PENDING"),
202,
)
.await;
assert_eq!(request.method, "POST");
assert_eq!(request.target, "/v1/projects/01900000-0000-7000-8000-000000000001/deployments");
assert_eq!(
request.headers.get("idempotency-key").unwrap(),
"register-17"
);
assert_eq!(
request.body,
json!({"serviceId":"svc_web", "completedBuildId":"build-17"})
);
let result = success(&output);
assert_eq!(result["operation"]["state"], "OPERATION_STATE_PENDING");
assert!(result.get("status").is_none());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_keep_deployment_release_receipt_and_decision_identities_separate() {
let deployment = json!({"id":"deployment-17", "projectId":"project/one", "serviceId":"svc_web",
"provenance": {"buildId":"build-17", "artifact":{"digest":format!("sha256:{}", "a".repeat(64))}},
"createdTime":"2026-09-08T00:00:00Z"});
let release = json!({"id":"release-23", "projectId":"project/one", "environmentId":"env_prod",
"status":"RELEASE_STATUS_HOLD", "policyRevision":"policy-5", "selectionGeneration":"7"});
let cases: Vec<(Vec<&str>, &str, Value)> = vec![
(vec!["deployments","list","--service","svc_web","--page-token","page one","--page-size","5"],
"/v1/projects/01900000-0000-7000-8000-000000000001/deployments?pageToken=page%20one&pageSize=5&serviceId=svc_web",
json!({"object":"list", "data":[deployment.clone()]})),
(vec!["deployments","get","deployment-17"], "/v1/projects/01900000-0000-7000-8000-000000000001/deployments/deployment-17",
json!({"deployment":deployment})),
(vec!["releases","list","--environment","env_prod"], "/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases",
json!({"object":"list", "data":[release.clone()]})),
(vec!["releases","get","release-23","--environment","env_prod"], "/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases/release-23",
json!({"release":release})),
(vec!["releases","receipts","release-23","--environment","env_prod"], "/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases/release-23/receipts",
json!({"object":"list", "data":[{"id":"receipt-29","conclusion":"DELIVERY_PROOF_CONCLUSION_SUCCESS"}]})),
(vec!["releases","decisions","release-23","--environment","env_prod"], "/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases/release-23/decisions",
json!({"object":"list", "data":[{"id":"decision-31","releaseId":"release-23","outcome":"DELIVERY_DECISION_OUTCOME_HOLD"}]})),
(vec!["delivery","configuration","get","--environment","env_prod","--service","svc_web"],
"/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/services/svc_web/delivery-configuration",
json!({"configuration":{"organizationId":"org_linked", "projectId":"project/one",
"environmentId":"env_prod", "serviceId":"svc_web",
"environmentServiceId":"22222222-2222-4222-8222-222222222222",
"authoringRevision":"33333333-3333-4333-8333-333333333333"}})),
(vec!["delivery","policy","get","--environment","env_prod"], "/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/delivery-policy",
json!({"policy":{"id":"policy-5","environmentId":"env_prod","policyRevision":"policy-5","version":"2","promote":"PROMOTE_MODE_MANUAL"}})),
];
for (args, target, response) in cases {
let (output, request) = invoke(&args, response.clone(), 200).await;
assert_eq!(request.method, "GET");
assert_eq!(request.target, target);
assert_eq!(request.body, Value::Null);
assert_eq!(success(&output), response);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn recovery_release_binds_existing_deployments_to_inspected_authoring_revisions() {
let members = json!([{
"serviceId":"11111111-1111-4111-8111-111111111111",
"deploymentId":"22222222-2222-4222-8222-222222222222",
"selection":"RELEASE_MEMBER_SELECTION_SELECTED",
"expectedAuthoringRevision":"33333333-3333-4333-8333-333333333333"
}]);
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), members.to_string()).unwrap();
let (output, request) = invoke(
&[
"releases",
"create",
"--environment",
"env_prod",
"--policy-revision",
"policy-5",
"--previous-release",
"release-current",
"--members",
file.path().to_str().unwrap(),
"--idempotency-key",
"recover-2",
],
operation("OPERATION_STATE_PENDING"),
202,
)
.await;
success(&output);
assert_eq!(request.method, "POST");
assert_eq!(
request.target,
"/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases"
);
assert_eq!(request.headers.get("idempotency-key").unwrap(), "recover-2");
assert_eq!(
request.body,
json!({"policyRevision":"policy-5","previousReleaseId":"release-current","members":members})
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn promotion_preserves_the_generation_fence_and_plan_mode() {
for plan in [false, true] {
let mut args = vec![
"releases",
"promote",
"release-23",
"--environment",
"env_prod",
"--expected-selection-generation",
"7",
"--idempotency-key",
"promote-23",
"--expected-policy-revision",
"policy-5",
];
if plan {
args.push("--plan");
}
let (output, request) = invoke(&args, operation("OPERATION_STATE_PENDING"), 202).await;
success(&output);
assert_eq!(request.method, "POST");
assert_eq!(
request.target,
"/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/releases/release-23:promote"
);
assert_eq!(
request.headers.get("idempotency-key").unwrap(),
"promote-23"
);
assert_eq!(
request.body,
json!({"expectedSelectionGeneration":"7", "expectedPolicyRevision":"policy-5", "mode":
if plan {"OPERATION_MODE_PLAN"} else {"OPERATION_MODE_APPLY"}})
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn policy_patch_preserves_false_empty_proofs_and_both_revision_fences() {
let patch = tempfile::NamedTempFile::new().unwrap();
std::fs::write(patch.path(), r#"{"autoDeploy":false,"requiredProofs":{}}"#).unwrap();
let (output, request) = invoke(
&[
"delivery",
"policy",
"patch",
"--environment",
"env_prod",
"--expected-policy-revision",
"policy-5",
"--expected-policy-version",
"2",
"--patch",
patch.path().to_str().unwrap(),
"--idempotency-key",
"policy-6",
],
operation("OPERATION_STATE_SUCCEEDED"),
200,
)
.await;
success(&output);
assert_eq!(request.method, "PATCH");
assert_eq!(
request.target,
"/v1/projects/01900000-0000-7000-8000-000000000001/environments/env_prod/delivery-policy"
);
assert_eq!(request.headers.get("idempotency-key").unwrap(), "policy-6");
assert_eq!(
request.body,
json!({"expectedPolicyRevision":"policy-5","expectedPolicyVersion":"2",
"patch":{"autoDeploy":false,"requiredProofs":{}}})
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn promotion_rejection_preserves_typed_recovery_and_terminal_failure_exits_nonzero() {
let args = [
"releases",
"promote",
"release-23",
"--environment",
"env_prod",
"--expected-selection-generation",
"7",
"--idempotency-key",
"promote-23",
"--expected-policy-revision",
"policy-5",
];
let (output, _) = invoke(
&args,
json!({
"code":"DELIVERY_REQUEST_ERROR_CODE_REVISION_CONFLICT", "message":"selection changed",
"field":"expectedSelectionGeneration", "currentRevision":"8",
"recoveryAction":"Read release-23 again before retrying promotion."
}),
409,
)
.await;
assert!(!output.status.success());
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(stderr.contains("Read release-23 again"), "{stderr}");
assert!(stderr.contains("currentRevision"), "{stderr}");
let (output, _) = invoke(&args, operation("OPERATION_STATE_FAILED"), 200).await;
assert!(!output.status.success());
let stdout: Value = serde_json::from_slice(&output.stdout).unwrap();
assert_eq!(stdout["operation"]["state"], "OPERATION_STATE_FAILED");
}
#[test]
fn retired_source_deploy_flags_and_unfenced_promotions_are_rejected_by_the_parser() {
let invalid: &[&[&str]] = &[
&["deploy", "project-one", "--ref", "main"],
&[
"deployments",
"create",
"--service",
"svc_web",
"--completed-build",
"build-17",
"--idempotency-key",
"x",
"--force",
],
&[
"deployments",
"create",
"--service",
"svc_web",
"--ref",
"main",
"--idempotency-key",
"x",
],
&["releases", "list"],
&[
"releases",
"promote",
"release-23",
"--environment",
"env_prod",
"--idempotency-key",
"x",
],
&[
"releases",
"promote",
"release-23",
"--environment",
"env_prod",
"--expected-selection-generation",
"7",
],
];
for args in invalid {
let output = std::process::Command::new(cargo_bin!("sylphx"))
.args(*args)
.output()
.unwrap();
assert_eq!(
output.status.code(),
Some(2),
"args={args:?}: {}",
String::from_utf8_lossy(&output.stderr)
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn succeeded_creation_retains_the_typed_result_identity_needed_for_release_creation() {
let mut response = operation("OPERATION_STATE_SUCCEEDED");
response["operation"]["result"] = json!({
"@type":"type.googleapis.com/sylphx.app.v1.CreateDeploymentResult",
"deployment":{
"id":"deployment-created-17", "projectId":"project/one", "serviceId":"svc_web",
"provenance":{"buildId":"build-17"}, "createdTime":"2026-09-08T00:00:00Z"
}
});
let (output, _) = invoke(
&[
"deployments",
"create",
"--service",
"svc_web",
"--completed-build",
"build-17",
"--idempotency-key",
"register-17",
],
response,
200,
)
.await;
let result = success(&output);
assert_eq!(
result["operation"]["result"]["deployment"]["id"],
"deployment-created-17"
);
assert_eq!(
result["operation"]["result"]["deployment"]["provenance"]["buildId"],
"build-17"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unknown_nested_file_fields_cannot_silently_clear_policy_or_rewrite_member_intent() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}/v1", listener.local_addr().unwrap());
let config = tempfile::tempdir().unwrap();
let file = tempfile::NamedTempFile::new().unwrap();
let cases = [
(
r#"{"requiredProofs":{"itmes":[{"id":"must-run"}]}}"#,
vec![
"delivery",
"policy",
"patch",
"--expected-policy-revision",
"policy-5",
"--expected-policy-version",
"2",
"--patch",
],
),
(
r#"[{"serviceId":"svc_web","deploymentId":"deployment-17","configuration":{"runtimeConfigDigest":"retired"}}]"#,
vec![
"releases",
"create",
"--policy-revision",
"policy-5",
"--members",
],
),
];
for (body, args) in cases {
std::fs::write(file.path(), body).unwrap();
let output = Command::new(cargo_bin!("sylphx"))
.env("SYLPHX_CONFIG_DIR", config.path())
.args(["--token", "svc_test", "--api-url", &base, "--json"])
.args(args)
.arg(file.path())
.args([
"--project",
"project-one",
"--environment",
"env_prod",
"--idempotency-key",
"attempt-1",
])
.output()
.await
.unwrap();
assert!(!output.status.success());
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(stderr.contains("unknown delivery fields"), "{stderr}");
}
assert!(
tokio::time::timeout(Duration::from_millis(100), listener.accept())
.await
.is_err(),
"malformed delivery intent must not reach the API"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unavailable_linked_project_stops_before_delivery_mutation() {
let (output, request) = invoke_with_lookup(
&["deployments", "create", "--service", "svc_web", "--completed-build", "build-17", "--idempotency-key", "register-17"],
operation("OPERATION_STATE_SUCCEEDED"), 200, 404,
).await;
assert!(!output.status.success());
assert_eq!(request.method, "GET");
assert_eq!(request.target, "/v1/projects/project%2Fone");
}