oc2_hlapi_v2 0.2.0

//! The bus daemon client transport: handshake, refusals, failures, and the full stack on top.

mod common;

use std::collections::VecDeque;
use std::io;
use std::sync::atomic::Ordering;
use std::time::Duration;

use common::*;
use oc2_hlapi_v2::Transport;
use oc2_hlapi_v2::client::HlapiClient;
use oc2_hlapi_v2::codecs::BaseCodec;
use oc2_hlapi_v2::transports::{BusdClient, BusdClientError, BusdStream};
use serde_json::{Value, json};

const TOKEN: &str = "00112233445566778899aabbccddeeff";

/// Plays the daemon's side of the unix socket: every `open_streams` hands the client one end of
/// a fresh connection, and the test holds the other ends.
struct Daemon {
    conns: VecDeque<MemStream>,
    opens: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}

impl Transport for Daemon {
    type Streams = BusdStream<MemStream>;
    type Error = io::Error;
    fn open_streams(&mut self) -> io::Result<Self::Streams> {
        self.opens.fetch_add(1, Ordering::SeqCst);
        self.conns
            .pop_front()
            .map(BusdStream)
            .ok_or_else(|| io::Error::other("daemon unreachable"))
    }
}

/// Three connections (rpc, blob, event). The daemon end of the rpc connection already holds
/// `hello_reply` (None: the daemon says nothing).
fn session(kind: TimeoutKind, hello_reply: Option<&[u8]>) -> (Daemon, Host) {
    let (rpc_c, rpc_h) = duplex(kind);
    let (blob_c, blob_h) = duplex(kind);
    let (event_c, event_h) = duplex(kind);
    if let Some(reply) = hello_reply {
        rpc_h.send(reply).unwrap();
    }
    (
        Daemon {
            conns: [rpc_c, blob_c, event_c].into(),
            opens: Default::default(),
        },
        Host {
            rpc: rpc_h,
            blob: blob_h,
            event: event_h,
        },
    )
}

fn hello(extra: Value) -> Vec<u8> {
    let mut v = json!({"type": "hello", "version": 1, "token": TOKEN});
    for (k, val) in extra.as_object().unwrap() {
        v[k] = val.clone();
    }
    framed(&v)
}

fn connect(
    d: Daemon,
) -> Result<BaseCodec<MemStream, BusdClient<MemStream, Daemon>>, BusdClientError<io::Error>> {
    BaseCodec::new(BusdClient::new(d))
}

#[test]
fn the_handshake_sends_the_hello_then_opens_blob_and_event_with_the_token() {
    for kind in KINDS {
        let (d, host) = session(kind, Some(&hello(json!({"gen": 3}))));
        let _codec = connect(d).unwrap();
        let sent = host.rpc.available();
        assert_eq!(
            &sent[..2],
            b"R\0",
            "role byte, then the NUL framing the hello"
        );
        assert_eq!(*sent.last().unwrap(), 0);
        let hello: Value = serde_json::from_slice(&sent[2..sent.len() - 1]).unwrap();
        assert_eq!(hello["type"], "hello");
        assert_eq!(hello["version"], 1);
        assert_eq!(hello["roles"], json!(["rpc", "blob", "event"]));
        assert_eq!(host.blob.available(), format!("B{TOKEN}").as_bytes());
        assert_eq!(host.event.available(), format!("E{TOKEN}").as_bytes());
    }
}

#[test]
fn a_hello_without_gen_is_fine() {
    let (d, _host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({}))));
    connect(d).unwrap();
}

#[test]
fn the_hello_reply_may_arrive_in_pieces() {
    let (d, host) = session(TimeoutKind::WouldBlock, None);
    let reply = hello(json!({"gen": 1}));
    let writer = std::thread::spawn(move || {
        for chunk in reply.chunks(5) {
            host.rpc.send(chunk).unwrap();
            std::thread::sleep(Duration::from_millis(3));
        }
        host
    });
    let r = within(move || connect(d).map(|_| ()));
    r.unwrap();
    drop(writer.join().unwrap());
}

#[test]
fn a_refusal_reports_the_daemons_reason() {
    let reply =
        framed(&json!({"type": "error", "gen": 1, "data": "unsupported protocol version 2"}));
    let (d, _host) = session(TimeoutKind::WouldBlock, Some(&reply));
    let err = connect(d).map(|_| ()).unwrap_err();
    assert!(
        matches!(&err, BusdClientError::ErrorOnHello(m) if m == "unsupported protocol version 2"),
        "{err:?}"
    );
    assert!(err.to_string().contains("unsupported protocol version 2"));
}

#[test]
fn a_refusal_opens_no_further_connections() {
    let reply = framed(&json!({"type": "error", "data": "nope"}));
    let (d, host) = session(TimeoutKind::WouldBlock, Some(&reply));
    let opens = d.opens.clone();
    assert!(connect(d).is_err());
    assert_eq!(opens.load(Ordering::SeqCst), 1);
    assert!(host.blob.available().is_empty());
}

