oc2_hlapi_v2 0.2.0

//! `HlapiClient`: caching, generations, events, calls, blobs, timeouts, error handling.

mod common;

use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use common::*;
use oc2_hlapi_v2::client::{DeviceBase, HlapiClient, HlapiClientError};
use oc2_hlapi_v2::codecs::BaseCodecError;
use oc2_hlapi_v2::devices::GenericDevice;
use oc2_hlapi_v2::messages::*;
use oc2_hlapi_v2::{Codec, uuid::Uuid};
use serde_json::{Value, json};

fn ms(n: u64) -> Option<Duration> {
    Some(Duration::from_millis(n))
}

fn generic(client: &HlapiClient, n: u128) -> GenericDevice {
    GenericDevice::new(client.clone(), id(n)).unwrap()
}

/// A host with the given list reply (generation 1) that answers everything else with a result.
fn host_with_devices(
    devices: Vec<(Uuid, Vec<&'static str>)>,
) -> impl FnMut(&Value, &[u8], &Host) + Send + 'static {
    move |req, _, host| {
        let list: Vec<(Uuid, &[&str])> = devices.iter().map(|(u, n)| (*u, n.as_slice())).collect();
        let reply = match req["type"].as_str().unwrap() {
            "list" => list_reply(req, 1, &list),
            "methods" => methods_reply(req, 1, &["a"]),
            _ => result_reply(req, 1, json!(null)),
        };
        host.reply(&reply);
    }
}

// --- listing and caching ------------------------------------------------------------------------

#[test]
fn lists_devices() {
    let (client, _s) = client_serving(simple_host);
    let devices = client.list_devices().unwrap();
    assert_eq!(devices.len(), 2);
    assert_eq!(devices[0].device_id, id(1));
    assert_eq!(devices[0].type_names, ["redstone", "device"]);
    assert_eq!(devices[1].type_names, ["storage"]);
}

#[test]
fn the_device_list_is_cached_while_the_generation_is_unchanged() {
    let (client, s) = client_serving(simple_host);
    client.list_devices().unwrap();
    client.list_devices().unwrap();
    client.find_devices_by_type::<GenericDevice>().unwrap();
    client.get_device_by_uuid::<GenericDevice>(id(1)).unwrap();
    assert_eq!(s.count("list"), 1);
}

#[test]
fn a_new_generation_in_a_response_invalidates_the_cache() {
    let gen_ = Arc::new(AtomicUsize::new(1));
    let g = gen_.clone();
    let (client, s) = client_serving(move |req, _, host| {
        let g = g.load(Ordering::SeqCst) as u32;
        host.reply(&match req["type"].as_str().unwrap() {
            "list" => list_reply(req, g, &[(id(1), &["a"])]),
            _ => result_reply(req, g, json!(null)),
        });
    });
    client.list_devices().unwrap();
    gen_.store(2, Ordering::SeqCst);
    // This reply carries generation 2
    generic(&client, 1).call::<()>("poke", &[], &[]).unwrap();
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2, "cached again under generation 2");
}

#[test]
fn a_devices_changed_event_invalidates_the_cache() {
    let (client, s) = client_serving(simple_host);
    client.list_devices().unwrap();
    s.host.emit(&devices_changed(2));
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
}

#[test]
fn an_event_with_the_current_generation_does_not() {
    let (client, s) = client_serving(simple_host);
    client.list_devices().unwrap();
    s.host.emit(&devices_changed(1));
    s.host.emit(&device_event(1, "tick", id(1), json!(null)));
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 1);
}

#[test]
fn any_event_with_a_new_generation_invalidates_the_cache() {
    // Unknown events and device events carry the generation too
    for event in [
        device_event(2, "tick", id(1), json!(null)),
        json!({"type": "someFutureEvent", "gen": 2}),
    ] {
        let (client, s) = client_serving(simple_host);
        client.list_devices().unwrap();
        s.host.emit(&event);
        client.list_devices().unwrap();
        assert_eq!(s.count("list"), 2, "{event}");
    }
}

#[test]
fn dropped_events_make_the_cache_suspect_until_the_next_host_message() {
    let (client, s) = client_serving(simple_host);
    client.list_devices().unwrap();
    s.host.emit(&events_dropped(1, 5));
    // Same generation, but a dropped event may have been a devicesChanged
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
    // The refetch answered with a message: the cache is trusted again
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
}

