oc2_hlapi_v2 0.2.0

//! Linux-only pieces: the unix socket stream, `TimeoutFile`, `BusdUnixSocket` and the connection
//! helpers. Everything else is covered by the platform-independent tests.
#![cfg(target_os = "linux")]

use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::OwnedFd;
use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::Mutex;
use std::thread;
use std::time::{Duration, Instant};

use oc2_hlapi_v2::client::HlapiClient;
use oc2_hlapi_v2::codecs::{BaseCodec, BaseStreams};
use oc2_hlapi_v2::messages::*;
use oc2_hlapi_v2::transports::{BusdUnixSocket, LinuxVirtio, TimeoutFile};
use oc2_hlapi_v2::{Codec, Stream, Transport};
use serde_json::{Value, json};

/// Tests that touch process-wide environment variables must not overlap.
static ENV: Mutex<()> = Mutex::new(());

fn framed(v: &Value) -> Vec<u8> {
    let mut out = vec![0];
    out.extend(serde_json::to_vec(v).unwrap());
    out.push(0);
    out
}

fn read_frame(s: &mut impl Read) -> io::Result<Value> {
    let mut frame = Vec::new();
    loop {
        let mut b = [0];
        s.read_exact(&mut b)?;
        if b[0] == 0 {
            if frame.is_empty() {
                continue;
            }
            return Ok(serde_json::from_slice(&frame).unwrap());
        }
        frame.push(b[0]);
    }
}

// --- UnixStream ---------------------------------------------------------------------------------

struct PairTransport(Option<BaseStreams<UnixStream>>);

impl Transport for PairTransport {
    type Streams = BaseStreams<UnixStream>;
    type Error = io::Error;
    fn open_streams(&mut self) -> io::Result<Self::Streams> {
        self.0
            .take()
            .ok_or_else(|| io::Error::other("no more connections"))
    }
}

#[test]
fn the_codec_works_over_unix_stream_pairs() {
    let (rpc, mut rpc_h) = UnixStream::pair().unwrap();
    let (blob, _blob_h) = UnixStream::pair().unwrap();
    let (event, mut event_h) = UnixStream::pair().unwrap();
    let mut codec = BaseCodec::new(PairTransport(Some(BaseStreams { rpc, blob, event }))).unwrap();

    codec
        .send_request(
            Request {
                id: Some(1),
                data: RequestData::List,
            },
            &[],
        )
        .unwrap();
    assert_eq!(
        read_frame(&mut rpc_h).unwrap(),
        json!({"type": "list", "id": 1})
    );
    rpc_h
        .write_all(&framed(
            &json!({"type": "list", "id": 1, "gen": 4, "data": []}),
        ))
        .unwrap();
    assert_eq!(
        codec
            .read_response(Some(Duration::from_secs(10)))
            .unwrap()
            .unwrap()
            .generation,
        Some(4)
    );

    assert!(
        codec
            .read_event(Some(Duration::from_millis(50)))
            .unwrap()
            .is_none()
    );
    assert!(codec.read_event(Some(Duration::ZERO)).unwrap().is_none());
    event_h
        .write_all(&framed(&json!({"type": "devicesChanged", "gen": 2})))
        .unwrap();
    assert_eq!(
        codec
            .read_event(Some(Duration::from_secs(10)))
            .unwrap()
            .unwrap()
            .generation,
        2
    );
}

// --- TimeoutFile --------------------------------------------------------------------------------

/// A readable and writable `File` on one end of a socket pair, and the other end.
fn file_pair() -> (TimeoutFile, UnixStream) {
    let (a, b) = UnixStream::pair().unwrap();
    (TimeoutFile::from(File::from(OwnedFd::from(a))), b)
}

#[test]
fn timeout_file_reads_and_writes() {
    let (mut f, mut peer) = file_pair();
    f.write_all(b"out").unwrap();
    f.flush().unwrap();
    let mut buf = [0; 3];
    peer.read_exact(&mut buf).unwrap();
    assert_eq!(&buf, b"out");
    peer.write_all(b"in!").unwrap();
    f.read_exact(&mut buf).unwrap();
    assert_eq!(&buf, b"in!");
}

#[test]
fn timeout_file_read_timeout() {
    let (mut f, mut peer) = file_pair();
    f.set_read_timeout(Some(Duration::from_millis(80))).unwrap();
    let start = Instant::now();
    let err = f.read(&mut [0; 4]).unwrap_err();
    assert_eq!(err.kind(), io::ErrorKind::TimedOut);
    assert!(start.elapsed() >= Duration::from_millis(70));
    // Data that arrives within the timeout is returned
    peer.write_all(b"x").unwrap();
    assert_eq!(f.read(&mut [0; 4]).unwrap(), 1);
}