#[test]
fn garbage_instead_of_a_hello_is_a_bad_hello() {
    for reply in [
        &b"\0not json at all\0"[..],
        &b"\0{\"type\":\"hello\"}\0"[..],          // missing fields
        &b"\0{\"type\":\"something else\"}\0"[..], // unknown message
        &b"\0[1,2,3]\0"[..],
    ] {
        let (d, _host) = session(TimeoutKind::WouldBlock, Some(reply));
        let err = connect(d).map(|_| ()).unwrap_err();
        assert!(
            matches!(err, BusdClientError::BadHello(_)),
            "{reply:?}: {err:?}"
        );
    }
}

#[test]
fn a_daemon_that_hangs_up_is_an_io_error() {
    let (d, host) = session(TimeoutKind::WouldBlock, None);
    drop(host);
    let err = connect(d).map(|_| ()).unwrap_err();
    // Writing to a closed socket or reading EOF, depending on the platform
    assert!(matches!(err, BusdClientError::Io(_)), "{err:?}");
}

#[test]
fn a_daemon_that_dies_mid_hello_is_an_error() {
    let (d, host) = session(TimeoutKind::WouldBlock, Some(b"\0{\"type\":\"hel"));
    let h = std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(30));
        drop(host);
    });
    let r = within(move || connect(d).map(|_| ()));
    h.join().unwrap();
    assert!(
        matches!(
            r,
            Err(BusdClientError::BadHello(_) | BusdClientError::Io(_))
        ),
        "{r:?}"
    );
}

#[test]
fn connection_failures_are_transport_errors() {
    // Cannot reach the daemon at all
    let d = Daemon {
        conns: VecDeque::new(),
        opens: Default::default(),
    };
    let err = connect(d).map(|_| ()).unwrap_err();
    assert!(matches!(err, BusdClientError::Transport(_)), "{err:?}");

    // Reaches it for the rpc connection but not for the other two
    let (mut d, _host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({}))));
    d.conns.truncate(1);
    let err = connect(d).map(|_| ()).unwrap_err();
    assert!(matches!(err, BusdClientError::Transport(_)), "{err:?}");
}

#[test]
fn the_streams_work_after_the_handshake() {
    let (d, host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({"gen": 1}))));
    let mut codec = connect(d).unwrap();
    host.rpc.available();
    host.blob.available();
    host.event.available();

    use oc2_hlapi_v2::Codec;
    use oc2_hlapi_v2::messages::*;
    codec
        .send_request(
            Request {
                id: Some(1),
                data: RequestData::List,
            },
            &[b"blob"],
        )
        .unwrap();
    assert_eq!(
        host.rpc.json(Some(LONG)).unwrap().unwrap(),
        json!({"type": "list", "id": 1})
    );
    assert_eq!(host.blob.available(), b"blob");

    host.reply(&json!({"type": "list", "id": 1, "data": []}));
    let r = codec.read_response(Some(LONG)).unwrap().unwrap();
    assert_eq!(r.id, 1);
    host.emit(&devices_changed(9));
    assert_eq!(codec.read_event(Some(LONG)).unwrap().unwrap().generation, 9);
}

#[test]
fn the_codec_redoes_the_whole_handshake_when_it_reconnects() {
    let (first, first_host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({}))));
    let (second, second_host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({"gen": 2}))));
    let opens = first.opens.clone();
    let mut conns = first.conns;
    conns.extend(second.conns);
    let mut codec = connect(Daemon {
        conns,
        opens: opens.clone(),
    })
    .unwrap();
    assert_eq!(opens.load(Ordering::SeqCst), 3);

    use oc2_hlapi_v2::Codec;
    drop(first_host); // the daemon restarts
    let err = codec.read_response(Some(LONG)).unwrap_err();
    assert!(
        err.to_string().contains("EOF") || err.to_string().contains("I/O"),
        "{err}"
    );
    assert_eq!(opens.load(Ordering::SeqCst), 6, "rpc + blob + event again");

    // Talks to the second daemon connection now
    assert!(second_host.rpc.available().starts_with(b"R\0"));
    assert_eq!(second_host.blob.available(), format!("B{TOKEN}").as_bytes());
    second_host.emit(&devices_changed(5));
    assert_eq!(codec.read_event(Some(LONG)).unwrap().unwrap().generation, 5);
}

#[test]
fn the_full_stack_works_over_the_daemon() {
    let (d, host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({"gen": 1}))));
    let codec = connect(d).unwrap();
    // Throw away the handshake bytes before the scripted host starts reading requests
    host.rpc.available();
    let _server = serve(host, simple_host);
    let client = HlapiClient::new(codec);
    assert_eq!(client.list_devices().unwrap().len(), 2);
    let d: Redstone = client.find_device_by_type().unwrap().unwrap();
    assert_eq!(d.get_input("left").unwrap(), 7);
}

#[test]
fn bus_daemon_replies_without_ids_are_accepted() {
    // busd leaves the `id` out of its reply when the request had none; with ids it echoes ours
    let (d, host) = session(TimeoutKind::WouldBlock, Some(&hello(json!({}))));
    let codec = connect(d).unwrap();
    host.rpc.available();
    let _server = serve(host, |req, _, host| {
        let mut reply = list_reply(req, 1, &[]);
        reply.as_object_mut().unwrap().remove("id");
        host.reply(&reply);
    });
    let client = HlapiClient::new(codec);
    assert!(client.list_devices().unwrap().is_empty());
}