liminal-server 0.24.5

//! SRV-038 R1 and R2: the network's root is asked of Lys once the grants are
//! complete, and an absent root is an error in the log.

use super::root_presence::{LysRoot, RootPresence};
use std::fmt::Write as _;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener};
use std::sync::mpsc;
use tracing_subscriber::prelude::*;

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

/// A Lys that answers each `POST /api/grants/who` with the next of `answers`
/// (status, body) and reports every request body it read.
fn issuer(answers: Vec<(u16, &'static str)>) -> TestResult<(SocketAddr, mpsc::Receiver<String>)> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    let (asked, asks) = mpsc::channel();
    std::thread::spawn(move || -> std::io::Result<()> {
        for (status, body) in answers {
            let (socket, _) = listener.accept()?;
            let mut reader = BufReader::new(socket);
            let mut first = String::new();
            reader.read_line(&mut first)?;
            let mut length = 0;
            loop {
                let mut header = String::new();
                if reader.read_line(&mut header)? == 0 || 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)?;
                }
            }
            let mut request = vec![0; length];
            reader.read_exact(&mut request)?;
            let _sent = asked.send(format!(
                "{} {}",
                first.trim_end(),
                String::from_utf8_lossy(&request)
            ));
            write!(
                reader.get_mut(),
                "HTTP/1.1 {status} fixture\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            )?;
            reader.get_mut().flush()?;
        }
        Ok(())
    });
    Ok((address, asks))
}

const NONE_HOLDS: &str = r#"{"holders":[],"revision":3,"complete":true,"next":null}"#;
const ONE_HOLDS: &str = r#"{"holders":[{"holder":"person-tom","kind":"person","grant":"g1"}],"revision":4,"complete":false,"next":"person-tom"}"#;

fn runtime() -> TestResult<tokio::runtime::Runtime> {
    Ok(tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()?)
}

/// The lines the presence writes, by level, captured for one test.
#[derive(Default)]
struct Lines(std::sync::Mutex<Vec<(tracing::Level, String)>>);
struct Capture(std::sync::Arc<Lines>);
#[derive(Default)]
struct Message(String);
impl tracing::field::Visit for Message {
    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        // Writing into a String cannot fail.
        let _ = write!(self.0, " {}={value:?}", field.name());
    }
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        let _ = write!(self.0, " {}={value}", field.name());
    }
}
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_root" {
            let mut message = Message::default();
            event.record(&mut message);
            if let Ok(mut lines) = self.0.0.lock() {
                lines.push((*event.metadata().level(), message.0));
            }
        }
    }
}

fn asked(issuer_at: SocketAddr, answers: &[LysRoot]) -> TestResult<Vec<(tracing::Level, String)>> {
    let lines = std::sync::Arc::new(Lines::default());
    let subscriber = tracing_subscriber::registry().with(Capture(std::sync::Arc::clone(&lines)));
    let runtime = runtime()?;
    let presence = RootPresence::new(
        "ablative",
        &format!("http://{issuer_at}/api/"),
        "fixture-app-credential".to_owned(),
        runtime.handle().clone(),
    )?;
    assert_eq!(presence.answer(), LysRoot::NotAsked);
    tracing::subscriber::with_default(subscriber, || {
        for expected in answers {
            presence.ask();
            assert_eq!(&presence.answer(), expected);
        }
    });
    let lines = lines
        .0
        .lock()
        .map_err(|_| "captured lines poisoned")?
        .clone();
    Ok(lines)
}

#[test]
fn a_network_with_no_root_is_absent_at_start() -> TestResult {
    let (at, asks) = issuer(vec![(200, NONE_HOLDS)])?;
    let lines = asked(at, &[LysRoot::Absent])?;
    let ask = asks.try_recv()?;
    assert!(ask.starts_with("POST /api/grants/who "), "{ask}");
    assert!(
        ask.contains("\"liminal.network\"") && ask.contains("\"ablative\""),
        "{ask}"
    );
    assert!(
        ask.contains("\"admin\"") && ask.contains("\"page_size\":1"),
        "{ask}"
    );
    assert!(asks.try_recv().is_err(), "exactly one who ask");
    assert_eq!(lines.len(), 1, "{lines:?}");
    assert_eq!(lines[0].0, tracing::Level::ERROR);
    assert!(
        lines[0].1.contains("ablative") && lines[0].1.contains(&at.to_string()),
        "{lines:?}"
    );
    Ok(())
}

#[test]
fn a_rooted_network_is_held_and_asked_once() -> TestResult {
    let (at, asks) = issuer(vec![(200, ONE_HOLDS)])?;
    let lines = asked(at, &[LysRoot::Held])?;
    assert!(asks.try_recv().is_ok());
    assert!(asks.try_recv().is_err(), "exactly one who ask");
    assert!(lines.is_empty(), "a held root logs nothing: {lines:?}");
    Ok(())
}

#[test]
fn a_reset_asks_again() -> TestResult {
    let (at, asks) = issuer(vec![(200, ONE_HOLDS), (200, NONE_HOLDS)])?;
    let lines = asked(at, &[LysRoot::Held, LysRoot::Absent])?;
    assert!(asks.try_recv().is_ok() && asks.try_recv().is_ok());
    assert_eq!(lines.len(), 1, "{lines:?}");
    assert_eq!(lines[0].0, tracing::Level::ERROR);
    Ok(())
}

#[test]
fn a_refused_question_is_unknown_with_its_cause() -> TestResult {
    let (at, _asks) = issuer(vec![(403, r#"{"error":"not yours to read"}"#)])?;
    let lines = asked(
        at,
        &[LysRoot::Unknown {
            cause: "lys_root_refused",
        }],
    )?;
    assert_eq!(lines.len(), 1, "{lines:?}");
    assert_eq!(lines[0].0, tracing::Level::WARN);
    assert!(
        !lines[0].1.contains("not yours to read"),
        "no response words in the log: {lines:?}"
    );
    Ok(())
}