haematite-cli 0.21.1

use super::{Limits, serve_incoming};
use crate::door::assistant_test_support::*;
use serde_json::json;
use std::{
    io::{BufRead, BufReader, Read, Write},
    net::{Shutdown, TcpListener, TcpStream},
    os::unix::net::UnixStream,
    sync::{Arc, atomic::Ordering, mpsc},
    time::Duration,
};

#[test]
fn r3_gap_sse_counts_in_max_connections_and_hangup_frees_the_same_counter() -> Result {
    let f = Fixture::new(json!({"door_max_connections":1}))?;
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let mut browser = TcpStream::connect(listener.local_addr()?)?;
    browser.set_read_timeout(Some(Duration::from_secs(5)))?;
    let (socket, _) = listener.accept()?;
    let (send, receive) = mpsc::channel();
    send.send(Ok(socket))?;
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    let limits = Limits {
        max_connections: 1,
        read_timeout: Duration::from_secs(5),
        head_cap: 16384,
        body_cap: 1_048_576,
    };
    let worker =
        std::thread::spawn(move || serve_incoming(receive.into_iter(), &door, &site, &limits));
    write!(
        browser,
        "GET /api/assistant/wards/events HTTP/1.1\r\nHost: studio.example\r\nCookie: haem_session={}\r\n\r\n",
        f.cookie
    )?;
    let mut browser = BufReader::new(browser);
    loop {
        let mut line = String::new();
        anyhow::ensure!(browser.read_line(&mut line)? > 0, "SSE ended");
        if line.starts_with("data:") {
            break;
        }
    }
    assert_eq!(
        f.door.connections.load(Ordering::SeqCst),
        1,
        "SSE not counted"
    );
    let blocked = UnixStream::connect(f.files.path().join("assistant.sock"))?;
    blocked.set_read_timeout(Some(Duration::from_secs(5)))?;
    let mut refusal = String::new();
    BufReader::new(blocked).read_line(&mut refusal)?;
    assert!(refusal.contains("max_connections=1; count=1"), "{refusal}");
    browser.get_ref().shutdown(Shutdown::Write)?;
    let mut ended = String::new();
    browser.read_to_string(&mut ended)?;
    assert_eq!(
        f.door.connections.load(Ordering::SeqCst),
        0,
        "hangup retained slot"
    );
    drop(f.seat("agent-pass", "wards")?);
    drop(send);
    worker
        .join()
        .map_err(|_| anyhow::anyhow!("listener panic"))?;
    Ok(())
}