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(())
}