wip-rs/crates/wip-fs-server/tests/end_to_end.rs

use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use tempfile::tempdir;
use wip_fs_server::{AccessMode, ServerConfig, WipFsServer};
use wip_http::{
    ClientResponseError, DecodedResponse, Endpoint, JSON_CONTENT_TYPE, Limits, Route,
    TransportFailureKind, decode_call_operation_response, decode_fetch_interface_response,
    decode_observe_response, encode_call_operation_request, encode_fetch_interface_request,
    encode_observe_request,
};
use wip_protocol::{
    CallOperationRequest, FetchInterfaceRequest, InterfaceTarget, ObserveRequest,
    ProtocolErrorCode, Target, Value,
};

#[test]
fn real_listener_serves_three_core_interactions_and_write() {
    let temporary = tempdir().expect("temporary root");
    std::fs::write(temporary.path().join("hello.txt"), b"hello world").expect("write file");
    std::fs::create_dir(temporary.path().join("nested")).expect("create directory");

    let mut config = ServerConfig::for_root(temporary.path());
    config.access = AccessMode::ReadWrite;
    let limits = config.limits.wire;
    let bound = Arc::new(
        WipFsServer::new(config)
            .expect("build server")
            .bind()
            .expect("bind"),
    );
    let address = bound.local_addr();
    let shutdown = Arc::new(AtomicBool::new(false));
    let worker = {
        let bound = Arc::clone(&bound);
        let shutdown = Arc::clone(&shutdown);
        thread::spawn(move || bound.serve_until(shutdown))
    };
    let endpoint = Endpoint::parse(&format!("http://localhost:{}/wip", address.port()))
        .expect("client endpoint");

    let file_request = ObserveRequest {
        path: "/hello.txt".into(),
        depth: 0,
    };
    let response = exchange(
        address,
        encode_observe_request(&endpoint, &file_request, limits).expect("encode observation"),
    );
    let file = success(
        decode_observe_response(&file_request, &response, limits).expect("decode observation"),
    )
    .object;
    assert_eq!(file.name, "hello.txt");
    assert_eq!(file.interfaces.len(), 2);
    assert!(file.validator.is_some());

    let tree_request = ObserveRequest {
        path: "/".into(),
        depth: 1,
    };
    let response = exchange(
        address,
        encode_observe_request(&endpoint, &tree_request, limits).expect("encode tree"),
    );
    let tree =
        success(decode_observe_response(&tree_request, &response, limits).expect("decode tree"));
    assert_eq!(
        tree.children
            .as_ref()
            .expect("root children observed")
            .len(),
        2
    );

    let read_interface_request = FetchInterfaceRequest {
        interface: file.interfaces[0].clone(),
    };
    let response = exchange(
        address,
        encode_fetch_interface_request(&endpoint, &read_interface_request, limits)
            .expect("encode interface fetch"),
    );
    let read_interface = success(
        decode_fetch_interface_response(&read_interface_request, &response, limits)
            .expect("decode interface"),
    );
    assert_eq!(read_interface.descriptor.operations[0].name, "read");

    let read_request = CallOperationRequest {
        target: Target {
            path: "/hello.txt".into(),
            validator: file.validator.clone(),
        },
        interface: InterfaceTarget {
            scope_ref: None,
            reference: file.interfaces[0].clone(),
            validator: None,
        },
        operation: "read".into(),
        arguments: BTreeMap::from([
            ("offset".into(), Value::Integer(6)),
            ("length".into(), Value::Integer(5)),
        ]),
    };
    let response = exchange(
        address,
        encode_call_operation_request(&endpoint, &read_request, &read_interface.descriptor, limits)
            .expect("encode read call"),
    );
    let read = success(
        decode_call_operation_response(
            &read_request,
            &read_interface.descriptor,
            &response,
            limits,
        )
        .expect("decode read call"),
    );
    assert_eq!(read.result, Value::Bytes(b"world".to_vec()));

    let write_interface_request = FetchInterfaceRequest {
        interface: file.interfaces[1].clone(),
    };
    let response = exchange(
        address,
        encode_fetch_interface_request(&endpoint, &write_interface_request, limits)
            .expect("encode write interface fetch"),
    );
    let write_interface = success(
        decode_fetch_interface_response(&write_interface_request, &response, limits)
            .expect("decode write interface"),
    );
    let write_request = CallOperationRequest {
        target: Target {
            path: "/hello.txt".into(),
            validator: file.validator,
        },
        interface: InterfaceTarget {
            scope_ref: None,
            reference: file.interfaces[1].clone(),
            validator: None,
        },
        operation: "replace".into(),
        arguments: BTreeMap::from([("contents".into(), Value::Bytes(b"updated".to_vec()))]),
    };
    let response = exchange(
        address,
        encode_call_operation_request(
            &endpoint,
            &write_request,
            &write_interface.descriptor,
            limits,
        )
        .expect("encode write call"),
    );
    let write = success(
        decode_call_operation_response(
            &write_request,
            &write_interface.descriptor,
            &response,
            limits,
        )
        .expect("decode write call"),
    );
    assert!(write.validator.is_some());
    assert_eq!(
        std::fs::read(temporary.path().join("hello.txt")).expect("read updated file"),
        b"updated"
    );

    shutdown.store(true, Ordering::Release);
    worker
        .join()
        .expect("join listener")
        .expect("listener exits cleanly");
}

