oc2_hlapi_v2 0.2.0

//! Shared test infrastructure.
//!
//! Everything here is in-memory (no sockets, no files), so the suite runs on every platform:
//!
//! * [`MemStream`]: one end of an in-memory duplex pipe implementing the crate's `Stream` trait,
//!   with real blocking / nonblocking / read-timeout semantics.
//! * [`Host`]: the other ends of the three HLAPI streams, i.e. the "VM host" side.
//! * [`serve`]: runs a scripted host on a thread, answering requests as they arrive.
//! * Builders for HLAPI messages, and a couple of device wrappers written the way a wrapper author
//!   would write them.
#![allow(dead_code)]

use std::collections::VecDeque;
use std::convert::Infallible;
use std::io::{self, Read, Write};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use oc2_hlapi_v2::client::{DeviceBase, HlapiClient, HlapiClientError};
use oc2_hlapi_v2::codecs::{BaseCodec, BaseStreams};
use oc2_hlapi_v2::messages::{BlobDescriptor, BlobMarker};
use oc2_hlapi_v2::uuid::Uuid;
use oc2_hlapi_v2::{Stream, Transport};
use serde_json::{Value, json};

/// Generous upper bound for "this must happen" waits. Only reached when a test is failing.
pub const LONG: Duration = Duration::from_secs(10);
/// How long to wait to be sure that something does *not* happen.
pub const SHORT: Duration = Duration::from_millis(150);

// ---------------------------------------------------------------------------------------------
// In-memory streams
// ---------------------------------------------------------------------------------------------

/// Which error kind a read timeout produces. Linux sockets report `WouldBlock`, Windows reports
/// `TimedOut`; the codec has to cope with both.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TimeoutKind {
    WouldBlock,
    TimedOut,
}

pub const KINDS: [TimeoutKind; 2] = [TimeoutKind::WouldBlock, TimeoutKind::TimedOut];

impl TimeoutKind {
    fn error(self) -> io::Error {
        match self {
            Self::WouldBlock => io::ErrorKind::WouldBlock.into(),
            Self::TimedOut => io::ErrorKind::TimedOut.into(),
        }
    }
}

#[derive(Default)]
struct PipeState {
    data: VecDeque<u8>,
    writer_gone: bool,
    reader_gone: bool,
}

#[derive(Default)]
struct Pipe {
    state: Mutex<PipeState>,
    cond: Condvar,
}

#[derive(Default)]
struct Config {
    nonblocking: bool,
    timeout: Option<Duration>,
}

/// A shared record of which stream was written to, in order.
pub type WriteLog = Arc<Mutex<Vec<String>>>;

/// One end of an in-memory duplex pipe.
///
/// Reads follow the semantics of the real streams the crate runs on: they return buffered data
/// first, then EOF (`Ok(0)`) once the other end is dropped, otherwise they block, fail with
/// `WouldBlock` (nonblocking mode) or fail with the configured timeout error kind (read timeout).
/// Writes fail with `BrokenPipe` once the other end is dropped.
pub struct MemStream {
    rx: Arc<Pipe>,
    tx: Arc<Pipe>,
    config: Mutex<Config>,
    kind: TimeoutKind,
    log: Option<(String, WriteLog)>,
}

impl std::fmt::Debug for MemStream {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemStream").finish_non_exhaustive()
    }
}

/// Create a connected pair of streams.
pub fn duplex(kind: TimeoutKind) -> (MemStream, MemStream) {
    let a_to_b = Arc::new(Pipe::default());
    let b_to_a = Arc::new(Pipe::default());
    let make = |rx: &Arc<Pipe>, tx: &Arc<Pipe>| MemStream {
        rx: rx.clone(),
        tx: tx.clone(),
        config: Mutex::new(Config::default()),
        kind,
        log: None,
    };
    (make(&b_to_a, &a_to_b), make(&a_to_b, &b_to_a))
}

impl MemStream {
    /// Record every write to this stream in `log` under `label`.
    pub fn logged(mut self, label: &str, log: &WriteLog) -> Self {
        self.log = Some((label.to_owned(), log.clone()));
        self
    }

