use super::secret_inspection::describe_selector;
use super::terminal_fitting::fit_rows;
use crate::cli::{FoldArgs, Format};
use crate::commands::rpc::{client_error, unexpected_response};
use crate::exit::ExitCode;
use crate::flourish;
use crate::host;
use crate::lookout::term;
use crate::output::{RolledSheep, RolledSheepRows, Streams, emit, emit_flock, write_outcome};
use crate::shutdown::Interrupt;
use crate::style::Presentation;
use crossterm::QueueableCommand as _;
use crossterm::cursor::{Hide, MoveTo, Show};
use crossterm::terminal::{Clear, ClearType};
use shep_client::Client;
use shep_core::paths::ShepPaths;
use shep_core::protocol::{HostUsage, ProcessInfo, Request, Response, SelectorSpec};
use shep_core::status::ProcStatus;
use shep_daemon::snapshot::FlockSnapshot;
use std::io::{self, Write as _};
use std::time::Duration;
pub(crate) fn read_roll(paths: &ShepPaths) -> Option<FlockSnapshot> {
std::fs::read(&paths.snapshot)
.ok()
.and_then(|bytes| serde_json::from_slice::<FlockSnapshot>(&bytes).ok())
}
pub fn flock_from_roll(streams: &mut Streams<'_>, paths: &ShepPaths) -> ExitCode {
let saved = read_roll(paths);
let mut sheep: Vec<RolledSheep> = saved
.map(|roll| {
roll.apps
.into_iter()
.map(|entry| RolledSheep {
name: entry.app.name.clone(),
instances: entry.instances_running,
status: "stopped",
})
.collect()
})
.unwrap_or_default();
sheep.sort_unstable_by(|a, b| a.name.cmp(&b.name));
if streams.fmt == Format::Table {
let _ = writeln!(
streams.err,
"no shepherd running. {}",
if sheep.is_empty() {
"nothing in the saved roll either.".to_owned()
} else {
format!(
"{} in the saved roll at {}:",
match sheep.len() {
1 => "1 sheep".to_owned(),
n => format!("{n} sheep"),
},
paths.snapshot.display()
)
}
);
}
let empty = sheep.is_empty();
if !(empty && streams.fmt == Format::Table) {
let _ = emit(
&mut *streams.out,
streams.fmt,
"flock",
RolledSheepRows(sheep),
streams.style,
);
}
if streams.fmt == Format::Table && !empty {
let _ = writeln!(streams.err, "`shep muster` brings them back.");
}
ExitCode::DaemonUnreachable
}
pub async fn flock(client: &Client, streams: &mut Streams<'_>) -> ExitCode {
match client.request(Request::ListFlock).await {
Ok(Response::Flock(procs)) => {
let art = (streams.fmt == Format::Table && streams.style.level.sheep())
.then(|| sheep_flourish(&procs))
.flatten();
if let Some(art) = &art {
let _ = write!(streams.out, "{art}");
}
let host = host_usage(client).await;
if streams.fmt == Format::Table
&& let Some(usage) = host
{
let strip = host::strip(usage, streams.style, streams.style.width);
let _ = writeln!(streams.out, "{strip}\n");
}
write_outcome(emit_flock(
&mut *streams.out,
streams.fmt,
"flock",
procs,
host,
streams.style,
))
}
Ok(_unrecognised) => unexpected_response(streams),
Err(err) => client_error(streams, &err),
}
}
async fn host_usage(client: &Client) -> Option<Option<HostUsage>> {
match client.request(Request::HostUsage).await {
Ok(Response::HostUsage(usage)) => Some(usage),
Ok(_unrecognised) => None,
Err(_unsupported) => None,
}
}
pub(super) fn sheep_flourish(listing: &[ProcessInfo]) -> Option<String> {
let sheep: Vec<&ProcessInfo> = listing.iter().filter(|p| p.dog.is_none()).collect();
if sheep.is_empty() {
return Some(flourish::empty_flock());
}
sheep
.iter()
.all(|p| p.status == ProcStatus::Stopped)
.then(|| flourish::all_asleep(sheep.len()))
}
pub(crate) async fn flock_follow(
client: &Client,
streams: &mut Streams<'_>,
interval: Duration,
) -> ExitCode {
let mut interrupt = match Interrupt::install() {
Ok(interrupt) => interrupt,
Err(err) => {
let message = format!("listening for an interrupt: {err}");
return streams.fail(ExitCode::Failure, &message);
}
};
term::install_panic_hook();
let _cursor = term::RestoreGuard::with_action(|| {
let _ = crossterm::execute!(io::stdout(), Show);
});
let _ = streams.out.queue(Hide);
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tokio::select! {
_ = ticker.tick() => {}
_ = interrupt.recv() => return ExitCode::Success,
}
let listing = tokio::select! {
listing = client.request(Request::ListFlock) => listing,
_ = interrupt.recv() => return ExitCode::Success,
};
let procs = match listing {
Ok(Response::Flock(procs)) => procs,
Ok(_unrecognised) => return unexpected_response(streams),
Err(err) => return client_error(streams, &err),
};
let host = tokio::select! {
host = host_usage(client) => host,
_ = interrupt.recv() => return ExitCode::Success,
};
let size = crossterm::terminal::size().ok();
let columns = size
.filter(|&(columns, _)| columns > 0)
.map_or(streams.style.width, |(columns, _)| usize::from(columns));
let frame = follow_frame(
procs,
streams.style,
host.map(|usage| host::strip(usage, streams.style, columns)),
);
let frame = match size {
Some((columns, rows)) => fit_rows(&frame, columns, rows),
None => frame,
};
let painted = paint(streams.out, &frame);
if painted.is_err() {
return write_outcome(painted);
}
}
}
fn paint(out: &mut dyn io::Write, frame: &str) -> io::Result<()> {
out.queue(MoveTo(0, 0))?;
out.queue(Clear(ClearType::FromCursorDown))?;
write!(out, "{frame}")?;
out.flush()
}
fn follow_frame(listing: Vec<ProcessInfo>, style: Presentation, strip: Option<String>) -> String {
let mut frame = Vec::new();
if let Some(strip) = strip {
let _ = writeln!(frame, "{strip}\n");
}
let _ = emit_flock(&mut frame, Format::Table, "flock", listing, None, style);
String::from_utf8_lossy(&frame).into_owned()
}
pub async fn fold(
client: &Client,
streams: &mut Streams<'_>,
paths: &ShepPaths,
args: &FoldArgs,
) -> ExitCode {
describe_selector(
client,
streams,
paths,
"fold",
None,
SelectorSpec::Fold(args.name.clone()),
)
.await
}
#[cfg(test)]
mod tests {
use shep_core::paths::ShepPaths;
use shep_core::protocol::{HostUsage, ProcessInfo, Request, SelectorSpec};
use shep_core::status::ProcStatus;
use crate::cli::{FoldArgs, Format};
use crate::exit::ExitCode;
use crate::host;
use crate::output::Streams;
use super::super::testing::*;
use super::*;
use crate::style::Presentation;
use shep_client::testing::{
fake_client_capturing_envelopes, fake_client_on, fake_client_replying_err,
fake_client_with_ack, sample_ack, sample_info,
};
use shep_core::protocol::RpcErrorCode;
#[test]
fn a_followed_frame_leads_with_the_host_strip() {
let usage = HostUsage {
cpu_percent: Some(11.0),
memory_used_bytes: 39_963_869_184,
memory_total_bytes: 51_539_607_552,
disk_bytes_per_second: Some((0, 0)),
network_bytes_per_second: Some((0, 0)),
};
let strip = host::strip(Some(usage), Presentation::BARE, 200);
let frame = follow_frame(vec![sample_info()], Presentation::BARE, Some(strip));
let mut lines = frame.lines();
assert!(lines.next().unwrap().starts_with("host cpu "));
assert_eq!(lines.next().unwrap(), "");
assert!(frame.contains("web"), "the table still follows: {frame}");
}
#[test]
fn a_frame_without_a_host_strip_is_the_tables_alone() {
let frame = follow_frame(vec![sample_info()], Presentation::BARE, None);
assert!(!frame.contains("host cpu"), "{frame}");
assert!(frame.contains("web"), "{frame}");
}
#[test]
fn a_followed_frame_of_an_empty_flock_carries_no_flourish() {
let frame = follow_frame(Vec::new(), Presentation::BARE, None);
assert!(!frame.contains("no sheep in the flock yet"), "{frame}");
}
#[tokio::test]
async fn a_shepherd_that_refuses_the_host_request_answers_for_no_strip() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _daemon) = fake_client_replying_err(
&path,
RpcErrorCode::Unsupported,
"this shepherd does not implement the request the client sent",
)
.await;
assert_eq!(host_usage(&client).await, None);
}
#[tokio::test]
async fn a_shepherd_answering_something_else_answers_for_no_strip() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _envelopes) = fake_client_capturing_envelopes(&path).await;
assert_eq!(host_usage(&client).await, None);
}
#[tokio::test]
async fn flock_asks_for_the_machine_as_well_as_the_flock() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, mut envelopes) = shepherd_answering(&path, Some(sample_host())).await;
let _ = listing(&client, Format::Table).await;
let mut sent = Vec::new();
for _ in 0..2 {
let envelope = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
.await
.expect("flock must reach the wire; it hung instead of sending a request")
.unwrap();
sent.push(envelope.body);
}
assert_eq!(sent, vec![Request::ListFlock, Request::HostUsage]);
}
#[tokio::test]
async fn a_one_shot_listing_leads_with_the_host_strip() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _envelopes) = shepherd_answering(&path, Some(sample_host())).await;
let (code, text) = listing(&client, Format::Table).await;
assert_eq!(code, ExitCode::Success);
let mut lines = text.lines();
assert!(
lines.next().unwrap().starts_with("host cpu "),
"got {text:?}"
);
assert_eq!(lines.next().unwrap(), "");
assert!(text.contains("web"), "the table still follows: {text:?}");
}
#[tokio::test]
async fn the_json_surface_carries_the_host_beside_the_flock() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _envelopes) = shepherd_answering(&path, Some(sample_host())).await;
let (_code, text) = listing(&client, Format::Json).await;
let json: serde_json::Value = serde_json::from_str(&text).unwrap();
assert_eq!(json["data"][0]["name"], "web");
assert_eq!(json["host"]["memory_used_bytes"], 39_963_869_184u64);
assert_eq!(json["host"]["cpu_percent"], 11.459_433);
assert!(!text.contains("host cpu"), "got {text:?}");
}
#[tokio::test]
async fn the_json_host_key_spells_its_three_answers_apart() {
let json = async |host: Option<HostUsage>| {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _envelopes) = shepherd_answering(&path, host).await;
let (_code, text) = listing(&client, Format::Json).await;
serde_json::from_str::<serde_json::Value>(&text).unwrap()
};
let read = json(Some(sample_host())).await;
assert_eq!(read["host"]["cpu_percent"], 11.459_433);
let unreadable = json(None).await;
assert!(unreadable["host"].is_null(), "got {unreadable:?}");
let unmeasured = json(Some(HostUsage {
cpu_percent: None,
disk_bytes_per_second: None,
network_bytes_per_second: None,
..sample_host()
}))
.await;
assert!(unmeasured["host"].is_object(), "got {unmeasured:?}");
assert!(unmeasured["host"]["cpu_percent"].is_null());
assert_eq!(unmeasured["host"]["memory_used_bytes"], 39_963_869_184u64);
}
#[tokio::test]
async fn a_refused_host_request_leaves_no_json_key_at_all() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
daemon.reply_to_list(vec![sample_info()]);
let (code, text) = listing(&client, Format::Json).await;
assert_eq!(code, ExitCode::Success);
let json: serde_json::Value = serde_json::from_str(&text).unwrap();
assert_eq!(json["data"][0]["name"], "web");
assert!(json.get("host").is_none(), "got {json:?}");
}
#[tokio::test]
async fn flock_asks_the_daemon_to_list_the_whole_flock() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
let mut out = Vec::new();
let mut err = Vec::new();
let mut streams = Streams {
out: &mut out,
err: &mut err,
style: crate::style::Presentation::BARE,
fmt: Format::Table,
};
let _ = flock(&client, &mut streams).await;
let sent = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
.await
.expect("flock must reach the wire; it hung instead of sending a request")
.unwrap();
assert_eq!(sent.body, Request::ListFlock);
}
#[tokio::test]
async fn fold_asks_the_daemon_for_that_fold_and_nothing_wider() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
let paths = ShepPaths::resolve(&|_| None, dir.path());
let mut out = Vec::new();
let mut err = Vec::new();
let mut streams = Streams {
out: &mut out,
err: &mut err,
style: crate::style::Presentation::BARE,
fmt: Format::Table,
};
let _ = fold(
&client,
&mut streams,
&paths,
&FoldArgs { name: "api".into() },
)
.await;
let sent = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
.await
.expect("fold must reach the wire; it hung instead of sending a request")
.unwrap();
assert_eq!(
sent.body,
Request::Describe {
selector: SelectorSpec::Fold("api".into())
}
);
}
#[tokio::test]
async fn flock_response_round_trips_into_rendered_flock_rows() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
daemon.reply_to_list(vec![sample_info()]);
let mut out = Vec::new();
let mut err = Vec::new();
let code = {
let mut streams = Streams {
out: &mut out,
err: &mut err,
style: crate::style::Presentation::BARE,
fmt: Format::Json,
};
flock(&client, &mut streams).await
};
assert_eq!(code, ExitCode::Success);
let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
assert_eq!(json["command"], "flock");
assert_eq!(json["data"][0]["name"], "web");
}
#[tokio::test]
async fn fold_never_computes_or_prints_a_secrets_section() {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
daemon.reply_to_describe(vec![
ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
]);
let paths = ShepPaths::resolve(&|_| None, dir.path());
let mut out = Vec::new();
let mut err = Vec::new();
let mut streams = Streams {
out: &mut out,
err: &mut err,
style: crate::style::Presentation::BARE,
fmt: Format::Json,
};
let code = fold(
&client,
&mut streams,
&paths,
&FoldArgs { name: "api".into() },
)
.await;
assert_eq!(code, ExitCode::Success);
let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
assert!(json.get("secrets").is_none(), "{json}");
}
#[test]
fn sheep_flourish_fires_empty_flock_on_a_truly_empty_listing() {
let art = sheep_flourish(&[]).expect("an empty listing must flourish");
assert!(art.contains("no sheep in the flock yet"), "{art}");
}
#[test]
fn sheep_flourish_treats_dogs_only_as_an_empty_flock() {
let art =
sheep_flourish(&[dog(1), dog(2)]).expect("dogs alone must read as an empty flock");
assert!(art.contains("no sheep in the flock yet"), "{art}");
}
#[test]
fn sheep_flourish_fires_all_asleep_when_every_sheep_is_stopped() {
let listing = [
sheep(1, ProcStatus::Stopped),
sheep(2, ProcStatus::Stopped),
dog(3),
];
let art = sheep_flourish(&listing).expect("an all-stopped flock must flourish");
assert!(art.contains("2 in the flock, all asleep"), "{art}");
}
#[test]
fn a_live_dog_does_not_block_all_asleep() {
let listing = [sheep(1, ProcStatus::Stopped), dog(2)];
let art = sheep_flourish(&listing).expect("a live dog must not suppress all_asleep");
assert!(art.contains("1 in the flock, all asleep"), "{art}");
}
#[test]
fn sheep_flourish_is_silent_on_a_mixed_flock() {
let listing = [sheep(1, ProcStatus::Online), sheep(2, ProcStatus::Stopped)];
assert_eq!(
sheep_flourish(&listing),
None,
"a mixed flock is not a flourish moment"
);
}
#[test]
fn stopping_does_not_count_as_asleep() {
let listing = [
sheep(1, ProcStatus::Stopping),
sheep(2, ProcStatus::Stopping),
];
assert_eq!(
sheep_flourish(&listing),
None,
"Stopping is a transient, not rest"
);
}
#[tokio::test]
async fn the_flourish_only_prints_under_table_format_and_a_sheep_drawing_level() {
use crate::style::{Presentation, StyleLevel};
for (fmt, level, expect_art) in [
(Format::Table, StyleLevel::Full, true),
(Format::Json, StyleLevel::Full, false),
(Format::Table, StyleLevel::Plain, false),
(Format::Table, StyleLevel::Bare, false),
] {
let dir = tempfile::tempdir().unwrap();
let path = shep_client::testing::control_address(dir.path());
let (client, _daemon) = fake_client_on(&path).await;
let mut out = Vec::new();
let mut err = Vec::new();
let mut streams = Streams {
out: &mut out,
err: &mut err,
style: Presentation::new(level, None, None, None, 80),
fmt,
};
let _ = flock(&client, &mut streams).await;
let printed = String::from_utf8_lossy(&out);
assert_eq!(
printed.contains("no sheep in the flock yet"),
expect_art,
"fmt={fmt:?} level={level:?}: {printed}"
);
}
}
}