#[test]
fn timeout_file_without_a_timeout_blocks() {
    let (mut f, mut peer) = file_pair();
    f.set_read_timeout(Some(Duration::from_millis(10))).unwrap();
    f.set_read_timeout(None).unwrap();
    let writer = thread::spawn(move || {
        thread::sleep(Duration::from_millis(100));
        peer.write_all(b"late").unwrap();
        peer
    });
    let start = Instant::now();
    assert_eq!(f.read(&mut [0; 8]).unwrap(), 4);
    assert!(start.elapsed() >= Duration::from_millis(80));
    drop(writer.join().unwrap());
}

#[test]
fn timeout_file_rejects_a_zero_timeout() {
    let (f, _peer) = file_pair();
    let err = f.set_read_timeout(Some(Duration::ZERO)).unwrap_err();
    assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
}

#[test]
fn timeout_file_nonblocking_mode() {
    let (mut f, mut peer) = file_pair();
    f.set_nonblocking(true).unwrap();
    assert_eq!(
        f.read(&mut [0; 4]).unwrap_err().kind(),
        io::ErrorKind::WouldBlock
    );
    peer.write_all(b"ab").unwrap();
    assert_eq!(f.read(&mut [0; 4]).unwrap(), 2);
    f.set_nonblocking(false).unwrap();
    f.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
    assert_eq!(
        f.read(&mut [0; 4]).unwrap_err().kind(),
        io::ErrorKind::TimedOut
    );
}

#[test]
fn timeout_file_reports_eof() {
    let (mut f, peer) = file_pair();
    f.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
    drop(peer);
    assert_eq!(f.read(&mut [0; 4]).unwrap(), 0);
}

struct FileTransport(Option<BaseStreams<TimeoutFile>>);

impl Transport for FileTransport {
    type Streams = BaseStreams<TimeoutFile>;
    type Error = io::Error;
    fn open_streams(&mut self) -> io::Result<Self::Streams> {
        self.0
            .take()
            .ok_or_else(|| io::Error::other("no more connections"))
    }
}

#[test]
fn the_codec_works_over_timeout_files() {
    // The shape of the virtio transport: plain files with timeouts layered on by poll()
    let (rpc, mut rpc_h) = file_pair();
    let (blob, mut blob_h) = file_pair();
    let (event, mut event_h) = file_pair();
    let mut codec = BaseCodec::new(FileTransport(Some(BaseStreams { rpc, blob, event }))).unwrap();

    assert!(
        codec
            .read_event(Some(Duration::from_millis(60)))
            .unwrap()
            .is_none()
    );
    assert!(codec.read_event(Some(Duration::ZERO)).unwrap().is_none());
    event_h
        .write_all(&framed(&json!({"type": "devicesChanged", "gen": 6})))
        .unwrap();
    assert_eq!(
        codec
            .read_event(Some(Duration::from_secs(10)))
            .unwrap()
            .unwrap()
            .generation,
        6
    );

    codec
        .send_request(
            Request {
                id: Some(2),
                data: RequestData::List,
            },
            &[b"b"],
        )
        .unwrap();
    assert_eq!(read_frame(&mut rpc_h).unwrap()["id"], 2);
    let mut one = [0];
    blob_h.read_exact(&mut one).unwrap();
    rpc_h
        .write_all(&framed(&json!({"type": "list", "id": 2, "data": []})))
        .unwrap();
    assert_eq!(
        codec
            .read_response(Some(Duration::from_secs(10)))
            .unwrap()
            .unwrap()
            .id,
        2
    );

    blob_h.write_all(b"stale").unwrap();
    codec.clear_blob_stream().unwrap();
    blob_h.write_all(b"fresh").unwrap();
    assert_eq!(
        codec
            .read_blob(BlobDescriptor::describe(b"fresh").unwrap())
            .unwrap(),
        b"fresh"
    );
}

// --- BusdUnixSocket and the connection helpers --------------------------------------------------

fn socket_path(tag: &str) -> std::path::PathBuf {
    let mut p = std::env::temp_dir();
    p.push(format!(
        "oc2_hlapi_v2_test_{}_{tag}.sock",
        std::process::id()
    ));
    let _ = std::fs::remove_file(&p);
    p
}

