use std::fs;
use std::io::{self, BufRead, BufReader, Cursor, Read, Write};
use std::path::PathBuf;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use acativa::runtime::protocol::{OPERATIONS, PROTOCOL, Request, serve};
use acativa::store::{StoreDialect, StrategyStore, schema_sql};
use acativa::util::file_security::{ensure_private_dir, harden_private_path};
use acativa::{
ActorSlot, BindingValue, CommandKind, CommandStatus, FailureClass, GENERIC_PROJECTION_SCHEMA,
GENERIC_WORKFLOW_SCHEMA, GraphState, GraphStateKind, ReducerEvent, RetryPolicy, RunCommand,
RunSnapshot, StrategyDefinition, StrategyRunStatus, Transition, TransitionEvent,
TransitionMode, WorkflowDefinition, WorkflowLimits, WorkflowMetadata,
};
use rusqlite::Connection;
use serde::de::DeserializeOwned;
use serde_json::{Value, json};
use uuid::Uuid;
const REVISION: &str = "1111111111111111111111111111111111111111111111111111111111111111";
const SEMANTICS: &str = "2222222222222222222222222222222222222222222222222222222222222222";
const LEASE: i64 = i64::MAX - 1;
const SENTINEL: &str = "synthetic-private-sentinel";
struct Fixture {
root: PathBuf,
database: PathBuf,
}
impl Fixture {
fn new() -> Self {
let root = std::env::temp_dir().join(format!("acativa-protocol-{}", Uuid::new_v4()));
ensure_private_dir(&root).expect("private synthetic directory");
let database = root.join("synthetic.sqlite3");
let connection = Connection::open(&database).expect("synthetic database");
harden_private_path(&database).expect("private synthetic file");
connection
.execute_batch(&schema_sql().expect("canonical schema"))
.expect("explicit host initialization");
drop(connection);
Self { root, database }
}
fn store(&self) -> StrategyStore {
StrategyStore::open_existing_database(&self.database, &StoreDialect::generic())
.expect("explicit existing synthetic database")
}
fn cli(&self) -> Cli {
let mut child = Command::new(env!("CARGO_BIN_EXE_acativa"))
.arg("serve")
.arg("--database")
.arg(&self.database)
.args([
"--workflow-schema",
GENERIC_WORKFLOW_SCHEMA,
"--projection-schema",
GENERIC_PROJECTION_SCHEMA,
"--ephemeral-prefix",
"acativa-temporary",
])
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("synthetic protocol subprocess");
let input = child.stdin.take();
let output = BufReader::new(child.stdout.take().expect("private stdout pipe"));
Cli {
child: Some(child),
input,
output,
next_id: 1,
}
}
fn expire(&self, command: &RunCommand) {
let connection = Connection::open(&self.database).expect("synthetic database");
assert_eq!(
connection
.execute(
"UPDATE strategy_commands SET lease_until=0 WHERE command_id=?1",
[&command.id],
)
.expect("synthetic lease clock"),
1
);
}
}
impl Drop for Fixture {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.root);
}
}
struct Cli {
child: Option<Child>,
input: Option<ChildStdin>,
output: BufReader<ChildStdout>,
next_id: u64,
}
impl Cli {
fn frame(&mut self, frame: &[u8]) -> Value {
let input = self.input.as_mut().expect("open input pipe");
input.write_all(frame).expect("write synthetic request");
input.write_all(b"\n").expect("frame delimiter");
input.flush().expect("flush synthetic request");
let mut response = String::new();
assert_ne!(
self.output
.read_line(&mut response)
.expect("read response frame"),
0
);
assert!(!response.contains(SENTINEL));
let response: Value = serde_json::from_str(&response).expect("NDJSON response");
assert_eq!(response["protocol"], PROTOCOL);
assert_ne!(
response.get("result").is_some(),
response.get("error").is_some()
);
response
}
fn request(&mut self, operation: &str, payload: Value) -> Value {
let id = self.next_id;
self.next_id += 1;
let request = Request {
protocol: PROTOCOL.into(),
id,
operation: operation.into(),
payload,
};
let response = self.frame(&serde_json::to_vec(&request).expect("request encodes"));
assert_eq!(response["id"], id);
response
}
fn ok(&mut self, operation: &str, payload: Value) -> Value {
let response = self.request(operation, payload);
assert!(response.get("error").is_none(), "operation must succeed");
response.get("result").expect("result field").clone()
}
fn refuse(&mut self, operation: &str, payload: Value, reason: Option<&str>) {
let response = self.request(operation, payload);
assert_error(&response, "runtime_refused", reason);
}
fn admit(&mut self, key: &str) -> RunSnapshot {
let result = self.ok("admit", admission(key));
assert_eq!(result["created"], true);
decode(result["snapshot"].clone())
}
fn claim(&mut self, run_id: &str, claimant: &str) -> Option<RunCommand> {
decode(self.ok(
"claim",
json!({
"runId": run_id,
"claimant": claimant,
"leaseUntilUnixMs": LEASE,
}),
))
}
fn start_and_authorize(&mut self, run_id: &str, command: &RunCommand, claimant: &str) {
let started: RunSnapshot = decode(self.ok(
"event",
event_payload(run_id, claimant, started_event(command)),
));
assert_eq!(started.commands[&command.id].status, CommandStatus::Running);
let definition: StrategyDefinition =
decode(self.ok("definition", json!({ "revisionDigest": REVISION })));
let grant = definition
.authorization
.expect("actual granted authorization");
assert!(grant.active);
assert_eq!(grant.revision, 1);
assert_eq!(
self.ok(
"authorize",
authorization(run_id, command, claimant, &grant.authorization_digest),
),
Value::Null
);
}
fn finish(mut self) {
drop(self.input.take());
let mut remainder = String::new();
self.output
.read_to_string(&mut remainder)
.expect("EOF response drain");
assert!(remainder.is_empty());
let output = self
.child
.take()
.expect("child process")
.wait_with_output()
.expect("EOF terminates child");
assert!(output.status.success());
assert!(output.stderr.is_empty());
}
}
impl Drop for Cli {
fn drop(&mut self) {
drop(self.input.take());
if let Some(child) = self.child.take() {
let _ = child.wait_with_output();
}
}
}
fn decode<T: DeserializeOwned>(value: Value) -> T {
serde_json::from_value(value).expect("actual SDK response type")
}
fn assert_error(response: &Value, code: &str, reason: Option<&str>) {
assert!(response.get("result").is_none());
let expected = match reason {
Some(reason) => json!({ "code": code, "reason": reason }),
None => json!({ "code": code }),
};
assert_eq!(response["error"], expected);
}
fn workflow() -> WorkflowDefinition {
let state = |id: &str, kind, binding| GraphState {
id: id.into(),
kind,
label: id.into(),
instruction: String::new(),
binding,
runtime: None,
entry: None,
workset: None,
retry: RetryPolicy {
max_attempts: 2,
transient_only: true,
},
};
let edge = |to: &str, event| Transition {
id: to.into(),
from: "work".into(),
to: to.into(),
event,
mode: TransitionMode::Flow,
guard: None,
};
WorkflowDefinition {
schema: GENERIC_WORKFLOW_SCHEMA.into(),
metadata: WorkflowMetadata {
id: "acativa-temporary-protocol".into(),
name: "Synthetic protocol fixture".into(),
version: "1".into(),
description: String::new(),
},
limits: WorkflowLimits {
max_attempts: 2,
..WorkflowLimits::default()
},
actor_slots: vec![ActorSlot::required_actor("worker", "Synthetic worker")],
runtimes: vec![],
worksets: vec![],
initial: "work".into(),
states: vec![
state("work", GraphStateKind::Actor, Some("worker".into())),
state("done", GraphStateKind::Succeed, None),
state("failed", GraphStateKind::Fail, None),
],
transitions: vec![
edge("done", TransitionEvent::Success),
edge("failed", TransitionEvent::Failure),
],
}
}
fn admission(key: &str) -> Value {
json!({
"revisionDigest": REVISION,
"semanticsDigest": SEMANTICS,
"workflow": workflow(),
"bindings": [BindingValue {
slot_id: "worker".into(),
ordinal: 0,
value_id: "agent:synthetic".into(),
model: String::new(),
reasoning_effort: String::new(),
revision: 1,
}],
"input": { "synthetic": true },
"idempotencyKey": key,
"contextId": "synthetic-context",
"runBinding": { "host": "synthetic" },
})
}
fn started_event(command: &RunCommand) -> ReducerEvent {
ReducerEvent::CommandStarted {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
}
}
fn success_event(command: &RunCommand) -> ReducerEvent {
ReducerEvent::CommandSucceeded {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
output: json!({ "syntheticOutput": true }),
}
}
fn event_payload(run_id: &str, claimant: &str, event: ReducerEvent) -> Value {
json!({ "runId": run_id, "claimant": claimant, "event": event })
}
fn completion(run_id: &str, command: &RunCommand, receipt: Value) -> Value {
let mut payload = event_payload(run_id, "first-owner", success_event(command));
payload["receipt"] = receipt;
payload
}
fn receipt_query(run_id: &str, command: &RunCommand) -> Value {
json!({
"runId": run_id,
"commandId": command.id,
"attemptToken": command.attempt_token,
})
}
fn authorization(run_id: &str, command: &RunCommand, claimant: &str, digest: &str) -> Value {
json!({
"runId": run_id,
"commandId": command.id,
"attemptToken": command.attempt_token,
"expectedAuthorizationDigest": digest,
"claimant": claimant,
"leaseUntilUnixMs": LEASE,
})
}
#[test]
fn child_recovers_frame_boundaries_and_sanitizes_failures() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
let description = cli.ok("describe", json!({}));
assert_eq!(
description,
json!({
"protocol": PROTOCOL,
"version": env!("CARGO_PKG_VERSION"),
"workflowSchema": GENERIC_WORKFLOW_SCHEMA,
"projectionSchema": GENERIC_PROJECTION_SCHEMA,
"operations": OPERATIONS,
})
);
assert!(
!description
.to_string()
.contains(fixture.database.to_string_lossy().as_ref())
);
for frame in [
format!("{{{SENTINEL}").into_bytes(),
vec![0xff],
serde_json::to_vec(&json!([PROTOCOL, 9, "describe", {}])).unwrap(),
serde_json::to_vec(&json!({
"protocol": PROTOCOL, "id": 10, "operation": "describe",
"payload": {}, "unexpected": SENTINEL,
}))
.unwrap(),
serde_json::to_vec(&json!({
"protocol": PROTOCOL, "id": 11, "operation": "describe",
}))
.unwrap(),
] {
let response = cli.frame(&frame);
assert_eq!(response["id"], Value::Null);
assert_error(&response, "invalid_request", None);
}
let wrong_protocol = cli.frame(
&serde_json::to_vec(&json!({
"protocol": SENTINEL, "id": 12, "operation": "describe", "payload": {},
}))
.unwrap(),
);
assert_eq!(wrong_protocol["id"], 12);
assert_error(&wrong_protocol, "invalid_request", None);
assert_error(
&cli.request(SENTINEL, json!({ "private": SENTINEL })),
"unsupported_operation",
None,
);
for (operation, payload) in [
("snapshot", json!({ "runId": 7 })),
("describe", json!({ "unexpected": SENTINEL })),
("describe", Value::Null),
("describe", json!([])),
("snapshot", json!([SENTINEL])),
("describe", json!({ "private": fixture.database })),
(
"claim",
json!({ "runId": SENTINEL, "claimant": "synthetic", "leaseUntilUnixMs": "later" }),
),
] {
assert_error(&cli.request(operation, payload), "invalid_payload", None);
}
for missing in ["input", "runBinding"] {
let mut payload = admission("missing-value");
payload.as_object_mut().unwrap().remove(missing);
assert_error(&cli.request("admit", payload), "invalid_payload", None);
}
assert_error(
&cli.request(
"complete",
json!({
"runId": SENTINEL,
"claimant": "synthetic",
"event": ReducerEvent::CancelRequested,
}),
),
"invalid_payload",
None,
);
cli.refuse("lookup", json!({ "idempotencyKey": "" }), None);
cli.refuse(
"snapshot",
json!({ "runId": SENTINEL }),
Some("strategy_run_not_found"),
);
cli.refuse(
"snapshot",
json!({ "runId": fixture.database }),
Some("strategy_run_not_found"),
);
cli.refuse(
"definition",
json!({ "revisionDigest": SENTINEL }),
Some("strategy_definition_not_found"),
);
assert_eq!(cli.ok("describe", json!({})), description);
cli.finish();
}
#[test]
fn wire_admission_authorization_completion_and_receipt_survive_restart() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
let snapshot = cli.admit("wire-success");
let run_id = &snapshot.run_id;
let duplicate = cli.ok("admit", admission("wire-success"));
assert_eq!(duplicate["created"], false);
assert_eq!(
decode::<RunSnapshot>(duplicate["snapshot"].clone()),
snapshot
);
let mut conflict = admission("wire-success");
conflict["input"] = json!({ "different": true });
cli.refuse("admit", conflict, Some("strategy_idempotency_conflict"));
assert_eq!(
cli.ok("lookup", json!({ "idempotencyKey": "wire-success" })),
json!(run_id)
);
assert_eq!(
cli.ok("lookup", json!({ "idempotencyKey": "not-admitted" })),
Value::Null
);
let definition: StrategyDefinition =
decode(cli.ok("definition", json!({ "revisionDigest": REVISION })));
let grant = definition.authorization.expect("persisted grant");
assert!(grant.active);
assert_eq!(grant.revision, 1);
let command = cli.claim(run_id, "first-owner").expect("one actor command");
assert_eq!(command.kind, CommandKind::Actor);
assert_eq!(command.attempt, 1);
assert!(cli.claim(run_id, "second-owner").is_none());
assert_eq!(
cli.ok("receipt", receipt_query(run_id, &command)),
Value::Null
);
cli.refuse(
"event",
event_payload(run_id, "second-owner", started_event(&command)),
Some("strategy_lease_lost"),
);
let mut stale = command.clone();
stale.attempt_token = "synthetic-stale-token".into();
cli.refuse(
"event",
event_payload(run_id, "first-owner", started_event(&stale)),
Some("strategy_callback_stale"),
);
cli.refuse(
"authorize",
authorization(run_id, &command, "first-owner", &grant.authorization_digest),
Some("strategy_lease_lost"),
);
cli.start_and_authorize(run_id, &command, "first-owner");
for (attempt, claimant, digest, reason) in [
(
&command,
"second-owner",
grant.authorization_digest.as_str(),
"strategy_lease_lost",
),
(
&stale,
"first-owner",
grant.authorization_digest.as_str(),
"strategy_lease_lost",
),
(
&command,
"first-owner",
SEMANTICS,
"strategy_authorization_stale",
),
] {
cli.refuse(
"authorize",
authorization(run_id, attempt, claimant, digest),
Some(reason),
);
}
assert_eq!(
cli.ok(
"renew",
json!({ "commandId": command.id, "claimant": "first-owner", "leaseUntilUnixMs": LEASE }),
),
Value::Null
);
let receipt = json!({ "synthetic": true, "opaque": ["kept", 7], "partial": false });
let complete = completion(run_id, &command, receipt.clone());
let settled: RunSnapshot = decode(cli.ok("complete", complete.clone()));
assert_eq!(settled.status, StrategyRunStatus::Completed);
assert_eq!(
settled.commands[&command.id].status,
CommandStatus::Succeeded
);
assert_eq!(decode::<RunSnapshot>(cli.ok("complete", complete)), settled);
cli.refuse(
"complete",
completion(run_id, &command, json!({ "altered": true })),
Some("strategy_receipt_conflict"),
);
cli.refuse(
"complete",
completion(run_id, &stale, receipt.clone()),
Some("strategy_callback_stale"),
);
assert_eq!(cli.ok("receipt", receipt_query(run_id, &command)), receipt);
assert!(cli.claim(run_id, "second-owner").is_none());
cli.finish();
let mut restarted = fixture.cli();
assert_eq!(
restarted.ok("lookup", json!({ "idempotencyKey": "wire-success" })),
json!(run_id)
);
assert_eq!(
decode::<RunSnapshot>(restarted.ok("snapshot", json!({ "runId": run_id }))),
settled
);
assert_eq!(
restarted.ok("receipt", receipt_query(run_id, &command)),
receipt
);
assert!(restarted.claim(run_id, "second-owner").is_none());
restarted.finish();
}
struct LostOutput;
impl Write for LostOutput {
fn write(&mut self, _: &[u8]) -> io::Result<usize> {
Err(io::ErrorKind::BrokenPipe.into())
}
fn flush(&mut self) -> io::Result<()> {
Err(io::ErrorKind::BrokenPipe.into())
}
}
#[test]
fn response_loss_does_not_erase_completion_or_process_the_next_request() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
let snapshot = cli.admit("response-loss");
let run_id = &snapshot.run_id;
let command = cli.claim(run_id, "first-owner").expect("synthetic command");
cli.start_and_authorize(run_id, &command, "first-owner");
cli.finish();
let receipt = json!({ "syntheticCompleted": true });
let mut frames = Vec::new();
for (id, operation, payload) in [
(1, "complete", completion(run_id, &command, receipt.clone())),
(2, "admit", admission("must-not-be-processed")),
] {
serde_json::to_writer(
&mut frames,
&Request {
protocol: PROTOCOL.into(),
id,
operation: operation.into(),
payload,
},
)
.unwrap();
frames.push(b'\n');
}
assert!(serve(Cursor::new(frames), LostOutput, fixture.store()).is_err());
let mut restarted = fixture.cli();
let settled: RunSnapshot = decode(restarted.ok("snapshot", json!({ "runId": run_id })));
assert_eq!(settled.status, StrategyRunStatus::Completed);
assert_eq!(
settled.commands[&command.id].status,
CommandStatus::Succeeded
);
assert_eq!(
restarted.ok("receipt", receipt_query(run_id, &command)),
receipt
);
assert!(restarted.claim(run_id, "second-owner").is_none());
assert_eq!(
restarted.ok(
"lookup",
json!({ "idempotencyKey": "must-not-be-processed" })
),
Value::Null
);
restarted.finish();
}
#[test]
fn recovery_preserves_held_claims_and_fences_only_expired_attempts() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
let never_started = cli.admit("never-started");
let run_id = &never_started.run_id;
let old = cli.claim(run_id, "first-owner").expect("initial claim");
cli.finish();
let mut restarted = fixture.cli();
let held = restarted.ok("recover", json!({ "runId": run_id }));
assert_eq!(held["nextClaimExpiryUnixMs"], LEASE);
let held_snapshot: RunSnapshot = decode(held["snapshot"].clone());
assert_eq!(held_snapshot.commands[&old.id], old);
assert!(restarted.claim(run_id, "second-owner").is_none());
fixture.expire(&old);
let recovered = restarted.ok("recover", json!({ "runId": run_id }));
assert_eq!(recovered["nextClaimExpiryUnixMs"], Value::Null);
let recovered_snapshot: RunSnapshot = decode(recovered["snapshot"].clone());
assert_eq!(
recovered_snapshot.commands[&old.id].status,
CommandStatus::Cancelled
);
let pending = recovered_snapshot
.commands
.values()
.find(|command| command.status == CommandStatus::Pending)
.expect("the reclaim emitted the next attempt");
assert_ne!(pending.id, old.id);
assert_eq!(pending.attempt, old.attempt + 1);
assert_ne!(pending.attempt_token, old.attempt_token);
restarted.refuse(
"event",
event_payload(run_id, "first-owner", started_event(&old)),
Some("strategy_callback_conflict"),
);
let reclaimed = restarted
.claim(run_id, "second-owner")
.expect("reclaimed command");
assert_eq!(reclaimed.id, pending.id);
assert_eq!(reclaimed.attempt, pending.attempt);
assert_eq!(reclaimed.attempt_token, pending.attempt_token);
let started = restarted.admit("started-expiry");
let running_id = &started.run_id;
let running = restarted
.claim(running_id, "first-owner")
.expect("another command");
restarted.start_and_authorize(running_id, &running, "first-owner");
restarted.finish();
let mut restarted = fixture.cli();
let held = restarted.ok("recover", json!({ "runId": running_id }));
assert_eq!(held["nextClaimExpiryUnixMs"], LEASE);
assert_eq!(
decode::<RunSnapshot>(held["snapshot"].clone()).commands[&running.id].status,
CommandStatus::Running
);
assert!(restarted.claim(running_id, "second-owner").is_none());
fixture.expire(&running);
let recovered = restarted.ok("recover", json!({ "runId": running_id }));
assert_eq!(recovered["nextClaimExpiryUnixMs"], Value::Null);
let in_doubt: RunSnapshot = decode(recovered["snapshot"].clone());
assert_eq!(
in_doubt.commands[&running.id].status,
CommandStatus::InDoubt
);
assert_eq!(in_doubt.commands[&running.id].attempt, running.attempt);
assert_eq!(in_doubt.commands.len(), 1);
assert!(restarted.claim(running_id, "second-owner").is_none());
assert_eq!(
restarted.ok("receipt", receipt_query(running_id, &running)),
Value::Null
);
assert_eq!(
restarted.ok("recover", json!({ "runId": running_id })),
recovered
);
restarted.finish();
}
#[test]
fn failed_effect_keeps_its_partial_receipt_without_reporting_success() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
let snapshot = cli.admit("permanent-failure");
let run_id = &snapshot.run_id;
let command = cli.claim(run_id, "first-owner").expect("synthetic command");
cli.start_and_authorize(run_id, &command, "first-owner");
let receipt = json!({ "partial": true, "syntheticProgress": [1, 2] });
let mut payload = event_payload(
run_id,
"first-owner",
ReducerEvent::CommandFailed {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
class: FailureClass::Permanent,
code: "synthetic_failure".into(),
},
);
payload["receipt"] = receipt.clone();
let failed: RunSnapshot = decode(cli.ok("complete", payload.clone()));
assert_eq!(failed.status, StrategyRunStatus::Failed);
assert_eq!(failed.commands[&command.id].status, CommandStatus::Failed);
assert_eq!(
failed.commands[&command.id].failure_class,
Some(FailureClass::Permanent)
);
assert_eq!(
failed.commands[&command.id].failure_code.as_deref(),
Some("synthetic_failure")
);
assert!(!failed.completed_states.contains("done"));
assert_eq!(decode::<RunSnapshot>(cli.ok("complete", payload)), failed);
assert_eq!(cli.ok("receipt", receipt_query(run_id, &command)), receipt);
assert!(cli.claim(run_id, "second-owner").is_none());
cli.finish();
}
#[test]
fn prune_bounds_terminal_retention_and_reports_deletions() {
let fixture = Fixture::new();
let mut cli = fixture.cli();
for key in ["prune-first", "prune-second"] {
let snapshot = cli.admit(key);
let run_id = &snapshot.run_id;
let command = cli.claim(run_id, "first-owner").expect("synthetic command");
cli.start_and_authorize(run_id, &command, "first-owner");
cli.ok(
"complete",
completion(run_id, &command, json!({"key": key})),
);
}
cli.refuse(
"prune",
json!({ "keepTerminalRuns": 8193, "keepEventsPerRun": 1, "vacuum": false }),
Some("strategy_prune_bounds_invalid"),
);
let report = cli.ok(
"prune",
json!({ "keepTerminalRuns": 1, "keepEventsPerRun": 64, "vacuum": true }),
);
assert_eq!(report["runsPruned"], 1);
assert_eq!(report["receiptsPruned"], 1);
assert_eq!(report["vacuumed"], true);
cli.finish();
}
#[test]
fn cli_exports_only_schema_and_rejects_implicit_or_ambiguous_configuration() {
let schema = Command::new(env!("CARGO_BIN_EXE_acativa"))
.arg("schema")
.output()
.expect("schema subprocess exits");
assert!(schema.status.success());
assert!(schema.stderr.is_empty());
assert_eq!(schema.stdout, schema_sql().unwrap().as_bytes());
let help = Command::new(env!("CARGO_BIN_EXE_acativa"))
.arg("--help")
.output()
.expect("help subprocess exits");
assert!(help.status.success());
assert!(help.stderr.is_empty());
assert!(
String::from_utf8(help.stdout)
.unwrap()
.contains("--database PATH")
);
for args in [
vec![],
vec!["serve"],
vec!["schema", SENTINEL],
vec!["serve", "--database"],
vec!["serve", "--unknown", SENTINEL],
vec!["serve", "--database", SENTINEL, "--database", SENTINEL],
vec!["serve", "--database", ""],
vec![
"serve",
"--database",
SENTINEL,
"--workflow-schema",
"",
"--projection-schema",
GENERIC_PROJECTION_SCHEMA,
"--ephemeral-prefix",
"acativa-temporary",
],
] {
let output = Command::new(env!("CARGO_BIN_EXE_acativa"))
.args(args)
.output()
.expect("invalid usage subprocess exits");
assert!(!output.status.success());
assert!(output.stdout.is_empty());
assert_eq!(output.stderr, b"acativa: invalid usage; use --help\n");
}
let fixture = Fixture::new();
let absent = fixture.root.join(SENTINEL);
let output = Command::new(env!("CARGO_BIN_EXE_acativa"))
.args(["serve", "--database"])
.arg(&absent)
.args([
"--workflow-schema",
GENERIC_WORKFLOW_SCHEMA,
"--projection-schema",
GENERIC_PROJECTION_SCHEMA,
"--ephemeral-prefix",
"acativa-temporary",
])
.output()
.expect("missing database subprocess exits");
assert!(!output.status.success());
assert!(output.stdout.is_empty());
assert_eq!(output.stderr, b"acativa: runtime refused\n");
assert!(!absent.exists());
}