#[test]
fn requests_get_increasing_positive_ids() {
    let (client, s) = client_serving(simple_host);
    let dev = generic(&client, 1);
    for _ in 0..3 {
        dev.call::<i64>("m", &[], &[]).unwrap();
    }
    dev.list_methods_for_test();
    let ids: Vec<u64> = s
        .requests()
        .iter()
        .map(|r| r["id"].as_u64().unwrap())
        .collect();
    assert!(ids[0] >= 1);
    assert!(ids.windows(2).all(|w| w[1] == w[0] + 1), "{ids:?}");
}

trait ListMethodsForTest {
    fn list_methods_for_test(&self);
}
impl ListMethodsForTest for GenericDevice {
    fn list_methods_for_test(&self) {
        use oc2_hlapi_v2::client::Device;
        self.list_methods().unwrap();
    }
}

// --- finding devices ----------------------------------------------------------------------------

fn world() -> Vec<(Uuid, Vec<&'static str>)> {
    vec![
        (id(30), vec!["redstone", "device", "left"]),
        (id(10), vec!["redstone", "right"]),
        (id(20), vec!["storage", "left"]),
    ]
}

#[test]
fn find_by_type_filters_and_sorts_by_uuid() {
    let (client, _s) = client_serving(host_with_devices(world()));
    let found = client.find_devices_by_type::<Redstone>().unwrap();
    let ids: Vec<Uuid> = found.iter().map(|d| d.uuid_base()).collect();
    assert_eq!(ids, [id(10), id(30)]);
    let all = client.find_devices_by_type::<GenericDevice>().unwrap();
    let ids: Vec<Uuid> = all.iter().map(|d| d.uuid_base()).collect();
    assert_eq!(ids, [id(10), id(20), id(30)]);
}

#[test]
fn find_one_returns_the_lowest_uuid() {
    let (client, _s) = client_serving(host_with_devices(world()));
    assert_eq!(
        client
            .find_device_by_type::<Redstone>()
            .unwrap()
            .unwrap()
            .uuid_base(),
        id(10)
    );
    assert_eq!(
        client
            .find_device_by_names::<Redstone>(&["device"])
            .unwrap()
            .unwrap()
            .uuid_base(),
        id(30)
    );
}

#[test]
fn find_by_names_requires_all_names() {
    let (client, _s) = client_serving(host_with_devices(world()));
    let uuids = |names: &[&str]| -> Vec<Uuid> {
        client
            .find_devices_by_names::<GenericDevice>(names)
            .unwrap()
            .iter()
            .map(|d| d.uuid_base())
            .collect()
    };
    assert_eq!(uuids(&["left"]), [id(20), id(30)]);
    assert_eq!(uuids(&["left", "redstone"]), [id(30)]);
    assert_eq!(uuids(&["left", "right"]), Vec::<Uuid>::new());
    assert_eq!(uuids(&["nonexistent"]), Vec::<Uuid>::new());
    assert_eq!(uuids(&[]), [id(10), id(20), id(30)]);
    // An empty name matches every device
    assert_eq!(uuids(&[""]), [id(10), id(20), id(30)]);
    assert_eq!(uuids(&["", "left"]), [id(20), id(30)]);
}

#[test]
fn find_by_names_also_applies_the_wrappers_type() {
    let (client, _s) = client_serving(host_with_devices(world()));
    let found = client.find_devices_by_names::<Redstone>(&["left"]).unwrap();
    assert_eq!(found.len(), 1);
    assert_eq!(found[0].uuid_base(), id(30));
    assert!(
        client
            .find_device_by_names::<Storage>(&["right"])
            .unwrap()
            .is_none()
    );
}

#[test]
fn nothing_found_is_none_not_an_error() {
    let (client, _s) = client_serving(host_with_devices(vec![]));
    assert!(client.find_device_by_type::<Redstone>().unwrap().is_none());
    assert!(
        client
            .find_devices_by_type::<Redstone>()
            .unwrap()
            .is_empty()
    );
    assert!(
        client
            .get_device_by_uuid::<GenericDevice>(id(1))
            .unwrap()
            .is_none()
    );
}

