running-process 4.10.16

Subprocess and PTY runtime for the running-process project
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

//! End-to-end test for the daemon SESSION backend endpoint (soldr#2365): over a
//! real `interprocess` local socket, the endpoint accepts a connection, reads
//! the client's `SessionStart`, runs the contained compile, and proxies its
//! stdio back. The command is carried on the wire; this crosses a real IPC
//! boundary through the production serve surface. Unix-first.

use futures_util::{SinkExt, StreamExt};

use super::serve_session_endpoint;
use crate::broker::protocol_v2::{session_frame, SessionFrame, SessionStart};
use crate::daemon::compile_session::session_framed;
use crate::platform::ipc::{AsyncListener, AsyncStream, Endpoint};

fn fixture_program() -> String {
    let exe = std::env::current_exe().expect("test executable path");
    let dir = exe
        .parent()
        .and_then(std::path::Path::parent)
        .expect("test binary should live in <profile>/deps/");
    dir.join(format!(
        "testbin-stdio-scripted{}",
        std::env::consts::EXE_SUFFIX
    ))
    .to_string_lossy()
    .into_owned()
}

fn frame(kind: session_frame::Kind) -> SessionFrame {
    SessionFrame { kind: Some(kind) }
}

#[cfg(unix)]
#[tokio::test]
async fn session_endpoint_serves_a_session_from_start_frame() {
    let path = std::env::temp_dir().join(format!("rp-endpoint-{}.sock", std::process::id()));
    let _ = std::fs::remove_file(&path);

    let ipc_endpoint = Endpoint::new(path.to_string_lossy().into_owned()).expect("IPC endpoint");
    let listener = AsyncListener::bind(&ipc_endpoint).expect("bind session endpoint");

    let endpoint = tokio::spawn(serve_session_endpoint(listener));

    // Client dials the real endpoint and speaks the SESSION wire.
    let stream = AsyncStream::connect(&ipc_endpoint)
        .await
        .expect("connect session endpoint");
    let mut client = session_framed(stream);

    client
        .send(frame(session_frame::Kind::Start(SessionStart {
            program: fixture_program(),
            args: vec![
                "out:HELLO".to_owned(),
                "err:WORLD".to_owned(),
                "exit:7".to_owned(),
            ],
            cwd: String::new(),
            env: Vec::new(),
            clear_inherited_env: false,
            environment_policy: 0,
        })))
        .await
        .expect("send start");
    client
        .send(frame(session_frame::Kind::StdinEof(true)))
        .await
        .expect("send stdin eof");

    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    let mut code = None;
    while let Some(Ok(sf)) = client.next().await {
        match sf.kind {
            Some(session_frame::Kind::Stdout(b)) => stdout.extend_from_slice(&b),
            Some(session_frame::Kind::Stderr(b)) => stderr.extend_from_slice(&b),
            Some(session_frame::Kind::Exit(e)) => {
                code = Some(e.code);
                break;
            }
            _ => panic!("unexpected inbound-only frame on the outbound lane"),
        }
    }

    endpoint.abort();
    let _ = std::fs::remove_file(&path);

    assert_eq!(stdout, b"HELLO", "stdout from the wire-carried command");
    assert_eq!(stderr, b"WORLD", "stderr from the wire-carried command");
    assert_eq!(code, Some(7), "exit code proxied over the real endpoint");
}