    /// Read at least one byte. `timeout`: `None` blocks, `Some(ZERO)` polls.
    pub fn recv(&self, buf: &mut [u8], timeout: Option<Duration>) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        let deadline = timeout.map(|t| Instant::now() + t);
        let mut st = self.rx.state.lock().unwrap();
        loop {
            if !st.data.is_empty() {
                let n = buf.len().min(st.data.len());
                for slot in &mut buf[..n] {
                    *slot = st.data.pop_front().unwrap();
                }
                return Ok(n);
            }
            if st.writer_gone {
                return Ok(0);
            }
            match deadline {
                None => st = self.rx.cond.wait(st).unwrap(),
                Some(d) => {
                    let now = Instant::now();
                    if now >= d {
                        return Err(self.kind.error());
                    }
                    st = self.rx.cond.wait_timeout(st, d - now).unwrap().0;
                }
            }
        }
    }

    /// Write from a shared reference (so a [`Host`] can be used from several threads).
    pub fn send(&self, bytes: &[u8]) -> io::Result<()> {
        let mut st = self.tx.state.lock().unwrap();
        if st.reader_gone {
            return Err(io::ErrorKind::BrokenPipe.into());
        }
        st.data.extend(bytes);
        self.tx.cond.notify_all();
        drop(st);
        if let Some((label, log)) = &self.log {
            log.lock().unwrap().push(label.clone());
        }
        Ok(())
    }

    /// Everything that is currently readable, without waiting.
    pub fn available(&self) -> Vec<u8> {
        let mut st = self.rx.state.lock().unwrap();
        st.data.drain(..).collect()
    }

    /// Wait for exactly `n` bytes.
    pub fn read_n(&self, n: usize, timeout: Duration) -> io::Result<Vec<u8>> {
        let deadline = Instant::now() + timeout;
        let mut out = Vec::new();
        while out.len() < n {
            let mut buf = vec![0; n - out.len()];
            let left = deadline.saturating_duration_since(Instant::now());
            let got = self.recv(&mut buf, Some(left))?;
            if got == 0 {
                return Err(io::ErrorKind::UnexpectedEof.into());
            }
            out.extend_from_slice(&buf[..got]);
        }
        Ok(out)
    }

    /// Read one NUL-terminated frame (empty frames are skipped). `Ok(None)` on timeout, `Err` on
    /// EOF. `None` as the timeout waits forever.
    pub fn frame(&self, timeout: Option<Duration>) -> io::Result<Option<Vec<u8>>> {
        let deadline = timeout.map(|t| Instant::now() + t);
        let mut frame = Vec::new();
        loop {
            let left = deadline.map(|d| d.saturating_duration_since(Instant::now()));
            let mut byte = [0];
            match self.recv(&mut byte, left) {
                Ok(0) => return Err(io::ErrorKind::UnexpectedEof.into()),
                Ok(_) if byte[0] == 0 => {
                    if !frame.is_empty() {
                        return Ok(Some(frame));
                    }
                }
                Ok(_) => frame.push(byte[0]),
                Err(e)
                    if matches!(
                        e.kind(),
                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
                    ) =>
                {
                    return Ok(None);
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// Like [`frame`](Self::frame) but parsed as JSON.
    pub fn json(&self, timeout: Option<Duration>) -> io::Result<Option<Value>> {
        Ok(self
            .frame(timeout)?
            .map(|f| serde_json::from_slice(&f).expect("peer sent a frame that is not JSON")))
    }

    /// Write `{NUL}{json}{NUL}`.
    pub fn send_json(&self, value: &Value) {
        self.send(&framed(value)).unwrap();
    }
}

/// `NUL json NUL`, the way the host frames messages.
pub fn framed(value: &Value) -> Vec<u8> {
    let mut v = vec![0];
    v.extend(serde_json::to_vec(value).unwrap());
    v.push(0);
    v
}

impl Read for MemStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let (nonblocking, timeout) = {
            let c = self.config.lock().unwrap();
            (c.nonblocking, c.timeout)
        };
        if nonblocking {
            return self
                .recv(buf, Some(Duration::ZERO))
                .map_err(|_| io::ErrorKind::WouldBlock.into());
        }
        self.recv(buf, timeout)
    }
}

impl Write for MemStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.send(buf)?;
        Ok(buf.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl Stream for MemStream {
    fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
        self.config.lock().unwrap().nonblocking = nonblocking;
        Ok(())
    }
    fn set_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
        // Same contract as the std socket types
        if timeout == Some(Duration::ZERO) {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "zero timeout"));
        }
        self.config.lock().unwrap().timeout = timeout;
        Ok(())
    }
}