#[test]
fn real_listener_keeps_route_method_and_malformed_body_boundaries_distinct() {
    let temporary = tempdir().expect("temporary root");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.wire = Limits::new(64, 4_096, 64).expect("wire limits");
    let bound = Arc::new(
        WipFsServer::new(config)
            .expect("build server")
            .bind()
            .expect("bind"),
    );
    let address = bound.local_addr();
    let shutdown = Arc::new(AtomicBool::new(false));
    let worker = {
        let bound = Arc::clone(&bound);
        let shutdown = Arc::clone(&shutdown);
        thread::spawn(move || bound.serve_until(shutdown))
    };

    for uri in [
        "/wip/v1/unknown",
        "/wip/v1/batch",
        "/wip/v1/observe?unexpected=true",
        "/wip/v1/observe/extra",
    ] {
        let response = exchange(address, json_request(http::Method::POST, uri, b"{}"));
        assert_eq!(response.status(), http::StatusCode::NOT_FOUND);
        assert!(response.body().is_empty());
        assert!(!response.headers().contains_key(http::header::CONTENT_TYPE));
        assert!(!response.headers().contains_key(http::header::ALLOW));
    }

    let wrong_method = exchange(
        address,
        json_request(http::Method::GET, "/wip/v1/observe", b""),
    );
    assert_eq!(wrong_method.status(), http::StatusCode::METHOD_NOT_ALLOWED);
    assert_eq!(wrong_method.headers()[http::header::ALLOW], "POST");
    assert!(wrong_method.body().is_empty());
    assert!(
        !wrong_method
            .headers()
            .contains_key(http::header::CONTENT_TYPE)
    );

    let mut unsupported_coding = json_request(
        http::Method::POST,
        "/wip/v1/observe",
        br#"{"path":"/","depth":0}"#,
    );
    unsupported_coding
        .headers_mut()
        .insert(http::header::CONTENT_ENCODING, "gzip".parse().unwrap());
    let unsupported_coding = exchange(address, unsupported_coding);
    assert_eq!(
        unsupported_coding.status(),
        http::StatusCode::UNSUPPORTED_MEDIA_TYPE
    );
    assert!(unsupported_coding.body().is_empty());
    assert!(
        !unsupported_coding
            .headers()
            .contains_key(http::header::CONTENT_TYPE)
    );

    let malformed = exchange(
        address,
        json_request(http::Method::POST, "/wip/v1/observe", b"{"),
    );
    assert_eq!(malformed.status(), http::StatusCode::BAD_REQUEST);
    assert_eq!(
        malformed.headers()[http::header::CONTENT_TYPE],
        JSON_CONTENT_TYPE
    );
    assert!(
        std::str::from_utf8(malformed.body())
            .expect("UTF-8 error envelope")
            .contains("\"invalid_request\"")
    );

    let oversized = exchange(
        address,
        json_request(http::Method::POST, "/wip/v1/observe", &[b'x'; 65]),
    );
    assert_eq!(oversized.status(), http::StatusCode::PAYLOAD_TOO_LARGE);
    assert!(oversized.body().is_empty());
    assert!(!oversized.headers().contains_key(http::header::CONTENT_TYPE));

    let mut header_only = TcpStream::connect(address).expect("connect header-only client");
    header_only
        .set_read_timeout(Some(Duration::from_secs(1)))
        .expect("set header-only timeout");
    write!(
        header_only,
        "POST /wip/v1/observe HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 65\r\n\r\n"
    )
    .expect("write oversized request head");
    header_only.flush().expect("flush oversized request head");
    let mut raw = Vec::new();
    header_only
        .read_to_end(&mut raw)
        .expect("read immediate limit response");
    assert!(raw.starts_with(b"HTTP/1.1 413 "));
    assert!(raw.ends_with(b"\r\n\r\n"));
    assert!(
        !raw.windows(b"resource_limit_exceeded".len())
            .any(|window| { window == b"resource_limit_exceeded" })
    );

    shutdown.store(true, Ordering::Release);
    worker
        .join()
        .expect("join listener")
        .expect("listener exits cleanly");
}

