dre_protocol/

conformance.rs

//! The protocol conformance suite. Every first-party plugin runs it in its tests; third-party
//! plugin authors can run it too. It only checks protocol behaviour common to every plugin kind;
//! each plugin's own tests cover what it does with real data. A package executable
//! (`dre-plugin-<package>`) is checked once per plugin it provides.

// Each check is an immediately-invoked closure so `?` works inside it.
#![allow(clippy::redundant_closure_call)]

use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use crate::host::{HostError, Incoming, LogSink, PluginProcess};
use serde_json::{Map, json};

use crate::msg::{DeliveryFile, Request, Response};
use crate::{MAX_VERSION, MIN_VERSION, PluginId, parse_executable_name, parse_package_executable_name};

const TIMEOUT: Duration = Duration::from_secs(20);

#[derive(Debug)]
pub struct Check {
    /// The plugin checked (`<kind>/<name>`), or the package for checks of the package itself.
    pub plugin: String,
    pub name: &'static str,
    pub passed: bool,
    pub detail: String,
}

fn quiet() -> LogSink {
    Arc::new(|_, _| {})
}

fn start_versions(
    path: &Path,
    plugin: Option<&PluginId>,
    env: &[(&str, &str)],
    versions: (u32, u32),
) -> Result<PluginProcess, HostError> {
    let mut p = PluginProcess::spawn_env(path, quiet(), env)?;
    if plugin.is_some() {
        p.ask_for(plugin.cloned());
    }
    p.handshake(versions, TIMEOUT)?;
    Ok(p)
}

/// Run every check against the plugin executable at `path`.
pub fn run(path: &Path) -> Vec<Check> {
    run_with_env(path, &[])
}

/// Like [`run`], with extra environment variables for every plugin process (for plugins whose
/// capabilities depend on their environment).
pub fn run_with_env(path: &Path, env: &[(&str, &str)]) -> Vec<Check> {
    let file = path.file_name().unwrap().to_string_lossy().to_string();
    if let Some((kind, name)) = parse_executable_name(&file) {
        return run_plugin(path, &PluginId::new(kind, name), false, env);
    }
    let fail = |detail: String| {
        vec![Check {
            plugin: file.clone(),
            name: "the executable is named dre-plugin-<package> or dre-<kind>-<name>",
            passed: false,
            detail,
        }]
    };
    let Some(package) = parse_package_executable_name(&file) else {
        return fail(format!("`{file}` is neither"));
    };
    let provides = match start_versions(path, None, env, (MIN_VERSION, MAX_VERSION)) {
        Ok(p) => {
            let provides = p.info().provides.clone();
            let _ = p.close();
            provides
        }
        Err(e) => return fail(format!("the handshake failed: {e}")),
    };
    let mut out = Vec::new();
    let unknown = PluginId::new(crate::Kind::Format, "dre_conformance_unknown");
    let refused = match start_versions(path, Some(&unknown), env, (MIN_VERSION, MAX_VERSION)) {
        Err(HostError::Plugin { .. }) => Ok(()),
        Err(e) => Err(format!("expected an error reply, got: {e}")),
        Ok(_) => Err(format!("the package served {unknown}, which it doesn't provide")),
    };
    out.push(Check {
        plugin: package,
        name: "asking for a plugin the package doesn't provide gets an error reply",
        passed: refused.is_ok(),
        detail: refused.err().unwrap_or_default(),
    });
    for id in &provides {
        out.extend(run_plugin(path, id, true, env));
    }
    out
}