#[test]
fn get_by_uuid_checks_existence_and_type() {
    let (client, _s) = client_serving(host_with_devices(world()));
    assert_eq!(
        client
            .get_device_by_uuid::<Redstone>(id(30))
            .unwrap()
            .unwrap()
            .uuid_base(),
        id(30)
    );
    assert!(
        client
            .get_device_by_uuid::<Redstone>(id(20))
            .unwrap()
            .is_none(),
        "wrong type"
    );
    assert!(
        client
            .get_device_by_uuid::<Redstone>(id(99))
            .unwrap()
            .is_none(),
        "unknown"
    );
    assert!(
        client
            .get_device_by_uuid::<GenericDevice>(id(20))
            .unwrap()
            .is_some()
    );
}

// --- calls --------------------------------------------------------------------------------------

#[test]
fn a_call_sends_the_invocation_and_returns_the_value() {
    let (client, s) =
        client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!({"v": [1, 2]}))));
    let (value, blobs): (Value, _) = generic(&client, 5)
        .call("setThing", &[&"left", &15, &true, &json!({"a": null})], &[])
        .unwrap();
    assert_eq!(value, json!({"v": [1, 2]}));
    assert!(blobs.is_empty());
    let req = &s.requests()[0];
    assert_eq!(req["type"], "invoke");
    assert_eq!(req["data"]["deviceId"], json!(id(5)));
    assert_eq!(req["data"]["name"], "setThing");
    assert_eq!(
        req["data"]["parameters"],
        json!(["left", 15, true, {"a": null}])
    );
}

#[test]
fn a_call_without_arguments_sends_an_empty_parameter_list() {
    let (client, s) = client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!(1))));
    generic(&client, 5).call::<i64>("get", &[], &[]).unwrap();
    assert_eq!(s.requests()[0]["data"]["parameters"], json!([]));
}

#[test]
fn return_values_are_deserialized_into_the_requested_type() {
    let replies = [
        json!(42),
        json!("text"),
        json!([1, 2, 3]),
        json!(null),
        json!({"k": 1.5}),
    ];
    let next = Arc::new(Mutex::new(0usize));
    let n = next.clone();
    let (client, _s) = client_serving(move |req, _, host| {
        let mut n = n.lock().unwrap();
        host.reply(&result_reply(req, 1, replies[*n].clone()));
        *n += 1;
    });
    let d = generic(&client, 1);
    assert_eq!(d.call::<i64>("a", &[], &[]).unwrap().0, 42);
    assert_eq!(d.call::<String>("a", &[], &[]).unwrap().0, "text");
    assert_eq!(d.call::<Vec<u8>>("a", &[], &[]).unwrap().0, [1, 2, 3]);
    assert_eq!(d.call::<Option<i64>>("a", &[], &[]).unwrap().0, None);
    assert_eq!(
        d.call::<std::collections::HashMap<String, f64>>("a", &[], &[])
            .unwrap()
            .0["k"],
        1.5
    );
}

#[test]
fn a_result_without_a_value_deserializes_as_unit_or_null() {
    let (client, _s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let d = generic(&client, 1);
    d.call::<()>("a", &[], &[]).unwrap();
    assert_eq!(d.call::<Value>("a", &[], &[]).unwrap().0, Value::Null);
    assert_eq!(d.call::<Option<String>>("a", &[], &[]).unwrap().0, None);
}

#[test]
fn a_return_value_of_the_wrong_type_is_a_serde_error() {
    let (client, _s) =
        client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!("not a number"))));
    let err = generic(&client, 1).call::<i64>("a", &[], &[]).unwrap_err();
    assert!(matches!(err, HlapiClientError::Serde(_)), "{err:?}");
}

#[test]
fn hlapi_errors_carry_the_hosts_message() {
    let (client, _s) =
        client_serving(|req, _, host| host.reply(&error_reply(req, 1, "no such method")));
    let err = generic(&client, 1).call::<()>("a", &[], &[]).unwrap_err();
    assert!(
        matches!(&err, HlapiClientError::Hlapi(m) if m == "no such method"),
        "{err:?}"
    );
    assert!(err.to_string().contains("no such method"));
    // and the client keeps working
    assert!(client.list_devices().is_err());
}