#[test]
fn raw_decoder_limits_are_transport_failures() {
    let temporary = tempdir().expect("temporary root");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.wire = Limits::new(4_096, 4_096, 8).expect("wire limits");
    let server = WipFsServer::new(config).expect("build server");

    let nested = br#"{"path":[[[[[[[[["/"]]]]]]]]],"depth":0}"#;
    let response = server.handle_http(json_request(http::Method::POST, "/wip/v1/observe", nested));
    assert_eq!(response.status(), http::StatusCode::PAYLOAD_TOO_LARGE);
    assert!(response.body().is_empty());
    assert!(!response.headers().contains_key(http::header::CONTENT_TYPE));
}

#[test]
fn logical_observation_limit_remains_protocol_resource_failure() {
    let temporary = tempdir().expect("temporary root");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.max_observation_depth = 0;
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("http://localhost/wip").expect("endpoint");
    let request = ObserveRequest {
        path: "/".into(),
        depth: 1,
    };

    let response = server.handle_http(
        encode_observe_request(&endpoint, &request, limits).expect("encode observation"),
    );
    assert_eq!(response.status(), http::StatusCode::UNPROCESSABLE_ENTITY);
    assert!(matches!(
        decode_observe_response(&request, &response, limits).expect("decode logical limit"),
        DecodedResponse::ProtocolFailure {
            error,
            status_mismatch: None,
        } if error.code == ProtocolErrorCode::ResourceLimitExceeded
    ));
}

#[test]
fn stalled_request_body_times_out_and_releases_the_only_worker() {
    let temporary = tempdir().expect("temporary root");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.max_concurrent_requests = 1;
    config.limits.request_timeout = Duration::from_millis(200);
    let limits = config.limits.wire;
    let bound = Arc::new(
        WipFsServer::new(config)
            .expect("build server")
            .bind()
            .expect("bind"),
    );
    let address = bound.local_addr();
    let shutdown = Arc::new(AtomicBool::new(false));
    let worker = {
        let bound = Arc::clone(&bound);
        let shutdown = Arc::clone(&shutdown);
        thread::spawn(move || bound.serve_until(shutdown))
    };

    let mut stalled = TcpStream::connect(address).expect("connect stalled client");
    stalled
        .set_read_timeout(Some(Duration::from_secs(1)))
        .expect("set stalled read timeout");
    write!(
        stalled,
        "POST /wip/v1/observe HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: 100\r\n\r\n{{"
    )
    .expect("write partial request");
    stalled.flush().expect("flush partial request");
    thread::sleep(Duration::from_millis(25));

    let endpoint = Endpoint::parse(&format!("http://localhost:{}/wip", address.port()))
        .expect("client endpoint");
    let observe = ObserveRequest {
        path: "/".into(),
        depth: 0,
    };
    let started = Instant::now();
    let response = exchange(
        address,
        encode_observe_request(&endpoint, &observe, limits).expect("encode observation"),
    );
    let _root =
        success(decode_observe_response(&observe, &response, limits).expect("decode observation"));
    assert!(
        started.elapsed() < Duration::from_secs(1),
        "the valid request remained starved after the intake deadline"
    );

    let mut byte = [0_u8; 1];
    match stalled.read(&mut byte) {
        Ok(0) => {}
        Err(error)
            if matches!(
                error.kind(),
                std::io::ErrorKind::ConnectionReset
                    | std::io::ErrorKind::ConnectionAborted
                    | std::io::ErrorKind::BrokenPipe
            ) => {}
        outcome => panic!("stalled connection was not closed after its deadline: {outcome:?}"),
    }

    shutdown.store(true, Ordering::Release);
    worker
        .join()
        .expect("join listener")
        .expect("listener exits cleanly");
}

#[test]
fn call_argument_decoding_does_not_mask_missing_target() {
    let temporary = tempdir().expect("temporary root");
    let config = ServerConfig::for_root(temporary.path());
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("http://localhost/wip").expect("endpoint");
    let body = br#"{"target":{"path":"/missing"},"interface":{"reference":{"scope":"/","name":"unknown"}},"operation":"missing","arguments":{"bytes":"not-base64"}}"#;
    let request = http::Request::builder()
        .method(http::Method::POST)
        .uri(endpoint.route(Route::CallOperation).as_str())
        .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE)
        .header(http::header::ACCEPT, "application/json")
        .body(body.to_vec())
        .expect("build request");
    let response = server.handle_http(request);
    assert_eq!(response.status(), http::StatusCode::NOT_FOUND);
    assert!(
        std::str::from_utf8(response.body())
            .expect("UTF-8 error")
            .contains("\"not_found\"")
    );
}

