waos-dev-protocol 0.1.0

//! The messages over a real socket pair, after the two sides agreed on the protocol, a client of
//! another build refused, and what the listener does with the socket file: a stale one replaced,
//! a second WaOS refused, and the connection closing seen as none.

#![cfg(unix)]

use std::fs;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream;
use waos_dev_protocol::{Connection, FromWaOS, Listener, OutputStream, PROTOCOL, RunEnd, ToWaOS};

/// A folder of this test's own, with its socket's path inside; removed with the folder.
struct TestFolder(PathBuf);

impl TestFolder {
    fn new(test: &str) -> TestFolder {
        let folder = std::env::temp_dir().join(format!("waos-dev-{}-{test}", std::process::id()));
        let _ = fs::remove_dir_all(&folder);
        TestFolder(folder)
    }

    fn socket(&self) -> PathBuf {
        self.0.join("dev.sock")
    }
}

impl Drop for TestFolder {
    fn drop(&mut self) {
        let _ = fs::remove_dir_all(&self.0);
    }
}

/// A listener at `path` and a connection to it, with the listener's end of that connection,
/// the two sides agreed on the protocol.
async fn connected(path: &Path) -> (Listener, Connection<UnixStream>, Connection<UnixStream>) {
    let listener = Listener::bind(path).await.unwrap();
    let (client, server) = tokio::join!(Connection::connect(path), listener.accept());
    let (mut client, mut server) = (client.unwrap(), server.unwrap());
    let (agreed, agreed_too) = tokio::join!(client.agree_on_protocol(), server.agree_on_protocol());
    agreed.unwrap();
    agreed_too.unwrap();
    (listener, client, server)
}

/// A client built from another waos-dev-protocol, one that speaks protocol `PROTOCOL + 1`: it sends its
/// number and gets WaOS's back, the one frame every build reads, and WaOS refuses it naming
/// both numbers and closes.
#[tokio::test]
async fn a_client_of_another_build_is_refused_with_both_numbers() {
    let folder = TestFolder::new("other-build");
    let socket = folder.socket();
    let listener = Listener::bind(&socket).await.unwrap();
    let (client, server) = tokio::join!(UnixStream::connect(&socket), listener.accept());
    let (mut client, mut server) = (client.unwrap(), server.unwrap());
    let theirs = PROTOCOL + 1;
    let mut their_frame = 4u32.to_le_bytes().to_vec();
    their_frame.extend_from_slice(&theirs.to_le_bytes());
    client.write_all(&their_frame).await.unwrap();

    let refused = server.agree_on_protocol().await.unwrap_err();
    assert_eq!(refused.kind(), ErrorKind::Unsupported);
    assert!(
        refused.to_string().contains(&format!("protocol {theirs}"))
            && refused
                .to_string()
                .contains(&format!("protocol {PROTOCOL}")),
        "{refused}"
    );
    drop(server);

    let mut ours = [0; 8];
    client.read_exact(&mut ours).await.unwrap();
    assert_eq!(&ours[..4], 4u32.to_le_bytes());
    assert_eq!(u32::from_le_bytes(ours[4..].try_into().unwrap()), PROTOCOL);
    assert_eq!(client.read(&mut [0; 1]).await.unwrap(), 0, "WaOS closed");
}

fn every_from_waos() -> Vec<FromWaOS> {
    vec![
        FromWaOS::Started,
        FromWaOS::Output {
            stream: OutputStream::Stdout,
            bytes: b"hello, ".to_vec(),
        },
        FromWaOS::Output {
            stream: OutputStream::Stderr,
            bytes: b"a warning\n".to_vec(),
        },
        FromWaOS::Ended(RunEnd::Returned { success: true }),
        FromWaOS::Ended(RunEnd::Returned { success: false }),
        FromWaOS::Ended(RunEnd::Failed("unreachable".into())),
        FromWaOS::Ended(RunEnd::Stopped),
        FromWaOS::Ended(RunEnd::Replaced),
        FromWaOS::Refused("not a component".into()),
    ]
}

#[tokio::test]
async fn every_message_comes_back_as_sent() {
    let folder = TestFolder::new("round-trip");
    let (_listener, mut client, mut server) = connected(&folder.socket()).await;

    let run = ToWaOS::Run {
        id: "dev.example.triangle".parse().unwrap(),
        component: (0..=255).cycle().take(3 << 20).collect(),
    };
    let (sent, received) = tokio::join!(client.send(&run), server.receive::<ToWaOS>());
    sent.unwrap();
    assert_eq!(received.unwrap().as_ref(), Some(&run));

    for message in every_from_waos() {
        let (sent, received) = tokio::join!(server.send(&message), client.receive::<FromWaOS>());
        sent.unwrap();
        assert_eq!(received.unwrap(), Some(message));
    }
}

#[tokio::test]
async fn the_halves_send_while_they_watch_for_the_close() {
    let folder = TestFolder::new("halves");
    let (_listener, mut client, server) = connected(&folder.socket()).await;
    let (mut server_in, mut server_out) = server.into_split();

    let watching = tokio::spawn(async move { server_in.receive::<ToWaOS>().await });
    server_out.send(&FromWaOS::Started).await.unwrap();
    assert_eq!(
        client.receive::<FromWaOS>().await.unwrap(),
        Some(FromWaOS::Started)
    );
    drop(client);
    assert_eq!(watching.await.unwrap().unwrap(), None, "the close is seen");
}

#[tokio::test]
async fn the_other_side_closing_is_none() {
    let folder = TestFolder::new("closing");
    let (_listener, client, mut server) = connected(&folder.socket()).await;
    drop(client);
    assert_eq!(server.receive::<ToWaOS>().await.unwrap(), None);
}

#[tokio::test]
async fn a_socket_file_a_dead_waos_left_is_replaced() {
    let folder = TestFolder::new("stale");
    let socket = folder.socket();
    fs::create_dir_all(&folder.0).unwrap();
    // Bound and dropped, as a WaOS that died leaves it: the file stays, nobody answers.
    drop(std::os::unix::net::UnixListener::bind(&socket).unwrap());
    assert!(socket.exists());

    let (_listener, mut client, mut server) = connected(&socket).await;
    let (sent, received) = tokio::join!(server.send(&FromWaOS::Started), client.receive());
    sent.unwrap();
    assert_eq!(received.unwrap(), Some(FromWaOS::Started));
}

#[tokio::test]
async fn a_second_waos_is_refused_while_the_first_answers() {
    let folder = TestFolder::new("second");
    let socket = folder.socket();
    let first = Listener::bind(&socket).await.unwrap();
    let (refused, probe) = tokio::join!(Listener::bind(&socket), first.accept());
    assert_eq!(refused.unwrap_err().kind(), ErrorKind::AddrInUse);
    let mut probe = probe.unwrap();
    assert_eq!(
        probe.receive::<ToWaOS>().await.unwrap(),
        None,
        "the probe is a connection that asked nothing"
    );
}

#[tokio::test]
async fn a_listener_dropped_leaves_no_socket_file() {
    let folder = TestFolder::new("dropped");
    let socket = folder.socket();
    drop(Listener::bind(&socket).await.unwrap());
    assert!(!socket.exists());
    assert_eq!(
        Connection::connect(&socket).await.unwrap_err().kind(),
        ErrorKind::NotFound
    );
}