liminal-server 0.25.4

//! SRV-044 R2: liminal binds its WebSocket address in the step that binds
//! the wire addresses, before Lys is asked and the store opens.
use std::net::{SocketAddr, TcpListener};

use super::{bind_listeners, bind_wire};
use crate::ServerError;
use crate::config::types::{LimitsConfig, ServerConfig, ServicesConfig, WebSocketConfig};
use crate::health::{ClusterReadiness, ReadinessState, SharedReadinessState};
use crate::server::connection::ConnectionSupervisor;
use crate::server::connection::websocket_bound_tests::upgrade_on_a_thread;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn config(websocket: Option<SocketAddr>) -> Result<ServerConfig, Box<dyn std::error::Error>> {
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 30_000,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: websocket.map(|listen_address| WebSocketConfig {
            listen_address,
            path: "/liminal".to_owned(),
            allowed_origins: Vec::new(),
            allowed_origin_ports: Vec::new(),
            ping_interval_ms: None,
        }),
    })
}

fn readiness() -> SharedReadinessState {
    SharedReadinessState::new(ReadinessState::new(
        false,
        false,
        ClusterReadiness::NotConfigured,
    ))
}

#[test]
fn a_websocket_address_in_use_refuses_in_the_wire_bind() -> TestResult {
    let taken = TcpListener::bind("127.0.0.1:0")?;
    let address = taken.local_addr()?;
    let refused = bind_wire(&config(Some(address))?);
    assert!(
        matches!(refused, Err(ServerError::ListenerBind { address: failed, .. }) if failed == address),
        "{:?}",
        refused.map(|_| ())
    );
    Ok(())
}

#[test]
fn a_websocket_client_during_the_start_is_upgraded_once_accepting_starts() -> TestResult {
    let config = config(Some("127.0.0.1:0".parse()?))?;
    let wire = bind_wire(&config)?;
    let address = wire
        .websocket
        .as_ref()
        .map(super::BoundWebSocket::local_addr)
        .ok_or("the [websocket] section binds its address")?;
    let client = upgrade_on_a_thread(address)?;
    let supervisor = ConnectionSupervisor::new()?;

    let (listener, websocket) = bind_listeners(wire, supervisor.clone(), &readiness())?;
    let status = client
        .join()
        .map_err(|_| "the upgrading client panicked")??;
    assert_eq!(status, 101, "the early client is upgraded");
    if let Some(websocket) = websocket {
        websocket.shutdown()?;
    }
    listener.shutdown()?;
    supervisor.shutdown();
    Ok(())
}

#[test]
fn no_websocket_section_binds_no_websocket_address() -> TestResult {
    let wire = bind_wire(&config(None)?)?;
    assert!(wire.websocket.is_none());
    Ok(())
}