clusterflux-public/crates/clusterflux-cli/src/tests.rs
Clusterflux release dry run 6f52bb46cd Public dry run dryrun-a43e907efd9d
Source commit: a43e907efd9d1561c23fe73499478e881f868355

Public tree identity: sha256:453dea30195485dd8939f575a69b39f4bb7acd84c4df23f8aa29b55d044f2673
2026-07-15 01:54:51 +02:00

4762 lines
177 KiB
Rust

#![allow(clippy::useless_concat)]
use std::fs;
use clap::CommandFactory;
use super::*;
fn parse(args: &[&str]) -> Cli {
Cli::parse_from(args)
}
fn write_runnable_wasm_project(project: &Path) {
let cli_manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
let sdk = cli_manifest_dir.parent().unwrap().join("clusterflux-sdk");
let shared_target = cli_manifest_dir
.parent()
.and_then(Path::parent)
.unwrap()
.join("target/cli-run-test-fixture");
fs::create_dir_all(project.join("src")).unwrap();
fs::create_dir_all(project.join(".cargo")).unwrap();
fs::write(
project.join("Cargo.toml"),
format!(
"[package]\nname = \"cli-run-fixture\"\nversion = \"0.0.0\"\nedition = \"2021\"\n\n[workspace]\n\n[lib]\ncrate-type = [\"cdylib\"]\n\n[dependencies]\nclusterflux = {{ package = \"clusterflux-sdk\", path = {} }}\n",
serde_json::to_string(&sdk.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(
project.join(".cargo/config.toml"),
format!(
"[build]\ntarget-dir = {}\n",
serde_json::to_string(&shared_target.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(
project.join("src/lib.rs"),
r#"
#[clusterflux::task]
#[unsafe(no_mangle)]
pub extern "C" fn task_add_one(value: i32) -> i32 { value + 1 }
#[clusterflux::main]
pub fn build_main() -> i32 { 7 }
#[clusterflux::main(name = "test")]
pub fn test_main() -> i32 { 8 }
"#,
)
.unwrap();
}
#[test]
fn cli_error_classifier_distinguishes_mvp_failure_categories() {
for (message, category, exit_code) in [
(
"login required before running this command",
"authentication",
20,
),
(
"CLI session credential has expired; run clusterflux login --browser again",
"authentication",
20,
),
(
"CLI session credential has been revoked",
"authentication",
20,
),
("unauthorized tenant action", "authorization", 21),
(
"quota unavailable: resource limit exceeded for api_calls",
"quota",
22,
),
("policy denied native command execution", "policy", 23),
(
"scheduler placement failed: no capable node for placement: project mismatch",
"capability",
24,
),
(
"scheduler placement failed: no capable node for placement: source snapshot unavailable and direct connectivity unavailable",
"connectivity",
25,
),
(
"failed to connect to coordinator: connection refused",
"connectivity",
25,
),
(
"missing environment envs/linux/Containerfile",
"environment",
26,
),
("task exited with status 101 after panic", "program", 27),
(
"project already has active virtual process vp-current",
"active_process",
28,
),
] {
let summary = cli_error_summary(message);
assert_eq!(summary["category"], category, "{message}");
assert_eq!(summary["stable_exit_code"], exit_code, "{message}");
assert_eq!(summary["safe_failure"], true);
assert_eq!(summary["process_exit_code_applied"], false);
assert!(summary["next_actions"].as_array().unwrap().len() >= 2);
if category == "quota" {
assert_eq!(summary["resource_category"], "api_calls");
assert_eq!(summary["community_tier_language"], true);
assert_eq!(summary["community_tier_label"], "community tier");
assert_eq!(summary["private_abuse_heuristics_exposed"], false);
let rendered = human_report(&json!({
"command": "run",
"machine_error": summary,
}));
assert!(rendered.contains("quota tier: community tier"));
let forbidden_tier = ["free", "tier"].join(" ");
assert!(!rendered.to_ascii_lowercase().contains(&forbidden_tier));
}
}
}
#[test]
fn command_report_exit_code_marks_command_failures_only() {
let run_start = run_start_summary(&json!({
"type": "error",
"message": "quota unavailable: resource limit exceeded for api_calls",
}));
let mut report = json!({
"command": "run",
"status": run_start["status"].clone(),
"run_start": run_start,
});
assert_eq!(apply_command_report_exit_code(&mut report), Some(22));
assert_eq!(
report["run_start"]["machine_error"]["process_exit_code_applied"],
true
);
let mut task_list = json!({
"command": "task list",
"tasks": [{
"task": "compile",
"state": "failed",
"machine_error": cli_error_summary_for_category("program", "task exited with status 1"),
}],
});
assert_eq!(apply_command_report_exit_code(&mut task_list), None);
assert_eq!(
task_list["tasks"][0]["machine_error"]["process_exit_code_applied"],
false
);
}
#[test]
fn top_level_version_is_available() {
let command = Cli::command();
assert_eq!(command.get_name(), "clusterflux");
assert!(command.get_version().is_some());
}
#[test]
fn run_defaults_to_current_project_and_build_entry() {
let Cli {
command: Commands::Run(args),
} = parse(&["clusterflux", "run"])
else {
panic!("wrong command");
};
let plan = run_plan(args, PathBuf::from("/repo"), CliSession::Anonymous).unwrap();
assert_eq!(plan.project, PathBuf::from("/repo"));
assert_eq!(plan.entry, "build");
assert_eq!(plan.coordinator, CoordinatorSelection::LocalOnly);
assert_eq!(plan.hosted_coordinator_endpoint, None);
assert_eq!(plan.session, CliSession::Anonymous);
}
#[test]
fn non_interactive_run_without_session_requires_explicit_auth_or_local() {
let Cli {
command: Commands::Run(args),
} = parse(&["clusterflux", "run", "--non-interactive", "--json"])
else {
panic!("wrong command");
};
let report = run_report(args, PathBuf::from("/repo"), CliSession::Anonymous).unwrap();
assert_eq!(report["command"], "run");
assert_eq!(report["status"], "authentication_required");
assert_eq!(report["non_interactive"], true);
assert_eq!(report["browser_opened"], false);
assert_eq!(report["private_website_required"], false);
assert_eq!(report["machine_error"]["category"], "authentication");
assert_eq!(report["machine_error"]["stable_exit_code"], 20);
assert_eq!(report["machine_error"]["browser_opened"], false);
let next_actions = report["machine_error"]["next_actions"]
.as_array()
.unwrap()
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>();
assert!(next_actions.contains(&"clusterflux login --browser"));
assert!(next_actions.contains(&"set CLUSTERFLUX_AGENT_PRIVATE_KEY for automation"));
assert!(next_actions.contains(&"pass --local to run against local services"));
}
#[test]
fn run_project_and_named_entry_are_respected() {
let Cli {
command: Commands::Run(args),
} = parse(&["clusterflux", "run", "test", "--project", "/other"])
else {
panic!("wrong command");
};
let plan = run_plan(args, PathBuf::from("/repo"), CliSession::HumanSession).unwrap();
assert_eq!(plan.project, PathBuf::from("/other"));
assert_eq!(plan.entry, "test");
assert_eq!(plan.coordinator, CoordinatorSelection::Hosted);
assert_eq!(
plan.hosted_coordinator_endpoint.as_deref(),
Some(DEFAULT_HOSTED_COORDINATOR_ENDPOINT)
);
assert_eq!(plan.session, CliSession::HumanSession);
}
#[test]
fn node_attach_detects_and_accepts_capability_overrides() {
let Cli {
command: Commands::Node {
command: NodeCommands::Attach(args),
},
} = parse(&["clusterflux", "node", "attach", "--cap", "quic-direct"])
else {
panic!("wrong command");
};
let plan = attach_plan(args);
assert!(plan
.capabilities
.capabilities
.contains(&Capability::QuicDirect));
assert!(!plan.capabilities.arch.is_empty());
assert!(plan.detection.auto_detected);
assert_eq!(plan.detection.os, plan.capabilities.os);
assert_eq!(plan.detection.arch, plan.capabilities.arch);
assert_eq!(plan.detection.command_backend, "native-command");
assert!(plan.detection.command_backend_available);
assert!(plan.detection.manual_capability_overrides_allowed);
assert_eq!(
plan.detection.manual_capability_overrides,
vec!["quic-direct".to_owned()]
);
assert!(plan
.detection
.recognized_capability_overrides
.contains(&Capability::QuicDirect));
assert!(plan.detection.unrecognized_capability_overrides.is_empty());
assert!(!plan.detection.os_arch_capabilities_require_manual_flags);
assert!(plan
.detection
.source_provider_backends
.iter()
.any(|provider| provider.provider == "filesystem" && provider.detected));
}
#[test]
fn node_attach_discloses_dangerous_capability_grants() {
let Cli {
command: Commands::Node {
command: NodeCommands::Attach(args),
},
} = parse(&[
"clusterflux",
"node",
"attach",
"--cap",
"network",
"--cap",
"host-filesystem",
"--cap",
"secrets",
])
else {
panic!("wrong command");
};
let plan = attach_plan(args);
let grants = plan
.grant_disclosures
.iter()
.map(|disclosure| disclosure.grant.as_str())
.collect::<Vec<_>>();
assert!(grants.contains(&"native_command_execution"));
assert!(grants.contains(&"source_access"));
assert!(grants.contains(&"network_access"));
assert!(grants.contains(&"host_filesystem_access"));
assert!(grants.contains(&"secret_access"));
assert!(plan
.grant_disclosures
.iter()
.all(|disclosure| disclosure.coordinator_policy_limited));
let rendered = human_report(&json!({
"command": "node attach",
"node": plan.node,
"grant_disclosures": plan.grant_disclosures,
}));
assert!(rendered.contains("grant native_command_execution"));
assert!(rendered.contains("grant network_access"));
assert!(rendered.contains("policy-limited"));
}
#[test]
fn agents_can_select_hosted_with_public_key_identity() {
let args = RunArgs {
entry: None,
project: None,
coordinator: None,
local: false,
non_interactive: true,
json: false,
};
let plan = run_plan(
args,
PathBuf::from("/repo"),
CliSession::AgentPublicKey {
agent: "agent-ci".to_owned(),
public_key: "agent-key".to_owned(),
public_key_fingerprint: Digest::sha256("agent-key"),
private_key: None,
browser_interaction_required: false,
},
)
.unwrap();
assert_eq!(plan.coordinator, CoordinatorSelection::Hosted);
assert_eq!(
plan.hosted_coordinator_endpoint.as_deref(),
Some(DEFAULT_HOSTED_COORDINATOR_ENDPOINT)
);
assert_eq!(
plan.session,
CliSession::AgentPublicKey {
agent: "agent-ci".to_owned(),
public_key: "agent-key".to_owned(),
public_key_fingerprint: Digest::sha256("agent-key"),
private_key: None,
browser_interaction_required: false,
}
);
}
#[test]
fn agent_environment_auth_requires_matching_private_key_possession() {
let public_only = agent_session_from_keys(
"agent-ci".to_owned(),
Some("ed25519:claimed-public-key".to_owned()),
None,
)
.unwrap_err();
assert!(public_only.to_string().contains("cannot authenticate"));
let private_key = "ed25519:BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc=";
let public_key =
clusterflux_core::agent_ed25519_public_key_from_private_key(private_key).unwrap();
let session = agent_session_from_keys(
"agent-ci".to_owned(),
Some(public_key.clone()),
Some(private_key.to_owned()),
)
.unwrap()
.unwrap();
assert_eq!(
session,
CliSession::AgentPublicKey {
agent: "agent-ci".to_owned(),
public_key: public_key.clone(),
public_key_fingerprint: Digest::sha256(public_key),
private_key: Some(private_key.to_owned()),
browser_interaction_required: false,
}
);
let mismatched = agent_session_from_keys(
"agent-ci".to_owned(),
Some("ed25519:different-public-key".to_owned()),
Some(private_key.to_owned()),
)
.unwrap_err();
assert!(mismatched.to_string().contains("does not match"));
}
#[test]
fn run_with_agent_public_key_sends_attributable_workflow_actor() {
let temp = tempfile::tempdir().unwrap();
write_runnable_wasm_project(temp.path());
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user-live".to_owned(),
coordinator: None,
},
)
.unwrap();
let private_key = "ed25519:BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc=";
let public_key =
clusterflux_core::agent_ed25519_public_key_from_private_key(private_key).unwrap();
let fingerprint = Digest::sha256(&public_key);
let expected_fingerprint = fingerprint.to_string();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"start_process""#));
assert!(line.contains(r#""tenant":"tenant-live""#));
assert!(line.contains(r#""project":"project-live""#));
assert!(line.contains(r#""actor_agent":"agent-ci""#));
assert!(line.contains(&format!(
r#""agent_public_key_fingerprint":"{expected_fingerprint}""#
)));
assert!(line.contains(r#""agent_signature""#));
assert!(line.contains(r#""nonce":"agent-signature-"#));
assert!(line.contains(r#""signature":"ed25519:"#));
assert!(!line.contains(r#""actor_user""#));
stream
.write_all(
format!(
r#"{{"type":"process_started","process":"vp-current","epoch":7,"actor":{{"kind":"agent","user":"user-live","agent":"agent-ci","credential_kind":"public_key","public_key_fingerprint":"{expected_fingerprint}","authenticated_without_browser":true,"scopes":["project:read","project:run"]}}}}"#
)
.as_bytes(),
)
.unwrap();
stream.write_all(b"\n").unwrap();
let mut launch_line = String::new();
reader.read_line(&mut launch_line).unwrap();
assert!(launch_line.contains(r#""type":"launch_task""#));
assert!(launch_line.contains(r#""tenant":"tenant-live""#));
assert!(launch_line.contains(r#""project":"project-live""#));
assert!(launch_line.contains(r#""process":"vp-current""#));
let launch: Value = serde_json::from_str(&launch_line).unwrap();
let task_definition = launch["payload"]["task_spec"]["task_definition"]
.as_str()
.expect("launch must use the selected entrypoint stable id");
assert!(task_definition.starts_with("sha256:"));
assert_eq!(
launch["payload"]["task_spec"]["task_instance"],
"ti:vp-current:main"
);
assert_eq!(
launch["payload"]["task_spec"]["failure_policy"],
"fail_fast"
);
assert!(launch_line.contains(r#""required_capabilities":[]"#));
assert!(launch_line.contains(r#""wasm_module_base64":""#));
assert!(launch_line.contains(r#""kind":"coordinator_node_wasm""#));
assert!(launch_line.contains(r#""export":"clusterflux_entry_v1_"#));
assert!(!launch_line.contains(r#""command":""#));
assert!(launch_line.contains(r#""actor_agent":"agent-ci""#));
assert!(launch_line.contains(&format!(
r#""agent_public_key_fingerprint":"{expected_fingerprint}""#
)));
assert!(launch_line.contains(r#""agent_signature""#));
assert!(launch_line.contains(r#""nonce":"agent-signature-"#));
assert!(launch_line.contains(r#""signature":"ed25519:"#));
assert!(!launch_line.contains(r#""actor_user""#));
stream
.write_all(
serde_json::to_string(&json!({
"type": "main_launched",
"process": "vp-current",
"task_definition": task_definition,
"task_instance": "ti:vp-current:main",
"placement": {"node": "agent-worker", "capabilities": []},
}))
.unwrap()
.as_bytes(),
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let report = run_report(
RunArgs {
entry: Some("build".to_owned()),
project: Some(temp.path().to_path_buf()),
coordinator: Some(format!("clusterflux+tcp://{addr}")),
local: false,
non_interactive: true,
json: false,
},
PathBuf::from("/unused"),
CliSession::AgentPublicKey {
agent: "agent-ci".to_owned(),
public_key,
public_key_fingerprint: fingerprint,
private_key: Some(private_key.to_owned()),
browser_interaction_required: false,
},
)
.unwrap();
server.join().unwrap();
assert_eq!(report["status"], "main_launched");
assert_eq!(report["workflow_actor"]["kind"], "agent");
assert_eq!(report["workflow_actor"]["agent"], "agent-ci");
assert_eq!(
report["workflow_actor"]["authenticated_without_browser"],
true
);
assert_eq!(report["run_start"]["actor"]["kind"], "agent");
assert_eq!(report["task_launch"]["type"], "main_launched");
assert_eq!(report["task_launch"]["placement"]["node"], "agent-worker");
}
#[test]
fn run_with_human_session_derives_workflow_scope_from_authenticated_envelope() {
let temp = tempfile::tempdir().unwrap();
write_runnable_wasm_project(temp.path());
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "spoofed-tenant".to_owned(),
project: "spoofed-project".to_owned(),
user: "spoofed-user".to_owned(),
coordinator: None,
},
)
.unwrap();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("run-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
for (expected_operation, response) in [
(
"start_process",
br#"{"type":"process_started","process":"vp-current","epoch":7,"actor":{"kind":"user","user":"user-session","agent":null,"credential_kind":"cli_device_session","public_key_fingerprint":null,"authenticated_without_browser":false,"scopes":["project:read","project:run"]}}"#.as_slice(),
),
(
"launch_task",
br#"{"type":"main_launched","process":"vp-current","task_definition":"entry-build","task_instance":"ti:vp-current:main","placement":{"node":"human-worker","capabilities":[]}}"#.as_slice(),
),
] {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let wire: Value = serde_json::from_str(&line).unwrap();
assert_eq!(wire["type"], "coordinator_request");
assert_eq!(wire["operation"], "authenticated");
assert_eq!(wire["authentication"]["kind"], "cli_session");
assert_eq!(wire["payload"]["type"], "authenticated");
assert_eq!(wire["payload"]["session_secret"], "run-session-secret");
assert_eq!(wire["payload"]["request"]["type"], expected_operation);
assert!(wire["payload"]["request"].get("tenant").is_none());
assert!(wire["payload"]["request"].get("project").is_none());
assert!(wire["payload"]["request"].get("actor_user").is_none());
stream.write_all(response).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let report = run_report(
RunArgs {
entry: Some("build".to_owned()),
project: Some(temp.path().to_path_buf()),
coordinator: None,
local: false,
non_interactive: true,
json: false,
},
PathBuf::from("/unused"),
CliSession::HumanSession,
)
.unwrap();
server.join().unwrap();
assert_eq!(report["status"], "main_launched");
assert_eq!(report["workflow_actor"]["user"], "user-session");
assert_eq!(report["task_launch"]["placement"]["node"], "human-worker");
}
#[test]
fn run_local_flag_overrides_logged_in_hosted_selection() {
let Cli {
command: Commands::Run(args),
} = parse(&["clusterflux", "run", "--local"])
else {
panic!("wrong command");
};
let plan = run_plan(args, PathBuf::from("/repo"), CliSession::HumanSession).unwrap();
assert_eq!(plan.coordinator, CoordinatorSelection::LocalOnly);
assert_eq!(plan.hosted_coordinator_endpoint, None);
}
#[test]
fn run_contacts_configured_coordinator_and_reports_active_process_conflicts() {
let temp = tempfile::tempdir().unwrap();
write_runnable_wasm_project(temp.path());
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user-live".to_owned(),
coordinator: None,
},
)
.unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for (index, response) in [
r#"{"type":"process_started","process":"vp-current","epoch":7}"#,
r#"{"type":"error","message":"coordinator request failed: unauthorized coordinator action: project already has active virtual process vp-current; attach to or restart it, request cooperative cancellation, abort it, or use another Coordinator Project"}"#,
]
.into_iter()
.enumerate()
{
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"start_process""#));
assert!(line.contains(r#""tenant":"tenant-live""#));
assert!(line.contains(r#""project":"project-live""#));
assert!(line.contains(r#""process":"vp-current""#));
assert!(line.contains(r#""restart":false"#));
stream.write_all(response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
if index == 0 {
let mut launch_line = String::new();
reader.read_line(&mut launch_line).unwrap();
assert!(launch_line.contains(r#""type":"launch_task""#));
assert!(launch_line.contains(r#""tenant":"tenant-live""#));
assert!(launch_line.contains(r#""project":"project-live""#));
assert!(launch_line.contains(r#""process":"vp-current""#));
let launch: Value = serde_json::from_str(&launch_line).unwrap();
let task_definition = launch["payload"]["task_spec"]["task_definition"]
.as_str()
.expect("launch must use the selected entrypoint stable id");
assert!(task_definition.starts_with("sha256:"));
assert_eq!(
launch["payload"]["task_spec"]["task_instance"],
"ti:vp-current:main"
);
assert!(launch_line.contains(r#""actor_user":"user-live""#));
assert!(launch_line.contains(r#""required_capabilities":[]"#));
assert!(launch_line.contains(r#""wasm_module_base64":""#));
assert!(launch_line.contains(r#""kind":"coordinator_node_wasm""#));
assert!(launch_line.contains(r#""export":"clusterflux_entry_v1_"#));
assert!(!launch_line.contains(r#""command":""#));
assert!(!launch_line.contains("--manifest-path"));
stream
.write_all(
serde_json::to_string(&json!({
"type": "main_launched",
"process": "vp-current",
"task_definition": task_definition,
"task_instance": "ti:vp-current:main",
"placement": {"node": "worker-live", "capabilities": []},
}))
.unwrap()
.as_bytes(),
)
.unwrap();
stream.write_all(b"\n").unwrap();
}
}
});
let started = run_report(
RunArgs {
entry: Some("build".to_owned()),
project: Some(temp.path().to_path_buf()),
coordinator: Some(format!("clusterflux+tcp://{addr}")),
local: false,
non_interactive: false,
json: false,
},
PathBuf::from("/unused"),
CliSession::Anonymous,
)
.unwrap();
let blocked = run_report(
RunArgs {
entry: Some("test".to_owned()),
project: Some(temp.path().to_path_buf()),
coordinator: Some(format!("clusterflux+tcp://{addr}")),
local: false,
non_interactive: false,
json: false,
},
PathBuf::from("/unused"),
CliSession::Anonymous,
)
.unwrap();
server.join().unwrap();
assert_eq!(started["status"], "main_launched");
assert_eq!(started["entry"], "build");
assert_eq!(started["tenant"], "tenant-live");
assert_eq!(started["project"], "project-live");
assert_eq!(started["process"], "vp-current");
assert_eq!(started["run_start"]["restart"], false);
assert_eq!(started["run_start"]["single_active_process_boundary"], true);
assert_eq!(started["task_launch"]["type"], "main_launched");
assert_eq!(started["task_launch"]["placement"]["node"], "worker-live");
assert_eq!(blocked["status"], "blocked_active_process");
assert_eq!(blocked["entry"], "test");
assert_eq!(
blocked["run_start"]["category"],
"active_process_already_running"
);
assert_eq!(blocked["run_start"]["error_category"], "active_process");
assert_eq!(blocked["run_start"]["stable_exit_code"], 28);
assert_eq!(
blocked["run_start"]["machine_error"]["category"],
"active_process"
);
assert_eq!(
blocked["run_start"]["machine_error"]["stable_exit_code"],
28
);
assert_eq!(blocked["run_start"]["safe_failure"], true);
assert!(blocked["run_start"]["next_actions"]
.as_array()
.unwrap()
.iter()
.any(|action| action == "clusterflux process restart --yes"));
}
#[test]
fn run_rejection_reports_machine_readable_error_category() {
let rejected = run_start_summary(&json!({
"type": "error",
"message": "quota unavailable: resource limit exceeded for api_calls"
}));
assert_eq!(rejected["status"], "coordinator_rejected");
assert_eq!(rejected["error_category"], "quota");
assert_eq!(rejected["stable_exit_code"], 22);
assert_eq!(rejected["machine_error"]["category"], "quota");
assert_eq!(rejected["machine_error"]["resource_category"], "api_calls");
assert_eq!(rejected["machine_error"]["community_tier_language"], true);
assert_eq!(
rejected["machine_error"]["private_abuse_heuristics_exposed"],
false
);
assert!(rejected["machine_error"]["next_actions"]
.as_array()
.unwrap()
.iter()
.any(|action| action == "clusterflux quota status"));
}
#[test]
fn local_only_run_executes_ephemeral_local_services() {
let Cli {
command: Commands::Run(args),
} = parse(&["clusterflux", "run", "--local"])
else {
panic!("wrong command");
};
let plan = run_plan(args, PathBuf::from("/repo"), CliSession::Anonymous).unwrap();
assert!(should_execute_local_node(&plan));
}
#[test]
fn login_defaults_to_the_server_owned_browser_flow() {
let Cli {
command: Commands::Login(args),
} = parse(&["clusterflux", "login"])
else {
panic!("wrong command");
};
let plan = login_plan(args);
assert_eq!(plan.coordinator, DEFAULT_HOSTED_COORDINATOR_ENDPOINT);
let LoginFlowPlan::Browser(flow) = plan.human_flow;
assert_eq!(flow.authorization_url, None);
assert!(flow.server_owns_state);
assert!(flow.server_owns_nonce);
assert!(flow.pkce_required);
assert!(flow.hosted_callback);
assert!(!flow.cli_receives_provider_authorization_code);
assert!(!flow.cli_submits_identity_claims);
}
#[test]
fn browser_login_flow_is_available_for_humans() {
let Cli {
command: Commands::Login(args),
} = parse(&["clusterflux", "login", "--browser"])
else {
panic!("wrong command");
};
let plan = login_plan(args);
let LoginFlowPlan::Browser(flow) = plan.human_flow;
assert_eq!(flow.authorization_url, None);
assert!(flow.server_owns_state);
assert!(flow.server_owns_nonce);
assert!(flow.pkce_required);
assert!(flow.hosted_callback);
assert!(!flow.cli_receives_provider_authorization_code);
assert!(!flow.cli_submits_identity_claims);
}
#[test]
fn login_inherits_the_current_project_scope() {
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "workspace-tenant".to_owned(),
project: "workspace-project".to_owned(),
user: "workspace-user".to_owned(),
coordinator: Some("https://workspace.example:9443".to_owned()),
},
)
.unwrap();
let Cli {
command: Commands::Login(args),
} = parse(&["clusterflux", "login", "--browser"])
else {
panic!("wrong command");
};
let args = login_args_for_project(args, temp.path()).unwrap();
assert_eq!(args.project, "workspace-project");
assert_eq!(args.coordinator, "https://workspace.example:9443");
}
#[test]
fn auth_status_reports_a_session_that_does_not_match_the_current_project() {
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "workspace-tenant".to_owned(),
project: "workspace-project".to_owned(),
user: "workspace-user".to_owned(),
coordinator: Some("https://workspace.example:9443".to_owned()),
},
)
.unwrap();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: "https://workspace.example:9443".to_owned(),
tenant: "other-tenant".to_owned(),
project: "other-project".to_owned(),
user: "other-user".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("other-session".to_owned()),
token_expiry_posture: "expires_at".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let Cli {
command: Commands::Auth {
command: AuthCommands::Status(args),
},
} = parse(&["clusterflux", "auth", "status"])
else {
panic!("wrong command");
};
let report = auth_status_report(args, temp.path().to_path_buf()).unwrap();
assert_eq!(report["session_matches_current_project"], false);
assert_eq!(
report["coordinator_account_status"]["reason"],
"stored CLI session does not match the current project"
);
assert_eq!(
report["coordinator_account_status"]["authenticated_for_current_project"],
false
);
}
#[test]
fn browser_login_non_interactive_fails_before_opening_browser() {
let Cli {
command: Commands::Login(args),
} = parse(&["clusterflux", "login", "--browser", "--non-interactive"])
else {
panic!("wrong command");
};
let report = non_interactive_browser_login_report(&args);
assert_eq!(report["command"], "login");
assert_eq!(report["status"], "authentication_required");
assert_eq!(report["non_interactive"], true);
assert_eq!(report["browser_requested"], true);
assert_eq!(report["browser_opened"], false);
assert_eq!(report["machine_error"]["category"], "authentication");
assert_eq!(report["machine_error"]["stable_exit_code"], 20);
assert_eq!(report["machine_error"]["browser_opened"], false);
let next_actions = report["machine_error"]["next_actions"]
.as_array()
.unwrap()
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>();
assert!(next_actions.contains(&"rerun without --non-interactive to open the browser"));
assert!(next_actions.contains(&"use CLUSTERFLUX_AGENT_PRIVATE_KEY for automation"));
}
#[test]
fn browser_login_completion_detects_raw_provider_token_fields() {
assert!(contains_provider_token_field(&json!({
"session": {
"access_token": "secret"
}
})));
assert!(!contains_provider_token_field(&json!({
"session": {
"cli_session_credential_kind": "CliDeviceSession",
"oidc_token_exchange": {
"received_access_token": true,
"received_id_token": true
}
}
})));
}
#[test]
fn stored_browser_login_session_omits_provider_token_values() {
let stored = stored_cli_session_from_login_response(
"https://coord.example.test",
&json!({
"session": {
"tenant": "tenant-live",
"project": "project-live",
"user": "user-live",
"cli_session_credential_kind": "CliDeviceSession",
"cli_session_secret": "clusterflux-cli-session-secret",
"expires_at": "2026-07-04T00:00:00Z",
"access_token": "provider-secret",
"id_token": "provider-id-token",
"provider_tokens_sent_to_nodes": false
}
}),
true,
false,
)
.unwrap();
let serialized = serde_json::to_string(&stored).unwrap();
assert_eq!(stored.kind, "human");
assert_eq!(stored.coordinator, "https://coord.example.test");
assert_eq!(stored.tenant, "tenant-live");
assert_eq!(stored.project, "project-live");
assert_eq!(stored.user, "user-live");
assert_eq!(
stored.session_secret.as_deref(),
Some("clusterflux-cli-session-secret")
);
assert_eq!(stored.token_expiry_posture, "expires_at");
assert!(stored.provider_tokens_exposed_to_cli);
assert!(!stored.provider_tokens_sent_to_nodes);
assert!(!serialized.contains("provider-secret"));
assert!(!serialized.contains("provider-id-token"));
assert!(!serialized.contains("access_token"));
assert!(!serialized.contains("id_token"));
}
#[test]
fn stored_browser_login_session_accepts_hosted_epoch_expiry() {
let stored = stored_cli_session_from_login_response(
"https://hosted.example.test",
&json!({
"session": {
"tenant": "tenant-live",
"project": "project-live",
"user": "user-live",
"cli_session_credential_kind": "CliDeviceSession",
"cli_session_secret": "clusterflux-cli-session-secret",
"expires_at_epoch_seconds": 1_800_000_000_u64,
"provider_tokens_sent_to_nodes": false
}
}),
false,
false,
)
.unwrap();
assert_eq!(stored.token_expiry_posture, "expires_at");
assert_eq!(stored.expires_at.as_deref(), Some("1800000000"));
}
#[test]
fn agent_enroll_uses_public_key_without_browser_each_run() {
let Cli {
command: Commands::Agent {
command: AgentCommands::Enroll(args),
},
} = parse(&[
"clusterflux",
"agent",
"enroll",
"--public-key",
"agent-key",
])
else {
panic!("wrong command");
};
let plan = agent_enrollment_plan(args);
assert!(!plan.browser_interaction_required_each_run);
assert!(plan.public_key_fingerprint.as_str().starts_with("sha256:"));
}
#[test]
fn bundle_inspect_discovers_environments_selected_inputs_and_source_providers() {
let temp = tempfile::tempdir().unwrap();
fs::create_dir_all(temp.path().join("envs/linux")).unwrap();
fs::create_dir_all(temp.path().join("envs/docker")).unwrap();
fs::create_dir_all(temp.path().join("src")).unwrap();
fs::write(
temp.path().join("envs/linux/Containerfile"),
"FROM alpine\n",
)
.unwrap();
fs::write(
temp.path().join("envs/docker/Dockerfile"),
"FROM debian:stable-slim\n",
)
.unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
fs::write(
temp.path().join("src/main.rs"),
"fn main() {}\n\n#[clusterflux::task]\nfn compile_linux() {}\n\n#[clusterflux::main]\nfn build_main() {}\n",
)
.unwrap();
let Cli {
command: Commands::Bundle {
command: BundleCommands::Inspect(args),
},
} = parse(&[
"clusterflux",
"bundle",
"inspect",
"--project",
temp.path().to_str().unwrap(),
])
else {
panic!("wrong command");
};
let inspection = bundle_inspection(args, PathBuf::from("/unused")).unwrap();
assert_eq!(inspection.project, temp.path());
assert!(inspection
.default_source_providers
.contains(&SourceProviderKind::Git));
assert!(inspection.source_provider_statuses.iter().any(|status| {
status.provider == "filesystem" && status.status == "enabled" && status.active
}));
assert!(inspection
.source_provider_statuses
.iter()
.any(|status| status.provider == "git" && status.status == "missing"));
assert!(inspection
.source_provider_manifest
.description
.contains("node task"));
assert!(
!inspection
.source_provider_manifest
.coordinator_requires_checkout_access
);
assert!(
!inspection
.source_provider_manifest
.transfer_policy
.default_full_repo_tarball
);
assert!(inspection
.metadata
.environments
.iter()
.any(|environment| environment.name == "linux"));
assert!(inspection
.metadata
.environments
.iter()
.any(|environment| environment.name == "docker"));
assert_eq!(
inspection.metadata.source_metadata.source_provider_manifest,
inspection.source_provider_manifest.digest
);
assert!(
inspection
.metadata
.source_metadata
.transfer_policy
.local_source_bytes_remain_node_local
);
assert!(
!inspection
.metadata
.source_metadata
.transfer_policy
.coordinator_receives_source_bytes_by_default
);
assert!(
!inspection
.metadata
.source_metadata
.transfer_policy
.default_full_repo_tarball
);
assert!(inspection
.metadata
.task_metadata
.entrypoints
.contains(&"build".to_owned()));
assert_eq!(
inspection.metadata.task_metadata.default_entrypoint,
"build"
);
assert_eq!(
inspection.metadata.task_metadata.task_abi,
task_abi_digest(&ProjectModel::discover_without_config(temp.path()).unwrap())
);
assert!(inspection
.metadata
.wasm_code
.as_str()
.starts_with("sha256:"));
assert!(inspection.metadata.debug_metadata.available);
assert!(inspection.metadata.debug_metadata.source_level_breakpoints);
assert!(inspection.metadata.debug_metadata.variables_pane_supported);
assert!(inspection
.metadata
.debug_metadata
.probes
.iter()
.any(|probe| probe.source_path == "src/main.rs"
&& probe.function == "compile_linux"
&& probe.task.as_str() == "compile_linux"));
assert!(
inspection
.metadata
.large_input_policy
.selected_inputs_are_content_digests
);
assert!(
!inspection
.metadata
.large_input_policy
.selected_input_bytes_included
);
assert!(
!inspection
.metadata
.large_input_policy
.full_repository_bytes_included
);
assert!(
!inspection
.metadata
.large_input_policy
.silent_task_argument_serialization
);
assert!(inspection
.metadata
.large_input_policy
.supported_handle_types
.contains(&"SourceSnapshot".to_owned()));
assert!(
inspection
.metadata
.restart_compatibility
.source_edits_can_restart_from_clean_task_boundary
);
assert!(
inspection
.metadata
.restart_compatibility
.requires_clean_checkpoint_boundary
);
assert_eq!(
inspection.metadata.restart_compatibility.compares_task_abi,
inspection.metadata.task_metadata.task_abi
);
assert!(
inspection
.metadata
.restart_compatibility
.incompatible_changes_require_whole_process_restart
);
assert!(!inspection.metadata.embeds_full_container_images);
assert!(inspection
.metadata
.selected_inputs
.iter()
.any(|input| input.path == "Cargo.toml"));
assert!(inspection
.metadata
.selected_inputs
.iter()
.any(|input| input.path == "src/main.rs"));
assert!(inspection.environment_diagnostics.is_empty());
assert!(inspection
.pre_schedule_diagnostics
.iter()
.any(|diagnostic| diagnostic.category == "capability"
&& diagnostic.message.contains("environment `linux`")));
}
#[test]
fn bundle_inspect_reports_missing_environment_references_before_schedule() {
let temp = tempfile::tempdir().unwrap();
fs::create_dir_all(temp.path().join("src")).unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
fs::write(
temp.path().join("src/main.rs"),
"fn main() { let _target = env!(\"linux\"); }\n",
)
.unwrap();
let inspection = bundle_inspection(
BundleInspectArgs {
project: Some(temp.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert_eq!(inspection.environment_diagnostics.len(), 1);
assert_eq!(inspection.environment_diagnostics[0].path, "src/main.rs");
assert_eq!(
inspection.environment_diagnostics[0].reference.name,
"linux"
);
assert!(inspection.environment_diagnostics[0]
.message
.contains("missing Clusterflux environment `linux`"));
assert!(inspection
.pre_schedule_diagnostics
.iter()
.any(|diagnostic| {
diagnostic.severity == "error"
&& diagnostic.category == "environment"
&& diagnostic.code == "missing_environment"
}));
}
#[test]
fn bundle_inspect_reports_source_provider_overrides_before_schedule() {
let temp = tempfile::tempdir().unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
let missing_git = bundle_inspection(
BundleInspectArgs {
project: Some(temp.path().to_path_buf()),
source_provider: Some("git".to_owned()),
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert!(missing_git
.source_provider_statuses
.iter()
.any(|status| { status.provider == "git" && status.status == "missing" && status.active }));
assert!(missing_git
.pre_schedule_diagnostics
.iter()
.any(|diagnostic| {
diagnostic.category == "source_provider" && diagnostic.code == "source_provider_missing"
}));
let unsupported = bundle_inspection(
BundleInspectArgs {
project: Some(temp.path().to_path_buf()),
source_provider: Some("custom-lfs".to_owned()),
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert!(unsupported.source_provider_statuses.iter().any(|status| {
status.provider == "custom-lfs" && status.status == "unsupported" && status.active
}));
assert!(unsupported
.pre_schedule_diagnostics
.iter()
.any(|diagnostic| {
diagnostic.category == "source_provider"
&& diagnostic.code == "source_provider_unsupported"
}));
}
#[test]
fn bundle_identity_changes_when_selected_input_file_changes() {
let temp = tempfile::tempdir().unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
let first = bundle_inspection(
BundleInspectArgs {
project: Some(temp.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
fs::write(
temp.path().join("Cargo.toml"),
"[package]\nname='changed'\n",
)
.unwrap();
let second = bundle_inspection(
BundleInspectArgs {
project: Some(temp.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert_ne!(first.metadata.identity, second.metadata.identity);
}
#[test]
fn bundle_rebuild_after_source_edit_keeps_restart_compatibility_contract() {
let temp = tempfile::tempdir().unwrap();
fs::create_dir_all(temp.path().join("src")).unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
fs::write(
temp.path().join("src/main.rs"),
"fn main() { println!(\"first\"); }\n",
)
.unwrap();
let inspect = |project: &Path| {
bundle_inspection(
BundleInspectArgs {
project: Some(project.to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: true,
},
PathBuf::from("/unused"),
)
.unwrap()
};
let first = inspect(temp.path());
fs::write(
temp.path().join("src/main.rs"),
"fn main() { println!(\"second\"); }\n",
)
.unwrap();
let rebuilt = inspect(temp.path());
assert_ne!(first.metadata.identity, rebuilt.metadata.identity);
assert_ne!(
first
.metadata
.source_metadata
.selected_inputs
.iter()
.find(|input| input.path == "src/main.rs")
.unwrap()
.digest,
rebuilt
.metadata
.source_metadata
.selected_inputs
.iter()
.find(|input| input.path == "src/main.rs")
.unwrap()
.digest
);
assert_eq!(
first.metadata.task_metadata.task_abi,
rebuilt.metadata.task_metadata.task_abi
);
assert_eq!(
first.metadata.restart_compatibility.compares_task_abi,
rebuilt.metadata.restart_compatibility.compares_task_abi
);
assert!(
rebuilt
.metadata
.restart_compatibility
.source_edits_can_restart_from_clean_task_boundary
);
assert!(
rebuilt
.metadata
.restart_compatibility
.requires_clean_checkpoint_boundary
);
assert!(
rebuilt
.metadata
.restart_compatibility
.discards_unflushed_task_local_changes
);
assert!(
rebuilt
.metadata
.restart_compatibility
.incompatible_changes_require_whole_process_restart
);
}
#[test]
fn source_provider_manifest_digest_does_not_include_local_project_path() {
let first = tempfile::tempdir().unwrap();
let second = tempfile::tempdir().unwrap();
fs::write(first.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
fs::write(second.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
let first = bundle_inspection(
BundleInspectArgs {
project: Some(first.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
let second = bundle_inspection(
BundleInspectArgs {
project: Some(second.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert_eq!(
first.source_provider_manifest.digest,
second.source_provider_manifest.digest
);
}
#[test]
fn node_attach_can_exchange_enrollment_grant() {
let Cli {
command: Commands::Node {
command: NodeCommands::Attach(args),
},
} = parse(&[
"clusterflux",
"node",
"attach",
"--enrollment-grant",
"grant",
"--public-key",
"node-key",
])
else {
panic!("wrong command");
};
let plan = attach_plan(args);
assert!(
plan.enrollment
.unwrap()
.exchanges_short_lived_grant_for_long_lived_node_identity
);
}
#[test]
fn node_attach_enrollment_uses_default_public_key_when_not_explicit() {
let Cli {
command: Commands::Node {
command: NodeCommands::Attach(args),
},
} = parse(&[
"clusterflux",
"node",
"attach",
"--node",
"node-default-key",
"--enrollment-grant",
"grant",
])
else {
panic!("wrong command");
};
let plan = attach_plan(args);
let enrollment = plan.enrollment.unwrap();
assert_eq!(enrollment.grant, "grant");
assert!(enrollment
.public_key_fingerprint
.as_str()
.starts_with("sha256:"));
}
#[test]
fn node_attach_local_credential_is_durable_and_project_scoped() {
let temp = tempfile::tempdir().unwrap();
let node = "node/durable";
let first = node::load_or_create_local_node_credential(temp.path(), node).unwrap();
let second = node::load_or_create_local_node_credential(temp.path(), node).unwrap();
let credential_file = node::local_node_credential_file(temp.path(), node);
assert_eq!(first, second);
assert!(credential_file.exists());
assert!(
credential_file
.strip_prefix(temp.path().join(".clusterflux").join("nodes"))
.unwrap()
.components()
.count()
== 1
);
let public_key = clusterflux_core::node_ed25519_public_key_from_private_key(&first).unwrap();
let bytes = fs::read(credential_file).unwrap();
let stored: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(stored["kind"], "clusterflux_node_credential");
assert_eq!(stored["node"], node);
assert_eq!(stored["public_key"], public_key);
assert_eq!(stored["credential_scope"], "local_project_node_identity");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let file_mode = fs::metadata(node::local_node_credential_file(temp.path(), node))
.unwrap()
.permissions()
.mode()
& 0o777;
let directory_mode = fs::metadata(temp.path().join(".clusterflux").join("nodes"))
.unwrap()
.permissions()
.mode()
& 0o777;
assert_eq!(file_mode, 0o600);
assert_eq!(directory_mode, 0o700);
}
}
#[cfg(unix)]
#[test]
fn node_attach_refuses_a_symlink_credential_target() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().unwrap();
let node = "node/symlink";
let file = node::local_node_credential_file(temp.path(), node);
fs::create_dir_all(file.parent().unwrap()).unwrap();
let target = temp.path().join("attacker-controlled.json");
fs::write(&target, b"{}").unwrap();
symlink(&target, &file).unwrap();
let error = node::load_or_create_local_node_credential(temp.path(), node).unwrap_err();
assert!(error.to_string().contains("symbolic link"));
}
#[test]
fn hosted_coordinator_remains_a_real_https_control_endpoint() {
assert_eq!(
control_endpoint_identity(DEFAULT_HOSTED_COORDINATOR_ENDPOINT).unwrap(),
"https://clusterflux.michelpaulissen.com/api/v1/control"
);
assert_eq!(
control_endpoint_identity("https://clusterflux.michelpaulissen.com/api/v1/control")
.unwrap(),
"https://clusterflux.michelpaulissen.com/api/v1/control"
);
assert!(control_endpoint_identity("http://operator.example.test").is_err());
assert_eq!(
control_endpoint_identity("127.0.0.1:7999").unwrap(),
"clusterflux+tcp://127.0.0.1:7999"
);
}
#[test]
fn doctor_reports_unchecked_coordinator_reachability_without_config() {
let temp = tempfile::tempdir().unwrap();
let report = doctor_report(
DoctorArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(report["command"], "doctor");
assert!(report["coordinator"].is_null());
assert_eq!(report["coordinator_reachability"]["checked"], false);
assert_eq!(
report["coordinator_reachability"]["status"],
"not_configured"
);
assert!(matches!(
report["node_readiness_summary"]["status"].as_str(),
Some("ready_to_attach") | Some("local_dependencies_missing") | Some("limited_capabilities")
));
assert_eq!(
report["node_readiness_summary"]["explicit_attach_required"],
true
);
assert_eq!(
report["node_readiness_summary"]["command_execution_capability"],
true
);
assert!(report["node_readiness_summary"]["missing_local_dependencies"].is_array());
assert!(
report["node_readiness_summary"]["next_actions"]
.as_array()
.unwrap()
.len()
>= 2
);
}
#[test]
fn doctor_pings_configured_coordinator() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let request: serde_json::Value = serde_json::from_str(&line).unwrap();
assert_eq!(request["type"], "coordinator_request");
assert_eq!(request["protocol_version"], 1);
assert_eq!(request["request_id"], "doctor-1");
assert_eq!(request["operation"], "ping");
assert_eq!(request["authentication"]["kind"], "none");
assert_eq!(request["payload"]["type"], "ping");
stream
.write_all(b"{\"type\":\"pong\",\"epoch\":42}\n")
.unwrap();
});
let temp = tempfile::tempdir().unwrap();
let report = doctor_report(
DoctorArgs {
scope: CliScopeArgs {
coordinator: Some(addr.clone()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["coordinator"], addr);
assert_eq!(report["coordinator_reachability"]["checked"], true);
assert_eq!(report["coordinator_reachability"]["status"], "reachable");
assert_eq!(
report["coordinator_reachability"]["response"]["type"],
"pong"
);
assert_eq!(report["coordinator_reachability"]["response"]["epoch"], 42);
}
#[test]
fn auth_status_reads_stored_cli_session_without_provider_tokens() {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"clusterflux-cli-session-secret""#));
assert!(line.contains(r#""type":"auth_status""#));
assert!(!line.contains(r#""actor_user":"user-session""#));
stream
.write_all(
br#"{"type":"auth_status","tenant":"tenant-session","project":"project-session","actor":"user-session","authenticated":true,"account_status":"active","suspended":false,"disabled":false,"sanitized_reason":null,"next_actions":[],"private_moderation_details_exposed":false,"signup_failure_details_exposed":false}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("clusterflux-cli-session-secret".to_owned()),
token_expiry_posture: "expires_at".to_owned(),
expires_at: Some("2026-07-04T00:00:00Z".to_owned()),
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let report = auth_status_report(
AuthStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["active_coordinator"], addr);
assert_eq!(report["principal"], "user-session");
assert_eq!(report["tenant"], "tenant-session");
assert_eq!(report["project"], "project-session");
assert_eq!(report["session"]["kind"], "human");
assert_eq!(report["session"]["source"], "session_file");
assert_eq!(report["session"]["authenticated"], true);
assert_eq!(
report["session"]["cli_session_credential_kind"],
"CliDeviceSession"
);
assert_eq!(
report["coordinator_account_status"]["used_cli_session_credential"],
true
);
assert_eq!(report["session"]["token_expiry_posture"], "expires_at");
assert_eq!(report["session"]["provider_tokens_exposed_to_cli"], false);
assert_eq!(report["session"]["provider_tokens_exposed_to_nodes"], false);
assert_eq!(report["coordinator_account_status"]["checked"], true);
assert_eq!(
report["coordinator_account_status"]["account_status"],
"active"
);
assert_eq!(
report["coordinator_account_status"]["private_moderation_details_exposed"],
false
);
}
#[test]
fn auth_status_reports_expired_or_revoked_cli_session_as_login_required() {
for message in [
"unauthorized coordinator action: CLI session credential has expired; run clusterflux login --browser again",
"unauthorized coordinator action: CLI session credential has been revoked",
] {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server_message = message.to_owned();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"stale-cli-session-secret""#));
assert!(line.contains(r#""type":"auth_status""#));
stream
.write_all(
format!(r#"{{"type":"error","message":{}}}"#, json!(server_message))
.as_bytes(),
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("stale-cli-session-secret".to_owned()),
token_expiry_posture: "expires_at".to_owned(),
expires_at: Some("2026-07-04T00:00:00Z".to_owned()),
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let report = auth_status_report(
AuthStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["coordinator_account_status"]["checked"], true);
assert_eq!(report["coordinator_account_status"]["reachable"], true);
assert_eq!(
report["coordinator_account_status"]["machine_error"]["category"],
"authentication"
);
assert!(
report["coordinator_account_status"]["next_actions"]
.as_array()
.unwrap()
.iter()
.any(|action| action == "clusterflux login --browser")
);
}
}
#[test]
fn auth_status_queries_coordinator_account_state_without_private_moderation_details() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"auth_status""#));
assert!(line.contains(r#""tenant":"tenant-live""#));
assert!(line.contains(r#""project":"project-live""#));
assert!(line.contains(r#""actor_user":"user-live""#));
stream
.write_all(
br#"{"type":"auth_status","tenant":"tenant-live","project":"project-live","actor":"user-live","authenticated":true,"account_status":"suspended","suspended":true,"disabled":false,"sanitized_reason":"account or tenant is suspended by hosted policy","next_actions":["contact the hosted operator"],"private_moderation_details_exposed":false,"signup_failure_details_exposed":false,"abuse_score":99,"moderation_notes":"private moderation note"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let temp = tempfile::tempdir().unwrap();
let report = auth_status_report(
AuthStatusArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user-live".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["command"], "auth status");
assert_eq!(report["coordinator_account_status"]["checked"], true);
assert_eq!(report["coordinator_account_status"]["reachable"], true);
assert_eq!(
report["coordinator_account_status"]["source"],
"public_coordinator_api"
);
assert_eq!(
report["coordinator_account_status"]["account_status"],
"suspended"
);
assert_eq!(report["coordinator_account_status"]["suspended"], true);
assert_eq!(report["coordinator_account_status"]["disabled"], false);
assert_eq!(
report["coordinator_account_status"]["sanitized_reason"],
"account or tenant is suspended by hosted policy"
);
assert_eq!(
report["coordinator_account_status"]["private_moderation_details_exposed"],
false
);
assert_eq!(
report["coordinator_account_status"]["signup_failure_details_exposed"],
false
);
assert_eq!(
report["coordinator_account_status"]["coordinator_response_type"],
"auth_status"
);
assert_eq!(
report["coordinator_account_status"]["coordinator_session_requests"],
1
);
let serialized = serde_json::to_string(&report).unwrap();
assert!(!serialized.contains("abuse_score"));
assert!(!serialized.contains("moderation_notes"));
assert!(!serialized.contains("private moderation note"));
let rendered = human_report(&report);
assert!(rendered.contains("account status: suspended"));
assert!(rendered.contains("account suspended: true"));
assert!(rendered.contains("private moderation details exposed: false"));
assert!(!rendered.contains("private moderation note"));
}
#[test]
fn auth_status_reports_disabled_deleted_and_manual_review_safely() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for (tenant, status, reason, flags) in [
(
"tenant-disabled",
"disabled",
"account or tenant is disabled by hosted policy",
(false, true, false, false),
),
(
"tenant-deleted",
"deleted",
"account or tenant is no longer active",
(false, false, true, false),
),
(
"tenant-review",
"manual_review",
"account or tenant is pending hosted review",
(false, false, false, true),
),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"auth_status""#));
assert!(line.contains(&format!(r#""tenant":"{tenant}""#)));
let (suspended, disabled, deleted, manual_review) = flags;
writeln!(
stream,
"{}",
json!({
"type": "auth_status",
"tenant": tenant,
"project": "project-live",
"actor": "user-live",
"authenticated": true,
"account_status": status,
"suspended": suspended,
"disabled": disabled,
"deleted": deleted,
"manual_review": manual_review,
"sanitized_reason": reason,
"next_actions": ["contact the hosted operator"],
"private_moderation_details_exposed": false,
"signup_failure_details_exposed": false,
"abuse_score": 99,
"moderation_notes": "private moderation note",
"signup_policy_trace": "private signup trace",
})
)
.unwrap();
}
});
for (tenant, status, rendered_marker) in [
("tenant-disabled", "disabled", "account disabled: true"),
("tenant-deleted", "deleted", "account deleted: true"),
(
"tenant-review",
"manual_review",
"account manual review: true",
),
] {
let temp = tempfile::tempdir().unwrap();
let report = auth_status_report(
AuthStatusArgs {
scope: CliScopeArgs {
coordinator: Some(addr.clone()),
tenant: tenant.to_owned(),
project: "project-live".to_owned(),
user: "user-live".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(
report["coordinator_account_status"]["account_status"],
status
);
assert_eq!(
report["coordinator_account_status"]["account_state_known"],
true
);
assert_eq!(
report["coordinator_account_status"]["private_moderation_details_exposed"],
false
);
assert_eq!(
report["coordinator_account_status"]["signup_failure_details_exposed"],
false
);
let serialized = serde_json::to_string(&report).unwrap();
assert!(!serialized.contains("abuse_score"));
assert!(!serialized.contains("moderation_notes"));
assert!(!serialized.contains("signup_policy_trace"));
assert!(!serialized.contains("private moderation note"));
let rendered = human_report(&report);
assert!(rendered.contains(&format!("account status: {status}")));
assert!(rendered.contains(rendered_marker));
assert!(!rendered.contains("private moderation note"));
}
server.join().unwrap();
}
#[test]
fn cli_first_mvp_command_surface_parses() {
for args in [
&["clusterflux", "doctor"][..],
&["clusterflux", "auth", "status"],
&["clusterflux", "logout", "--yes"],
&["clusterflux", "auth", "logout", "--yes"],
&["clusterflux", "login", "--browser", "--non-interactive"],
&[
"clusterflux",
"key",
"add",
"--agent",
"agent",
"--public-key",
"key",
],
&["clusterflux", "key", "list"],
&["clusterflux", "key", "revoke", "--agent", "agent", "--yes"],
&["clusterflux", "project", "init", "--yes"],
&["clusterflux", "project", "status"],
&["clusterflux", "project", "list"],
&["clusterflux", "project", "select", "project"],
&["clusterflux", "inspect"],
&["clusterflux", "build"],
&["clusterflux", "run", "--non-interactive"],
&["clusterflux", "node", "enroll"],
&["clusterflux", "node", "list"],
&["clusterflux", "node", "status"],
&["clusterflux", "node", "revoke", "--node", "node", "--yes"],
&["clusterflux", "process", "list"],
&["clusterflux", "process", "status"],
&["clusterflux", "process", "restart", "--yes"],
&["clusterflux", "process", "cancel", "--yes"],
&["clusterflux", "process", "abort", "--yes"],
&["clusterflux", "task", "list"],
&["clusterflux", "task", "restart", "compile-linux", "--yes"],
&["clusterflux", "logs"],
&["clusterflux", "artifact", "list"],
&["clusterflux", "artifact", "download", "artifact"],
&[
"clusterflux",
"artifact",
"export",
"artifact",
"--to",
"/tmp/out",
],
&["clusterflux", "dap", "--plan"],
&["clusterflux", "debug", "attach"],
&["clusterflux", "quota", "status"],
&["clusterflux", "admin", "status"],
&["clusterflux", "admin", "bootstrap", "--yes"],
&[
"clusterflux",
"admin",
"revoke-node",
"--node",
"node",
"--yes",
],
&["clusterflux", "admin", "stop-process", "--yes"],
&["clusterflux", "admin", "suspend-tenant", "--yes"],
] {
let _ = parse(args);
}
}
#[test]
fn cli_has_no_direct_hosted_account_creation_command() {
for args in [
&["clusterflux", "signup"][..],
&["clusterflux", "account", "create"],
&["clusterflux", "login", "--create-account"],
] {
let error = Cli::try_parse_from(args).unwrap_err().to_string();
assert!(
error.contains("unrecognized subcommand") || error.contains("unexpected argument"),
"expected no direct account creation command for {args:?}, got {error}"
);
}
let mut command = Cli::command();
let help = command.render_help().to_string();
assert!(help.contains("Hosted account creation happens in the browser login flow."));
assert!(!help.contains("clusterflux signup"));
assert!(!help.contains("account create"));
}
#[test]
fn admin_bootstrap_reports_self_hosted_cli_only_path() {
let temp = tempfile::tempdir().unwrap();
let report = admin_bootstrap_report(
AdminBootstrapArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "team".to_owned(),
project: "self-hosted".to_owned(),
user: "admin".to_owned(),
json: false,
},
name: "Self Hosted".to_owned(),
yes: true,
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(report["command"], "admin bootstrap");
assert_eq!(report["mode"], "self_hosted_local");
assert_eq!(report["private_website_required"], false);
assert_eq!(report["self_hosted_cli_only"], true);
assert_eq!(report["project_config_written"], true);
assert_eq!(report["project_init"]["command"], "project init");
assert_eq!(report["project_init"]["private_website_required"], false);
assert_eq!(
report["project_init"]["project_config"]["project"],
"self-hosted"
);
assert_eq!(
report["admin_surfaces"]["node"],
"clusterflux node enroll/list/status/revoke"
);
let steps = report["bootstrap_sequence"].as_array().unwrap();
for expected in [
"start_self_hosted_coordinator",
"create_or_link_project",
"create_node_enrollment_grant",
"attach_worker_node",
"run_process",
"inspect_status_logs_artifacts",
"revoke_access",
] {
assert!(
steps.iter().any(|step| step["step"] == expected),
"missing bootstrap step {expected}"
);
}
assert!(steps.iter().all(|step| {
!step
.get("private_website_required")
.and_then(Value::as_bool)
.unwrap_or(false)
}));
assert!(steps.iter().any(|step| step
.get("command")
.and_then(Value::as_str)
.unwrap_or("")
.contains("clusterflux node enroll")));
assert!(steps.iter().any(|step| step
.get("command")
.and_then(Value::as_str)
.unwrap_or("")
.contains("clusterflux admin revoke-node")));
}
#[test]
fn top_level_help_exposes_primary_workflow_without_auth() {
let mut command = Cli::command();
let help = command.render_help().to_string();
for expected in [
"Primary workflow:",
"clusterflux login --browser",
"clusterflux project init",
"clusterflux node enroll",
"clusterflux node attach",
"clusterflux-node --worker",
"clusterflux run [entry] --project <path>",
"Clusterflux: Launch Virtual Process",
"clusterflux dap",
"clusterflux process status",
"task list",
"logs",
"artifact list",
"--json",
"Hosted account creation happens in the browser login flow.",
] {
assert!(help.contains(expected), "help output missing {expected}");
}
}
#[test]
fn top_level_logout_alias_removes_only_cli_session_state() {
let temp = tempfile::tempdir().unwrap();
let session_file = session_config_file(temp.path());
fs::create_dir_all(session_file.parent().unwrap()).unwrap();
fs::write(&session_file, br#"{"kind":"human","token":"local"}"#).unwrap();
let unconfirmed = logout_report(
AuthLogoutArgs {
yes: false,
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
"logout",
)
.unwrap();
assert_eq!(unconfirmed["status"], "confirmation_required");
assert_eq!(unconfirmed["coordinator_request_sent"], false);
assert_eq!(unconfirmed["machine_error"]["category"], "policy");
assert!(session_file.exists());
let report = logout_report(
AuthLogoutArgs {
yes: true,
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
"logout",
)
.unwrap();
assert_eq!(report["command"], "logout");
assert_eq!(report["requires_confirmation"], false);
assert_eq!(report["removed_cli_session_file"], true);
assert_eq!(report["node_credentials_untouched"], true);
assert!(!session_file.exists());
}
#[test]
fn logout_revokes_stored_cli_session_on_coordinator_before_local_removal() {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("logout-cli-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let session_file = session_config_file(temp.path());
assert!(session_file.exists());
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"logout-cli-session-secret""#));
assert!(line.contains(r#""type":"revoke_cli_session""#));
assert!(!line.contains(r#""actor_user":"user-session""#));
stream
.write_all(
br#"{"type":"cli_session_revoked","tenant":"tenant-session","project":"project-session","actor":"user-session"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let report = logout_report(
AuthLogoutArgs {
yes: true,
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
"logout",
)
.unwrap();
server.join().unwrap();
assert_eq!(report["removed_cli_session_file"], true);
assert_eq!(report["server_session_revocation"]["attempted"], true);
assert_eq!(report["server_session_revocation"]["revoked"], true);
assert_eq!(
report["server_session_revocation"]["coordinator_response_type"],
"cli_session_revoked"
);
assert!(!session_file.exists());
}
#[test]
fn mutating_commands_require_yes_before_side_effects() {
let temp = tempfile::tempdir().unwrap();
let scope = CliScopeArgs {
coordinator: Some("127.0.0.1:9".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let reports = [
key_revoke_report(KeyRevokeArgs {
scope: scope.clone(),
agent: "agent-ci".to_owned(),
yes: false,
})
.unwrap(),
node_revoke_report(
NodeRevokeArgs {
scope: scope.clone(),
node: "node-a".to_owned(),
yes: false,
},
temp.path().to_path_buf(),
)
.unwrap(),
process_restart_report(ProcessRestartArgs {
scope: scope.clone(),
process: "vp".to_owned(),
yes: false,
})
.unwrap(),
process_cancel_report(ProcessCancelArgs {
scope: scope.clone(),
process: "vp".to_owned(),
node: None,
task: None,
yes: false,
})
.unwrap(),
task_restart_report(TaskRestartArgs {
scope: scope.clone(),
task: "compile-linux".to_owned(),
process: "vp".to_owned(),
yes: false,
})
.unwrap(),
admin_suspend_tenant_report(AdminSuspendTenantArgs {
scope,
target_tenant: Some("tenant".to_owned()),
admin_token: None,
yes: false,
})
.unwrap(),
];
for report in reports {
assert_eq!(report["status"], "confirmation_required");
assert_eq!(report["requires_confirmation"], true);
assert_eq!(report["coordinator_request_sent"], false);
assert_eq!(report["safe_failure"], true);
assert_eq!(report["machine_error"]["category"], "policy");
assert_eq!(report["machine_error"]["confirmation_required"], true);
assert!(report["next_actions"]
.as_array()
.unwrap()
.iter()
.any(|action| action.as_str().unwrap_or_default().contains("--yes")));
}
}
#[test]
fn cli_first_json_mode_parses_for_primary_commands() {
for args in [
&["clusterflux", "doctor", "--json"][..],
&["clusterflux", "login", "--json"],
&[
"clusterflux",
"login",
"--browser",
"--non-interactive",
"--json",
],
&["clusterflux", "logout", "--yes", "--json"],
&["clusterflux", "auth", "status", "--json"],
&[
"clusterflux",
"agent",
"enroll",
"--public-key",
"key",
"--json",
],
&[
"clusterflux",
"key",
"add",
"--agent",
"agent",
"--public-key",
"key",
"--json",
],
&["clusterflux", "project", "init", "--yes", "--json"],
&["clusterflux", "inspect", "--json"],
&["clusterflux", "build", "--json"],
&["clusterflux", "bundle", "inspect", "--json"],
&["clusterflux", "run", "--json"],
&["clusterflux", "run", "--non-interactive", "--json"],
&["clusterflux", "node", "attach", "--json"],
&["clusterflux", "node", "enroll", "--json"],
&["clusterflux", "process", "status", "--json"],
&["clusterflux", "task", "list", "--json"],
&["clusterflux", "logs", "--json"],
&["clusterflux", "artifact", "list", "--json"],
&["clusterflux", "artifact", "download", "artifact", "--json"],
&[
"clusterflux",
"artifact",
"export",
"artifact",
"--to",
"/tmp/out",
"--json",
],
&["clusterflux", "dap", "--plan", "--json"],
&["clusterflux", "debug", "attach", "--json"],
&["clusterflux", "quota", "status", "--json"],
&["clusterflux", "admin", "status", "--json"],
] {
let _ = parse(args);
}
}
#[test]
fn key_lifecycle_reports_project_scoped_agent_credentials() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let add_response = concat!(
r#"{"type":"agent_public_key","actor":"user","record":{"tenant":"tenant","project":"project","user":"user","agent":"agent-ci","public_key":"agent-key-v1","public_key_fingerprint":"sha256:agent-v1","version":1,"revoked":false,"scopes":["project:read","project:run"],"human_account_creation_privilege":false,"browser_interaction_required_each_run":false}}"#
);
let list_response = concat!(
r#"{"type":"agent_public_keys","actor":"user","records":[{"tenant":"tenant","project":"project","user":"user","agent":"agent-ci","public_key":"agent-key-v1","public_key_fingerprint":"sha256:agent-v1","version":1,"revoked":false,"scopes":["project:read","project:run"],"human_account_creation_privilege":false,"browser_interaction_required_each_run":false}]}"#
);
let revoke_response = concat!(
r#"{"type":"agent_public_key","actor":"user","record":{"tenant":"tenant","project":"project","user":"user","agent":"agent-ci","public_key":"agent-key-v1","public_key_fingerprint":"sha256:agent-v1","version":1,"revoked":true,"scopes":["project:read","project:run"],"human_account_creation_privilege":false,"browser_interaction_required_each_run":false}}"#
);
for (expected, response) in [
("register_agent_public_key", add_response),
("list_agent_public_keys", list_response),
("revoke_agent_public_key", revoke_response),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(&format!(r#""type":"{expected}""#)));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""user":"user""#));
if expected != "list_agent_public_keys" {
assert!(line.contains(r#""agent":"agent-ci""#));
}
if expected == "register_agent_public_key" {
assert!(line.contains(r#""public_key":"agent-key-v1""#));
}
stream.write_all(response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let added = key_add_report(KeyAddArgs {
scope: scope.clone(),
agent: "agent-ci".to_owned(),
public_key: "agent-key-v1".to_owned(),
})
.unwrap();
let listed = key_list_report(KeyListArgs {
scope: scope.clone(),
})
.unwrap();
let revoked = key_revoke_report(KeyRevokeArgs {
scope,
agent: "agent-ci".to_owned(),
yes: true,
})
.unwrap();
server.join().unwrap();
assert_eq!(added["command"], "key add");
assert_eq!(added["agent"], "agent-ci");
assert_eq!(added["credential_scope"]["actions"][0], "project:read");
assert_eq!(
added["credential_scope"]["human_account_creation_privilege"],
false
);
assert_eq!(added["browser_interaction_required_each_run"], false);
assert_eq!(added["attribution"]["registered_by_user"], "user");
assert_eq!(listed["records"].as_array().unwrap().len(), 1);
assert_eq!(listed["credential_scope"]["listed_for_user"], "user");
assert_eq!(revoked["revoked"], true);
assert_eq!(revoked["attribution"]["revoked_by_user"], "user");
}
#[test]
fn key_lifecycle_uses_coordinator_bound_client_session_without_claimed_scope() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let stored_session = StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("key-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
};
let server = std::thread::spawn(move || {
for (expected, response) in [
(
"register_agent_public_key",
br#"{"type":"agent_public_key","actor":"user-session","record":{"public_key_fingerprint":"sha256:agent-v1","scopes":["project:read","project:run"],"revoked":false}}"#.as_slice(),
),
(
"list_agent_public_keys",
br#"{"type":"agent_public_keys","actor":"user-session","records":[]}"#.as_slice(),
),
(
"revoke_agent_public_key",
br#"{"type":"agent_public_key","actor":"user-session","record":{"revoked":true}}"#.as_slice(),
),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let wire: Value = serde_json::from_str(&line).unwrap();
let payload = &wire["payload"];
assert_eq!(payload["type"], "authenticated");
assert_eq!(payload["session_secret"], "key-session-secret");
assert_eq!(payload["request"]["type"], expected);
assert!(payload["request"].get("tenant").is_none());
assert!(payload["request"].get("project").is_none());
assert!(payload["request"].get("user").is_none());
stream.write_all(response).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let ignored_scope = CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
};
let added = key_add_report_with_session(
KeyAddArgs {
scope: ignored_scope.clone(),
agent: "agent-ci".to_owned(),
public_key: "agent-key-v1".to_owned(),
},
Some(&stored_session),
)
.unwrap();
let listed = key_list_report_with_session(
KeyListArgs {
scope: ignored_scope.clone(),
},
Some(&stored_session),
)
.unwrap();
let revoked = key_revoke_report_with_session(
KeyRevokeArgs {
scope: ignored_scope,
agent: "agent-ci".to_owned(),
yes: true,
},
Some(&stored_session),
)
.unwrap();
server.join().unwrap();
assert_eq!(added["tenant"], "tenant-session");
assert_eq!(added["project"], "project-session");
assert_eq!(added["user"], "user-session");
assert_eq!(listed["tenant"], "tenant-session");
assert_eq!(revoked["user"], "user-session");
}
#[test]
fn client_session_secret_is_never_sent_to_a_different_coordinator() {
let stored_session = StoredCliSession {
kind: "human".to_owned(),
coordinator: "https://trusted.example:9443".to_owned(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("must-not-leak".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
};
let authenticated = crate::client::authenticated_or_local_trusted_request(
"https://trusted.example:9443",
Some(&stored_session),
json!({"type": "list_projects"}),
json!({"type": "local_trusted"}),
)
.unwrap();
let untrusted = crate::client::authenticated_or_local_trusted_request(
"https://attacker.example:9443",
Some(&stored_session),
json!({"type": "list_projects"}),
json!({"type": "local_trusted"}),
);
assert_eq!(authenticated["type"], "authenticated");
assert_eq!(authenticated["session_secret"], "must-not-leak");
let error = untrusted.unwrap_err().to_string();
assert!(error.contains("no authenticated CLI session matches coordinator"));
assert!(!error.contains("must-not-leak"));
}
#[test]
fn node_revoke_reports_scoped_credential_revocation() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"revoke_node_credential""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""actor_user":"user""#));
assert!(line.contains(r#""node":"node-a""#));
stream
.write_all(
br#"{"type":"node_credential_revoked","node":"node-a","tenant":"tenant","project":"project","actor":"user","descriptor_removed":true,"queued_assignments_removed":2}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let temp = tempfile::tempdir().unwrap();
let revoked = node_revoke_report(
NodeRevokeArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
node: "node-a".to_owned(),
yes: true,
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(revoked["command"], "node revoke");
assert_eq!(revoked["node"], "node-a");
assert_eq!(revoked["credential_revoked"], true);
assert_eq!(revoked["descriptor_removed"], true);
assert_eq!(revoked["queued_assignments_removed"], 2);
assert_eq!(revoked["node_credentials_separate_from_user_session"], true);
}
#[test]
fn admin_status_and_suspend_use_public_coordinator_api() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for (expected, response) in [
(
"admin_status",
r#"{"type":"admin_status","tenant":"tenant","actor":"admin","suspended":false,"safe_default":"read_only"}"#,
),
(
"suspend_tenant",
r#"{"type":"tenant_suspended","tenant":"tenant","actor":"admin","policy":{"tenant":"tenant","name":"tenant:suspended","digest":"sha256:suspension"}}"#,
),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(&format!(r#""type":"{expected}""#)));
assert!(
line.contains(r#""tenant":"admin-tenant""#)
|| line.contains(r#""tenant":"tenant""#)
);
assert!(line.contains(r#""actor_user":"admin""#));
assert!(!line.contains(r#""admin_token""#));
let wire: Value = serde_json::from_str(&line).unwrap();
let payload = &wire["payload"];
let tenant = payload["tenant"].as_str().unwrap();
let actor_user = payload["actor_user"].as_str().unwrap();
let target_tenant = payload["target_tenant"].as_str().unwrap_or(tenant);
let admin_nonce = payload["admin_nonce"].as_str().unwrap();
let issued_at_epoch_seconds = payload["issued_at_epoch_seconds"].as_u64().unwrap();
assert_eq!(
payload["admin_proof"],
clusterflux_core::admin_request_proof(
"admin-token",
expected,
tenant,
actor_user,
target_tenant,
admin_nonce,
issued_at_epoch_seconds,
)
.to_string()
);
if expected == "suspend_tenant" {
assert!(line.contains(r#""target_tenant":"tenant""#));
}
stream.write_all(response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "admin-tenant".to_owned(),
project: "project".to_owned(),
user: "admin".to_owned(),
json: false,
};
let status = admin_status_report(AdminStatusArgs {
scope: scope.clone(),
admin_token: Some("admin-token".to_owned()),
})
.unwrap();
let suspended = admin_suspend_tenant_report(AdminSuspendTenantArgs {
scope,
target_tenant: Some("tenant".to_owned()),
admin_token: Some("admin-token".to_owned()),
yes: true,
})
.unwrap();
server.join().unwrap();
assert_eq!(status["command"], "admin status");
assert_eq!(status["safe_default"], "read_only");
assert_eq!(status["private_website_required"], false);
assert_eq!(status["suspended"], false);
assert_eq!(suspended["command"], "admin suspend-tenant");
assert_eq!(suspended["tenant"], "tenant");
assert_eq!(suspended["actor_tenant"], "admin-tenant");
assert_eq!(suspended["suspended"], true);
assert_eq!(suspended["private_website_required"], false);
}
#[test]
fn admin_commands_require_explicit_admin_token_for_coordinator_requests() {
let error = admin_status_report(AdminStatusArgs {
scope: CliScopeArgs {
coordinator: Some("127.0.0.1:9".to_owned()),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "admin".to_owned(),
json: false,
},
admin_token: None,
})
.unwrap_err();
assert!(error.to_string().contains("--admin-token"));
}
#[test]
fn debug_attach_reports_public_authorization() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"debug_attach""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""actor_user":"user""#));
assert!(line.contains(r#""process":"vp""#));
stream
.write_all(
br#"{"type":"debug_attach","process":"vp","actor":"user","authorization":{"allowed":true,"reason":"debug attach authorized for project"},"audit_event":{"tenant":"tenant","project":"project","process":"vp","task":null,"actor":"user","operation":"debug_attach","allowed":true,"reason":"debug attach authorized for project","charged_debug_read_bytes":1024,"used_debug_read_bytes":1024},"charged_debug_read_bytes":1024,"used_debug_read_bytes":1024}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let report = debug_attach_report_with_dap(
DebugAttachArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
process: "vp".to_owned(),
},
"/tmp/clusterflux-debug-dap-test".to_owned(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["command"], "debug attach");
assert_eq!(report["process"], "vp");
assert_eq!(report["authorized"], true);
assert_eq!(
report["authorization"]["reason"],
"debug attach authorized for project"
);
assert_eq!(report["audit_event"]["operation"], "debug_attach");
assert_eq!(report["charged_debug_read_bytes"], 1024);
assert_eq!(report["used_debug_read_bytes"], 1024);
assert_eq!(report["debug_reads_quota_limited"], true);
assert_eq!(report["private_website_required"], false);
}
#[test]
fn user_control_commands_use_authenticated_envelope_with_stored_cli_session() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for (expected, response) in [
(
"list_processes",
br#"{"type":"process_statuses","processes":[{"process":"vp","state":"running","connected_nodes":[],"coordinator_epoch":42}],"actor":"user-session"}"#.as_slice(),
),
(
"list_task_events",
br#"{"type":"task_events","events":[]}"#.as_slice(),
),
(
"list_processes",
br#"{"type":"process_statuses","processes":[{"process":"vp","state":"running","connected_nodes":[],"coordinator_epoch":42}],"actor":"user-session"}"#.as_slice(),
),
(
"start_process",
br#"{"type":"process_started","process":"vp","epoch":42}"#.as_slice(),
),
(
"cancel_process",
br#"{"type":"process_cancellation_requested","process":"vp","cancelled_tasks":[],"affected_nodes":[]}"#.as_slice(),
),
(
"abort_process",
br#"{"type":"process_aborted","process":"vp","aborted_tasks":[],"affected_nodes":[]}"#.as_slice(),
),
(
"restart_task",
br#"{"type":"task_restart","process":"vp","task":"task-a","actor":"user-session","accepted":false,"clean_boundary_available":false,"active_task":false,"completed_event_observed":false,"requires_whole_process_restart":true,"message":"restart requires checkpoint","charged_debug_read_bytes":1024,"used_debug_read_bytes":1024,"audit_event":{"tenant":"tenant-session","project":"project-session","process":"vp","task":"task-a","actor":"user-session","operation":"restart_task","allowed":true,"reason":"restart requires checkpoint","charged_debug_read_bytes":1024,"used_debug_read_bytes":1024}}"#.as_slice(),
),
(
"create_artifact_download_link",
br#"{"type":"artifact_download_link","link":{"artifact":"app.txt","source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant-session/project-session/vp/app.txt","scoped_token_digest":"sha256:token","expires_at_epoch_seconds":60,"tenant":"tenant-session","project":"project-session","process":"vp","actor":{"User":"user-session"},"max_bytes":2048,"policy_context_digest":"sha256:policy"}}"#.as_slice(),
),
(
"debug_attach",
br#"{"type":"debug_attach","process":"vp","actor":"user-session","authorization":{"allowed":true,"reason":"debug attach authorized for project"},"audit_event":{"tenant":"tenant-session","project":"project-session","process":"vp","task":null,"actor":"user-session","operation":"debug_attach","allowed":true,"reason":"debug attach authorized for project","charged_debug_read_bytes":1024,"used_debug_read_bytes":1024},"charged_debug_read_bytes":1024,"used_debug_read_bytes":1024}"#.as_slice(),
),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"coordinator_request""#));
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"control-cli-session-secret""#));
assert!(line.contains(&format!(r#""type":"{expected}""#)));
assert!(!line.contains("ignored-tenant"));
assert!(!line.contains("ignored-project"));
assert!(!line.contains("ignored-user"));
stream.write_all(response).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let session = StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("control-cli-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
};
let scope = CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
};
let coordinator_scope = CliScopeArgs {
coordinator: Some(addr),
..scope.clone()
};
process_status_report_with_session(
ProcessStatusArgs {
scope: scope.clone(),
process: "vp".to_owned(),
},
Some(&session),
)
.unwrap();
process_list_report_with_session(
ProcessListArgs {
scope: scope.clone(),
},
Some(&session),
)
.unwrap();
process_restart_report_with_session(
ProcessRestartArgs {
scope: scope.clone(),
process: "vp".to_owned(),
yes: true,
},
Some(&session),
)
.unwrap();
process_cancel_report_with_session(
ProcessCancelArgs {
scope: scope.clone(),
process: "vp".to_owned(),
node: None,
task: None,
yes: true,
},
Some(&session),
)
.unwrap();
process_abort_report_with_session(
ProcessAbortArgs {
scope: scope.clone(),
process: "vp".to_owned(),
yes: true,
},
Some(&session),
)
.unwrap();
task_restart_report_with_session(
TaskRestartArgs {
scope: coordinator_scope.clone(),
task: "task-a".to_owned(),
process: "vp".to_owned(),
yes: true,
},
Some(&session),
)
.unwrap();
artifact_download_report_with_session(
ArtifactDownloadArgs {
scope: coordinator_scope.clone(),
artifact: "app.txt".to_owned(),
to: None,
max_bytes: 2048,
},
Some(&session),
)
.unwrap();
debug_attach_report_with_dap_and_session(
DebugAttachArgs {
scope: coordinator_scope,
process: "vp".to_owned(),
},
"/tmp/clusterflux-debug-dap-test".to_owned(),
Some(&session),
)
.unwrap();
server.join().unwrap();
}
#[test]
fn human_report_is_text_not_json() {
let report = json!({
"command": "doctor",
"status": "ok",
"coordinator": "127.0.0.1:9443",
"next_actions": ["clusterflux login --browser", "clusterflux project init"],
});
let human = human_report(&report);
assert!(!human.trim_start().starts_with('{'));
assert!(human.contains("Clusterflux doctor"));
assert!(human.contains("status: ok"));
assert!(human.contains("coordinator: 127.0.0.1:9443"));
assert!(human.contains("clusterflux login --browser"));
}
#[test]
fn project_init_select_and_status_use_local_project_config() {
let temp = tempfile::tempdir().unwrap();
let init = project_init_report(
ProjectInitArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "ignored".to_owned(),
user: "user".to_owned(),
json: false,
},
new_project: "project-a".to_owned(),
name: "Project A".to_owned(),
yes: true,
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(init["command"], "project init");
assert_eq!(init["source"], "local_project_config");
assert_eq!(init["project_config_written"], true);
assert_eq!(
init["current_directory_link"]["config_format"],
"clusterflux_project_config_v1"
);
assert_eq!(
init["current_directory_link"]["links_current_directory"],
true
);
assert_eq!(
init["current_directory_link"]["writes_current_directory_only"],
true
);
assert_eq!(init["safe_defaults"]["project"], "project-a");
assert_eq!(init["safe_defaults"]["tenant"], "tenant");
assert_eq!(init["safe_defaults"]["browser_interaction_required"], false);
assert_eq!(init["coordinator_create_before_local_write"], false);
let rendered = human_report(&init);
assert!(rendered.contains("current directory linked: true"));
assert!(rendered.contains("current directory config:"));
let config = read_project_config(temp.path()).unwrap().unwrap();
assert_eq!(config.project, "project-a");
assert_eq!(config.tenant, "tenant");
let selected = project_select_report(
ProjectSelectArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "ignored".to_owned(),
user: "user".to_owned(),
json: false,
},
selected_project: "project-b".to_owned(),
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(selected["command"], "project select");
assert_eq!(
read_project_config(temp.path()).unwrap().unwrap().project,
"project-b"
);
let status = project_status_report(
ProjectStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(status["command"], "project status");
assert_eq!(status["project_identity"]["project"], "project-b");
assert_eq!(status["project_identity"]["tenant"], "tenant");
assert_eq!(status["active_process"], "unknown_without_coordinator");
assert_eq!(status["attached_nodes"]["checked"], false);
}
#[test]
fn project_init_uses_public_create_before_writing_local_config() {
let temp_success = tempfile::tempdir().unwrap();
let temp_rejected = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for index in 0..2 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"create_project""#));
assert!(line.contains(r#""tenant":"tenant-live""#));
assert!(line.contains(r#""actor_user":"user-live""#));
match index {
0 => {
assert!(line.contains(r#""project":"project-created""#));
stream
.write_all(
br#"{"type":"project_created","project":{"id":"project-created","tenant":"tenant-live","name":"Created Project"},"actor":"user-live"}"#,
)
.unwrap();
}
1 => {
assert!(line.contains(r#""project":"foreign-project""#));
stream
.write_all(
br#"{"type":"error","message":"project id is outside the signed-in tenant scope"}"#,
)
.unwrap();
}
_ => unreachable!(),
}
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant-live".to_owned(),
project: "ignored".to_owned(),
user: "user-live".to_owned(),
json: false,
};
let created = project_init_report(
ProjectInitArgs {
scope: scope.clone(),
new_project: "project-created".to_owned(),
name: "Created Project".to_owned(),
yes: true,
},
temp_success.path().to_path_buf(),
)
.unwrap();
assert_eq!(created["command"], "project init");
assert_eq!(created["source"], "public_coordinator_api");
assert_eq!(created["coordinator_create_before_local_write"], true);
assert_eq!(
created["project_config_write_after_coordinator_acceptance"],
true
);
assert_eq!(created["coordinator_session_requests"], 1);
assert_eq!(
created["created_or_linked_project"]["id"],
"project-created"
);
assert_eq!(
created["current_directory_link"]["links_current_directory"],
true
);
assert_eq!(
created["safe_defaults"]["browser_interaction_required"],
false
);
assert_eq!(created["private_website_required"], false);
assert_eq!(
read_project_config(temp_success.path())
.unwrap()
.unwrap()
.project,
"project-created"
);
let rejected = project_init_report(
ProjectInitArgs {
scope,
new_project: "foreign-project".to_owned(),
name: "Foreign Project".to_owned(),
yes: true,
},
temp_rejected.path().to_path_buf(),
)
.unwrap_err();
server.join().unwrap();
assert!(rejected.to_string().contains("tenant scope"));
assert!(read_project_config(temp_rejected.path()).unwrap().is_none());
}
#[test]
fn project_list_and_select_use_public_api_without_website() {
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-original".to_owned(),
user: "user-live".to_owned(),
coordinator: None,
},
)
.unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for index in 0..3 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""tenant":"tenant-live""#));
assert!(line.contains(r#""actor_user":"user-live""#));
match index {
0 => {
assert!(line.contains(r#""type":"list_projects""#));
stream
.write_all(
br#"{"type":"projects","projects":[{"id":"project-a","tenant":"tenant-live","name":"Project A"}],"actor":"user-live"}"#,
)
.unwrap();
}
1 => {
assert!(line.contains(r#""type":"select_project""#));
assert!(line.contains(r#""project":"project-a""#));
stream
.write_all(
br#"{"type":"project_selected","project":{"id":"project-a","tenant":"tenant-live","name":"Project A"},"actor":"user-live"}"#,
)
.unwrap();
}
2 => {
assert!(line.contains(r#""type":"select_project""#));
assert!(line.contains(r#""project":"project-b""#));
stream
.write_all(
br#"{"type":"error","message":"project is outside the signed-in tenant scope"}"#,
)
.unwrap();
}
_ => unreachable!(),
}
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant-live".to_owned(),
project: "project".to_owned(),
user: "user-live".to_owned(),
json: false,
};
let list = project_list_report(
ProjectListArgs {
scope: scope.clone(),
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(list["command"], "project list");
assert_eq!(list["source"], "public_coordinator_api");
assert_eq!(list["project_count"], 1);
assert_eq!(list["projects"][0]["id"], "project-a");
assert_eq!(list["private_website_required"], false);
assert_eq!(list["coordinator_session_requests"], 1);
let selected = project_select_report(
ProjectSelectArgs {
scope: scope.clone(),
selected_project: "project-a".to_owned(),
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(selected["command"], "project select");
assert_eq!(selected["source"], "public_coordinator_api");
assert_eq!(selected["selected_project"]["id"], "project-a");
assert_eq!(selected["project_config_written"], true);
assert_eq!(selected["private_website_required"], false);
assert_eq!(
read_project_config(temp.path()).unwrap().unwrap().project,
"project-a"
);
let rejected = project_select_report(
ProjectSelectArgs {
scope,
selected_project: "project-b".to_owned(),
},
temp.path().to_path_buf(),
)
.unwrap_err();
server.join().unwrap();
assert!(rejected.to_string().contains("tenant scope"));
assert_eq!(
read_project_config(temp.path()).unwrap().unwrap().project,
"project-a"
);
}
#[test]
fn project_list_uses_authenticated_envelope_with_stored_cli_session() {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("project-list-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"project-list-session-secret""#));
assert!(line.contains(r#""type":"list_projects""#));
assert!(!line.contains(r#""actor_user":"user-session""#));
stream
.write_all(
br#"{"type":"projects","projects":[{"id":"project-session","tenant":"tenant-session","name":"Session Project"}],"actor":"user-session"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let report = project_list_report(
ProjectListArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["source"], "public_coordinator_api");
assert_eq!(report["tenant"], "tenant-session");
assert_eq!(report["user"], "user-session");
assert_eq!(report["projects"][0]["id"], "project-session");
}
#[test]
fn project_status_queries_public_coordinator_state() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for _ in 0..3 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
if line.contains("\"type\":\"list_node_descriptors\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"node_descriptors","descriptors":[{"id":"node-a","online":true}],"actor":"user"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else if line.contains("\"type\":\"list_task_events\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"task_events","events":[{"process":"vp-live","task":"task-a"}]}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else if line.contains("\"type\":\"list_processes\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"process_statuses","processes":[{"process":"vp-live","state":"running","connected_nodes":[],"coordinator_epoch":42}],"actor":"user"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else {
panic!("unexpected coordinator request: {line}");
}
}
});
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user".to_owned(),
coordinator: Some(addr.clone()),
},
)
.unwrap();
let status = project_status_report(
ProjectStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(status["coordinator"], addr);
assert_eq!(status["project_identity"]["project"], "project-live");
assert_eq!(status["attached_nodes"]["checked"], true);
assert_eq!(status["attached_nodes"]["count"], 1);
assert_eq!(status["attached_nodes"]["online"], 1);
assert_eq!(status["active_process"], "vp-live");
assert_eq!(
status["quota_posture"]["current_usage"]["observed_task_events"],
1
);
}
#[test]
fn project_status_uses_authenticated_client_session_for_nodes_and_events() {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("status-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let server = std::thread::spawn(move || {
for _ in 0..3 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let wire: Value = serde_json::from_str(&line).unwrap();
let payload = &wire["payload"];
assert_eq!(payload["type"], "authenticated");
assert_eq!(payload["session_secret"], "status-session-secret");
assert!(payload["request"].get("tenant").is_none());
assert!(payload["request"].get("project").is_none());
assert!(payload["request"].get("actor_user").is_none());
match payload["request"]["type"].as_str().unwrap() {
"list_node_descriptors" => stream
.write_all(
br#"{"type":"node_descriptors","descriptors":[{"id":"node-session","online":true}],"actor":"user-session"}"#,
)
.unwrap(),
"list_task_events" => stream
.write_all(
br#"{"type":"task_events","events":[{"process":"vp-session","task":"task-session"}]}"#,
)
.unwrap(),
"list_processes" => stream
.write_all(
br#"{"type":"process_statuses","processes":[{"process":"vp-session","state":"running","connected_nodes":[],"coordinator_epoch":42}],"actor":"user-session"}"#,
)
.unwrap(),
operation => panic!("unexpected coordinator operation: {operation}"),
}
stream.write_all(b"\n").unwrap();
}
});
let status = project_status_report(
ProjectStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(status["coordinator"], addr);
assert_eq!(status["project_identity"]["tenant"], "tenant-session");
assert_eq!(status["project_identity"]["project"], "project-session");
assert_eq!(status["project_identity"]["user"], "user-session");
assert_eq!(status["attached_nodes"]["count"], 1);
assert_eq!(status["active_process"], "vp-session");
}
#[test]
fn quota_status_uses_project_config_and_generic_public_limits() {
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-quota".to_owned(),
project: "project-quota".to_owned(),
user: "user".to_owned(),
coordinator: None,
},
)
.unwrap();
let status = quota_status_report(
QuotaStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(status["command"], "quota status");
assert_eq!(status["tenant"], "tenant-quota");
assert_eq!(status["project"], "project-quota");
assert_eq!(status["current_usage"]["attached_nodes"], 0);
assert_eq!(status["limits"]["configured"], false);
assert_eq!(status["quota_configuration_source"], "unavailable_offline");
assert!(status["quota_tier"].is_null());
let rendered = human_report(&status);
assert!(!rendered.contains("quota tier:"));
let forbidden_tier = ["free", "tier"].join(" ");
assert!(!rendered.to_ascii_lowercase().contains(&forbidden_tier));
assert_eq!(
status["next_blocked_action"]["action"],
"node_work_requires_online_attached_node"
);
assert_eq!(
status["next_blocked_action"]["machine_error"]["category"],
"capability"
);
assert_eq!(
status["next_blocked_action"]["machine_error"]["stable_exit_code"],
24
);
}
#[test]
fn quota_status_queries_public_coordinator_usage() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
for _ in 0..3 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
if line.contains("\"type\":\"list_node_descriptors\"") {
assert!(line.contains("\"tenant\":\"tenant-live\""));
stream
.write_all(
br#"{"type":"node_descriptors","descriptors":[{"id":"node-a","online":true},{"id":"node-b","online":false}],"actor":"user"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else if line.contains("\"type\":\"list_task_events\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"task_events","events":[{"process":"vp-live","task":"task-a"},{"process":"vp-live","task":"task-b"}]}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else if line.contains("\"type\":\"quota_status\"") {
stream
.write_all(
br#"{"type":"quota_status","tenant":"tenant-live","project":"project-live","actor":"user","policy_label":"community tier","limits":{"limits":{"Spawn":64,"ArtifactDownloadBytes":268435456}},"window_seconds":{"Spawn":60,"ArtifactDownloadBytes":86400},"usage":{"Spawn":2,"ArtifactDownloadBytes":123},"window_started_epoch_seconds":{"Spawn":120,"ArtifactDownloadBytes":0}}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else {
panic!("unexpected coordinator request: {line}");
}
}
});
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user".to_owned(),
coordinator: Some(addr.clone()),
},
)
.unwrap();
let status = quota_status_report(
QuotaStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(status["coordinator"], addr);
assert_eq!(status["current_usage"]["attached_nodes"], 2);
assert_eq!(status["current_usage"]["online_nodes"], 1);
assert_eq!(status["current_usage"]["observed_task_events"], 2);
assert_eq!(status["current_usage"]["scoped_resource_usage"]["Spawn"], 2);
assert_eq!(status["limits"]["Spawn"], 64);
assert_eq!(status["limits"]["ArtifactDownloadBytes"], 268_435_456);
assert_eq!(status["window_seconds"]["Spawn"], 60);
assert_eq!(status["quota_configuration_source"], "coordinator");
assert_eq!(status["quota_tier"], "community tier");
assert!(status["next_blocked_action"].is_null());
assert_eq!(status["task_events"]["response"]["type"], "task_events");
}
#[test]
fn process_task_log_and_artifact_reports_summarize_task_events() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let response = concat!(
r#"{"type":"task_events","events":["#,
r#"{"tenant":"tenant","project":"project","process":"vp","node":"node-a","task":"task-a","placement":{"node":"node-a","score":120,"reasons":["warm environment cache","source snapshot already local"]},"terminal_state":"completed","status_code":0,"stdout_bytes":12,"stderr_bytes":0,"stdout_tail":"ok","stderr_tail":"","stdout_truncated":false,"stderr_truncated":false,"artifact_path":"/vfs/artifacts/app.txt","artifact_digest":"sha256:artifact","artifact_size_bytes":12},"#,
r#"{"tenant":"tenant","project":"project","process":"vp","node":"node-b","task":"task-b","terminal_state":"failed","status_code":1,"stdout_bytes":0,"stderr_bytes":7,"stdout_tail":"","stderr_tail":"boom","stdout_truncated":false,"stderr_truncated":false,"artifact_path":null,"artifact_digest":null,"artifact_size_bytes":null},"#,
r#"{"tenant":"tenant","project":"project","process":"vp","node":"node-c","task":"task-c","terminal_state":"failed","status_code":1,"stdout_bytes":0,"stderr_bytes":71,"stdout_tail":"","stderr_tail":"source snapshot unavailable and direct connectivity unavailable","stdout_truncated":false,"stderr_truncated":false,"artifact_path":null,"artifact_digest":null,"artifact_size_bytes":null}"#,
r#"]}"#
);
for _ in 0..5 {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
if line.contains("\"type\":\"list_processes\"") {
stream
.write_all(
br#"{"type":"process_statuses","processes":[{"process":"vp","state":"running","connected_nodes":[],"coordinator_epoch":42}],"actor":"user"}"#,
)
.unwrap();
} else {
assert!(line.contains("\"type\":\"list_task_events\""));
assert!(line.contains("\"process\":\"vp\""));
stream.write_all(response.as_bytes()).unwrap();
}
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let process = process_status_report(ProcessStatusArgs {
scope: scope.clone(),
process: "vp".to_owned(),
})
.unwrap();
let tasks = task_list_report(TaskListArgs {
scope: scope.clone(),
process: Some("vp".to_owned()),
})
.unwrap();
let logs = logs_report(LogsArgs {
scope: scope.clone(),
process: Some("vp".to_owned()),
task: Some("task-a".to_owned()),
})
.unwrap();
let artifacts = artifact_list_report(ArtifactListArgs {
scope,
process: Some("vp".to_owned()),
})
.unwrap();
server.join().unwrap();
assert_eq!(process["state"], "running");
assert_eq!(process["current_task_count"], 3);
assert_eq!(
process["current_tasks"][0]["node_placement"]["node"],
"node-a"
);
assert_eq!(
process["current_tasks"][0]["node_placement"]["reasons"][0],
"warm environment cache"
);
assert_eq!(tasks["tasks"][0]["node_placement"]["score"], 120);
let rendered_tasks = human_report(&tasks);
assert!(rendered_tasks.contains("placement task-a: node-a"));
assert!(rendered_tasks.contains("source snapshot already local"));
assert_eq!(tasks["tasks"][1]["failure_reason"], "boom");
assert_eq!(tasks["tasks"][1]["machine_error"]["category"], "program");
assert_eq!(tasks["tasks"][1]["machine_error"]["stable_exit_code"], 27);
assert_eq!(
tasks["tasks"][2]["locality_failure"]["affected_data"],
"source_snapshot"
);
assert_eq!(
tasks["tasks"][2]["locality_failure"]["coordinator_bulk_relay_used"],
false
);
assert_eq!(
tasks["tasks"][2]["machine_error"]["category"],
"connectivity"
);
assert_eq!(tasks["tasks"][2]["machine_error"]["stable_exit_code"], 25);
assert_eq!(tasks["tasks"][2]["machine_error"]["locality_failure"], true);
assert!(tasks["tasks"][2]["machine_error"]["next_actions"]
.as_array()
.unwrap()
.iter()
.any(|action| action == "rerun source preparation on an attached node"));
assert!(rendered_tasks.contains("locality task-c: source_snapshot"));
assert!(rendered_tasks.contains("do not rely on coordinator bulk source relay"));
assert_eq!(logs["log_entries"].as_array().unwrap().len(), 1);
assert_eq!(logs["log_entries"][0]["task"], "task-a");
assert_eq!(logs["log_entries"][0]["stdout_tail"], "ok");
assert_eq!(artifacts["artifacts"].as_array().unwrap().len(), 1);
assert_eq!(artifacts["artifacts"][0]["artifact"], "app.txt");
assert_eq!(artifacts["artifacts"][0]["size_bytes"], 12);
assert_eq!(artifacts["artifacts"][0]["known_locations"][0], "node-a");
assert_eq!(artifacts["default_durable_store_assumed"], false);
}
#[test]
fn log_and_task_reports_redact_secret_like_values() {
let events = json!({
"response": {
"type": "task_events",
"events": [{
"tenant": "tenant",
"project": "project",
"process": "vp",
"node": "node-a",
"task": "task-secret",
"terminal_state": "failed",
"status_code": 1,
"stdout_bytes": 128,
"stderr_bytes": 64,
"stdout_tail": "upload token=abc123 Authorization: Bearer bearer-secret",
"stderr_tail": "failed password=hunter2 access_token=provider-secret",
"stdout_truncated": true,
"stderr_truncated": false
}]
}
});
let entries = log_entries(Some(&events), Some("task-secret"));
let entry = &entries.as_array().unwrap()[0];
assert_eq!(
entry["stdout_tail"],
"upload token=[redacted] Authorization: Bearer [redacted]"
);
assert_eq!(
entry["stderr_tail"],
"failed password=[redacted] access_token=[redacted]"
);
assert_eq!(entry["stdout_bytes"], 128);
assert_eq!(entry["stdout_truncated"], true);
assert_eq!(entry["secret_like_values_redacted"], true);
assert_eq!(entry["redacted_fields"][0], "stdout_tail");
assert_eq!(entry["redacted_fields"][1], "stderr_tail");
let tasks = task_summaries(Some(&events));
let task = &tasks.as_array().unwrap()[0];
assert_eq!(
task["failure_reason"],
"failed password=[redacted] access_token=[redacted]"
);
assert_eq!(
task["machine_error"]["message"],
"failed password=[redacted] access_token=[redacted]"
);
assert!(!serde_json::to_string(&entries)
.unwrap()
.contains("provider-secret"));
assert!(!serde_json::to_string(&tasks).unwrap().contains("hunter2"));
}
#[test]
fn artifact_download_and_export_reports_expose_safe_session_boundaries() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let download_response = concat!(
r#"{"type":"artifact_download_link","link":{"artifact":"app.txt","artifact_digest":"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c","artifact_size_bytes":9,"source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":2048,"policy_context_digest":"sha256:policy"}}"#
);
let export_response = concat!(
r#"{"type":"artifact_export_plan","source_node":"node-a","receiver_node":"node-b","artifact_size_bytes":9,"plan":{"transport":"NativeQuic","scope":{"tenant":"tenant","project":"project","process":"vp","object":{"Artifact":"app.txt"},"authorization_subject":"artifact-export:node-a-to-node-b"},"source":{"node":"node-a","advertised_addr":"quic://node-a","public_key_fingerprint":"sha256:source"},"destination":{"node":"node-b","advertised_addr":"quic://node-b","public_key_fingerprint":"sha256:destination"},"authorization_digest":"sha256:auth","coordinator_assisted_rendezvous":true,"coordinator_bulk_relay_allowed":false}}"#
);
let export_download_response = concat!(
r#"{"type":"artifact_download_link","link":{"artifact":"app.txt","artifact_digest":"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c","artifact_size_bytes":9,"source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:export-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":67108864,"policy_context_digest":"sha256:policy"}}"#
);
let export_stream_response = concat!(
r#"{"type":"artifact_download_stream","link":{"artifact":"app.txt","artifact_digest":"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c","artifact_size_bytes":9,"source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:export-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":67108864,"policy_context_digest":"sha256:policy"},"streamed_bytes":9,"charged_download_bytes":9,"content_bytes_available":true,"content_offset":0,"content_eof":true,"content_base64":"YXBwLWJ5dGVz","content_source":"retaining_node_reverse_stream"}"#
);
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"create_artifact_download_link""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""artifact":"app.txt""#));
assert!(line.contains(r#""max_bytes":2048"#));
stream.write_all(download_response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
for (expected, phase) in [
("export_artifact_to_node", "export-plan"),
("create_artifact_download_link", "export-download"),
("open_artifact_download_stream", "export-stream"),
] {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(&format!(r#""type":"{expected}""#)));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""artifact":"app.txt""#));
if expected == "create_artifact_download_link" {
assert_eq!(phase, "export-download");
assert!(line.contains(&format!(
r#""max_bytes":{}"#,
DEFAULT_ARTIFACT_EXPORT_MAX_BYTES
)));
stream
.write_all(export_download_response.as_bytes())
.unwrap();
} else if expected == "export_artifact_to_node" {
assert!(line.contains(r#""receiver_node":"node-b""#));
stream.write_all(export_response.as_bytes()).unwrap();
} else {
assert!(line.contains(r#""chunk_bytes":9"#));
assert!(line.contains(r#""token_digest":"sha256:export-token""#));
stream.write_all(export_stream_response.as_bytes()).unwrap();
}
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let temp = tempfile::tempdir().unwrap();
let export_path = temp.path().join("exports/app.txt");
let download = artifact_download_report(ArtifactDownloadArgs {
scope: scope.clone(),
artifact: "app.txt".to_owned(),
to: None,
max_bytes: 2048,
})
.unwrap();
let export = artifact_export_report(ArtifactExportArgs {
scope,
artifact: "app.txt".to_owned(),
to: export_path.clone(),
receiver_node: "node-b".to_owned(),
})
.unwrap();
server.join().unwrap();
assert_eq!(
download["download_session"]["status"],
"download_link_issued"
);
assert_eq!(download["download_session"]["tenant"], "tenant");
assert_eq!(download["download_session"]["project"], "project");
assert_eq!(download["download_session"]["process"], "vp");
assert_eq!(
download["download_session"]["source"]["RetainedNode"],
"node-a"
);
assert_eq!(download["download_session"]["max_bytes"], 2048);
assert_eq!(
download["download_session"]["token_material_returned"],
false
);
assert_eq!(
download["download_session"]["scoped_token_digest_present"],
true
);
assert_eq!(download["download_session"]["authorization_required"], true);
assert_eq!(download["download_session"]["short_lived"], true);
assert_eq!(download["download_session"]["guessable_public_url"], false);
assert_eq!(download["download_session"]["cross_tenant_usable"], false);
assert_eq!(
download["download_session"]["unauthorized_project_usable"],
false
);
assert_eq!(
download["download_session"]["default_durable_store_assumed"],
false
);
assert_eq!(
download["grant_disclosures"][0]["grant"],
"artifact_download"
);
assert_eq!(
download["grant_disclosures"][0]["coordinator_policy_limited"],
true
);
assert_eq!(
download["grant_disclosures"][0]["scoped_token_digest_present"],
true
);
assert_eq!(
download["grant_disclosures"][0]["token_material_returned"],
false
);
assert_eq!(
download["grant_disclosures"][0]["guessable_public_url"],
false
);
assert_eq!(
download["grant_disclosures"][0]["cross_tenant_reuse_allowed"],
false
);
assert_eq!(
download["grant_disclosures"][0]["unauthorized_project_reuse_allowed"],
false
);
let rendered_download = human_report(&download);
assert!(rendered_download.contains("grant artifact_download"));
assert!(rendered_download.contains("policy-limited"));
assert_eq!(export["export_plan"]["status"], "transfer_plan_created");
assert_eq!(export["export_plan"]["source_node"], "node-a");
assert_eq!(export["export_plan"]["receiver_node"], "node-b");
assert_eq!(export["export_plan"]["artifact"], "app.txt");
assert_eq!(
export["export_plan"]["local_path"].as_str().unwrap(),
export_path.to_string_lossy().as_ref()
);
assert_eq!(export["export_plan"]["artifact_size_bytes"], 9);
assert_eq!(export["export_plan"]["local_bytes_written_by_cli"], true);
assert_eq!(
export["export_plan"]["local_export_status"],
"local_bytes_written"
);
assert_eq!(export["export_plan"]["bytes_written"], 9);
assert_eq!(
export["local_export"]["stream"]["content_material_returned_in_report"],
false
);
assert_eq!(
export["local_export"]["download_session"]["scoped_token_digest_present"],
true
);
assert_eq!(
export["local_export"]["download_session"]["guessable_public_url"],
false
);
assert_eq!(export["grant_disclosures"][0]["grant"], "artifact_download");
assert_eq!(
export["grant_disclosures"][0]["cross_tenant_reuse_allowed"],
false
);
assert_eq!(
std::fs::read(&export_path).unwrap().as_slice(),
b"app-bytes"
);
assert_eq!(
export["export_plan"]["default_durable_store_assumed"],
false
);
assert_eq!(
export["export_plan"]["coordinator_bulk_relay_allowed"],
false
);
assert_eq!(export["export_plan"]["authorization_digest_present"], true);
}
#[test]
fn artifact_download_to_path_streams_and_verifies_published_digest() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut link_request = String::new();
reader.read_line(&mut link_request).unwrap();
assert!(link_request.contains(r#""type":"create_artifact_download_link""#));
stream
.write_all(
br#"{"type":"artifact_download_link","link":{"artifact":"app.txt","artifact_digest":"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c","artifact_size_bytes":9,"source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:download-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":2048,"policy_context_digest":"sha256:policy"}}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
let mut stream_request = String::new();
reader.read_line(&mut stream_request).unwrap();
assert!(stream_request.contains(r#""type":"open_artifact_download_stream""#));
assert!(stream_request.contains(r#""chunk_bytes":9"#));
stream
.write_all(
br#"{"type":"artifact_download_stream","link":{"artifact":"app.txt","artifact_digest":"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c","artifact_size_bytes":9,"source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:download-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":2048,"policy_context_digest":"sha256:policy"},"streamed_bytes":9,"charged_download_bytes":9,"content_bytes_available":true,"content_offset":0,"content_eof":true,"content_base64":"YXBwLWJ5dGVz","content_source":"retaining_node_reverse_stream"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let temp = tempfile::tempdir().unwrap();
let destination = temp.path().join("app.txt");
let report = artifact_download_report(ArtifactDownloadArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
artifact: "app.txt".to_owned(),
to: Some(destination.clone()),
max_bytes: 2048,
})
.unwrap();
server.join().unwrap();
assert_eq!(report["local_download"]["status"], "local_bytes_written");
assert_eq!(report["local_download"]["bytes_written"], 9);
assert_eq!(
report["local_download"]["verified_digest"],
"sha256:8e4c2e339a4a879ce6b1e89da60d5bc8a32d3c84dec737b56eb0e79d45e4432c"
);
assert_eq!(fs::read(destination).unwrap(), b"app-bytes");
}
#[test]
fn artifact_failure_reports_apply_stable_exit_codes() {
let mut download = json!({
"command": "artifact download",
"download_session": artifact_download_session_summary(&json!({
"type": "error",
"message": "artifact download unauthorized for project",
})),
});
assert_eq!(apply_command_report_exit_code(&mut download), Some(21));
assert_eq!(
download["download_session"]["machine_error"]["category"],
"authorization"
);
assert_eq!(
download["download_session"]["machine_error"]["process_exit_code_applied"],
true
);
let mut export = json!({
"command": "artifact export",
"export_plan": artifact_export_plan_summary(
&json!({
"type": "error",
"message": "direct connectivity unavailable for artifact export",
}),
Path::new("dist/app.txt"),
),
});
assert_eq!(apply_command_report_exit_code(&mut export), Some(25));
assert_eq!(
export["export_plan"]["machine_error"]["category"],
"connectivity"
);
assert_eq!(
export["export_plan"]["machine_error"]["process_exit_code_applied"],
true
);
let mut stream = json!({
"command": "artifact export",
"local_export": {
"stream": artifact_stream_summary(&json!({
"type": "error",
"message": "quota unavailable: resource limit exceeded for artifact_download_bytes",
})),
},
});
assert_eq!(apply_command_report_exit_code(&mut stream), Some(22));
assert_eq!(
stream["local_export"]["stream"]["machine_error"]["category"],
"quota"
);
assert_eq!(
stream["local_export"]["stream"]["machine_error"]["process_exit_code_applied"],
true
);
}
#[test]
fn process_restart_cancel_and_abort_reports_expose_control_boundaries() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let restart_response = r#"{"type":"process_started","process":"vp","epoch":42}"#;
let cancel_response = r#"{"type":"process_cancellation_requested","process":"vp","cancelled_tasks":[{"process":"vp","task":"compile-linux","node":"node-a"},{"process":"vp","task":"link-linux","node":"node-b"}],"affected_nodes":["node-a","node-b"]}"#;
let abort_response =
r#"{"type":"process_aborted","process":"vp","aborted_tasks":[],"affected_nodes":[]}"#;
for expected in ["start_process", "cancel_process", "abort_process"] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(&format!(r#""type":"{expected}""#)));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""process":"vp""#));
if expected == "start_process" {
stream.write_all(restart_response.as_bytes()).unwrap();
} else if expected == "cancel_process" {
assert!(line.contains(r#""actor_user":"user""#));
stream.write_all(cancel_response.as_bytes()).unwrap();
} else {
assert!(line.contains(r#""actor_user":"user""#));
stream.write_all(abort_response.as_bytes()).unwrap();
}
stream.write_all(b"\n").unwrap();
}
});
let scope = CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let restart = process_restart_report(ProcessRestartArgs {
scope: scope.clone(),
process: "vp".to_owned(),
yes: true,
})
.unwrap();
let cancel = process_cancel_report(ProcessCancelArgs {
scope: scope.clone(),
process: "vp".to_owned(),
node: None,
task: None,
yes: true,
})
.unwrap();
let abort = process_abort_report_with_session(
ProcessAbortArgs {
scope,
process: "vp".to_owned(),
yes: true,
},
None,
)
.unwrap();
server.join().unwrap();
assert_eq!(restart["restart_request"]["status"], "process_started");
assert_eq!(
restart["restart_request"]["operation"],
"restart_virtual_process"
);
assert_eq!(restart["restart_request"]["accepted"], true);
assert_eq!(restart["restart_request"]["process"], "vp");
assert_eq!(restart["restart_request"]["coordinator_epoch"], 42);
assert_eq!(restart["restart_request"]["requires_confirmation"], false);
assert_eq!(restart["restart_request"]["website_required"], false);
assert_eq!(
cancel["cancel_request"]["status"],
"process_cancellation_requested"
);
assert_eq!(
cancel["cancel_request"]["operation"],
"cancel_virtual_process"
);
assert_eq!(cancel["cancel_request"]["accepted"], true);
assert_eq!(cancel["cancel_request"]["process"], "vp");
assert_eq!(cancel["cancel_request"]["cancelled_task_count"], 2);
assert_eq!(
cancel["cancel_request"]["cancelled_tasks"][0]["task"],
"compile-linux"
);
assert_eq!(cancel["cancel_request"]["affected_nodes"][1], "node-b");
assert_eq!(cancel["cancel_request"]["requires_confirmation"], false);
assert_eq!(cancel["cancel_request"]["website_required"], false);
assert_eq!(
cancel["cancel_request"]["whole_process_cancel_available"],
true
);
assert_eq!(
cancel["cancel_request"]["node_must_poll_task_control"],
true
);
assert_eq!(cancel["cancel_request"]["new_task_launches_blocked"], true);
assert_eq!(abort["status"], "aborted");
assert_eq!(abort["abort_request"]["accepted"], true);
assert_eq!(abort["abort_request"]["forced"], true);
assert_eq!(abort["abort_request"]["cooperative"], false);
assert_eq!(abort["abort_request"]["process_slot_released"], true);
}
#[test]
fn task_restart_reports_clean_boundary_requirements() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"restart_task""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""actor_user":"user""#));
assert!(line.contains(r#""process":"vp""#));
assert!(line.contains(r#""task":"compile-linux""#));
stream
.write_all(
br#"{"type":"task_restart","process":"vp","task":"compile-linux","actor":"user","accepted":false,"clean_boundary_available":false,"active_task":true,"completed_event_observed":false,"requires_whole_process_restart":true,"message":"selected task is still active; clean task restart requires a captured checkpoint boundary","audit_event":{"tenant":"tenant","project":"project","process":"vp","task":"compile-linux","actor":"user","operation":"restart_task","allowed":true,"reason":"selected task is still active; clean task restart requires a captured checkpoint boundary","charged_debug_read_bytes":1024,"used_debug_read_bytes":1024},"charged_debug_read_bytes":1024,"used_debug_read_bytes":1024}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let report = task_restart_report(TaskRestartArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
task: "compile-linux".to_owned(),
process: "vp".to_owned(),
yes: true,
})
.unwrap();
server.join().unwrap();
assert_eq!(report["command"], "task restart");
assert_eq!(
report["restart_request"]["operation"],
"restart_selected_task"
);
assert_eq!(report["restart_request"]["accepted"], false);
assert_eq!(report["restart_request"]["clean_boundary_available"], false);
assert_eq!(
report["restart_request"]["requires_whole_process_restart"],
true
);
assert_eq!(report["restart_request"]["active_task"], true);
assert_eq!(
report["restart_request"]["audit_event"]["operation"],
"restart_task"
);
assert_eq!(report["restart_request"]["charged_debug_read_bytes"], 1024);
assert_eq!(report["restart_request"]["used_debug_read_bytes"], 1024);
assert_eq!(report["restart_request"]["debug_reads_quota_limited"], true);
assert_eq!(report["restart_request"]["website_required"], false);
assert_eq!(report["coordinator_session_requests"], 1);
}
#[test]
fn build_command_reuses_bundle_inspection_without_full_repo_upload() {
let temp = tempfile::tempdir().unwrap();
let project = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../..")
.join("examples/launch-build-demo");
let output = temp.path().join("bundle");
let report = build_report(
BuildArgs {
project: Some(project),
source_provider: None,
disabled_source_providers: Vec::new(),
output: Some(output.clone()),
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert_eq!(report["command"], "build");
assert_eq!(report["content_addressed"], true);
assert_eq!(report["contains_full_repository_upload"], false);
assert_eq!(report["bundle_artifact"]["task_descriptor_count"], 9);
assert_eq!(report["bundle_artifact"]["entrypoint_count"], 4);
assert!(output.join("module.wasm").is_file());
assert!(output.join("manifest.json").is_file());
assert!(output.join("task-descriptors.json").is_file());
let task_descriptors: Value =
serde_json::from_slice(&fs::read(output.join("task-descriptors.json")).unwrap()).unwrap();
assert!(task_descriptors
.as_array()
.unwrap()
.iter()
.any(|descriptor| {
descriptor["name"] == "task_add_one"
&& descriptor["argument_schema"] == "input : i32"
&& descriptor["result_schema"] == "i32"
&& descriptor["restart_compatibility_hash"]
.as_str()
.unwrap()
.starts_with("sha256:")
}));
assert!(report["bundle"]["metadata"]["identity"]
.as_str()
.unwrap()
.starts_with("sha256:"));
assert!(report["bundle"]["metadata"]["wasm_code"]
.as_str()
.unwrap()
.starts_with("sha256:"));
assert_eq!(
report["bundle"]["metadata"]["task_metadata"]["default_entrypoint"],
"build"
);
assert!(report["bundle"]["metadata"]["task_metadata"]["entrypoints"]
.as_array()
.unwrap()
.iter()
.any(|entrypoint| entrypoint == "build"));
assert!(
!report["bundle"]["metadata"]["task_metadata"]["entrypoints"]
.as_array()
.unwrap()
.iter()
.any(|entrypoint| entrypoint == "release")
);
assert_eq!(
report["bundle"]["metadata"]["source_metadata"]["transfer_policy"]
["coordinator_receives_source_bytes_by_default"],
false
);
assert_eq!(
report["bundle"]["metadata"]["source_metadata"]["transfer_policy"]
["default_full_repo_tarball"],
false
);
assert_eq!(
report["bundle"]["metadata"]["debug_metadata"]["dap_virtual_process"],
true
);
assert_eq!(
report["bundle"]["metadata"]["large_input_policy"]["selected_inputs_are_content_digests"],
true
);
assert_eq!(
report["bundle"]["metadata"]["large_input_policy"]["selected_input_bytes_included"],
false
);
assert_eq!(
report["bundle"]["metadata"]["large_input_policy"]["full_repository_bytes_included"],
false
);
assert_eq!(
report["bundle"]["metadata"]["large_input_policy"]["silent_task_argument_serialization"],
false
);
assert!(
report["bundle"]["metadata"]["large_input_policy"]["supported_handle_types"]
.as_array()
.unwrap()
.iter()
.any(|handle| handle == "Artifact")
);
assert_eq!(
report["bundle"]["metadata"]["restart_compatibility"]
["source_edits_can_restart_from_clean_task_boundary"],
true
);
assert_eq!(
report["bundle"]["metadata"]["restart_compatibility"]["requires_clean_checkpoint_boundary"],
true
);
assert_eq!(
report["bundle"]["metadata"]["restart_compatibility"]["compares_task_abi"],
report["bundle"]["metadata"]["task_metadata"]["task_abi"]
);
assert_eq!(
report["bundle"]["metadata"]["restart_compatibility"]
["incompatible_changes_require_whole_process_restart"],
true
);
assert_eq!(report["status"], "built");
assert_eq!(report["scheduled_work"], false);
}
#[test]
fn build_blocks_before_schedule_on_missing_environment_reference() {
let temp = tempfile::tempdir().unwrap();
fs::create_dir_all(temp.path().join("src")).unwrap();
fs::write(temp.path().join("Cargo.toml"), "[package]\nname='demo'\n").unwrap();
fs::write(
temp.path().join("src/main.rs"),
"fn main() { let _target = env!(\"linux\"); }\n",
)
.unwrap();
let report = build_report(
BuildArgs {
project: Some(temp.path().to_path_buf()),
source_provider: None,
disabled_source_providers: Vec::new(),
output: None,
json: false,
},
PathBuf::from("/unused"),
)
.unwrap();
assert_eq!(report["status"], "blocked_before_schedule");
assert_eq!(report["scheduled_work"], false);
assert_eq!(report["machine_error"]["category"], "environment");
assert_eq!(report["diagnostics"][0]["code"], "missing_environment");
}
#[test]
fn node_enroll_reports_short_lived_public_api_grant() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"create_node_enrollment_grant""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""actor_user":"user""#));
assert!(!line.contains(r#""grant":""#));
assert!(!line.contains("now_epoch_seconds"));
assert!(line.contains(r#""ttl_seconds":300"#));
stream
.write_all(
br#"{"type":"node_enrollment_grant_created","tenant":"tenant","project":"project","grant":"grant-live","scope":"node:attach","expires_at_epoch_seconds":300}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
});
let temp = tempfile::tempdir().unwrap();
let report = node_enroll_report(
NodeEnrollArgs {
scope: CliScopeArgs {
coordinator: Some(addr),
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
ttl_seconds: 300,
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(report["command"], "node enroll");
assert_eq!(report["status"], "created");
assert_eq!(report["private_website_required"], false);
assert_eq!(report["tenant"], "tenant");
assert_eq!(report["project"], "project");
assert_eq!(report["user"], "user");
assert_eq!(report["enrollment_grant"]["grant"], "grant-live");
assert_eq!(report["enrollment_grant"]["scope"], "node:attach");
assert_eq!(report["enrollment_grant"]["ttl_seconds"], 300);
assert_eq!(report["enrollment_grant"]["expires_at_epoch_seconds"], 300);
assert_eq!(report["enrollment_grant"]["short_lived"], true);
assert_eq!(
report["enrollment_grant"]["exchange_for_persistent_node_identity"],
true
);
assert_eq!(
report["enrollment_grant"]["node_credentials_separate_from_user_session"],
true
);
assert_eq!(report["coordinator_session_requests"], 1);
}
#[test]
fn node_commands_use_authenticated_envelope_with_stored_cli_session() {
let temp = tempfile::tempdir().unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap().to_string();
write_cli_session(
temp.path(),
&StoredCliSession {
kind: "human".to_owned(),
coordinator: addr.clone(),
tenant: "tenant-session".to_owned(),
project: "project-session".to_owned(),
user: "user-session".to_owned(),
cli_session_credential_kind: "CliDeviceSession".to_owned(),
session_secret: Some("node-cli-session-secret".to_owned()),
token_expiry_posture: "unknown_coordinator_session".to_owned(),
expires_at: None,
provider_tokens_exposed_to_cli: false,
provider_tokens_sent_to_nodes: false,
created_at_unix_seconds: 1,
},
)
.unwrap();
let server = std::thread::spawn(move || {
for (request_type, response) in [
(
"create_node_enrollment_grant",
br#"{"type":"node_enrollment_grant_created","tenant":"tenant-session","project":"project-session","grant":"grant-session","scope":"node:attach","expires_at_epoch_seconds":300}"#.as_slice(),
),
(
"list_node_descriptors",
br#"{"type":"node_descriptors","descriptors":[{"id":"node-session","online":true}],"actor":"user-session"}"#.as_slice(),
),
(
"revoke_node_credential",
br#"{"type":"node_credential_revoked","node":"node-session","tenant":"tenant-session","project":"project-session","actor":"user-session","descriptor_removed":true,"queued_assignments_removed":0}"#.as_slice(),
),
] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"authenticated""#));
assert!(line.contains(r#""session_secret":"node-cli-session-secret""#));
assert!(line.contains(&format!(r#""type":"{request_type}""#)));
assert!(!line.contains(r#""actor_user":"ignored-user""#));
assert!(!line.contains(r#""tenant":"ignored-tenant""#));
assert!(!line.contains(r#""project":"ignored-project""#));
stream.write_all(response).unwrap();
stream.write_all(b"\n").unwrap();
}
});
let enrolled = node_enroll_report(
NodeEnrollArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
},
ttl_seconds: 300,
},
temp.path().to_path_buf(),
)
.unwrap();
let listed = node_list_report(
NodeListArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
let revoked = node_revoke_report(
NodeRevokeArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "ignored-tenant".to_owned(),
project: "ignored-project".to_owned(),
user: "ignored-user".to_owned(),
json: false,
},
node: "node-session".to_owned(),
yes: true,
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(enrolled["tenant"], "tenant-session");
assert_eq!(enrolled["project"], "project-session");
assert_eq!(enrolled["user"], "user-session");
assert_eq!(listed["response"]["descriptors"][0]["id"], "node-session");
assert_eq!(revoked["credential_revoked"], true);
}
#[test]
fn node_enroll_and_process_commands_have_safe_plan_without_coordinator() {
let scope = CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
};
let temp = tempfile::tempdir().unwrap();
let enroll = node_enroll_report(
NodeEnrollArgs {
scope: scope.clone(),
ttl_seconds: 60,
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(enroll["status"], "requires_coordinator");
assert_eq!(enroll["private_website_required"], false);
assert_eq!(enroll["enrollment_grant"], serde_json::Value::Null);
assert_eq!(enroll["requested_ttl_seconds"], 60);
let cancel = process_cancel_report(ProcessCancelArgs {
scope,
process: "vp".to_owned(),
node: None,
task: None,
yes: false,
})
.unwrap();
assert_eq!(cancel["status"], "confirmation_required");
assert_eq!(cancel["requires_confirmation"], true);
assert_eq!(cancel["coordinator_request_sent"], false);
}