use crate::door::assistant_test_support::*;
use serde_json::{Value, json};
use std::{
sync::{Arc, atomic::Ordering, mpsc},
time::Duration,
};
fn decision(f: &Fixture, draft: &str) -> Result<String> {
written(f.answer(
"POST",
"/decide",
&json!({"draft":draft,"decision":"approve"}),
))
}
fn post(f: &Fixture, body: &Value) -> std::thread::JoinHandle<Result<String>> {
let request = f.request("POST", "", body);
let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
std::thread::spawn(move || {
written(crate::door::route::answer(
&door,
&site,
&request,
crate::serve::pass::now(),
))
})
}
fn recorded(text: &str, code: &str) -> Result {
anyhow::ensure!(
text.starts_with("HTTP/1.1 200 ")
&& text.contains("\"seat_told\":false")
&& text.contains("approved")
&& text.contains(code)
&& text.contains("/changes"),
"recorded decision misreported: {text}"
);
Ok(())
}
#[test]
fn r3_03_pharmacy_draft_at_wards_never_calls_lys_decide() -> Result {
let f = Fixture::new(json!({}))?;
let read = ack(f.seat("agent-pass", "wards")?);
let text = decision(&f, "draft-pharmacy")?;
assert!(
text.starts_with("HTTP/1.1 409 ") && text.contains("draft_other_seat"),
"{text}"
);
assert_eq!(
f.door.issuer().decisions.load(Ordering::SeqCst),
0,
"Lys was asked to decide another seat's draft"
);
check(
f.answer("POST", "", &json!({"text":"still here"})),
202,
"id",
)?;
assert_eq!(read.recv()?["op"]["text"], "still here");
Ok(())
}
#[test]
fn r3_04_recorded_decision_survives_seat_departure() -> Result {
let f = Fixture::new(json!({}))?;
let seat = f.seat("agent-pass", "wards")?;
let mut stream = Stream::open(f.answer("GET", "/events", &json!({})))?;
stream.event()?;
*f.door
.issuer()
.after_decision
.lock()
.map_err(|_| anyhow::anyhow!("action"))? = Some(Box::new(move || {
drop(seat);
assert!(stream.event().is_ok_and(|e| e.contains("false")));
assert!(stream.close().is_ok());
}));
recorded(&decision(&f, "draft-accepted")?, "assistant_not_started")
}
#[test]
fn r3_04_recorded_decision_survives_seat_refusal() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let worker = std::thread::spawn(move || -> Result {
let op = seat.read()?;
seat.send(&json!({"refused":{"id":op["op"]["id"],"code":"seat_busy","message":"turn already closed"}}))
});
let text = decision(&f, "draft-accepted")?;
worker.join().map_err(|_| anyhow::anyhow!("seat"))??;
recorded(&text, "seat_busy")
}
#[test]
fn r3_04_recorded_decision_survives_pending_operation_bound() -> Result {
let f = Fixture::new(json!({"max_per_person":1}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let first = post(&f, &json!({"text":"held"}));
let op = seat.read()?;
let text = decision(&f, "draft-accepted")?;
seat.send(&json!({"ack":{"id":op["op"]["id"]}}))?;
first.join().map_err(|_| anyhow::anyhow!("post"))??;
recorded(&text, "assistant_bound")
}
#[test]
fn r3_04_recorded_decision_survives_ack_deadline() -> Result {
let f = Fixture::new(json!({"door_timeout_ms":1000}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let request = f.request(
"POST",
"/decide",
&json!({"draft":"draft-accepted","decision":"approve"}),
);
let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
let worker = std::thread::spawn(move || {
written(crate::door::route::answer(
&door,
&site,
&request,
crate::serve::pass::now(),
))
});
let op = seat.read()?;
assert!(op["op"].get("decided").is_some());
let text = worker.join().map_err(|_| anyhow::anyhow!("post"))??;
recorded(&text, "assistant_ack_unknown")
}
#[test]
fn r3_05_lys_transport_failures_are_502() -> Result {
let f = Fixture::new(json!({}))?;
let seat = f.seat("agent-pass", "wards")?;
for draft in ["draft-unavailable", "draft-malformed"] {
let text = decision(&f, draft)?;
assert!(
text.starts_with("HTTP/1.1 502 ") && text.contains("lys_unavailable"),
"{text}"
);
}
drop(seat);
Ok(())
}
async fn native_answer(status: u16, body: &[u8]) -> Result<crate::serve::storage::Failure> {
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
let issuer = format!("http://{}", listener.local_addr()?);
let bytes = body.to_vec();
let worker = std::thread::spawn(move || -> Result {
let (socket, _) = listener.accept()?;
crate::door::http::read(&mut std::io::BufReader::new(&socket), 16384, 1_048_576)?;
crate::door::http::Response::new(status, "application/json", bytes).write_to(&socket)?;
Ok(())
});
let client = crate::door::assistant_lys::Client {
issuer,
http: reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()?,
};
let result = client
.decide("person-pass", "draft-native", "approve")
.await;
worker.join().map_err(|_| anyhow::anyhow!("issuer"))??;
result
.err()
.ok_or_else(|| anyhow::anyhow!("invalid Lys reply admitted"))
}
#[tokio::test(flavor = "current_thread")]
async fn r3_05_lys_refusal_keeps_its_own_words_and_code() -> Result {
let error = native_answer(
403,
br#"{"code":"DraftNotPending","message":"The pharmacist already refused this draft."}"#,
)
.await?;
assert_eq!(error.code, "DraftNotPending");
assert_eq!(error.message, "The pharmacist already refused this draft.");
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn r3_05_malformed_lys_bytes_are_unavailable() -> Result {
assert_eq!(native_answer(200, b"{").await?.code, "lys_unavailable");
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn r3_05_unreachable_lys_is_unavailable() -> Result {
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
let issuer = format!("http://{}", listener.local_addr()?);
drop(listener);
let client = crate::door::assistant_lys::Client {
issuer,
http: reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()?,
};
let error = client
.decide("person-pass", "draft-native", "approve")
.await
.err()
.ok_or_else(|| anyhow::anyhow!("unreachable Lys answered"))?;
assert_eq!(error.code, "lys_unavailable");
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn r3_15_draft_lookup_has_a_page_budget() -> Result {
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
let issuer = format!("http://{}", listener.local_addr()?);
let (sent, read) = mpsc::channel();
let worker = std::thread::spawn(move || -> Result {
let mut count = 0;
for step in 0..=64 {
listener.set_nonblocking(false)?;
let (socket, _) = listener.accept()?;
count += 1;
let request =
crate::door::http::read(&mut std::io::BufReader::new(&socket), 16384, 1_048_576)?;
assert_eq!(
request.parameter("limit").as_deref(),
Some("64"),
"unbounded Lys page"
);
let response = if step == 64 {
crate::door::http::Response::json(
503,
&json!({"code":"fixture_budget_exceeded","message":"too many pages"}),
)
} else {
crate::door::http::Response::json(
200,
&json!({"drafts":[],"next":format!("page-{step}")}),
)
};
response.write_to(&socket)?;
if step == 63 {
sent.send(count)?;
break;
}
}
Ok(())
});
let client = crate::door::assistant_lys::Client {
issuer,
http: reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()?,
};
let answer = client
.decide("person-pass", "draft-native", "approve")
.await;
let joined = worker.join();
let error = answer
.err()
.ok_or_else(|| anyhow::anyhow!("unlimited draft search"))?;
assert_eq!(error.code, "lys_draft_pages_exceeded");
joined.map_err(|_| anyhow::anyhow!("issuer"))??;
assert_eq!(read.recv()?, 64);
Ok(())
}