impl Drop for MemStream {
    fn drop(&mut self) {
        self.rx.state.lock().unwrap().reader_gone = true;
        self.tx.state.lock().unwrap().writer_gone = true;
        self.rx.cond.notify_all();
        self.tx.cond.notify_all();
    }
}

// ---------------------------------------------------------------------------------------------
// Host side and transports
// ---------------------------------------------------------------------------------------------

/// The far end of the three HLAPI streams.
#[derive(Debug)]
pub struct Host {
    pub rpc: MemStream,
    pub blob: MemStream,
    pub event: MemStream,
}

impl Host {
    /// Send a framed message on the rpc stream.
    pub fn reply(&self, value: &Value) {
        self.rpc.send_json(value);
    }
    /// Send a framed message on the event stream.
    pub fn emit(&self, value: &Value) {
        self.event.send_json(value);
    }
    /// Send raw bytes on the blob stream.
    pub fn blob_out(&self, bytes: &[u8]) {
        self.blob.send(bytes).unwrap();
    }
}

/// A fresh set of connected streams: the codec side and the host side.
pub fn connection(kind: TimeoutKind) -> (BaseStreams<MemStream>, Host) {
    let (rpc_c, rpc_h) = duplex(kind);
    let (blob_c, blob_h) = duplex(kind);
    let (event_c, event_h) = duplex(kind);
    (
        BaseStreams {
            rpc: rpc_c,
            blob: blob_c,
            event: event_c,
        },
        Host {
            rpc: rpc_h,
            blob: blob_h,
            event: event_h,
        },
    )
}

type Opener = Box<dyn FnMut() -> io::Result<BaseStreams<MemStream>> + Send>;

/// A transport handing out in-memory connections.
pub struct MockTransport {
    next: Opener,
    pub opens: Arc<AtomicUsize>,
}

impl MockTransport {
    /// Hands out the given connections in order, then fails.
    pub fn sequence(conns: Vec<BaseStreams<MemStream>>) -> Self {
        let mut conns: VecDeque<_> = conns.into();
        Self::with(move || {
            conns
                .pop_front()
                .ok_or_else(|| io::Error::other("no more connections"))
        })
    }
    /// Calls `f` for every connection attempt.
    pub fn with(f: impl FnMut() -> io::Result<BaseStreams<MemStream>> + Send + 'static) -> Self {
        Self {
            next: Box::new(f),
            opens: Arc::new(AtomicUsize::new(0)),
        }
    }
}

impl Transport for MockTransport {
    type Streams = BaseStreams<MemStream>;
    type Error = io::Error;
    fn open_streams(&mut self) -> io::Result<Self::Streams> {
        self.opens.fetch_add(1, Ordering::SeqCst);
        (self.next)()
    }
}

pub type MockCodec = BaseCodec<MemStream, MockTransport>;

/// A codec on one connection, plus the host end of it.
pub fn codec(kind: TimeoutKind) -> (MockCodec, Host) {
    let (streams, host) = connection(kind);
    (
        BaseCodec::new(MockTransport::sequence(vec![streams])).unwrap(),
        host,
    )
}

/// A codec with `spares` further connections available for reconnecting. Returns the host ends of
/// all connections (the first is the one in use) and the open counter.
pub fn codec_with_spares(
    kind: TimeoutKind,
    spares: usize,
) -> (MockCodec, Vec<Host>, Arc<AtomicUsize>) {
    let mut streams = Vec::new();
    let mut hosts = Vec::new();
    for _ in 0..=spares {
        let (s, h) = connection(kind);
        streams.push(s);
        hosts.push(h);
    }
    let transport = MockTransport::sequence(streams);
    let opens = transport.opens.clone();
    (BaseCodec::new(transport).unwrap(), hosts, opens)
}

/// A codec whose write order across the rpc and blob streams is recorded.
pub fn codec_logged(kind: TimeoutKind) -> (MockCodec, Host, WriteLog) {
    let (streams, host) = connection(kind);
    let log = WriteLog::default();
    let streams = BaseStreams {
        rpc: streams.rpc.logged("rpc", &log),
        blob: streams.blob.logged("blob", &log),
        event: streams.event,
    };
    (
        BaseCodec::new(MockTransport::sequence(vec![streams])).unwrap(),
        host,
        log,
    )
}

