boxmaker/crates/loopd/tests/broker_port.rs

//! Tests for `BrokerPort` against a fake `brokerd` on a temporary socket: answers, approvals and
//! timeouts. Frames that break the protocol are in `broker_port_bad.rs`. Do not edit.

#[path = "support/broker.rs"]
mod fake;

use std::io::Write;
use std::os::unix::net::UnixListener;
use std::thread;
use std::time::Duration;

use fake::{assert_unavailable, broker, call, frame, in_ms, request, result, send, socket_path};
use loopd::broker_port::{BrokerPort, NoBroker, POINTER, TOO_LARGE};
use loopd::tools::{Pending, ToolPort};
use proto::{
    DenyReason, Message, PROTOCOL_VERSION, Timestamp, ToolResponse, read_frame, write_frame,
};

#[test]
fn a_result_comes_back_as_it_was_sent() {
    let (socket, broker) = broker(|stream, request| {
        send(stream, request.id, true, result("hello\n"));
    });
    let got = call(socket, 2_000, &request());
    assert_eq!(got.response, result("hello\n"));
    assert!(got.pending.is_empty());
    assert!(got.lines.is_empty(), "nothing to report: {:?}", got.lines);

    // What the broker received: one final frame holding exactly the request.
    let sent = broker.join().unwrap();
    assert_eq!(sent.v, PROTOCOL_VERSION);
    assert!(sent.r#final, "a request is a single final frame");
    assert_eq!(sent.msg, Message::ToolRequest(request()));
}

#[test]
fn every_denial_and_a_failure_come_back_as_they_were_sent() {
    let mut answers: Vec<ToolResponse> = [
        DenyReason::NoGrant,
        DenyReason::GrantExpired,
        DenyReason::TaintTooHigh,
        DenyReason::DeniedByGrant,
        DenyReason::ApprovalRefused,
        DenyReason::ApprovalExpired,
        DenyReason::GrantsInvalid,
        DenyReason::AuditUnavailable,
        DenyReason::InvalidArguments,
        DenyReason::StateUnreadable,
    ]
    .into_iter()
    .map(|reason| ToolResponse::Denied { reason })
    .collect();
    answers.push(ToolResponse::Failed {
        message: "the runner arrives in M3b".to_string(),
    });
    for answer in answers {
        let reply = answer.clone();
        let (socket, broker) = broker(move |stream, request| {
            send(stream, request.id, true, reply);
        });
        let got = call(socket, 2_000, &request());
        assert_eq!(got.response, answer);
        assert!(
            got.lines.is_empty(),
            "a denial is not an outage: {:?}",
            got.lines
        );
        broker.join().unwrap();
    }
}

#[test]
fn a_pending_frame_reaches_the_callback_and_the_final_frame_is_the_answer() {
    let expires = in_ms(60_000);
    let (socket, broker) = broker(move |stream, request| {
        let pending = ToolResponse::PendingApproval {
            approval: 41,
            expires,
        };
        send(stream, request.id, false, pending);
        thread::sleep(Duration::from_millis(150));
        send(stream, request.id, true, result("approved"));
    });
    let got = call(socket, 2_000, &request());
    assert_eq!(got.response, result("approved"));
    assert_eq!(
        got.pending,
        [Pending {
            approval: 41,
            expires
        }],
        "called once, with the frame's values"
    );
    assert!(got.lines.is_empty(), "{:?}", got.lines);
    broker.join().unwrap();
}

#[test]
fn an_answer_after_expires_but_within_the_timeout_of_it_is_delivered() {
    // expires in 100 ms, timeout 1,500 ms: the port waits until 1,600 ms. The answer comes at
    // about 400 ms, well after `expires`: an approval given at the last moment still gets the
    // whole timeout to run.
    let (socket, broker) = broker(|stream, request| {
        let pending = ToolResponse::PendingApproval {
            approval: 1,
            expires: in_ms(100),
        };
        send(stream, request.id, false, pending);
        thread::sleep(Duration::from_millis(400));
        send(stream, request.id, true, result("late but good"));
    });
    let got = call(socket, 1_500, &request());
    assert_eq!(got.response, result("late but good"));
    assert!(got.lines.is_empty(), "{:?}", got.lines);
    broker.join().unwrap();
}

#[test]
fn an_expiry_that_has_already_passed_still_leaves_the_timeout() {
    let (socket, broker) = broker(|stream, request| {
        let pending = ToolResponse::PendingApproval {
            approval: 1,
            expires: Timestamp::parse("2020-01-01T00:00:00.000Z").unwrap(),
        };
        send(stream, request.id, false, pending);
        thread::sleep(Duration::from_millis(200));
        send(stream, request.id, true, result("fine"));
    });
    let got = call(socket, 1_500, &request());
    assert_eq!(got.response, result("fine"));
    broker.join().unwrap();
}

#[test]
fn no_socket_is_unavailable() {
    let socket = socket_path();
    let _ = std::fs::remove_file(&socket);
    let got = call(socket.to_path_buf(), 2_000, &request());
    assert_unavailable(&got, "no socket file");
    assert!(
        got.lines[0].contains(&socket.display().to_string()),
        "the line names the socket: {}",
        got.lines[0]
    );
    assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took);
}

