haematite-cli 0.21.1

use std::io;
use std::net::Shutdown;
use std::sync::mpsc;
use std::time::Duration;

use serde_json::Value;

use super::fixtures::{
    LIST, Reply, Server, TestResult, config, refusal, request, request_with_headers,
};
use super::native_call_fixture::Store;
use super::{Config, listener};

const CAP: usize = 131_072;

fn setup() -> Result<(Store, Server), Box<dyn std::error::Error + Send + Sync>> {
    let store = Store::start()?;
    let mcp = Server::mcp(config()?, store.path.clone(), CAP, 2)?;
    Ok((store, mcp))
}

#[test]
fn a_caller_whose_pass_the_store_takes_sees_exactly_the_existing_eleven_tools() -> TestResult {
    let (store, mcp) = setup()?;
    let response = Reply::read(request(&mcp, "POST", LIST, true, &[])?)?;
    assert_eq!(response.status, 200);
    let body: Value = serde_json::from_slice(&response.body)?;
    let tools = body
        .pointer("/result/tools")
        .and_then(Value::as_array)
        .ok_or_else(|| io::Error::other("tools missing"))?;
    let names = tools
        .iter()
        .map(|tool| tool.get("name").and_then(Value::as_str))
        .collect::<Vec<_>>();
    assert_eq!(
        names,
        crate::mcp::schema::tools()
            .iter()
            .map(|tool| Some(tool.name))
            .collect::<Vec<_>>()
    );
    assert_eq!(tools.len(), 11);
    assert_eq!(store.presented()?, 1);
    mcp.finish()?;
    store.finish()
}

#[test]
fn every_post_presents_its_pass_including_notifications_and_get_is_refused_first() -> TestResult {
    let (store, mcp) = setup()?;
    assert_eq!(
        Reply::read(request(&mcp, "POST", LIST, true, &[])?)?.status,
        200
    );
    let notification = r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#;
    let response = Reply::read(request(&mcp, "POST", notification, true, &[])?)?;
    assert_eq!(response.status, 202);
    assert!(response.body.is_empty());
    let response = Reply::read(request(&mcp, "GET", "", true, &[])?)?;
    refusal(&response, 405, "mcp_http_method_unsupported")?;
    assert_eq!(
        response.headers.get("allow").and_then(|v| v.to_str().ok()),
        Some("POST")
    );
    assert_eq!(store.presented()?, 2);
    mcp.finish()?;
    store.finish()
}

#[test]
fn unlisted_origin_refuses_before_the_store() -> TestResult {
    let (store, mcp) = setup()?;
    refusal(
        &Reply::read(request(
            &mcp,
            "POST",
            LIST,
            true,
            &[("Origin", "https://other.example")],
        )?)?,
        403,
        "mcp_http_origin_refused",
    )?;
    assert!(store.asked()?.is_empty());
    assert_eq!(
        Reply::read(request(
            &mcp,
            "POST",
            LIST,
            true,
            &[("Origin", "https://agent.example")]
        )?)?
        .status,
        200
    );
    assert_eq!(store.presented()?, 1);
    mcp.finish()?;
    store.finish()
}

#[test]
fn frame_limit_and_protocol_version_are_named_before_the_store() -> TestResult {
    let store = Store::start()?;
    let mcp = Server::mcp(config()?, store.path.clone(), 4096, 2)?;
    refusal(
        &Reply::read(request(&mcp, "POST", &"x".repeat(4097), true, &[])?)?,
        413,
        "mcp_http_body_refused",
    )?;
    refusal(
        &Reply::read(request(
            &mcp,
            "POST",
            "{}",
            true,
            &[("mcp-protocol-version", "unsupported")],
        )?)?,
        400,
        "mcp_http_protocol_version_unsupported",
    )?;
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

#[test]
fn config_refuses_a_public_listen_address_and_insecure_metadata_urls() -> TestResult {
    let text = r#"{"listen":"127.0.0.1:9041","allowed_origins":[],"resource_metadata_url":"https://identity.example/resource-metadata"}"#;
    let mut config: Config = serde_json::from_str(text)?;
    config.validate()?;
    for url in [
        "/resource-metadata",
        "http://identity.example/resource-metadata",
        "https://user:pass@identity.example/resource-metadata",
        "https://identity.example/resource-metadata#part",
    ] {
        config.resource_metadata_url = Some(url.to_owned());
        assert!(config.validate().is_err(), "{url} was accepted");
    }
    config.resource_metadata_url = None;
    config.validate()?;
    config.listen = "0.0.0.0:9041".parse()?;
    assert!(config.validate().is_err());
    Ok(())
}

#[test]
fn duplicate_authorization_refuses_without_reaching_the_store() -> TestResult {
    let (store, mcp) = setup()?;
    refusal(
        &Reply::read(request(
            &mcp,
            "POST",
            LIST,
            true,
            &[("Authorization", "Bearer fixture-token")],
        )?)?,
        401,
        "mcp_sign_in_required",
    )?;
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

#[test]
fn accepted_quality_parameters_reach_the_existing_tool_list() -> TestResult {
    let (store, mcp) = setup()?;
    let headers = [
        ("Content-Type", "application/json"),
        (
            "Accept",
            "application/json;q=0.8, text/event-stream;q=1.000",
        ),
        ("Authorization", "Bearer fixture-token"),
    ];
    assert_eq!(
        Reply::read(request_with_headers(&mcp, "POST", LIST, &headers)?)?.status,
        200
    );
    assert_eq!(store.presented()?, 1);
    mcp.finish()?;
    store.finish()
}

#[test]
fn protocol_header_refusals_never_reach_the_store() -> TestResult {
    let (store, mcp) = setup()?;
    for (content_type, accept, status, code) in [
        (
            "text/plain",
            "application/json, text/event-stream",
            415,
            "mcp_http_content_type_invalid",
        ),
        (
            "application/json",
            "application/json;q=0, text/event-stream",
            406,
            "mcp_http_accept_invalid",
        ),
    ] {
        let headers = [
            ("Content-Type", content_type),
            ("Accept", accept),
            ("Authorization", "Bearer fixture-token"),
        ];
        refusal(
            &Reply::read(request_with_headers(&mcp, "POST", LIST, &headers)?)?,
            status,
            code,
        )?;
    }
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

#[test]
fn an_oversized_body_keeps_its_named_refusal_after_worker_exit() -> TestResult {
    let cap = 4096;
    let store = Store::start()?;
    let (logged, lines) = mpsc::channel();
    let server = listener::Server::new(
        config()?,
        store.path.clone(),
        Duration::from_secs(1),
        cap,
        1,
    )?
    .with_diagnostics(logged)?;
    let mcp = Server::from_mcp(server)?;
    let socket = request(&mcp, "POST", &"x".repeat(cap * 4), true, &[])?;
    socket.shutdown(Shutdown::Write)?;
    let line = lines.recv()?;
    assert!(line.contains("mcp_http_body_refused"), "{line}");
    mcp.finish()?;
    refusal(&Reply::read(socket)?, 413, "mcp_http_body_refused")?;
    assert!(store.asked()?.is_empty());
    store.finish()
}