// ---------------------------------------------------------------------------------------------
// Scripted host
// ---------------------------------------------------------------------------------------------

/// A request as the scripted host saw it.
#[derive(Clone, Debug)]
pub struct Seen {
    pub request: Value,
    /// Blob bytes that had been written by the time the request arrived.
    pub blobs: Vec<u8>,
}

/// A host running on its own thread, answering each request with the handler.
pub struct Server {
    pub host: Arc<Host>,
    seen: Arc<Mutex<Vec<Seen>>>,
    _thread: JoinHandle<()>,
}

impl Server {
    pub fn requests(&self) -> Vec<Value> {
        self.seen
            .lock()
            .unwrap()
            .iter()
            .map(|s| s.request.clone())
            .collect()
    }
    pub fn seen(&self) -> Vec<Seen> {
        self.seen.lock().unwrap().clone()
    }
    /// How many requests of the given `type` arrived.
    pub fn count(&self, typ: &str) -> usize {
        self.requests().iter().filter(|r| r["type"] == typ).count()
    }
    pub fn total(&self) -> usize {
        self.seen.lock().unwrap().len()
    }
}

/// Answer requests arriving on `host` until the codec side is dropped.
///
/// The handler gets the parsed request, the blob bytes that preceded it, and the host (to reply,
/// emit events, send blobs).
pub fn serve(host: Host, mut handler: impl FnMut(&Value, &[u8], &Host) + Send + 'static) -> Server {
    let host = Arc::new(host);
    let seen = Arc::new(Mutex::new(Vec::new()));
    let thread = {
        let host = host.clone();
        let seen = seen.clone();
        thread::spawn(move || {
            while let Ok(Some(raw)) = host.rpc.frame(None) {
                let request: Value = serde_json::from_slice(&raw).unwrap();
                let blobs = host.blob.available();
                seen.lock().unwrap().push(Seen {
                    request: request.clone(),
                    blobs: blobs.clone(),
                });
                handler(&request, &blobs, &host);
            }
        })
    };
    Server {
        host,
        seen,
        _thread: thread,
    }
}

/// A client talking to a scripted host.
pub fn client_serving(
    handler: impl FnMut(&Value, &[u8], &Host) + Send + 'static,
) -> (HlapiClient, Server) {
    client_serving_kind(TimeoutKind::WouldBlock, handler)
}

pub fn client_serving_kind(
    kind: TimeoutKind,
    handler: impl FnMut(&Value, &[u8], &Host) + Send + 'static,
) -> (HlapiClient, Server) {
    let (codec, host) = codec(kind);
    (HlapiClient::new(codec), serve(host, handler))
}

/// A client talking to a host that never answers (but can emit events).
pub fn client_silent() -> (HlapiClient, Server) {
    client_serving(|_, _, _| {})
}

/// Poll until `cond` holds, or panic after [`LONG`].
pub fn wait_until(what: &str, mut cond: impl FnMut() -> bool) {
    let deadline = Instant::now() + LONG;
    while !cond() {
        assert!(Instant::now() < deadline, "timed out waiting for {what}");
        thread::sleep(Duration::from_millis(2));
    }
}

/// Run `f` on a thread and fail instead of hanging if it takes longer than [`LONG`].
pub fn within<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = std::sync::mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    rx.recv_timeout(LONG).expect("operation hung")
}

// ---------------------------------------------------------------------------------------------
// Message builders. `req` is the request being answered, its id is echoed.
// ---------------------------------------------------------------------------------------------

pub fn id(n: u128) -> Uuid {
    Uuid::from_u128(n)
}

fn with_id(req: &Value, mut v: Value) -> Value {
    if let Some(i) = req.get("id") {
        v["id"] = i.clone();
    }
    v
}

pub fn list_reply(req: &Value, gen_: u32, devices: &[(Uuid, &[&str])]) -> Value {
    let data: Vec<_> = devices
        .iter()
        .map(|(u, names)| json!({"deviceId": u, "typeNames": names}))
        .collect();
    with_id(req, json!({"type": "list", "gen": gen_, "data": data}))
}