#[test]
fn an_error_reply_without_an_id_belongs_to_the_request_in_flight() {
    // The host answers errors it can't attribute with id 0 (busd leaves the id out)
    let (client, _s) = client_serving(|_, _, host| {
        host.reply(&json!({"type": "error", "id": 0, "gen": 1, "data": "malformed"}));
    });
    let err = client.list_devices().unwrap_err();
    assert!(
        matches!(&err, HlapiClientError::Hlapi(m) if m == "malformed"),
        "{err:?}"
    );
    let (client, _s) = client_serving(|_, _, host| {
        host.reply(&json!({"type": "error", "data": "malformed"}));
    });
    assert!(matches!(
        client.list_devices().unwrap_err(),
        HlapiClientError::Hlapi(_)
    ));
}

#[test]
fn a_reply_of_the_wrong_kind_is_reported() {
    let (client, _s) = client_serving(|req, _, host| host.reply(&list_reply(req, 1, &[])));
    let err = generic(&client, 1).call::<()>("a", &[], &[]).unwrap_err();
    assert!(
        matches!(err, HlapiClientError::UnexpectedResponseType),
        "{err:?}"
    );
    let (client, _s) = client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!(1))));
    let err = client.list_devices().unwrap_err();
    assert!(
        matches!(err, HlapiClientError::UnexpectedResponseType),
        "{err:?}"
    );
}

#[test]
fn the_generation_of_an_error_reply_is_still_noted() {
    let (client, s) = client_serving(|req, _, host| {
        host.reply(&match req["type"].as_str().unwrap() {
            "list" => list_reply(req, 1, &[]),
            _ => error_reply(req, 2, "gone"),
        })
    });
    client.list_devices().unwrap();
    generic(&client, 1).call::<()>("a", &[], &[]).unwrap_err();
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
}

// --- blobs --------------------------------------------------------------------------------------

#[test]
fn blobs_are_sent_with_descriptors_before_the_request() {
    let (client, s) = client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!(null))));
    generic(&client, 1)
        .call::<()>(
            "write",
            &[&BlobMarker { blob_idx: 0 }, &BlobMarker { blob_idx: 1 }],
            &[b"hello", b"x\0y"],
        )
        .unwrap();
    let seen = &s.seen()[0];
    assert_eq!(seen.blobs, b"hellox\0y");
    assert_eq!(
        seen.request["blobs"],
        json!([descriptor(b"hello"), descriptor(b"x\0y")])
    );
    assert_eq!(
        seen.request["data"]["parameters"],
        json!([{"$blob": 0}, {"$blob": 1}])
    );
}

#[test]
fn blobs_in_results_are_read_and_verified() {
    let (client, _s) = client_serving(|req, _, host| {
        host.blob_out(b"first");
        host.blob_out(b"second!");
        host.reply(&with_blobs(
            req,
            json!([{"$blob": 0}, {"$blob": 1}]),
            &[b"first", b"second!"],
        ));
    });
    let (value, blobs): (Vec<BlobMarker>, _) = generic(&client, 1).call("read", &[], &[]).unwrap();
    assert_eq!(value.iter().map(|m| m.blob_idx).collect::<Vec<_>>(), [0, 1]);
    assert_eq!(blobs, [b"first".to_vec(), b"second!".to_vec()]);
}

fn with_blobs(req: &Value, data: Value, blobs: &[&[u8]]) -> Value {
    let mut v = result_reply(req, 1, data);
    v["blobs"] = Value::Array(blobs.iter().map(|b| descriptor(b)).collect());
    v
}

#[test]
fn a_corrupt_result_blob_is_an_error() {
    let (client, _s) = client_serving(|req, _, host| {
        host.blob_out(b"firsu");
        host.reply(&with_blobs(req, json!({"$blob": 0}), &[b"first"]));
    });
    let err = generic(&client, 1)
        .call::<Value>("read", &[], &[])
        .unwrap_err();
    let HlapiClientError::Codec(inner) = err else {
        panic!("wrong error")
    };
    assert!(matches!(
        inner.downcast_ref::<BaseCodecError<io::Error>>(),
        Some(BaseCodecError::BlobChecksumInvalid)
    ));
}

