use std::io;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};
use muxio_tokio_rpc_ipc_client::RpcCallPrebuffered;
use term_session_muxio_service_definitions::{
Attach, AttachRequest, ChannelName, ListChannels, ShutdownGateway, Spawn, SpawnRequest,
SpawnResponse, path_wire, probe_ipc_endpoint,
};
fn bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_term-session"))
}
fn mock_bin() -> PathBuf {
term_session_mock::get_mock_bin()
}
fn unique_gateway(tag: &str) -> String {
static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
let id = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
format!("term-wm/dtest-{tag}-{id}")
}
fn spawn_daemon(gateway: &str, selfcheck: bool) -> (Child, Option<PathBuf>) {
let marker = if selfcheck {
let path = std::env::temp_dir().join(format!(
"term-session-selfcheck-{}.txt",
gateway.replace('/', "-")
));
let _ = std::fs::remove_file(&path);
Some(path)
} else {
None
};
let mut cmd = Command::new(bin());
cmd.env("TERM_WM_GATEWAY", gateway).arg("--daemon");
if let Some(ref m) = marker {
cmd.arg("--daemon-selfcheck").arg(m);
}
cmd.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
let child = cmd.spawn().expect("spawn daemon");
(child, marker)
}
fn spawn_daemon_in(gateway: &str, cwd: &std::path::Path) -> Child {
let mut cmd = Command::new(bin());
cmd.env("TERM_WM_GATEWAY", gateway)
.arg("--daemon")
.current_dir(cwd)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
cmd.spawn().expect("spawn daemon")
}
async fn attach_to(
client: &muxio_tokio_rpc_ipc_client::RpcIpcClient,
channel: &str,
hostname: &str,
) -> usize {
Attach::call(
client,
AttachRequest {
channel: channel.to_string(),
hostname: hostname.to_string(),
pid: std::process::id() as u64,
user: "test-user".to_string(),
version: "test-version".to_string(),
ssh_ip: None,
},
)
.await
.expect("attach")
}
async fn wait_connectable(gateway: &str) -> Arc<muxio_tokio_rpc_ipc_client::RpcIpcClient> {
let start = Instant::now();
loop {
match muxio_tokio_rpc_ipc_client::RpcIpcClient::new(gateway).await {
Ok(c) => return c,
Err(_) if start.elapsed() < Duration::from_secs(20) => {
tokio::time::sleep(Duration::from_millis(50)).await;
}
Err(e) => panic!("gateway {gateway} not reachable after 20s: {e}"),
}
}
}
#[tokio::test]
async fn daemon_detaches_and_reports_proof() {
let gateway = unique_gateway("detach");
let (mut child, marker) = spawn_daemon(&gateway, true);
let marker = marker.expect("marker requested");
let start = Instant::now();
let proof = loop {
if let Ok(content) = std::fs::read_to_string(&marker) {
break content.trim().to_string();
}
assert!(
start.elapsed() < Duration::from_secs(8),
"daemon never wrote selfcheck marker"
);
tokio::time::sleep(Duration::from_millis(50)).await;
};
#[cfg(windows)]
assert_eq!(proof, "windows-no-console", "marker: {proof}");
#[cfg(unix)]
assert_eq!(proof, "unix-session-leader", "marker: {proof}");
let client = wait_connectable(&gateway).await;
ShutdownGateway::call(&*client, true).await.unwrap();
let _ = child.wait();
}
#[tokio::test]
async fn daemon_survives_all_clients_disconnecting() {
let gateway = unique_gateway("survive");
let (mut child, _marker) = spawn_daemon(&gateway, false);
let client = wait_connectable(&gateway).await;
let channel = "test/daemon_survive";
attach_to(&client, channel, "t").await;
Spawn::call(
&*client,
SpawnRequest {
cmd: Some(vec![
mock_bin().to_string_lossy().to_string(),
"sleep".into(),
"60000".into(),
]),
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
drop(client);
let client2 = wait_connectable(&gateway).await;
attach_to(&client2, channel, "t").await;
Spawn::call(
&*client2,
SpawnRequest {
cmd: None,
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
ShutdownGateway::call(&*client2, true).await.unwrap();
let _ = child.wait();
}
#[tokio::test]
async fn daemon_survives_parent_death() {
let gateway = unique_gateway("parent_death");
let channel = "test/daemon_parent_death";
let mock = mock_bin().to_string_lossy().to_string();
let mut client = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["--channel", channel, "--", &mock, "sleep", "60000"])
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.expect("spawn auto-attach client");
tokio::time::sleep(Duration::from_millis(2000)).await;
let _ = client.kill();
let _ = client.wait();
let client = wait_connectable(&gateway).await;
attach_to(&client, channel, "t").await;
let SpawnResponse {
id,
cols: _,
rows: _,
} = Spawn::call(
&*client,
SpawnRequest {
cmd: None,
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
assert_eq!(id, 1, "session from the orphaned daemon must persist");
ShutdownGateway::call(&*client, true).await.unwrap();
tokio::time::sleep(Duration::from_millis(1000)).await;
}
#[tokio::test]
async fn session_starts_in_client_cwd() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let gateway = format!("term-wm/dtest-client_cwd-{nonce}");
let channel = "test/client_cwd";
let daemon_dir = tempfile::tempdir().expect("daemon tempdir");
let client_dir = tempfile::tempdir().expect("client tempdir");
let report_dir = tempfile::tempdir().expect("report tempdir");
let report = report_dir.path().join("pwd.txt");
let mut child = spawn_daemon_in(&gateway, daemon_dir.path());
let client = wait_connectable(&gateway).await;
attach_to(&client, channel, "cwd").await;
Spawn::call(
&*client,
SpawnRequest {
cmd: Some(vec![
mock_bin().to_string_lossy().to_string(),
"pwd".into(),
report.to_string_lossy().to_string(),
]),
cols: 80u16,
rows: 24u16,
cwd: Some(path_wire::encode_path(client_dir.path())),
},
)
.await
.unwrap();
let start = Instant::now();
let got = loop {
if let Ok(content) = std::fs::read(&report) {
break content;
}
assert!(
start.elapsed() < Duration::from_secs(10),
"mock pwd never wrote the report"
);
tokio::time::sleep(Duration::from_millis(50)).await;
};
let expected = std::fs::canonicalize(client_dir.path()).unwrap();
assert_eq!(
path_wire::decode_path(&path_wire::PathWire::from(got)),
expected,
"session must start in the client's launch directory, not the daemon's \
startup directory ({:?})",
daemon_dir.path()
);
ShutdownGateway::call(&*client, true).await.unwrap();
let _ = child.wait();
}
#[tokio::test]
async fn daemon_does_not_inherit_parent_handles() {
use term_session::auto_spawn::connect_or_spawn_server;
let gateway = unique_gateway("no_inherit");
unsafe {
std::env::set_var("TERM_WM_GATEWAY", &gateway);
}
#[cfg(windows)]
let (read_end, write_end) = create_inheritable_pipe();
#[cfg(unix)]
let (read_end, write_end) = create_cloexec_pipe();
#[cfg(not(any(unix, windows)))]
panic!("handle-inheritance test not supported on this platform");
connect_or_spawn_server(Some(&bin())).expect("auto-spawn daemon");
close_write_end(write_end);
assert_eof_on_read_end(read_end, Duration::from_secs(5))
.expect("daemon inherited the parent's pipe write end");
close_read_end(read_end);
let client = wait_connectable(&gateway).await;
ShutdownGateway::call(&*client, true).await.unwrap();
}
#[cfg(windows)]
fn create_inheritable_pipe() -> (
windows_sys::Win32::Foundation::HANDLE,
windows_sys::Win32::Foundation::HANDLE,
) {
use windows_sys::Win32::Security::SECURITY_ATTRIBUTES;
use windows_sys::Win32::System::Pipes::CreatePipe;
let sa = SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: std::ptr::null_mut(),
bInheritHandle: 1,
};
let mut read = std::ptr::null_mut();
let mut write = std::ptr::null_mut();
let ok = unsafe { CreatePipe(&mut read, &mut write, &sa, 0) };
assert_ne!(ok, 0, "CreatePipe failed: {}", io::Error::last_os_error());
(read, write)
}
#[cfg(unix)]
fn create_cloexec_pipe() -> (libc::c_int, libc::c_int) {
use std::os::unix::io::RawFd;
let mut fds = [0 as RawFd; 2];
assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0, "pipe() failed");
for &fd in &fds {
let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
assert!(flags >= 0, "F_GETFD failed");
let rc = unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) };
assert_eq!(rc, 0, "F_SETFD failed");
}
(fds[0], fds[1])
}
#[cfg(windows)]
fn assert_eof_on_read_end(
read: windows_sys::Win32::Foundation::HANDLE,
timeout: Duration,
) -> io::Result<()> {
use windows_sys::Win32::Foundation::ERROR_BROKEN_PIPE;
use windows_sys::Win32::System::Pipes::PeekNamedPipe;
let start = Instant::now();
loop {
let mut total_avail: u32 = 0;
let ok = unsafe {
PeekNamedPipe(
read,
std::ptr::null_mut(),
0,
std::ptr::null_mut(),
&mut total_avail,
std::ptr::null_mut(),
)
};
if ok == 0 {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(ERROR_BROKEN_PIPE as i32) {
return Ok(());
}
return Err(err);
}
if start.elapsed() >= timeout {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"read end never reached EOF (daemon inherited the write handle)",
));
}
std::thread::sleep(Duration::from_millis(50));
}
}
#[cfg(unix)]
fn assert_eof_on_read_end(fd: libc::c_int, timeout: Duration) -> io::Result<()> {
let start = Instant::now();
loop {
let mut poll_fds = [libc::pollfd {
fd,
events: libc::POLLIN | libc::POLLHUP,
revents: 0,
}];
let n = unsafe { libc::poll(poll_fds.as_mut_ptr(), 1, 50) };
if n < 0 {
return Err(io::Error::last_os_error());
}
if n > 0 {
let mut buf = [0u8; 64];
loop {
let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) };
if r < 0 {
if io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
continue;
}
return Err(io::Error::last_os_error());
}
if r == 0 {
return Ok(());
}
if (r as usize) < buf.len() {
break;
}
}
}
if start.elapsed() >= timeout {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"read end never reached EOF (child inherited the write fd)",
));
}
std::thread::sleep(Duration::from_millis(10));
}
}
#[cfg(windows)]
fn close_read_end(read: windows_sys::Win32::Foundation::HANDLE) {
unsafe {
let _ = windows_sys::Win32::Foundation::CloseHandle(read);
}
}
#[cfg(windows)]
fn close_write_end(write: windows_sys::Win32::Foundation::HANDLE) {
unsafe {
let _ = windows_sys::Win32::Foundation::CloseHandle(write);
}
}
#[cfg(unix)]
fn close_read_end(read: libc::c_int) {
unsafe {
let _ = libc::close(read);
}
}
#[cfg(unix)]
fn close_write_end(write: libc::c_int) {
unsafe {
let _ = libc::close(write);
}
}
#[tokio::test]
async fn cli_kill_client_detaches_one_client() {
let gateway = unique_gateway("kill_client");
let channel = "test/kill_client";
let (mut child, _marker) = spawn_daemon(&gateway, false);
let c1 = wait_connectable(&gateway).await;
let c2 = wait_connectable(&gateway).await;
attach_to(&c1, channel, "one").await;
attach_to(&c2, channel, "two").await;
Spawn::call(
&*c1,
SpawnRequest {
cmd: Some(vec![
mock_bin().to_string_lossy().to_string(),
"sleep".into(),
"60000".into(),
]),
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
Spawn::call(
&*c2,
SpawnRequest {
cmd: None,
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
let resp = ListChannels::call(&*c1, ()).await.unwrap();
let ch = resp
.channels
.iter()
.find(|c| c.name == channel)
.expect("channel listed");
assert_eq!(ch.clients.len(), 2, "two clients attached");
let target = ch.clients[0].conn_id;
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["kill-client", channel, &target.to_string()])
.output()
.expect("run kill-client");
assert!(
out.status.success(),
"kill-client failed: {}",
String::from_utf8_lossy(&out.stderr)
);
let resp = ListChannels::call(&*c1, ()).await.unwrap();
let ch = resp
.channels
.iter()
.find(|c| c.name == channel)
.expect("channel listed");
assert_eq!(
ch.clients.len(),
1,
"one client should remain after kill-client"
);
ShutdownGateway::call(&*c1, true).await.unwrap();
let _ = child.wait();
}
#[test]
fn bare_term_session_shows_help_and_does_not_connect() {
let gateway = unique_gateway("bare");
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.output()
.expect("run bare term-session");
assert_eq!(
out.status.code(),
Some(2),
"bare run must exit 2 (help, not auto-connect), got: {:?}",
out.status.code()
);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stderr.contains("--channel"),
"help should mention --channel, got: {stderr}"
);
assert!(
stderr.contains("ls"),
"help should list the ls subcommand, got: {stderr}"
);
assert!(
stderr.contains("stop"),
"help should list stop, got: {stderr}"
);
let gw = ChannelName::parse(&gateway).expect("gateway name");
assert!(
!probe_ipc_endpoint(&gw),
"bare run must not auto-spawn a daemon"
);
}
#[tokio::test]
async fn top_level_channel_auto_attaches() {
let gateway = unique_gateway("autoattach");
let channel = "test/autoattach";
let mock = mock_bin().to_string_lossy().to_string();
let mut client = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["--channel", channel, "--", &mock, "sleep", "60000"])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("spawn auto-attach client");
let rpc = wait_connectable(&gateway).await;
let start = Instant::now();
loop {
let resp = ListChannels::call(&*rpc, ()).await.unwrap();
let live = resp
.channels
.iter()
.any(|c| c.name == channel && c.session.as_ref().is_some_and(|s| !s.exited));
if live {
break;
}
assert!(
start.elapsed() < Duration::from_secs(20),
"session never appeared on the auto-attached channel"
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
let _ = client.kill();
let _ = client.wait();
ShutdownGateway::call(&*rpc, true).await.unwrap();
}
#[tokio::test]
async fn dash_dash_disambiguates_command_from_subcommand() {
let gateway = unique_gateway("disambig");
let gw = ChannelName::parse(&gateway).expect("gateway name");
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.arg("list")
.output()
.expect("run list");
assert!(
String::from_utf8_lossy(&out.stderr).contains("No gateway"),
"`list` must parse as the admin subcommand, got: {}",
String::from_utf8_lossy(&out.stderr)
);
assert!(
!probe_ipc_endpoint(&gw),
"admin subcommand must not auto-spawn"
);
let mut client = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["--", "list"])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("run -- list");
let rpc = wait_connectable(&gateway).await;
ShutdownGateway::call(&*rpc, true).await.unwrap();
let _ = client.kill();
let _ = client.wait();
}
#[tokio::test]
async fn unknown_flag_errors_without_spawning_gateway() {
for flag in ["--list", "--bogus", "-x"] {
let gateway = unique_gateway("unknown_flag");
let gw = ChannelName::parse(&gateway).expect("gateway name");
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.arg(flag)
.output()
.expect("run with unknown flag");
assert_ne!(
out.status.code(),
Some(0),
"`{flag}` must not exit successfully"
);
assert!(
String::from_utf8_lossy(&out.stderr).contains("unexpected argument"),
"`{flag}` must be reported as an unexpected argument, got: {}",
String::from_utf8_lossy(&out.stderr)
);
assert!(
!probe_ipc_endpoint(&gw),
"`{flag}` must not auto-spawn a gateway"
);
}
}
#[tokio::test]
async fn cli_list_renders_client_identity() {
let gateway = unique_gateway("list_identity");
let channel = "test/list_identity";
let (mut daemon, _marker) = spawn_daemon(&gateway, false);
let client = wait_connectable(&gateway).await;
Attach::call(
&*client,
AttachRequest {
channel: channel.to_string(),
hostname: "render-host".to_string(),
pid: 4242,
user: "bob".to_string(),
version: "v7".to_string(),
ssh_ip: Some("203.0.113.9".to_string()),
},
)
.await
.unwrap();
Spawn::call(
&*client,
SpawnRequest {
cmd: None,
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.arg("list")
.output()
.expect("run list");
assert!(
out.status.success(),
"list failed: {}",
String::from_utf8_lossy(&out.stderr)
);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("user: bob"),
"list must render the client user, got: {stdout}"
);
assert!(
stdout.contains("version: v7"),
"list must render the client version, got: {stdout}"
);
assert!(
stdout.contains("ssh ip from: 203.0.113.9"),
"list must render the remote ssh ip, got: {stdout}"
);
ShutdownGateway::call(&*client, true).await.unwrap();
let _ = daemon.wait();
}
#[tokio::test]
async fn cli_stop_requires_force_when_live_sessions() {
let gateway = unique_gateway("stop_force");
let channel = "test/stop_force";
let (mut child, _marker) = spawn_daemon(&gateway, false);
let client = wait_connectable(&gateway).await;
attach_to(&client, channel, "cli").await;
Spawn::call(
&*client,
SpawnRequest {
cmd: Some(vec![
mock_bin().to_string_lossy().to_string(),
"sleep".into(),
"60000".into(),
]),
cols: 80u16,
rows: 24u16,
cwd: None,
},
)
.await
.unwrap();
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.arg("stop")
.output()
.expect("run stop");
assert!(
!out.status.success(),
"stop must refuse while a live session runs"
);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stderr.contains("live session"),
"refusal message should mention live sessions, got: {stderr}"
);
ListChannels::call(&*client, ())
.await
.expect("gateway alive");
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["stop", "--force"])
.output()
.expect("run stop --force");
assert!(
out.status.success(),
"stop --force failed: {}",
String::from_utf8_lossy(&out.stderr)
);
let _ = child.wait();
}
#[cfg(unix)]
#[test]
fn connection_error_is_printed_to_stderr() {
use interprocess::local_socket::{GenericNamespaced, ListenerOptions, ToNsName, prelude::*};
use std::io::Write;
let gateway = unique_gateway("silent-exit");
let name = gateway
.as_str()
.to_ns_name::<GenericNamespaced>()
.expect("gateway ns name");
let listener = ListenerOptions::new()
.name(name)
.try_overwrite(true)
.create_sync()
.expect("bind dummy gateway");
let acceptor = std::thread::spawn(move || {
while let Ok(mut stream) = listener.accept() {
let _ = stream.write_all(b"not-a-muxio-frame");
}
});
let out = Command::new(bin())
.env("TERM_WM_GATEWAY", &gateway)
.args(["--channel", "test/silent-exit"])
.output()
.expect("run client");
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
!stderr.trim().is_empty(),
"client must emit a diagnostic on a failed connection (silent-exit regression), got empty stderr"
);
assert!(
!out.status.success(),
"client must exit non-zero, got status: {:?}",
out.status.code()
);
drop(acceptor);
}