//! The compose shell-out engine (04-01, RIG-01): runner seam, version
//! check, LOCKED arg builders, and the LDJSON/array parsers — serde
//! models out, no printing (the TUI rides the actions layer in Phase 6).
//!
//! ## The runner seam
//!
//! [`ComposeRunner`] is the ONLY way any rig code spawns a process:
//! actions and discovery take `&dyn ComposeRunner`, so every decision
//! path is unit-testable against a scripted fake (no docker needed).
//! The production [`DockerCompose`] shells out via
//! `tokio::process::Command` (the workspace `process` feature — the one
//! Phase 4 dependency change). `run` prefixes the `compose` subcommand;
//! `run_docker` spawns the PLAIN docker CLI (volume ls / port
//! attribution — no `-p` prefix, those are not compose project ops);
//! `run_streaming` pipes stdout and forwards lines as they arrive (the
//! `logs -f` follow shape, 04-02).
//!
//! ## LOCKED invocation shapes (research §Compose invocation shapes)
//!
//! Every builder is a pure function whose exact output vector is
//! unit-pinned — `-p <resolved-name>` EXPLICIT on every project op
//! (Pitfall 8: no implicit directory-name projects), `--project-directory`
//! always on resolve (Pitfall 8: `.env` loading is cwd-sensitive),
//! `--remove-orphans` on up AND down (Pitfall 4), `--wait-timeout`
//! explicit on up (Pitfall 3: healthchecks + image pulls).
//!
//! ## The two output conventions (research Pitfall 1)
//!
//! `ps`/`volume ls`/`docker ps` emit ONE OBJECT PER LINE (LDJSON —
//! parsed with a `StreamDeserializer`); `config --format json` emits a
//! SINGLE doc (an object on current compose; an older array shape is
//! tolerated by unwrapping the first element). BOTH conventions are
//! fixture-pinned so the divergence can never regress into a naive
//! `from_str::<Vec<T>>`.
use std::path::Path;
use serde::Deserialize;
use serde_json::Value;
use crate::error::CoreError;
use super::RigPlan;
/// How many stderr lines ride a failed invocation's error message
/// (research Pitfall 3: compose's tail is the diagnosis).
const STDERR_TAIL_LINES: usize = 5;
/// One completed `docker …` invocation: captured output + exit code.
/// A spawn FAILURE (no docker binary) is `code: 127` with the spawn
/// error in `stderr` — one error path, no io::Error leakage.
#[derive(Debug, Clone, PartialEq)]
pub struct ComposeOutput {
/// Captured stdout (UTF-8 lossy).
pub stdout: String,
/// Captured stderr (UTF-8 lossy).
pub stderr: String,
/// Process exit code (`-1` when terminated by signal).
pub code: i32,
}
/// The process seam for everything rig-related. Actions NEVER spawn
/// processes directly — they script this trait (the GatewayApi
/// precedent), which is what makes the whole family testable without
/// docker.
#[async_trait::async_trait]
pub trait ComposeRunner: Send + Sync {
/// Run `docker compose <args…>` and capture stdout/stderr/exit.
async fn run(&self, args: &[String]) -> ComposeOutput;
/// Run the PLAIN docker CLI (`docker <args…>`) — the volume-ls and
/// port-attribution shapes that are NOT compose project ops.
async fn run_docker(&self, args: &[String]) -> ComposeOutput;
/// Stream `docker compose <args…>`: stdout forwards to `line_sink`
/// LINE-BY-LINE as it arrives (piped stdout — `logs -f` follow
/// mode), stderr is captured for diagnostics (a failed invocation
/// carries its tail in the error; a successful one's diagnostics
/// go to OUR stderr via tracing — NEVER the data stream). The
/// returned [`ComposeOutput`] carries EMPTY stdout: the lines
/// already went to the sink (fakes replay a preloaded stdout
/// through the sink — the keep-it-simple contract).
async fn run_streaming(
&self,
args: &[String],
line_sink: &mut (dyn for<'a> FnMut(&'a str) + Send),
) -> ComposeOutput;
}
/// Production [`ComposeRunner`]: shells out to the real `docker` binary
/// via `tokio::process::Command`.
pub struct DockerCompose;
impl DockerCompose {
/// Spawn `program` with `args`, capture everything. Spawn failure
/// (binary absent) maps to the shell's 127 with the reason in
/// stderr — callers translate nonzero exits uniformly.
async fn spawn(program: &str, args: &[String]) -> ComposeOutput {
let joined = std::iter::once(program.to_string())
.chain(args.iter().cloned())
.collect::<Vec<_>>()
.join(" ");
match tokio::process::Command::new(program)
.args(args)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.output()
.await
{
Ok(output) => ComposeOutput {
stdout: String::from_utf8_lossy(&output.stdout).into_owned(),
stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
code: output.status.code().unwrap_or(-1),
},
Err(err) => ComposeOutput {
stdout: String::new(),
stderr: format!("failed to spawn `{joined}`: {err}"),
code: 127,
},
}
}
/// The STREAMING spawn (04-02): piped stdout read line-by-line
/// into `line_sink` until EOF, stderr drained CONCURRENTLY (a full
/// stderr pipe would deadlock the stdout reader), then the wait —
/// the child streams until it exits (Ctrl-C kills the whole
/// foreground process group, the `logs -f` precedent).
async fn spawn_streaming(
program: &str,
args: &[String],
line_sink: &mut (dyn for<'a> FnMut(&'a str) + Send),
) -> ComposeOutput {
use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
let joined = std::iter::once(program.to_string())
.chain(args.iter().cloned())
.collect::<Vec<_>>()
.join(" ");
let mut child = match tokio::process::Command::new(program)
.args(args)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
{
Ok(child) => child,
Err(err) => {
return ComposeOutput {
stdout: String::new(),
stderr: format!("failed to spawn `{joined}`: {err}"),
code: 127,
};
}
};
let stdout = child.stdout.take().expect("stdout piped above");
let mut stderr = child.stderr.take().expect("stderr piped above");
// Concurrent stderr drain — the join only fails on task panic.
let stderr_task = tokio::spawn(async move {
let mut buffer = String::new();
let _ = stderr.read_to_string(&mut buffer).await;
buffer
});
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
line_sink(&line);
}
match child.wait().await {
Ok(status) => ComposeOutput {
stdout: String::new(),
stderr: stderr_task.await.unwrap_or_default(),
code: status.code().unwrap_or(-1),
},
Err(err) => ComposeOutput {
stdout: String::new(),
stderr: format!("failed to wait for `{joined}`: {err}"),
code: -1,
},
}
}
}
#[async_trait::async_trait]
impl ComposeRunner for DockerCompose {
async fn run(&self, args: &[String]) -> ComposeOutput {
let mut full = vec!["compose".to_string()];
full.extend(args.iter().cloned());
Self::spawn("docker", &full).await
}
async fn run_docker(&self, args: &[String]) -> ComposeOutput {
Self::spawn("docker", args).await
}
async fn run_streaming(
&self,
args: &[String],
line_sink: &mut (dyn for<'a> FnMut(&'a str) + Send),
) -> ComposeOutput {
let mut full = vec!["compose".to_string()];
full.extend(args.iter().cloned());
Self::spawn_streaming("docker", &full, line_sink).await
}
}
/// Map a completed invocation: exit 0 → the stdout str; nonzero →
/// [`CoreError::Rig`] carrying the stderr tail (last
/// [`STDERR_TAIL_LINES`] lines — research Pitfall 3).
pub fn check_output<'a>(output: &'a ComposeOutput, context: &str) -> Result<&'a str, CoreError> {
if output.code == 0 {
Ok(&output.stdout)
} else {
let tail = stderr_tail(&output.stderr);
let note = if tail.is_empty() {
String::new()
} else {
format!(": {tail}")
};
Err(CoreError::Rig(format!(
"{context} failed (exit {}){note}",
output.code
)))
}
}
/// The last `lines` stderr lines, in order, trimmed — the diagnosis
/// tail compose prints on failure.
pub(crate) fn stderr_tail(stderr: &str) -> String {
stderr
.lines()
.rev()
.take(STDERR_TAIL_LINES)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n")
.trim()
.to_string()
}
/// Verify `docker compose` answers with major version ≥ 2 (the v1
/// `docker-compose` Python binary never answers to `docker compose` —
/// absence IS the install-hint case). Returns the version string
/// (e.g. `"5.1.2"`).
pub async fn compose_version(runner: &dyn ComposeRunner) -> Result<String, CoreError> {
let output = runner.run(&["version".to_string()]).await;
if output.code != 0 {
return Err(CoreError::Rig(format!(
"docker compose is unavailable (exit {}): {} — install Docker Desktop or \
the compose v2 plugin; the legacy docker-compose v1 binary is not supported",
output.code,
stderr_tail(&output.stderr)
)));
}
parse_compose_version(&output.stdout).ok_or_else(|| {
CoreError::Rig(format!(
"cannot parse `docker compose version` output ({:?}) — compose ≥ v2 \
(the `docker compose` plugin) is required",
output.stdout.trim()
))
})
}
/// `"Docker Compose version v5.1.2"` → `Some("5.1.2")`; anything else —
/// including a major < 2 — is `None`. Pure; pinned by tests.
fn parse_compose_version(stdout: &str) -> Option<String> {
let rest = stdout.trim().strip_prefix("Docker Compose version ")?;
let version = rest.split_whitespace().next()?;
let version = version.strip_prefix('v').unwrap_or(version);
let mut parts = version.split('.');
let major: u32 = parts.next()?.parse().ok()?;
if major < 2 {
return None;
}
Some(version.to_string())
}
// ---------------------------------------------------------------------------
// Arg builders — pure, exact-vector-pinned below.
// ---------------------------------------------------------------------------
/// The resolve step (research Pattern 1): `docker compose -f <file>
/// --project-directory <dir> config --format json`. `--project-directory`
/// is ALWAYS explicit — `.env` (and thus `COMPOSE_PROJECT_NAME`)
/// loading is cwd-sensitive (Pitfall 8).
pub fn config_args(file: &Path, project_dir: &Path) -> Vec<String> {
vec![
"-f".into(),
file.display().to_string(),
"--project-directory".into(),
project_dir.display().to_string(),
"config".into(),
"--format".into(),
"json".into(),
]
}
/// `up` (research LOCKED shape): explicit `-p <name>` (never an
/// implicit directory-name project), detached + `--wait` with an
/// EXPLICIT timeout (Pitfall 3: `--wait` blocks on healthchecks and
/// image pulls), `--remove-orphans`.
pub fn up_args(plan: &RigPlan, wait_timeout_s: u64) -> Vec<String> {
vec![
"-p".into(),
plan.name.clone(),
"-f".into(),
plan.compose_file.display().to_string(),
"up".into(),
"-d".into(),
"--wait".into(),
"--wait-timeout".into(),
wait_timeout_s.to_string(),
"--remove-orphans".into(),
]
}
/// `down`: stop + remove containers/networks; `--remove-orphans` always
/// (Pitfall 4); `-v` (named+anonymous volume deletion) only for the
/// reset teardown half (04-02).
pub fn down_args(plan: &RigPlan, volumes: bool) -> Vec<String> {
let mut args = vec![
"-p".into(),
plan.name.clone(),
"-f".into(),
plan.compose_file.display().to_string(),
"down".into(),
"--remove-orphans".into(),
];
if volumes {
args.push("-v".into());
}
args
}
/// `ps` as LDJSON (research Pitfall 1): one object per service.
pub fn ps_args(plan: &RigPlan) -> Vec<String> {
vec![
"-p".into(),
plan.name.clone(),
"-f".into(),
plan.compose_file.display().to_string(),
"ps".into(),
"--format".into(),
"json".into(),
]
}
/// `logs` (human-form passthrough by design — the streaming exception
/// when `--follow`; wired by 04-02's `rig_logs` via `run` one-shot /
/// `run_streaming` follow). Invocation shape LOCKED from day one.
pub fn logs_args(plan: &RigPlan, tail: u32, follow: bool, service: Option<&str>) -> Vec<String> {
let mut args = vec![
"-p".into(),
plan.name.clone(),
"-f".into(),
plan.compose_file.display().to_string(),
"logs".into(),
"--tail".into(),
tail.to_string(),
];
if follow {
args.push("-f".into());
}
if let Some(service) = service {
args.push(service.to_string());
}
args
}
/// `docker volume ls` (04-01 status / 04-02 reset preview): PLAIN
/// docker CLI — no `compose` subcommand, no `-p` prefix (volumes are
/// labeled, not project-scoped, at this layer). Invoked via
/// [`ComposeRunner::run_docker`], which spawns `docker`, NOT
/// `docker compose`, for exactly this shape.
pub fn volume_ls_args(project: &str) -> Vec<String> {
vec![
"volume".into(),
"ls".into(),
"--filter".into(),
format!("label=com.docker.compose.project={project}"),
"--format".into(),
"json".into(),
]
}
/// `docker ps --filter publish=<port> --format json` — host-port
/// occupancy with attribution (research Pattern 3); also a PLAIN-docker
/// shape via [`ComposeRunner::run_docker`].
pub fn docker_ps_publish_args(port: u16) -> Vec<String> {
vec![
"ps".into(),
"--filter".into(),
format!("publish={port}"),
"--format".into(),
"json".into(),
]
}
// ---------------------------------------------------------------------------
// Parsers — pure functions over recorded-fixture strings.
// ---------------------------------------------------------------------------
/// One published-port mapping from a resolved compose config
/// (container `target` → host `published`).
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub struct PortMapping {
/// Container port.
pub target: u16,
/// Host port.
pub published: u16,
}
/// Parse `docker compose config --format json` output into a
/// [`RigPlan`] skeleton: `.name` is THE identity truth (honors the
/// rig's own `.env` `COMPOSE_PROJECT_NAME`), services are the service
/// map's keys, `port_mappings` the collected target→published pairs,
/// volumes the volume map's keys. Tolerant where compose is shape-shifty
/// (`published` arrives as string OR number; the doc may be a bare
/// object — current compose — or a single-element array — older
/// builds); loud where identity is at stake (no `.name` → error, never
/// an implicit directory-name project).
pub fn parse_config(
stdout: &str,
compose_file: &Path,
project_dir: &Path,
) -> Result<RigPlan, CoreError> {
let trimmed = stdout.trim();
let doc: Value = serde_json::from_str(trimmed).map_err(|err| {
CoreError::Rig(format!(
"cannot parse `docker compose config` output: {err}"
))
})?;
let root = match doc {
Value::Array(mut items) => items.pop().unwrap_or(Value::Null),
Value::Object(object) => Value::Object(object),
other => other,
};
let name = root
.get("name")
.and_then(Value::as_str)
.filter(|name| !name.is_empty())
.ok_or_else(|| {
CoreError::Rig(
"resolved compose config carries no `.name` — the project name is the \
rig's identity truth; refusing to guess (set COMPOSE_PROJECT_NAME in \
the rig's .env)"
.to_string(),
)
})?
.to_string();
let services: Vec<String> = root
.get("services")
.and_then(Value::as_object)
.map(|map| map.keys().cloned().collect())
.unwrap_or_default();
let mut port_mappings: Vec<PortMapping> = Vec::new();
if let Some(services) = root.get("services").and_then(Value::as_object) {
for service in services.values() {
let ports = service.get("ports").and_then(Value::as_array);
for port in ports.into_iter().flatten() {
// `published` is a STRING on current compose ("9088"),
// a number on some builds — tolerate both; entries
// without a published port (random host ports) don't
// map and are skipped.
let published = port.get("published").and_then(|value| {
value
.as_str()
.and_then(|s| s.parse().ok())
.or_else(|| value.as_u64().and_then(|n| u16::try_from(n).ok()))
});
let target = port
.get("target")
.and_then(|value| value.as_u64().and_then(|n| u16::try_from(n).ok()));
if let (Some(published), Some(target)) = (published, target)
&& !port_mappings
.iter()
.any(|mapping| mapping.published == published)
{
port_mappings.push(PortMapping { target, published });
}
}
}
}
let host_ports = port_mappings
.iter()
.map(|mapping| mapping.published)
.collect();
let volumes: Vec<String> = root
.get("volumes")
.and_then(Value::as_object)
.map(|map| map.keys().cloned().collect())
.unwrap_or_default();
Ok(RigPlan {
name,
compose_file: compose_file.to_path_buf(),
project_dir: project_dir.to_path_buf(),
services,
host_ports,
port_mappings,
volumes,
})
}
// ---------------------------------------------------------------------------
// Runner-scripted ops (04-02)
// ---------------------------------------------------------------------------
/// The reset preview (04-02, RIG-01): the named-volume names `rig
/// reset`'s `down -v` half will remove, reported in the result data so
/// agents see WHAT reset took before/as it acts. Label-filtered at the
/// docker layer ([`volume_ls_args`]) and name-filtered here — only
/// `<project>_-prefixed` volumes are reset's to take (defense in
/// depth; research Pitfall 4 shape: `Name` + `Labels`).
pub async fn reset_preview(
runner: &dyn ComposeRunner,
plan: &RigPlan,
) -> Result<Vec<String>, CoreError> {
let output = runner.run_docker(&volume_ls_args(&plan.name)).await;
let stdout = check_output(&output, "docker volume ls")?;
let prefix = format!("{}_", plan.name);
Ok(parse_volume_ls_ldjson(stdout)
.into_iter()
.map(|entry| entry.name)
.filter(|name| name.starts_with(&prefix))
.collect())
}
/// One `docker compose ps` publisher row (live-captured field names).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct Publisher {
/// Bind host (`"0.0.0.0"`).
#[serde(default, rename = "URL")]
pub url: Option<String>,
/// Container port.
#[serde(default, rename = "TargetPort")]
pub target_port: Option<u16>,
/// Host port.
#[serde(default, rename = "PublishedPort")]
pub published_port: Option<u16>,
/// `tcp`/`udp`.
#[serde(default, rename = "Protocol")]
pub protocol: Option<String>,
}
/// One `docker compose ps` service row (live-captured field names;
/// `Health`/`Publishers` optional — services without healthchecks or
/// ports omit them).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct ServiceStatus {
/// Container name (`whk-global-ignition-1`).
#[serde(default, rename = "Name")]
pub name: String,
/// Service name (`ignition`).
#[serde(default, rename = "Service")]
pub service: String,
/// `running` / `exited` / …
#[serde(default, rename = "State")]
pub state: String,
/// `healthy` / `starting` / absent.
#[serde(default, rename = "Health")]
pub health: Option<String>,
/// Last exit code.
#[serde(default, rename = "ExitCode")]
pub exit_code: Option<i64>,
/// Published ports with attribution.
#[serde(default, rename = "Publishers")]
pub publishers: Vec<Publisher>,
}
/// Parse LDJSON (one JSON object per line — research Pitfall 1). The
/// per-line split IS the delimiter contract (compose never wraps a row
/// across lines), so a stray non-JSON warning line WARNs and skips
/// without losing the rows after it; empty output is an empty vec (a
/// down rig has no rows).
fn parse_ldjson<T: serde::de::DeserializeOwned>(stdout: &str, what: &str) -> Vec<T> {
stdout
.lines()
.filter(|line| !line.trim().is_empty())
.filter_map(|line| match serde_json::from_str::<T>(line) {
Ok(item) => Some(item),
Err(err) => {
tracing::warn!(source = what, error = %err, "skipping unparseable line");
None
}
})
.collect()
}
/// Parse `docker compose ps --format json` LDJSON into
/// [`ServiceStatus`] rows.
pub fn parse_ps_ldjson(stdout: &str) -> Vec<ServiceStatus> {
parse_ldjson(stdout, "docker compose ps")
}
/// One `docker volume ls` row (only the name is consumed by status).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct VolumeEntry {
/// Volume name (`<project>_<volume>`).
#[serde(default, rename = "Name")]
pub name: String,
/// Labels verbatim (includes `com.docker.compose.project`).
#[serde(default, rename = "Labels")]
pub labels: Value,
}
/// Parse `docker volume ls --format json` LDJSON (research Pitfall 1).
pub fn parse_volume_ls_ldjson(stdout: &str) -> Vec<VolumeEntry> {
parse_ldjson(stdout, "docker volume ls")
}
/// One `docker ps --filter publish=` row, reduced to attribution: the
/// container's (first) name and its compose project label, when it has
/// one. `docker ps` JSON differs from `compose ps` JSON (`Names`
/// plural, `Labels` sometimes a `"k=v,k=v"` string instead of a map) —
/// both shapes are tolerated and fixture-pinned.
#[derive(Debug, Clone, PartialEq)]
pub struct DockerPsEntry {
/// Container name.
pub name: String,
/// `com.docker.compose.project` label, when present.
pub compose_project: Option<String>,
}
/// Parse `docker ps --format json` LDJSON into [`DockerPsEntry`] rows
/// (name + compose-project attribution only).
pub fn parse_docker_ps_ldjson(stdout: &str) -> Vec<DockerPsEntry> {
parse_ldjson::<serde_json::Value>(stdout, "docker ps")
.into_iter()
.filter_map(|value| {
// `Names` (docker ps, plural, comma-joined when multiple)
// with `Name` (compose ps shape) accepted as a fallback.
let name = value
.get("Names")
.or_else(|| value.get("Name"))
.and_then(Value::as_str)
.map(|names| names.split(',').next().unwrap_or(names).trim().to_string())
.unwrap_or_default();
if name.is_empty() {
return None;
}
let compose_project = match value.get("Labels") {
Some(Value::String(labels)) => labels.split(',').find_map(|pair| {
pair.trim()
.strip_prefix("com.docker.compose.project=")
.map(str::to_string)
}),
Some(labels @ Value::Object(_)) => labels
.get("com.docker.compose.project")
.and_then(Value::as_str)
.map(str::to_string),
_ => None,
};
Some(DockerPsEntry {
name,
compose_project,
})
})
.collect()
}
#[cfg(test)]
mod tests {
use std::path::Path;
use super::{
ComposeOutput, ComposeRunner, DockerCompose, PortMapping, check_output, config_args,
docker_ps_publish_args, down_args, logs_args, parse_compose_version, parse_config,
parse_docker_ps_ldjson, parse_ps_ldjson, parse_volume_ls_ldjson, ps_args, stderr_tail,
up_args, volume_ls_args,
};
use crate::rig::RigPlan;
fn sample_plan() -> RigPlan {
RigPlan {
name: "ignition-devops".into(),
compose_file: "/rigs/git-module/docker/docker-compose.yml".into(),
project_dir: "/rigs/git-module/docker".into(),
services: vec!["ignition".into(), "db".into()],
host_ports: vec![9088, 9443],
port_mappings: vec![
PortMapping {
target: 8088,
published: 9088,
},
PortMapping {
target: 443,
published: 9443,
},
],
volumes: vec!["gw-data".into()],
}
}
fn s(args: &[&str]) -> Vec<String> {
args.iter().map(|a| a.to_string()).collect()
}
// ----- builder pins (the LOCKED invocation shapes) -------------------
#[test]
fn config_args_pinned() {
assert_eq!(
config_args(
Path::new("/rigs/docker/compose.yml"),
Path::new("/rigs/docker"),
),
s(&[
"-f",
"/rigs/docker/compose.yml",
"--project-directory",
"/rigs/docker",
"config",
"--format",
"json"
]),
);
}
#[test]
fn up_args_pinned() {
assert_eq!(
up_args(&sample_plan(), 300),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"up",
"-d",
"--wait",
"--wait-timeout",
"300",
"--remove-orphans"
]),
);
}
#[test]
fn down_args_pinned_with_and_without_volumes() {
assert_eq!(
down_args(&sample_plan(), false),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"down",
"--remove-orphans"
]),
"plain down keeps volumes (reset's -v arrives in 04-02)"
);
assert_eq!(
down_args(&sample_plan(), true),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"down",
"--remove-orphans",
"-v"
]),
);
}
#[test]
fn ps_args_pinned() {
assert_eq!(
ps_args(&sample_plan()),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"ps",
"--format",
"json"
]),
);
}
#[test]
fn logs_args_pinned() {
assert_eq!(
logs_args(&sample_plan(), 200, false, None),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"logs",
"--tail",
"200"
]),
);
assert_eq!(
logs_args(&sample_plan(), 50, true, Some("ignition")),
s(&[
"-p",
"ignition-devops",
"-f",
"/rigs/git-module/docker/docker-compose.yml",
"logs",
"--tail",
"50",
"-f",
"ignition"
]),
);
}
#[test]
fn volume_ls_args_pinned_plain_docker_shape() {
assert_eq!(
volume_ls_args("ignition-devops"),
s(&[
"volume",
"ls",
"--filter",
"label=com.docker.compose.project=ignition-devops",
"--format",
"json"
]),
"plain docker CLI: no compose subcommand, no -p prefix"
);
}
#[test]
fn docker_ps_publish_args_pinned() {
assert_eq!(
docker_ps_publish_args(18088),
s(&["ps", "--filter", "publish=18088", "--format", "json"]),
);
}
// ----- version parse -------------------------------------------------
#[test]
fn version_parses_live_capture_and_rejects_old_or_garbage() {
// Live v5.1.2 capture (research).
assert_eq!(
parse_compose_version("Docker Compose version v5.1.2\n"),
Some("5.1.2".into())
);
assert_eq!(
parse_compose_version("Docker Compose version v2.24.6\n"),
Some("2.24.6".into())
);
// Unversioned builds exist ("Docker Compose version 2.34.0").
assert_eq!(
parse_compose_version("Docker Compose version 2.34.0\n"),
Some("2.34.0".into())
);
// v1-shaped output and garbage are unusable.
assert_eq!(parse_compose_version("1.29.2"), None);
assert_eq!(parse_compose_version(""), None);
assert_eq!(parse_compose_version("docker-compose 1.25.5"), None);
}
// ----- exit mapping ---------------------------------------------------
#[test]
fn check_output_maps_nonzero_to_rig_with_stderr_tail() {
let ok = ComposeOutput {
stdout: "{}".into(),
stderr: String::new(),
code: 0,
};
assert_eq!(check_output(&ok, "ctx").unwrap(), "{}");
let tail = (1..=9).map(|n| format!("line-{n}")).collect::<Vec<_>>();
let failed = ComposeOutput {
stdout: String::new(),
stderr: tail.join("\n"),
code: 1,
};
let err = check_output(&failed, "docker compose up").unwrap_err();
assert!(matches!(err, crate::error::CoreError::Rig(_)));
let message = err.to_string();
assert!(
message.contains("docker compose up failed (exit 1)"),
"{message}"
);
// The tail is the LAST ~5 lines only.
assert!(
!message.contains("line-4\n"),
"tail keeps at most 5 lines: {message}"
);
assert!(message.contains("line-5"));
assert!(message.contains("line-9"));
}
#[test]
fn stderr_tail_trims_and_keeps_order() {
assert_eq!(stderr_tail("b\na\n"), "b\na");
assert_eq!(stderr_tail(""), "");
assert_eq!(stderr_tail("\n\n"), "");
}
// ----- parse_config ---------------------------------------------------
/// The whk-environment-orchestration-shaped resolve fixture:
/// `.name` from the rig's own `.env`, string-form `published`, a
/// volumes map, and ignored top-level keys.
const CONFIG_FIXTURE: &str = r#"{
"name": "whk-global",
"services": {
"ignition": {
"image": "inductiveautomation/ignition:8.3.6",
"ports": [
{"mode": "host", "target": 8088, "published": "9088", "protocol": "tcp"},
{"mode": "host", "target": 443, "published": "9443", "protocol": "tcp"}
]
},
"git-server": {
"image": "git-server:latest",
"ports": [{"mode": "host", "target": 22, "published": "9022", "protocol": "tcp"}]
}
},
"volumes": {
"gw-data": {"name": "whk-global_gw-data", "driver": "local"},
"gw-tag-definition": {"name": "whk-global_gw-tag-definition"}
},
"networks": {"default": {"name": "whk-global_default"}},
"secrets": {}
}"#;
#[test]
fn parse_config_reads_name_services_ports_volumes() {
let plan = parse_config(
CONFIG_FIXTURE,
Path::new("/whk/whk-environment-orchestration/docker-compose.yml"),
Path::new("/whk/whk-environment-orchestration"),
)
.expect("fixture parses");
assert_eq!(plan.name, "whk-global");
assert_eq!(plan.services, vec!["git-server", "ignition"]);
// Services iterate sorted (serde_json's map) → ports arrive in
// service order: git-server's 9022 first.
assert_eq!(plan.host_ports, vec![9022, 9088, 9443]);
assert!(
plan.port_mappings.contains(&PortMapping {
target: 8088,
published: 9088
}),
"string-form `published` tolerated: {:?}",
plan.port_mappings
);
assert_eq!(plan.volumes, vec!["gw-data", "gw-tag-definition"]);
assert_eq!(
plan.compose_file,
Path::new("/whk/whk-environment-orchestration/docker-compose.yml")
);
assert_eq!(
plan.project_dir,
Path::new("/whk/whk-environment-orchestration")
);
}
#[test]
fn parse_config_tolerates_array_doc_and_numeric_published() {
// Older compose wraps the config doc in a single-element array,
// and some builds emit numeric `published`.
let array_form = format!("[{}]", CONFIG_FIXTURE.replace("\"9088\"", "9088"));
let plan = parse_config(&array_form, Path::new("/c.yml"), Path::new("/"))
.expect("array doc + numeric published parse");
assert_eq!(plan.name, "whk-global");
assert!(plan.host_ports.contains(&9088));
}
#[test]
fn parse_config_without_name_refuses() {
let fixture = r#"{"services": {"ignition": {}}}"#;
let err = parse_config(fixture, Path::new("/c.yml"), Path::new("/")).unwrap_err();
let message = err.to_string();
assert!(message.contains("no `.name`"), "{message}");
assert!(message.contains("COMPOSE_PROJECT_NAME"), "{message}");
}
#[test]
fn parse_config_empty_services_and_missing_sections() {
let plan = parse_config(r#"{"name": "bare"}"#, Path::new("/c.yml"), Path::new("/"))
.expect("minimal doc parses");
assert!(plan.services.is_empty());
assert!(plan.host_ports.is_empty());
assert!(plan.volumes.is_empty());
}
/// Research Open Question 4, resolved empirically: the rig's own
/// `.env` `COMPOSE_PROJECT_NAME` governs the resolved `.name` even
/// when `ign` runs from an unrelated cwd — because `config_args`
/// ALWAYS passes `--project-directory` (Pitfall 8). Runs the REAL
/// docker CLI (config is client-side; no daemon needed) and skips
/// quietly when docker is absent (CI).
#[tokio::test]
async fn config_resolves_env_project_name_from_project_directory() {
let dir = tempfile::tempdir().expect("tempdir");
let compose = dir.path().join("docker-compose.yml");
std::fs::write(
&compose,
"services:\n sidecar:\n image: alpine:latest\n",
)
.expect("write compose");
std::fs::write(
dir.path().join(".env"),
"COMPOSE_PROJECT_NAME=env-resolved-name\n",
)
.expect("write .env");
let runner = DockerCompose;
if runner.run(&["version".to_string()]).await.code != 0 {
eprintln!("skipping: docker compose unavailable");
return;
}
// The spawn's cwd is the TEST binary's dir — nowhere near the
// fixture — which is exactly the cwd-elsewhere case.
let output = runner.run(&config_args(&compose, dir.path())).await;
assert_eq!(output.code, 0, "config run: {}", output.stderr);
let plan = parse_config(&output.stdout, &compose, dir.path()).expect("resolve parses");
assert_eq!(
plan.name, "env-resolved-name",
"the .env name wins over the directory-derived default"
);
}
/// The STREAMING spawn (04-02): piped stdout forwarded line-by-line
/// to the sink, stderr drained, exit reported. Runs the REAL docker
/// CLI (`logs` on an absent project is exit 0 + empty output on
/// compose v2 — no daemon-side state needed) and skips quietly when
/// docker is absent (CI). Client-side proof of the follow seam.
#[tokio::test]
async fn run_streaming_forwards_lines_via_piped_stdout() {
let dir = tempfile::tempdir().expect("tempdir");
let compose = dir.path().join("docker-compose.yml");
std::fs::write(
&compose,
"services:\n sidecar:\n image: alpine:latest\n",
)
.expect("write compose");
let runner = DockerCompose;
if runner.run(&["version".to_string()]).await.code != 0 {
eprintln!("skipping: docker compose unavailable");
return;
}
// `compose version` is CLIENT-side; the `compose logs` below
// needs a LIVE daemon — its absent-project exit-0 contract only
// holds when the daemon answers. Probe the daemon (plain
// `docker version` exits nonzero when unreachable — e.g. an
// auto-stopped OrbStack) so a daemon-less machine skips as
// quietly as a docker-less one.
if runner.run_docker(&["version".to_string()]).await.code != 0 {
eprintln!("skipping: docker daemon unreachable");
return;
}
let plan = parse_config(
r#"{"name":"stream-fixture","services":{"sidecar":{"image":"alpine"}}}"#,
&compose,
dir.path(),
)
.expect("plan parses");
let mut streamed = 0usize;
let mut sink = |_: &str| streamed += 1;
let output = runner
.run_streaming(&logs_args(&plan, 5, false, None), &mut sink)
.await;
assert_eq!(
output.code, 0,
"compose logs on an absent project: {}",
output.stderr
);
assert_eq!(
streamed, 0,
"no containers → no lines, but the spawn/read/wait ran"
);
assert!(
output.stdout.is_empty(),
"streamed stdout never reports back"
);
}
// ----- LDJSON parsers (research Pitfall 1: BOTH conventions pinned) --
/// Live-captured compose ps row shape (whk-global-style): one
/// object per line, Health/Publishers present on this row.
const PS_ROW_FULL: &str = r#"{"Command":"\"/usr/bin/start-ignition.sh\"","CreatedAt":"2026-08-22 10:00:00 +0000 UTC","ExitCode":0,"Health":"healthy","Labels":{"com.docker.compose.project":"whk-global","com.docker.compose.service":"ignition"},"Name":"whk-global-ignition-1","Networks":"whk-global_default","Ports":"0.0.0.0:9088->8088/tcp, 0.0.0.0:9443->443/tcp","Publishers":[{"URL":"0.0.0.0","TargetPort":8088,"PublishedPort":9088,"Protocol":"tcp"},{"URL":"0.0.0.0","TargetPort":443,"PublishedPort":9443,"Protocol":"tcp"}],"RunningFor":"2 hours","Service":"ignition","State":"running","Status":"Up 2 hours (healthy)"}"#;
/// A second row WITHOUT Health/Publishers (service publishes no
/// ports, no healthcheck) — the optional-fields tolerance pin.
const PS_ROW_MINIMAL: &str = r#"{"Command":"sleep","ExitCode":0,"Name":"whk-global-sidecar-1","Service":"sidecar","State":"exited","Status":"Exited (0) 3 minutes ago"}"#;
#[test]
fn parse_ps_ldjson_reads_rows_and_tolerates_missing_optionals() {
let rows = parse_ps_ldjson(&format!("{PS_ROW_FULL}\n{PS_ROW_MINIMAL}\n"));
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].name, "whk-global-ignition-1");
assert_eq!(rows[0].service, "ignition");
assert_eq!(rows[0].state, "running");
assert_eq!(rows[0].health.as_deref(), Some("healthy"));
assert_eq!(rows[0].exit_code, Some(0));
assert_eq!(rows[0].publishers.len(), 2);
assert_eq!(rows[0].publishers[0].target_port, Some(8088));
assert_eq!(rows[0].publishers[0].published_port, Some(9088));
assert_eq!(rows[0].publishers[0].protocol.as_deref(), Some("tcp"));
assert_eq!(rows[1].health, None);
assert!(rows[1].publishers.is_empty());
}
#[test]
fn parse_ps_ldjson_empty_and_warning_lines() {
assert!(parse_ps_ldjson("").is_empty(), "a down rig has no rows");
// A stray non-JSON warning line skips (warn), the rows survive.
let rows = parse_ps_ldjson(&format!("[NOTE] something\n{PS_ROW_MINIMAL}\n"));
assert_eq!(rows.len(), 1);
}
#[test]
fn parse_volume_ls_ldjson_fixture() {
let stdout = concat!(
r#"{"CreatedAt":"2026-08-22T10:00:00Z","Driver":"local","Labels":{"com.docker.compose.project":"ignition-devops","com.docker.compose.volume":"gw-data"},"Mountpoint":"/var/lib/docker/volumes/ignition-devops_gw-data/_data","Name":"ignition-devops_gw-data","Options":null,"Scope":"local"}"#,
"\n",
r#"{"CreatedAt":"2026-08-22T10:00:00Z","Driver":"local","Labels":{"com.docker.compose.project":"ignition-devops","com.docker.compose.volume":"gw-tag-definition"},"Mountpoint":"/var/lib/docker/volumes/ignition-devops_gw-tag-definition/_data","Name":"ignition-devops_gw-tag-definition","Options":null,"Scope":"local"}"#,
"\n",
);
let volumes = parse_volume_ls_ldjson(stdout);
assert_eq!(volumes.len(), 2);
assert_eq!(volumes[0].name, "ignition-devops_gw-data");
assert_eq!(volumes[1].name, "ignition-devops_gw-tag-definition");
assert!(parse_volume_ls_ldjson("").is_empty());
}
#[test]
fn parse_docker_ps_labels_as_string_and_map() {
// docker ps emits `Names` (plural) and Labels as a k=v string.
let string_labels = r#"{"ID":"abc123","Image":"alpine","Labels":"com.docker.compose.project=other-project,com.docker.compose.service=sidecar","Names":"other-project-sidecar-1","Ports":"0.0.0.0:18088->8088/tcp","State":"running"}"#;
let rows = parse_docker_ps_ldjson(string_labels);
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].name, "other-project-sidecar-1");
assert_eq!(rows[0].compose_project.as_deref(), Some("other-project"));
// The compose-ps-style map labels + Name key are accepted too.
let map_labels =
r#"{"Name":"ign-research","Labels":{"com.docker.compose.project":"research"}}"#;
let rows = parse_docker_ps_ldjson(map_labels);
assert_eq!(rows[0].name, "ign-research");
assert_eq!(rows[0].compose_project.as_deref(), Some("research"));
// Non-compose containers attribute with no project.
let rows = parse_docker_ps_ldjson(r#"{"Names":"standalone-thing","Labels":""}"#);
assert_eq!(rows[0].compose_project, None);
assert!(parse_docker_ps_ldjson("").is_empty());
}
}