#[test]
fn wrappers_can_exchange_blobs() {
    let (client, s) = client_serving(|req, blobs, host| {
        if req["data"]["name"] == "read" {
            host.blob_out(b"stored bytes");
            host.reply(&with_blobs(req, json!({"$blob": 0}), &[b"stored bytes"]));
        } else {
            assert_eq!(blobs, b"to store");
            host.reply(&empty_result_reply(req, 1));
        }
    });
    let storage = Storage::new(client.clone(), id(2)).unwrap();
    assert_eq!(storage.read().unwrap(), b"stored bytes");
    storage.write(b"to store").unwrap();
    assert_eq!(s.total(), 2);
}

#[test]
fn stale_data_on_the_streams_is_discarded_before_each_request() {
    let (client, s) = client_serving(|req, _, host| {
        host.blob_out(b"real");
        host.reply(&with_blobs(req, json!({"$blob": 0}), &[b"real"]));
    });
    // Leftovers from some earlier, abandoned exchange
    s.host.blob_out(b"stale blob bytes");
    s.host
        .reply(&result_reply(&json!({"id": 0}), 1, json!("stale reply")));
    let (v, blobs): (BlobMarker, _) = generic(&client, 1).call("read", &[], &[]).unwrap();
    assert_eq!(v.blob_idx, 0);
    assert_eq!(blobs, [b"real".to_vec()]);
}

#[test]
fn a_request_that_is_too_big_fails_cleanly_and_the_client_survives() {
    let (client, s) = client_serving(|req, _, host| host.reply(&result_reply(req, 1, json!(1))));
    let big = "x".repeat(5000);
    let err = generic(&client, 1)
        .call::<i64>("m", &[&big], &[b"blob"])
        .unwrap_err();
    let HlapiClientError::Codec(inner) = err else {
        panic!("wrong error")
    };
    assert!(matches!(
        inner.downcast_ref::<BaseCodecError<io::Error>>(),
        Some(BaseCodecError::RequestTooBig)
    ));
    // Nothing reached the host, and the next call is not desynchronized by the blob
    assert_eq!(s.total(), 0);
    let (v, _) = generic(&client, 1).call::<i64>("m", &[], &[]).unwrap();
    assert_eq!(v, 1);
    assert!(s.seen()[0].blobs.is_empty());
}

// --- timeouts and late replies --------------------------------------------------------------------

#[test]
fn the_default_request_timeout_matches_the_lua_libraries() {
    assert_eq!(
        HlapiClient::DEFAULT_REQUEST_TIMEOUT,
        Duration::from_secs(35)
    );
}

#[test]
fn a_request_times_out() {
    let (client, s) = client_silent();
    client.set_request_timeout(ms(120)).unwrap();
    let start = Instant::now();
    let err = client.list_devices().unwrap_err();
    assert!(matches!(err, HlapiClientError::RequestTimeout), "{err:?}");
    let took = start.elapsed();
    assert!(
        took >= Duration::from_millis(100) && took < Duration::from_secs(5),
        "{took:?}"
    );
    assert_eq!(s.total(), 1);
}

#[test]
fn the_timeout_setting_is_shared_by_clones_and_can_be_disabled() {
    let (client, _s) = client_serving(|req, _, host| {
        thread::sleep(Duration::from_millis(250));
        host.reply(&list_reply(req, 1, &[]));
    });
    let other = client.clone();
    other.set_request_timeout(ms(50)).unwrap();
    assert!(matches!(
        client.list_devices(),
        Err(HlapiClientError::RequestTimeout)
    ));
    client.set_request_timeout(None).unwrap();
    assert!(other.list_devices().is_ok());
}

#[test]
fn a_late_reply_to_an_abandoned_request_is_not_mistaken_for_the_next_reply() {
    let first_id = Arc::new(Mutex::new(None));
    let mut count = 0;
    let (client, _s) = client_serving({
        let first_id = first_id.clone();
        move |req, _, host| {
            count += 1;
            if count == 1 {
                *first_id.lock().unwrap() = Some(req["id"].clone());
            } else {
                // The answer to request 1 finally arrives, right before the answer to request 2
                let stale_id = first_id.lock().unwrap().clone().unwrap();
                host.reply(&result_reply(&json!({"id": stale_id}), 1, json!("stale")));
                host.reply(&result_reply(req, 1, json!("fresh")));
            }
        }
    });
    client.set_request_timeout(ms(100)).unwrap();
    let dev = generic(&client, 1);
    assert!(matches!(
        dev.call::<String>("m", &[], &[]),
        Err(HlapiClientError::RequestTimeout)
    ));
    client.set_request_timeout(Some(LONG)).unwrap();
    assert_eq!(dev.call::<String>("m", &[], &[]).unwrap().0, "fresh");
}

