boxmaker/crates/gatewayd/tests/support/gateway.rs

//! Running `gatewayd`'s serve loop in a test, against the fake Mattermost and the fake `loopd`.
//! Do not edit.

#![allow(dead_code)] // each test file uses a different part of this module

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread::JoinHandle;
use std::time::Duration;

use gatewayd::config::Config;
use gatewayd::secrets::Secret;
use gatewayd::serve::{Stop, Tuning, run};
use proto::Turn;

use crate::fake_loop::{done, serve_loop};
use crate::fake_mm::{FakeMm, KYLE, WsPeer};
use crate::tmp::TempDir;

pub const SHARED: &str = "c0000000000000000000000000";
pub const OTHER: &str = "o0000000000000000000000000";
pub const WAIT: Duration = Duration::from_secs(5);

pub struct Running {
    pub stop: Arc<AtomicBool>,
    pub log: Arc<Mutex<Vec<String>>>,
    pub handle: Option<JoinHandle<Stop>>,
}

impl Running {
    pub fn log(&self) -> Vec<String> {
        self.log.lock().unwrap().clone()
    }

    pub fn wait_log(&self, part: &str) -> Vec<String> {
        let until = std::time::Instant::now() + WAIT;
        while !self.log().iter().any(|l| l.contains(part)) {
            assert!(
                std::time::Instant::now() < until,
                "no log line with {part:?}: {:?}",
                self.log()
            );
            std::thread::sleep(Duration::from_millis(10));
        }
        self.log()
    }

    /// The `Stop` `run` returns by itself within 5 s; after that it is stopped, and the test fails.
    pub fn join_within(mut self) -> Stop {
        let handle = self.handle.take().unwrap();
        let until = std::time::Instant::now() + WAIT;
        while !handle.is_finished() && std::time::Instant::now() < until {
            std::thread::sleep(Duration::from_millis(10));
        }
        self.stop.store(true, Ordering::SeqCst);
        let stop = handle.join().unwrap();
        assert!(!matches!(stop, Stop::Asked), "run did not stop by itself");
        stop
    }

    pub fn finish(mut self) -> Stop {
        self.stop.store(true, Ordering::SeqCst);
        self.handle.take().unwrap().join().unwrap()
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::SeqCst);
        // Wait for `run` to finish before the test's home is removed, or it can recreate its
        // socket directory there afterwards (tip T44). The wait has a limit (T27).
        if let Some(handle) = self.handle.take() {
            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
            while !handle.is_finished() && std::time::Instant::now() < deadline {
                std::thread::sleep(std::time::Duration::from_millis(10));
            }
            if handle.is_finished() {
                let _ = handle.join();
            }
        }
    }
}

pub fn config(home: &TempDir, url: &str, extra: &str) -> Config {
    let text = format!(
        r#"
[mattermost]
url = "{url}"
[secrets.mattermost_token]
env = "NOT_READ_BY_RUN"
[allow]
users = ["{KYLE}"]
channels = ["{SHARED}"]
[paths]
home = "{}"
[limits]
typing_every_ms = 100
{extra}
"#,
        home.path().display()
    );
    Config::parse(&text).unwrap()
}

pub fn start(config: Config) -> Running {
    let stop = Arc::new(AtomicBool::new(false));
    let log = Arc::new(Mutex::new(Vec::new()));
    let tuning = Tuning {
        backoff: vec![Duration::from_millis(50)],
        poll: Duration::from_millis(20),
        rest_timeout: WAIT,
        notice: Duration::from_millis(300),
    };
    let (s, l) = (Arc::clone(&stop), Arc::clone(&log));
    let handle = std::thread::spawn(move || {
        let sink: gatewayd::serve::Log =
            Arc::new(move |line: &str| l.lock().unwrap().push(line.to_string()));
        run(config, Secret::new("TOKEN".to_string()), tuning, sink, &s)
    });
    Running {
        stop,
        log,
        handle: Some(handle),
    }
}

/// A fake loop that answers every turn with "answer to <content>", after `delay`.
pub fn answering(home: &TempDir, delay: Duration) -> mpsc::Receiver<Turn> {
    serve_loop(&home.path().join("run/loop/loop.sock"), move |_, turn| {
        std::thread::sleep(delay);
        vec![done(&format!("answer to {}", turn.content))]
    })
}

pub fn read_state(home: &TempDir) -> serde_json::Value {
    let text = std::fs::read_to_string(home.path().join("gateway/state.json")).unwrap();
    serde_json::from_str(&text).unwrap()
}

pub fn loop_dir(home: &TempDir) {
    std::fs::create_dir_all(home.path().join("run/loop")).unwrap();
}

/// Start with a fake Mattermost and a fake loop; the first WebSocket is returned.
pub fn up(home: &TempDir, delay: Duration) -> (FakeMm, Running, mpsc::Receiver<Turn>, WsPeer) {
    loop_dir(home);
    let turns = answering(home, delay);
    let mm = FakeMm::start();
    let running = start(config(home, &mm.url(), ""));
    let ws = mm.next_ws(WAIT);
    running.wait_log("gatewayd: connected to ");
    (mm, running, turns, ws)
}