#[test]
fn a_broker_that_closes_without_answering_is_unavailable() {
    let (socket, broker) = broker(|_, _| {});
    let got = call(socket, 2_000, &request());
    assert_unavailable(&got, "closed before any frame");
    assert!(
        got.took < Duration::from_millis(1_500),
        "a close is seen at once: {:?}",
        got.took
    );
    broker.join().unwrap();
}

#[test]
fn a_broker_that_closes_while_pending_is_unavailable() {
    let (socket, broker) = broker(|stream, request| {
        let pending = ToolResponse::PendingApproval {
            approval: 3,
            expires: in_ms(60_000),
        };
        send(stream, request.id, false, pending);
        // brokerd was restarted: the connection just ends.
    });
    let got = call(socket, 2_000, &request());
    assert_unavailable(&got, "closed while pending");
    assert_eq!(got.pending.len(), 1, "the pending frame was reported first");
    assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took);
    broker.join().unwrap();
}

#[test]
fn a_broker_that_never_answers_is_unavailable_after_the_timeout() {
    let (socket, broker) = broker(|_, _| thread::sleep(Duration::from_millis(1_200)));
    let got = call(socket, 300, &request());
    assert_unavailable(&got, "silence");
    assert!(
        got.took >= Duration::from_millis(250),
        "gave up early: {:?}",
        got.took
    );
    assert!(
        got.took < Duration::from_millis(1_100),
        "gave up late: {:?}",
        got.took
    );
    broker.join().unwrap();
}

#[test]
fn a_pending_call_nobody_answers_ends_at_expires_plus_the_timeout() {
    let (socket, broker) = broker(|stream, request| {
        let pending = ToolResponse::PendingApproval {
            approval: 3,
            expires: in_ms(300),
        };
        send(stream, request.id, false, pending);
        thread::sleep(Duration::from_millis(1_800));
    });
    let got = call(socket, 300, &request());
    assert_unavailable(&got, "silence while pending");
    assert!(
        got.took >= Duration::from_millis(550),
        "it must wait for expires (300) plus the timeout (300): {:?}",
        got.took
    );
    assert!(
        got.took < Duration::from_millis(1_700),
        "gave up late: {:?}",
        got.took
    );
    broker.join().unwrap();
}

#[test]
fn the_timeout_is_a_deadline_for_the_frame_not_for_each_read() {
    // The four header bytes arrive 250 ms apart, the last at 1,000 ms. A port that only sets a
    // 600 ms read timeout sees every single read succeed and returns the result; a port with a
    // deadline gives up at 600 ms. The answer tells them apart, so no timing is asserted.
    let (socket, broker) = broker(|stream, request| {
        let mut bytes = Vec::new();
        write_frame(
            &mut bytes,
            &frame(request.id, true, Message::ToolResponse(result("slow"))),
        )
        .unwrap();
        for byte in bytes.iter().take(4) {
            thread::sleep(Duration::from_millis(250));
            if stream.write_all(&[*byte]).is_err() {
                return;
            }
        }
        let _ = stream.write_all(&bytes[4..]);
    });
    let got = call(socket, 600, &request());
    assert_unavailable(&got, "a trickled frame");
    broker.join().unwrap();
}

#[test]
fn a_zero_timeout_fails_closed_and_does_not_panic() {
    let (socket, _broker) = broker(|stream, request| {
        send(stream, request.id, true, result("too late"));
    });
    let got = call(socket, 0, &request());
    assert_unavailable(&got, "timeout_ms = 0");
}

#[test]
fn a_request_too_large_for_a_frame_is_its_own_failure() {
    // Nothing is sent, so the fake broker sees a connection that closes or none at all.
    let path = socket_path();
    let _ = std::fs::remove_file(&path);
    let _listener = UnixListener::bind(&path).unwrap();
    let mut big = request();
    big.arguments = format!(r#"{{"path":"/{}"}}"#, "a".repeat(proto::MAX_FRAME));
    let got = call(path.to_path_buf(), 1_000, &big);
    assert_eq!(
        got.response,
        ToolResponse::Failed {
            message: TOO_LARGE.to_string()
        }
    );
    assert!(
        got.lines.is_empty(),
        "the broker is fine; this is not an outage: {:?}",
        got.lines
    );
}

#[test]
fn every_call_is_its_own_connection() {
    let path = socket_path();
    let _ = std::fs::remove_file(&path);
    let listener = UnixListener::bind(&path).unwrap();
    let server = thread::spawn(move || {
        for n in 0..3u64 {
            let (mut stream, _) = listener.accept().unwrap();
            let request = read_frame(&mut stream).unwrap();
            send(
                &mut stream,
                request.id,
                true,
                result(&format!("answer {n}")),
            );
        }
    });
    let port = BrokerPort::new(path.to_path_buf(), Duration::from_millis(2_000));
    for n in 0..3 {
        let got = port.call(&request(), &mut |_| {});
        assert_eq!(got, result(&format!("answer {n}")));
    }
    server.join().unwrap();
}

#[test]
fn without_a_broker_every_call_fails_in_words_the_model_can_repeat() {
    let mut seen = 0;
    let got = NoBroker.call(&request(), &mut |_| seen += 1);
    assert_eq!(
        got,
        ToolResponse::Failed {
            message: "no tool broker is configured".to_string()
        }
    );
    assert_eq!(seen, 0);
    assert_eq!(POINTER, "see docs/runbook.md#broker-unavailable");
}