//! 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, pub log: Arc>>, pub handle: Option>, } impl Running { pub fn log(&self) -> Vec { self.log.lock().unwrap().clone() } pub fn wait_log(&self, part: &str) -> Vec { 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); } } 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, }; 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 ", after `delay`. pub fn answering(home: &TempDir, delay: Duration) -> mpsc::Receiver { 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, 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) }