#[test]
fn skipping_foreign_replies_does_not_extend_the_timeout() {
    let (client, _s) = client_serving(|_, _, host| {
        for _ in 0..40 {
            host.reply(&result_reply(&json!({"id": 4_000_000}), 1, json!(0)));
            thread::sleep(Duration::from_millis(10));
        }
    });
    client.set_request_timeout(ms(150)).unwrap();
    let start = Instant::now();
    assert!(matches!(
        client.list_devices(),
        Err(HlapiClientError::RequestTimeout)
    ));
    assert!(
        start.elapsed() < Duration::from_millis(350),
        "{:?}",
        start.elapsed()
    );
}

#[test]
fn the_client_works_again_after_a_timeout() {
    let mut count = 0;
    let (client, _s) = client_serving(move |req, _, host| {
        count += 1;
        if count > 1 {
            host.reply(&list_reply(req, 1, &[(id(1), &["a"])]));
        }
    });
    client.set_request_timeout(ms(80)).unwrap();
    assert!(client.list_devices().is_err());
    client.set_request_timeout(Some(LONG)).unwrap();
    assert_eq!(client.list_devices().unwrap().len(), 1);
}

/// --- misc --------------------------------------------------------------------------------------

#[test]
fn clones_share_one_connection() {
    let (client, s) = client_serving(simple_host);
    let other = client.clone();
    client.list_devices().unwrap();
    other.list_devices().unwrap();
    assert_eq!(s.count("list"), 1);
}

#[test]
fn concurrent_calls_from_many_threads_are_matched_to_their_replies() {
    let (client, s) = client_serving(|req, _, host| {
        // Echo the first parameter
        host.reply(&result_reply(req, 1, req["data"]["parameters"][0].clone()));
    });
    let threads: Vec<_> = (0..8u64)
        .map(|t| {
            let dev = generic(&client, t as u128);
            thread::spawn(move || {
                for i in 0..25u64 {
                    let v = t * 1000 + i;
                    let (got, _) = dev.call::<u64>("echo", &[&v], &[]).unwrap();
                    assert_eq!(got, v);
                }
            })
        })
        .collect();
    for t in threads {
        t.join().unwrap();
    }
    assert_eq!(s.total(), 200);
}

// --- failures -----------------------------------------------------------------------------------

#[test]
fn a_dead_connection_is_a_codec_error() {
    let (codec, host) = codec(TimeoutKind::WouldBlock);
    let client = HlapiClient::new(codec);
    drop(host); // the mock transport has no further connections to reconnect with
    let err = client.list_devices().unwrap_err();
    assert!(matches!(err, HlapiClientError::Codec(_)), "{err:?}");
    assert!(err.to_string().starts_with("Codec error"));
}

/// A codec that panics in `read_event`, to poison the client's lock.
struct PanickingCodec;

#[derive(Debug, thiserror::Error)]
#[error("never")]
struct Never;

impl Codec for PanickingCodec {
    type Error = Never;
    fn send_request(&mut self, _: Request, _: &[&[u8]]) -> Result<(), Never> {
        Ok(())
    }
    fn read_response(&mut self, _: Option<Duration>) -> Result<Option<Response>, Never> {
        Ok(None)
    }
    fn read_event(&mut self, _: Option<Duration>) -> Result<Option<Event>, Never> {
        panic!("codec bug");
    }
    fn read_blob(&mut self, _: BlobDescriptor) -> Result<Vec<u8>, Never> {
        Ok(vec![])
    }
    fn clear_response_stream(&mut self) -> Result<(), Never> {
        Ok(())
    }
    fn clear_blob_stream(&mut self) -> Result<(), Never> {
        Ok(())
    }
}