#[test]
fn backend_invalid_unicode_path_is_not_found_not_internal() {
    let temporary = tempdir().expect("temporary root");
    let config = ServerConfig::for_root(temporary.path());
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("http://localhost/wip").expect("endpoint");
    let request = ObserveRequest {
        path: "/bad\0name".into(),
        depth: 0,
    };
    let response = server.handle_http(
        encode_observe_request(&endpoint, &request, limits).expect("encode observation"),
    );
    assert_eq!(response.status(), http::StatusCode::NOT_FOUND);
    match decode_observe_response(&request, &response, limits).expect("decode missing response") {
        DecodedResponse::ProtocolFailure { error, .. } => {
            assert_eq!(error.code, ProtocolErrorCode::NotFound);
        }
        DecodedResponse::Success(_) => panic!("backend-invalid path must not resolve"),
    }
}

#[test]
fn read_response_size_failure_is_transport_local() {
    let temporary = tempdir().expect("temporary root");
    std::fs::write(temporary.path().join("large"), vec![b'x'; 2_000]).expect("write file");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.wire = Limits::new(4_096, 1_024, 64).expect("wire limits");
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("http://localhost/wip").expect("endpoint");

    let observe_request = ObserveRequest {
        path: "/large".into(),
        depth: 0,
    };
    let response = server.handle_http(
        encode_observe_request(&endpoint, &observe_request, limits).expect("encode observation"),
    );
    let file = success(
        decode_observe_response(&observe_request, &response, limits).expect("decode observation"),
    )
    .object;
    let interface_request = FetchInterfaceRequest {
        interface: file.interfaces[0].clone(),
    };
    let response = server.handle_http(
        encode_fetch_interface_request(&endpoint, &interface_request, limits)
            .expect("encode interface fetch"),
    );
    let interface = success(
        decode_fetch_interface_response(&interface_request, &response, limits)
            .expect("decode interface"),
    );
    let call = CallOperationRequest {
        target: Target {
            path: "/large".into(),
            validator: file.validator,
        },
        interface: InterfaceTarget {
            scope_ref: None,
            reference: file.interfaces[0].clone(),
            validator: None,
        },
        operation: "read".into(),
        arguments: BTreeMap::from([("length".into(), Value::Integer(1_500))]),
    };
    let response = server.handle_http(
        encode_call_operation_request(&endpoint, &call, &interface.descriptor, limits)
            .expect("encode read call"),
    );
    assert_eq!(response.status(), http::StatusCode::INTERNAL_SERVER_ERROR);
    assert!(response.body().is_empty());
    assert!(!response.headers().contains_key(http::header::CONTENT_TYPE));
    assert!(matches!(
        decode_call_operation_response(&call, &interface.descriptor, &response, limits)
            .expect_err("oversized representation is not a protocol failure"),
        ClientResponseError::TransportBinding(failure)
            if failure.status == http::StatusCode::INTERNAL_SERVER_ERROR
                && failure.kind == TransportFailureKind::HttpStatus
    ));
}

fn success<T>(response: DecodedResponse<T>) -> T {
    match response {
        DecodedResponse::Success(value) => value,
        DecodedResponse::ProtocolFailure { error, .. } => panic!("protocol failure: {error}"),
    }
}

fn json_request(method: http::Method, uri: &str, body: &[u8]) -> http::Request<Vec<u8>> {
    http::Request::builder()
        .method(method)
        .uri(uri)
        .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE)
        .header(http::header::ACCEPT, "application/json")
        .body(body.to_vec())
        .expect("build JSON request")
}

fn exchange(address: SocketAddr, request: http::Request<Vec<u8>>) -> http::Response<Vec<u8>> {
    let mut stream = TcpStream::connect(address).expect("connect listener");
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .expect("set read timeout");
    let path = request
        .uri()
        .path_and_query()
        .map_or("/", |value| value.as_str());
    write!(
        stream,
        "{} {} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n",
        request.method(),
        path
    )
    .expect("write request line");
    for (name, value) in request.headers() {
        write!(
            stream,
            "{}: {}\r\n",
            name,
            value.to_str().expect("ASCII request header")
        )
        .expect("write header");
    }
    write!(stream, "Content-Length: {}\r\n\r\n", request.body().len())
        .expect("write content length");
    stream.write_all(request.body()).expect("write body");
    stream.flush().expect("flush request");

    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).expect("read response");
    let split = raw
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .expect("HTTP response header");
    let head = std::str::from_utf8(&raw[..split]).expect("UTF-8 response head");
    let mut lines = head.split("\r\n");
    let status = lines
        .next()
        .and_then(|line| line.split_whitespace().nth(1))
        .and_then(|status| status.parse::<u16>().ok())
        .expect("response status");
    let mut builder = http::Response::builder().status(status);
    for line in lines {
        let (name, value) = line.split_once(':').expect("response header");
        builder = builder.header(name, value.trim());
    }
    builder
        .body(raw[split + 4..].to_vec())
        .expect("build response")
}