use super::*;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use tracing_subscriber::prelude::*;
type TestResult<T = ()> = Result<T, Box<dyn std::error::Error>>;
struct Issuer {
address: SocketAddr,
stopping: Arc<AtomicBool>,
accepts: Arc<AtomicU64>,
sockets: Arc<Mutex<Vec<TcpStream>>>,
worker: Option<std::thread::JoinHandle<std::io::Result<()>>>,
}
fn serve_connection(
socket: TcpStream,
statuses: &[u16],
requests: &AtomicU64,
) -> std::io::Result<()> {
let mut reader = BufReader::new(socket);
loop {
let mut first = String::new();
if reader.read_line(&mut first)? == 0 {
return Ok(());
}
if !first.starts_with("POST /api/grants/check/batch ") {
reader.get_ref().shutdown(Shutdown::Both)?;
return Err(std::io::Error::other("unexpected fixture route"));
}
let mut length = 0;
loop {
let mut header = String::new();
if reader.read_line(&mut header)? == 0 {
return Ok(());
}
if header == "\r\n" {
break;
}
if let Some((name, value)) = header.split_once(':')
&& name.eq_ignore_ascii_case("content-length")
{
length = value.trim().parse().map_err(std::io::Error::other)?;
}
}
assert!(length <= 16_384);
reader.read_exact(&mut vec![0; length])?;
let index = usize::try_from(requests.fetch_add(1, Ordering::Relaxed))
.map_err(std::io::Error::other)?;
let status = statuses
.get(index)
.ok_or_else(|| std::io::Error::other("unexpected fixture ask"))?;
let body = if *status == 200 {
r#"{"revision":1,"results":[{"allowed":true,"grant":"fixture-grant","mode":"outright","via":null,"refusal":null,"reason":null,"degraded":null}],"degraded":null}"#
} else {
""
};
write!(
reader.get_mut(),
"HTTP/1.1 {status} fixture\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: keep-alive\r\n\r\n{body}",
body.len()
)?;
reader.get_mut().flush()?;
}
}
impl Issuer {
fn origin(&self) -> String {
format!("http://{}/api/", self.address)
}
fn new(statuses: Vec<u16>) -> TestResult<Self> {
let listener = TcpListener::bind("127.0.0.1:0")?;
let address = listener.local_addr()?;
let stopping = Arc::new(AtomicBool::new(false));
let accepts = Arc::new(AtomicU64::new(0));
let sockets = Arc::new(Mutex::new(Vec::new()));
let stop = Arc::clone(&stopping);
let accepted = Arc::clone(&accepts);
let active = Arc::clone(&sockets);
let worker = std::thread::spawn(move || {
let statuses = Arc::new(statuses);
let requests = Arc::new(AtomicU64::new(0));
let mut workers = Vec::new();
loop {
let (socket, _) = listener.accept()?;
{
let mut sockets = active
.lock()
.map_err(|_| std::io::Error::other("fixture sockets poisoned"))?;
if stop.load(Ordering::Acquire) {
break;
}
accepted.fetch_add(1, Ordering::Release);
sockets.push(socket.try_clone()?);
}
let statuses = Arc::clone(&statuses);
let requests = Arc::clone(&requests);
workers.push(std::thread::spawn(move || {
serve_connection(socket, &statuses, &requests)
}));
}
for worker in workers {
worker
.join()
.map_err(|_| std::io::Error::other("fixture request worker panicked"))??;
}
Ok(())
});
Ok(Self {
address,
stopping,
accepts,
sockets,
worker: Some(worker),
})
}
fn stop(&mut self) -> TestResult {
let Some(worker) = self.worker.take() else {
return Ok(());
};
{
let sockets = self
.sockets
.lock()
.map_err(|_| "fixture sockets poisoned")?;
self.stopping.store(true, Ordering::Release);
for socket in sockets.iter() {
socket.shutdown(Shutdown::Both)?;
}
}
drop(TcpStream::connect(self.address)?);
worker
.join()
.map_err(|_| "fixture accept worker panicked")??;
Ok(())
}
}
impl Drop for Issuer {
fn drop(&mut self) {
if let Err(error) = self.stop() {
eprintln!("fixture shutdown failed: {error}");
}
}
}
#[derive(Clone, Debug, Default)]
struct AskLine {
route: String,
ok: Option<bool>,
elapsed_ms: Option<f64>,
}
impl tracing::field::Visit for AskLine {
fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
if field.name() == "route" {
self.route = value.to_owned();
}
}
fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
if field.name() == "ok" {
self.ok = Some(value);
}
}
fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
if field.name() == "elapsed_ms" {
self.elapsed_ms = Some(value);
}
}
fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
}
struct Capture(std::sync::mpsc::Sender<AskLine>);
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Capture {
fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
if event.metadata().target() == "liminal::lys_ask" {
assert_eq!(*event.metadata().level(), tracing::Level::DEBUG);
let mut fields = AskLine::default();
event.record(&mut fields);
assert!(self.0.send(fields).is_ok());
}
}
}
fn runtime() -> TestResult<tokio::runtime::Runtime> {
Ok(tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()?)
}
#[test]
fn a_direct_check_batch_against_the_fixture_answers() -> TestResult {
let mut issuer = Issuer::new(vec![200])?;
let runtime = runtime()?;
let ask = HttpLysAsk::new(&issuer.origin(), runtime.handle().clone())?;
let target = Target::new("liminal.network", "fixture", "serve")?;
let answer = runtime.block_on(ask.client.check_batch(
"fixture-pass",
"fixture-holder",
std::slice::from_ref(&target),
None,
))?;
assert!(answer.results.first().is_some_and(|result| result.allowed));
assert_eq!(issuer.accepts.load(Ordering::Acquire), 1);
issuer.stop()
}
#[test]
fn every_lys_ask_logs_its_route_result_and_elapsed_and_updates_statistics() -> TestResult {
let mut issuer = Issuer::new(vec![200, 403])?;
let runtime = runtime()?;
let ask = HttpLysAsk::new(&issuer.origin(), runtime.handle().clone())?;
let (sender, lines) = std::sync::mpsc::channel();
let subscriber = tracing_subscriber::registry().with(Capture(sender));
let target = Target::new("liminal.network", "fixture", "serve")?;
tracing::subscriber::with_default(subscriber, || -> TestResult {
let granted = ask.ask("fixture-pass", "fixture-holder", &target, 1)?;
assert!(matches!(granted, LiveAnswer::Outright { .. }));
assert_eq!(ask.statistics().count, 1);
let refused = ask.ask("fixture-pass", "fixture-holder", &target, 2);
assert!(refused.is_err());
assert_eq!(ask.statistics().count, 2);
Ok(())
})?;
let lines = lines.try_iter().collect::<Vec<_>>();
assert_eq!(lines.len(), 2);
assert_eq!(lines[0].ok, Some(true));
assert_eq!(lines[1].ok, Some(false));
let mut maximum = 0.0_f64;
for line in &lines {
assert_eq!(line.route, "/api/grants/check/batch");
maximum = maximum.max(line.elapsed_ms.ok_or("elapsed_ms was not logged")?);
}
assert_eq!(ask.statistics().max_elapsed_ms.to_bits(), maximum.to_bits());
let health = ask_statistics()?;
assert!(health.count >= 2);
assert!(health.max_elapsed_ms >= maximum);
issuer.stop()
}
#[test]
fn two_lys_asks_reuse_one_fixture_tcp_connection() -> TestResult {
let mut issuer = Issuer::new(vec![200, 200])?;
let runtime = runtime()?;
let ask = HttpLysAsk::new(&issuer.origin(), runtime.handle().clone())?;
let target = Target::new("liminal.network", "fixture", "serve")?;
for now in [1, 2] {
ask.ask("fixture-pass", "fixture-holder", &target, now)?;
}
assert_eq!(issuer.accepts.load(Ordering::Acquire), 1);
issuer.stop()
}