/// Every check, for one plugin. `ask`: name it in the handshake (a package's plugins).
fn run_plugin(path: &Path, id: &PluginId, ask: bool, env: &[(&str, &str)]) -> Vec<Check> {
    let asked = ask.then_some(id);
    let start = |path: &Path| start_versions(path, asked, env, (MIN_VERSION, MAX_VERSION));
    let mut out = Vec::new();
    let mut check = |name: &'static str, r: Result<(), String>| {
        out.push(Check {
            plugin: id.to_string(),
            name,
            passed: r.is_ok(),
            detail: r.err().unwrap_or_default(),
        });
    };

    check(
        "handshake negotiates a supported version and reports its identity",
        (|| {
            let p = start(path).map_err(|e| e.to_string())?;
            let info = p.info().clone();
            if info.kind != id.kind || info.name != id.name {
                return Err(format!(
                    "expected {id}, handshake says {}/{}",
                    info.kind, info.name
                ));
            }
            if !info.provides.contains(id) {
                return Err(format!("`provides` doesn't list {id}: {:?}", info.provides));
            }
            if info.version.is_empty() {
                return Err("empty plugin version".into());
            }
            p.close().map_err(|e| e.to_string())
        })(),
    );

    check(
        "an unsupported protocol range is refused, not hung on",
        (|| {
            let far = MAX_VERSION + 1000;
            match start_versions(path, asked, env, (far, far)) {
                Err(HostError::Incompatible { plugin_range, .. }) if plugin_range.1 < far => Ok(()),
                Err(e) => Err(format!("expected a version mismatch, got: {e}")),
                Ok(_) => Err("the plugin accepted a protocol version it can't speak".into()),
            }
        })(),
    );

    check(
        "protocol 0 is still spoken, for an older core",
        (|| {
            let p = start_versions(path, asked, env, (0, 0)).map_err(|e| e.to_string())?;
            match p.info().protocol_version {
                0 => p.close().map_err(|e| e.to_string()),
                v => Err(format!("offered only version 0, the plugin chose {v}")),
            }
        })(),
    );

    check(
        "protocol 1 is spoken",
        (|| {
            let p = start(path).map_err(|e| e.to_string())?;
            match p.info().protocol_version {
                v if v >= 1 => p.close().map_err(|e| e.to_string()),
                v => Err(format!(
                    "offered versions up to {MAX_VERSION}, the plugin chose {v}"
                )),
            }
        })(),
    );

    check(
        "replies carry the request's id, and a cancel for a request that isn't running is ignored",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            p.write_raw(&frame_json(r#"{"type":"cancel","id":999}"#))
                .map_err(|e| e.to_string())?;
            for (req, want) in [
                (r#"{"type":"describe","id":41}"#, "describe"),
                (r#"{"type":"frobnicate","id":42}"#, "error"),
            ] {
                p.write_raw(&frame_json(req)).map_err(|e| e.to_string())?;
                let v = p.recv_raw(TIMEOUT).map_err(|e| e.to_string())?;
                let id = req[req.len() - 3..req.len() - 1].parse::<u64>().unwrap();
                if v["type"] != want || v["id"] != id {
                    return Err(format!(
                        "sent {req}, expected a `{want}` reply with id {id}, got {v}"
                    ));
                }
            }
            p.close().map_err(|e| e.to_string())
        })(),
    );

    check(
        "describe is answered",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            p.describe().map_err(|e| e.to_string())?;
            p.close().map_err(|e| e.to_string())
        })(),
    );

    if id.kind == crate::Kind::Format {
        check(
            "a write of an empty result set makes a valid file (or asks for options)",
            (|| {
                let dir =
                    std::env::temp_dir().join(format!("dre-conformance-{}-{}", id.name, std::process::id()));
                std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
                let file = dir.join(format!("empty.{}", id.name));
                let schema = Arc::new(arrow::datatypes::Schema::new(vec![
                    arrow::datatypes::Field::new("id", arrow::datatypes::DataType::Int64, true),
                    arrow::datatypes::Field::new("name", arrow::datatypes::DataType::Utf8, true),
                ]));
                let meta = crate::msg::ResultSetMeta {
                    name: "empty".into(),
                    query: "empty".into(),
                    result_index: 1,
                    anchor: None,
                    header: None,
                    columns: Default::default(),
                    autofit: None,
                    style: None,
                };
                let mut p = start(path).map_err(|e| e.to_string())?;
                p.write_begin(&file.to_string_lossy(), &id.name, Map::new(), vec![meta], None)
                    .map_err(|e| e.to_string())?;
                let empty = arrow::array::RecordBatch::new_empty(schema.clone());
                let result = p
                    .write_result_set(&schema, vec![empty])
                    .and_then(|_| p.write_finish());
                let r = match result {
                    Ok(files) if files.iter().all(|f| Path::new(f).is_file()) && !files.is_empty() => Ok(()),
                    Ok(files) => Err(format!("`written` lists files that don't exist: {files:?}")),
                    // A format that needs options (fixed_width's `columns`) or a template context
                    // refuses as a config problem.
                    Err(HostError::Plugin { kind: Some(k), .. }) if k == "config" => Ok(()),
                    Err(e) => Err(format!("an empty result set failed: {e}")),
                };
                let _ = p.close();
                let _ = std::fs::remove_dir_all(&dir);
                r
            })(),
        );
    }

    if id.kind == crate::Kind::Source {
        check(
            "describe gives the identifier quote character",
            (|| {
                let mut p = start(path).map_err(|e| e.to_string())?;
                let d = p.description().map_err(|e| e.to_string())?;
                match d.identifier_quote.as_deref() {
                    Some(q) if q.chars().count() == 1 => {}
                    Some(q) => return Err(format!("`identifier_quote` must be one character, got {q:?}")),
                    None => return Err("a source's describe reply has no `identifier_quote`".into()),
                }
                p.close().map_err(|e| e.to_string())
            })(),
        );
    }

    check(
        "validate is advertised, answered, and refuses an option the plugin doesn't declare",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            if !p.has(crate::CAP_VALIDATE) {
                return Err("the plugin doesn't advertise `validate`".into());
            }
            let (_, fields) = p.describe_all().map_err(|e| e.to_string())?;
            p.validate(Map::new()).map_err(|e| e.to_string())?;
            let key = "dre_conformance_unknown_option";
            if fields.iter().any(|f| f.name == key) {
                return Err(format!("the plugin declares `{key}`"));
            }
            let errors = p
                .validate(json!({key: true}).as_object().unwrap().clone())
                .map_err(|e| e.to_string())?;
            if !errors.iter().any(|e| e.contains(key)) {
                return Err(format!("an unknown option was accepted: {errors:?}"));
            }
            p.close().map_err(|e| e.to_string())
        })(),
    );

    if id.kind != crate::Kind::Format {
        check(
            "validate_connection is answered without connecting, and an unknown key only warns",
            (|| {
                let mut p = start(path).map_err(|e| e.to_string())?;
                let key = "dre_conformance_unknown_key";
                let (_, warnings) = p
                    .validate_connection(json!({key: true}).as_object().unwrap().clone(), Vec::new())
                    .map_err(|e| e.to_string())?;
                if !warnings.iter().any(|w| w.contains(key)) {
                    return Err(format!("an unknown key gave no warning: {warnings:?}"));
                }
                p.close().map_err(|e| e.to_string())
            })(),
        );
    }

    check(
        "an unknown request gets an error reply and the plugin keeps serving",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            p.write_raw(&frame_json(r#"{"type":"frobnicate"}"#))
                .map_err(|e| e.to_string())?;
            match p
                .recv(Some(TIMEOUT), "an error reply")
                .map_err(|e| e.to_string())?
            {
                Incoming::Json(Response::Error { .. }) => {}
                other => return Err(format!("expected an error reply, got {other:?}")),
            }
            p.describe()
                .map_err(|e| format!("plugin stopped serving after an unknown request: {e}"))?;
            p.close().map_err(|e| e.to_string())
        })(),
    );

    check(
        "a request meant for another plugin kind gets an error reply",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            let req = match p.info().kind {
                crate::Kind::Source => Request::Finish {},
                _ => Request::Check {
                    sql: "select 1".into(),
                },
            };
            p.send(&req).map_err(|e| e.to_string())?;
            match p
                .recv(Some(TIMEOUT), "an error reply")
                .map_err(|e| e.to_string())?
            {
                Incoming::Json(Response::Error { .. }) => {}
                other => return Err(format!("expected an error reply, got {other:?}")),
            }
            p.close().map_err(|e| e.to_string())
        })(),
    );

    check(
        "a malformed frame is reported or ends the plugin, never hangs",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            let _ = p.write_raw(&[0, 0, 0, 2, b'X', b'!']);
            match p.recv(Some(TIMEOUT), "a reply to a malformed frame") {
                Ok(Incoming::Json(Response::Error { .. })) | Err(HostError::Crashed { .. }) => Ok(()),
                Err(HostError::Timeout { .. }) => Err("the plugin hung on a malformed frame".into()),
                other => Err(format!("unexpected reply to a malformed frame: {other:?}")),
            }
        })(),
    );

    // Destinations: `deliver` with options (and, with `multi_file`, several files) is parsed and
    // answered. The file doesn't exist, so a delivered or an error reply are both fine.
    let destination = start(path).ok().map(|p| {
        let info = p.info().clone();
        let _ = p.close();
        info
    });
    if let Some(info) = destination.filter(|i| i.kind == crate::Kind::Destination) {
        let missing = |n: &str| DeliveryFile {
            local_path: format!("/nonexistent/dre-conformance/{n}"),
            remote_path: Some(format!("dre-conformance/{n}")),
        };
        let one = missing("a.csv");
        let mut forms = vec![(
            "deliver with options gets a reply and the plugin keeps serving",
            Request::Deliver {
                local_path: Some(one.local_path),
                remote_path: one.remote_path,
                files: Vec::new(),
                connection: Map::new(),
                options: json!({"conformance": true}).as_object().unwrap().clone(),
                message: None,
            },
        )];
        if info.capabilities.iter().any(|c| c == crate::CAP_MULTI_FILE) {
            forms.push((
                "a multi-file deliver gets a reply and the plugin keeps serving",
                Request::Deliver {
                    local_path: None,
                    remote_path: None,
                    files: vec![missing("a.csv"), missing("b.csv")],
                    connection: Map::new(),
                    options: Map::new(),
                    message: None,
                },
            ));
        }
        let has = |c: &str| info.capabilities.iter().any(|x| x == c);
        if has(crate::CAP_MESSAGE) {
            forms.push((
                "a message deliver gets a reply and the plugin keeps serving",
                Request::Deliver {
                    local_path: None,
                    remote_path: None,
                    files: Vec::new(),
                    connection: Map::new(),
                    options: Map::new(),
                    message: Some(crate::msg::Message {
                        title: "Conformance".into(),
                        text: "**DRE** conformance check".into(),
                        html: None,
                        path: "/nonexistent/dre-conformance/message.md".into(),
                    }),
                },
            ));
        }
        if has(crate::CAP_MESSAGE_ONLY) {
            check(
                "a message_only plugin also advertises message",
                if has(crate::CAP_MESSAGE) {
                    Ok(())
                } else {
                    Err("advertises `message_only` without `message`".into())
                },
            );
            check(
                "a message_only plugin refuses a file deliver and keeps serving",
                (|| {
                    let mut p = start(path).map_err(|e| e.to_string())?;
                    let f = missing("a.csv");
                    p.send(&Request::Deliver {
                        local_path: Some(f.local_path),
                        remote_path: f.remote_path,
                        files: Vec::new(),
                        connection: Map::new(),
                        options: Map::new(),
                        message: None,
                    })
                    .map_err(|e| e.to_string())?;
                    match p
                        .recv(Some(TIMEOUT), "a deliver reply")
                        .map_err(|e| e.to_string())?
                    {
                        Incoming::Json(Response::Error { .. }) => {}
                        other => return Err(format!("expected an error, got {other:?}")),
                    }
                    p.describe()
                        .map_err(|e| format!("plugin stopped serving after a deliver: {e}"))?;
                    p.close().map_err(|e| e.to_string())
                })(),
            );
        }
        for (name, req) in forms {
            check(
                name,
                (|| {
                    let mut p = start(path).map_err(|e| e.to_string())?;
                    p.send(&req).map_err(|e| e.to_string())?;
                    match p
                        .recv(Some(TIMEOUT), "a deliver reply")
                        .map_err(|e| e.to_string())?
                    {
                        Incoming::Json(Response::Delivered { .. } | Response::Error { .. }) => {}
                        other => return Err(format!("expected delivered or error, got {other:?}")),
                    }
                    p.describe()
                        .map_err(|e| format!("plugin stopped serving after a deliver: {e}"))?;
                    p.close().map_err(|e| e.to_string())
                })(),
            );
        }
    }

    check(
        "close ends the process with exit code 0",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            p.send(&Request::Close {}).map_err(|e| e.to_string())?;
            let _ = p.recv(Some(TIMEOUT), "the close reply");
            match p.wait_exit(TIMEOUT) {
                Some(s) if s.success() => Ok(()),
                Some(s) => Err(format!("exited with {s}")),
                None => Err("still running after close".into()),
            }
        })(),
    );

    check(
        "end of input ends the process",
        (|| {
            let mut p = start(path).map_err(|e| e.to_string())?;
            match p.wait_exit(TIMEOUT) {
                Some(_) => Ok(()),
                None => Err("still running after stdin closed".into()),
            }
        })(),
    );

    out
}

/// Panic with a readable report unless every check passes. For use in plugin tests.
pub fn assert_conforms(path: &Path) {
    let checks = run(path);
    let failed: Vec<String> = checks
        .iter()
        .filter(|c| !c.passed)
        .map(|c| format!("  ✗ {}: {}: {}", c.plugin, c.name, c.detail))
        .collect();
    assert!(
        failed.is_empty(),
        "{} fails protocol conformance:\n{}",
        path.display(),
        failed.join("\n")
    );
}

fn frame_json(body: &str) -> Vec<u8> {
    let mut v = Vec::new();
    v.extend_from_slice(&((body.len() + 1) as u32).to_be_bytes());
    v.push(crate::frame::JSON_TAG);
    v.extend_from_slice(body.as_bytes());
    v
}