koh 0.2.0

koh — a resilient peer-to-peer remote shell: mosh, rewritten in Rust over iroh
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

//! Real-PTY / real-shell tests for [`koh::pty::Pty`].
//!
//! These live in their own integration-test binary (rather than inline `#[cfg(test)]`) on purpose:
//! each spawns a real child + PTY + two pump threads, and running them alongside the ~100 inline
//! unit/property tests in one massively-parallel binary starved the PTY reader thread under load
//! (a flaky timeout). A dedicated binary runs only these few in parallel, so they stay reliable.

// Integration test: a failed unwrap/expect/assert/timeout IS the test failing.
#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    reason = "integration test code; panics are assertion failures"
)]

use std::time::Duration;

use koh::pty::Pty;

#[tokio::test]
#[allow(
    clippy::match_wild_err_arm,
    reason = "a timeout in this test IS the test failing; panicking on the `Err(_)` deadline arm is the intended assertion"
)]
async fn spawns_and_streams_output() {
    // Run a one-shot command in the PTY and confirm we receive its output + reap it.
    let (mut pty, mut rx) = Pty::spawn(24, 80, Some("echo"), "xterm-256color").expect("spawn echo");
    // `echo` with no args prints just a newline; assert we get *something* and EOF.
    let mut collected = Vec::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
    loop {
        match tokio::time::timeout_at(deadline, rx.recv()).await {
            Ok(Some(chunk)) => collected.extend_from_slice(&chunk),
            Ok(None) => break, // channel closed: child exited and reader finished
            Err(_) => panic!("timed out waiting for pty output"),
        }
    }
    // `echo` prints a newline (CR/LF in a pty).
    assert!(
        collected.contains(&b'\n'),
        "expected a newline from echo, got {collected:?}"
    );
    // Child should be reapable.
    let status = pty.wait().expect("wait");
    assert!(status.success() || status.exit_code() == 0);
}

#[tokio::test]
#[allow(
    clippy::match_same_arms,
    reason = "channel-close (`Ok(None)`) and deadline (`Err(_)`) are conceptually distinct outcomes kept as separate arms for readability, even though both set `found = false`"
)]
async fn interactive_shell_echoes_input() {
    // Spawn the default shell, send a command, and verify the echoed output comes back.
    let (mut pty, mut rx) = Pty::spawn(24, 80, None, "xterm-256color").expect("spawn shell");
    // Give the shell a moment to start, then type a command that prints a marker.
    tokio::time::sleep(Duration::from_millis(300)).await;
    pty.write_input(b"printf KOH_MARKER_OK\n").expect("write");

    let mut collected = Vec::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
    let found = loop {
        match tokio::time::timeout_at(deadline, rx.recv()).await {
            Ok(Some(chunk)) => {
                collected.extend_from_slice(&chunk);
                if String::from_utf8_lossy(&collected).contains("KOH_MARKER_OK") {
                    break true;
                }
            }
            Ok(None) => break false,
            Err(_) => break false,
        }
    };
    // Resize should not error while the shell is live.
    let _ = pty.resize(40, 120);
    let _ = pty.kill();
    assert!(
        found,
        "did not observe the marker in shell output: {}",
        String::from_utf8_lossy(&collected)
    );
}

#[tokio::test]
#[allow(
    clippy::match_same_arms,
    reason = "channel-close (`Ok(None)`) and deadline (`Err(_)`) are conceptually distinct outcomes kept as separate arms for readability, even though both set `in_order = false`"
)]
async fn write_input_takes_shared_ref_and_preserves_order() {
    // `pty` is bound WITHOUT `mut`, proving write_input takes `&self`. Two separate enqueues must
    // reach the child in FIFO order: the concatenated marker only appears if the second chunk did
    // not overtake the first.
    let (pty, mut rx) = Pty::spawn(24, 80, None, "xterm-256color").expect("spawn shell");
    tokio::time::sleep(Duration::from_millis(300)).await;
    pty.write_input(b"printf ORDER_").expect("first enqueue");
    pty.write_input(b"AB_CD\n").expect("second enqueue");

    let mut collected = Vec::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
    let in_order = loop {
        match tokio::time::timeout_at(deadline, rx.recv()).await {
            Ok(Some(chunk)) => {
                collected.extend_from_slice(&chunk);
                if String::from_utf8_lossy(&collected).contains("ORDER_AB_CD") {
                    break true;
                }
            }
            Ok(None) => break false,
            Err(_) => break false,
        }
    };
    drop(pty);
    assert!(
        in_order,
        "FIFO ordering of two enqueues should yield ORDER_AB_CD; got: {}",
        String::from_utf8_lossy(&collected)
    );
}

#[tokio::test]
#[allow(
    clippy::needless_continue,
    clippy::match_wild_err_arm,
    reason = "the explicit `continue` documents the drain-and-keep-reading intent; the `Err(_)` deadline arm panics because a timeout here IS the test failing"
)]
async fn dropping_pty_eofs_child_and_stops_writer() {
    // `cat` blocks reading stdin. Dropping the Pty drops the writer-thread sender; the writer thread
    // then finishes and drops the PTY write handle, on which portable-pty sends EOT — so the child
    // sees EOF, exits, the slave closes, and the output channel ends. If the writer thread were
    // stuck (or never dropped its handle), the channel would never close.
    let (pty, mut rx) = Pty::spawn(24, 80, Some("cat"), "xterm-256color").expect("spawn cat");
    tokio::time::sleep(Duration::from_millis(200)).await;
    drop(pty); // no kill(): EOF must come purely from the writer handle being dropped

    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
    loop {
        match tokio::time::timeout_at(deadline, rx.recv()).await {
            Ok(Some(_)) => continue, // drain any echoed bytes
            Ok(None) => break,       // channel closed: child EOF'd + exited; writer thread ended
            Err(_) => panic!("dropping Pty did not EOF the child (writer stuck?)"),
        }
    }
}

#[tokio::test]
async fn shutdown_joins_both_io_threads_without_deadlock() {
    // Graceful teardown: shutdown() kills the child (so the reader's blocking read returns EOF) and
    // drops the writer sender (so the writer's recv returns), then joins BOTH pump threads. It must
    // return promptly — a hang would mean a thread never unblocked.
    let (pty, mut rx) = Pty::spawn(24, 80, Some("sh"), "xterm-256color").expect("spawn shell");
    // Keep the output channel drained so the reader thread never blocks on a full channel.
    let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} });
    tokio::time::sleep(Duration::from_millis(200)).await;

    tokio::time::timeout(
        Duration::from_secs(20),
        tokio::task::spawn_blocking(move || pty.shutdown()),
    )
    .await
    .expect("shutdown must not deadlock (both threads must unblock and join)")
    .expect("shutdown task panicked");
    let _ = drain.await;
}