pub fn methods_reply(req: &Value, gen_: u32, names: &[&str]) -> Value {
    let data: Vec<_> = names.iter().map(|n| json!({"name": n})).collect();
    with_id(req, json!({"type": "methods", "gen": gen_, "data": data}))
}

pub fn result_reply(req: &Value, gen_: u32, value: Value) -> Value {
    with_id(req, json!({"type": "result", "gen": gen_, "data": value}))
}

/// A result without a `data` field (a method returning nothing).
pub fn empty_result_reply(req: &Value, gen_: u32) -> Value {
    with_id(req, json!({"type": "result", "gen": gen_}))
}

pub fn error_reply(req: &Value, gen_: u32, msg: &str) -> Value {
    with_id(req, json!({"type": "error", "gen": gen_, "data": msg}))
}

pub fn descriptor(blob: &[u8]) -> Value {
    serde_json::to_value(BlobDescriptor::describe(blob).unwrap()).unwrap()
}

pub fn devices_changed(gen_: u32) -> Value {
    json!({"type": "devicesChanged", "gen": gen_})
}

pub fn events_dropped(gen_: u32, count: u32) -> Value {
    json!({"type": "eventsDropped", "gen": gen_, "data": count})
}

pub fn device_event(gen_: u32, typ: &str, device: Uuid, data: Value) -> Value {
    json!({"type": typ, "gen": gen_, "deviceId": device, "data": data})
}

/// A host that knows two devices and answers `list`/`methods`/`invoke` simply. Used by tests that
/// don't care about the details. Generation is always 1.
pub fn simple_host(req: &Value, _blobs: &[u8], host: &Host) {
    let redstone = id(1);
    let storage = id(2);
    let reply = match req["type"].as_str().unwrap() {
        "list" => list_reply(
            req,
            1,
            &[(redstone, &["redstone", "device"]), (storage, &["storage"])],
        ),
        "methods" => methods_reply(req, 1, &["getInput", "setOutput"]),
        "invoke" => result_reply(req, 1, json!(7)),
        other => panic!("unexpected request type {other}"),
    };
    host.reply(&reply);
}

// ---------------------------------------------------------------------------------------------
// Device wrappers, written the way the docs tell wrapper authors to write them
// ---------------------------------------------------------------------------------------------

#[derive(Debug, Clone)]
pub struct Redstone {
    client: HlapiClient,
    uuid: Uuid,
}

impl DeviceBase for Redstone {
    const TYPE: &'static str = "redstone";
    type Event = Infallible;
    fn new(client: HlapiClient, uuid: Uuid) -> Result<Self, HlapiClientError> {
        Ok(Self { client, uuid })
    }
    fn uuid_base(&self) -> Uuid {
        self.uuid
    }
    fn client(&self) -> &HlapiClient {
        &self.client
    }
}

impl Redstone {
    pub fn get_input(&self, side: &str) -> Result<i64, HlapiClientError> {
        Ok(self.call::<i64>("getInput", &[&side], &[])?.0)
    }
    pub fn set_output(&self, side: &str, level: u8) -> Result<(), HlapiClientError> {
        self.call::<()>("setOutput", &[&side, &level], &[])?;
        Ok(())
    }
}

/// A wrapper whose methods take and return blobs.
#[derive(Debug, Clone)]
pub struct Storage {
    client: HlapiClient,
    uuid: Uuid,
}

impl DeviceBase for Storage {
    const TYPE: &'static str = "storage";
    type Event = Infallible;
    fn new(client: HlapiClient, uuid: Uuid) -> Result<Self, HlapiClientError> {
        Ok(Self { client, uuid })
    }
    fn uuid_base(&self) -> Uuid {
        self.uuid
    }
    fn client(&self) -> &HlapiClient {
        &self.client
    }
}

impl Storage {
    pub fn read(&self) -> Result<Vec<u8>, HlapiClientError> {
        let (marker, mut blobs) = self.call::<BlobMarker>("read", &[], &[])?;
        Ok(blobs.swap_remove(marker.blob_idx as usize))
    }
    pub fn write(&self, data: &[u8]) -> Result<(), HlapiClientError> {
        self.call::<()>("write", &[&BlobMarker { blob_idx: 0 }], &[data])?;
        Ok(())
    }
}