/// A minimal bus daemon: does the handshake, then answers `list` requests.
fn fake_daemon(listener: UnixListener) -> thread::JoinHandle<()> {
    thread::spawn(move || {
        let (mut rpc, _) = listener.accept().unwrap();
        let mut role = [0];
        rpc.read_exact(&mut role).unwrap();
        assert_eq!(role[0], b'R');
        let hello = read_frame(&mut rpc).unwrap();
        assert_eq!(hello["type"], "hello");
        rpc.write_all(&framed(
            &json!({"type": "hello", "version": 1, "token": "tok", "gen": 1}),
        ))
        .unwrap();

        let (mut blob, _) = listener.accept().unwrap();
        let mut buf = [0; 4];
        blob.read_exact(&mut buf).unwrap();
        assert_eq!(&buf, b"Btok");
        let (mut event, _) = listener.accept().unwrap();
        event.read_exact(&mut buf).unwrap();
        assert_eq!(&buf, b"Etok");

        while let Ok(req) = read_frame(&mut rpc) {
            // busd gives the client back its own id
            let reply = json!({"type": "list", "id": req["id"], "gen": 1, "data": [
                {"deviceId": "00000000-0000-0000-0000-000000000009", "typeNames": ["redstone"]},
            ]});
            rpc.write_all(&framed(&reply)).unwrap();
        }
    })
}

#[test]
fn busd_unix_socket_connects_to_the_given_path() {
    let path = socket_path("explicit");
    let listener = UnixListener::bind(&path).unwrap();
    let accept = thread::spawn(move || listener.accept().map(|_| ()));
    let mut transport = BusdUnixSocket::new_with_path(&path);
    transport.open_streams().unwrap();
    accept.join().unwrap().unwrap();
    std::fs::remove_file(&path).unwrap();

    // Once the socket is gone, connecting fails with an I/O error
    assert!(transport.open_streams().is_err());
}

#[test]
fn busd_unix_socket_path_selection() {
    let _guard = ENV.lock().unwrap_or_else(|e| e.into_inner());
    assert_eq!(BusdUnixSocket::DEFAULT_PATH, "/run/oc2/bus");
    // SAFETY: serialized by ENV, and no other test in this binary reads the variable
    unsafe { std::env::remove_var("OC2_BUS_SOCKET") };
    assert!(format!("{:?}", BusdUnixSocket::new()).contains("/run/oc2/bus"));
    assert!(format!("{:?}", BusdUnixSocket::default()).contains("/run/oc2/bus"));
    unsafe { std::env::set_var("OC2_BUS_SOCKET", "/somewhere/else") };
    assert!(format!("{:?}", BusdUnixSocket::new()).contains("/somewhere/else"));
    unsafe { std::env::remove_var("OC2_BUS_SOCKET") };
}

#[test]
fn connect_busd_client_talks_to_a_daemon_end_to_end() {
    let _guard = ENV.lock().unwrap_or_else(|e| e.into_inner());
    let path = socket_path("e2e");
    let listener = UnixListener::bind(&path).unwrap();
    let daemon = fake_daemon(listener);
    unsafe { std::env::set_var("OC2_BUS_SOCKET", &path) };
    let client = HlapiClient::connect_busd_client();
    unsafe { std::env::remove_var("OC2_BUS_SOCKET") };

    let client = client.unwrap();
    let devices = client.list_devices().unwrap();
    assert_eq!(devices.len(), 1);
    assert_eq!(devices[0].type_names, ["redstone"]);
    drop(client);
    daemon.join().unwrap();
    let _ = std::fs::remove_file(&path);
}

#[test]
fn connect_busd_client_without_a_daemon_is_a_connection_error() {
    let _guard = ENV.lock().unwrap_or_else(|e| e.into_inner());
    unsafe { std::env::set_var("OC2_BUS_SOCKET", socket_path("missing")) };
    let err = HlapiClient::connect_busd_client().unwrap_err();
    unsafe { std::env::remove_var("OC2_BUS_SOCKET") };
    assert!(
        matches!(err, oc2_hlapi_v2::client::HlapiClientError::Connection(_)),
        "{err:?}"
    );
}

#[test]
fn virtio_discovery_fails_cleanly_when_there_are_no_ports() {
    // Whether the ports exist depends on the machine (this passes inside an OC2 VM too), so only
    // check that neither call panics, and that failures are errors rather than hangs.
    let _ = LinuxVirtio::new();
    let _ = HlapiClient::connect_virtio();
}