#[test]
fn a_poisoned_client_reports_an_error_instead_of_panicking() {
    use oc2_hlapi_v2::client::Device;
    let client = HlapiClient::new(PanickingCodec);
    let c = client.clone();
    assert!(thread::spawn(move || c.list_devices()).join().is_err());
    let dev = generic(&client, 1);
    assert!(matches!(
        client.list_devices(),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(matches!(
        client.set_request_timeout(None),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(matches!(
        client.find_device_by_type::<Redstone>(),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(matches!(
        client.get_device_by_uuid::<Redstone>(id(1)),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(matches!(
        dev.call::<()>("m", &[], &[]),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(matches!(
        dev.list_methods(),
        Err(HlapiClientError::MutexPoisoned)
    ));
    assert!(format!("{client:?}").contains("HlapiClient"));
}

/// A codec that is not byte based at all: the client only depends on the `Codec` trait.
struct ScriptedCodec {
    sent: Arc<Mutex<Vec<Request>>>,
    fail_sends: bool,
}

#[derive(Debug, thiserror::Error)]
#[error("scripted failure {0}")]
struct ScriptedError(u32);

impl Codec for ScriptedCodec {
    type Error = ScriptedError;
    fn send_request(&mut self, r: Request, _: &[&[u8]]) -> Result<(), ScriptedError> {
        self.sent.lock().unwrap().push(r);
        if self.fail_sends {
            Err(ScriptedError(7))
        } else {
            Ok(())
        }
    }
    fn read_response(&mut self, _: Option<Duration>) -> Result<Option<Response>, ScriptedError> {
        let id = self
            .sent
            .lock()
            .unwrap()
            .last()
            .and_then(|r| r.id)
            .unwrap_or(0);
        Ok(Some(
            serde_json::from_value(json!({"type": "list", "id": id, "gen": 1, "data": []}))
                .unwrap(),
        ))
    }
    fn read_event(&mut self, _: Option<Duration>) -> Result<Option<Event>, ScriptedError> {
        Ok(None)
    }
    fn read_blob(&mut self, _: BlobDescriptor) -> Result<Vec<u8>, ScriptedError> {
        Ok(vec![])
    }
    fn clear_response_stream(&mut self) -> Result<(), ScriptedError> {
        Ok(())
    }
    fn clear_blob_stream(&mut self) -> Result<(), ScriptedError> {
        Ok(())
    }
}

#[test]
fn any_codec_can_drive_the_client() {
    let sent = Arc::new(Mutex::new(Vec::new()));
    let client = HlapiClient::new(ScriptedCodec {
        sent: sent.clone(),
        fail_sends: false,
    });
    assert!(client.list_devices().unwrap().is_empty());
    assert!(matches!(sent.lock().unwrap()[0].data, RequestData::List));
}

#[test]
fn codec_errors_keep_their_type_inside_the_client_error() {
    let client = HlapiClient::new(ScriptedCodec {
        sent: Default::default(),
        fail_sends: true,
    });
    let err = client.list_devices().unwrap_err();
    let HlapiClientError::Codec(inner) = &err else {
        panic!("{err:?}")
    };
    assert_eq!(inner.downcast_ref::<ScriptedError>().unwrap().0, 7);
    assert_eq!(err.to_string(), "Codec error: scripted failure 7");
    // `source()` reaches the codec error too
    assert!(std::error::Error::source(&err).is_some());
}

// --- static guarantees --------------------------------------------------------------------------

fn assert_traits<T: Clone + std::fmt::Debug + Send + Sync + 'static>() {}
fn assert_error<T: std::error::Error + Send + Sync + 'static>() {}

#[test]
fn public_types_have_the_expected_bounds() {
    assert_traits::<HlapiClient>();
    assert_traits::<GenericDevice>();
    assert_traits::<Redstone>(); // a wrapper with plain derives, no manual impls
    assert_error::<HlapiClientError>();
    assert_error::<BaseCodecError<io::Error>>();
}

#[test]
fn error_messages_are_readable() {
    assert_eq!(
        HlapiClientError::RequestTimeout.to_string(),
        "Request timeout"
    );
    assert_eq!(HlapiClientError::BlobTooBig.to_string(), "Blob too big");
    assert_eq!(
        HlapiClientError::Hlapi("x".into()).to_string(),
        "HLAPI error: x"
    );
    assert_eq!(
        HlapiClientError::MutexPoisoned.to_string(),
        "HLAPI client state mutex poisoned"
    );
}