diff --git a/crates/ipx/src/daemon.rs b/crates/ipx/src/daemon.rs index 94e5974..93dc60d 100644 --- a/crates/ipx/src/daemon.rs +++ b/crates/ipx/src/daemon.rs @@ -358,6 +358,16 @@ mod tests { use super::*; use crate::test_util::{feed, test_ctx}; + #[tokio::test] + async fn the_daemon_sleeps_at_least_30_s_and_at_most_10_minutes() { + let ctx = test_ctx(config::Config::default()).await; + assert_eq!(until_next_scan(&ctx).await, Duration::from_secs(600), "nothing to check: the ceiling"); + let mut cfg = config::Config::default(); + cfg.feeds.insert("never-checked".into(), feed()); + let ctx = test_ctx(cfg).await; + assert_eq!(until_next_scan(&ctx).await, Duration::from_secs(30), "due now: the floor, not a spin"); + } + #[tokio::test] async fn a_reload_takes_the_catalogue_another_ipx_wrote() { let with = |id: &str| { diff --git a/crates/ipx/tests/daemon.rs b/crates/ipx/tests/daemon.rs new file mode 100644 index 0000000..a095dca --- /dev/null +++ b/crates/ipx/tests/daemon.rs @@ -0,0 +1,218 @@ +//! The daemon as production runs it: the built binary, a scratch config and SQLite database, a +//! feed served from this process, and SIGTERM sent as Docker sends it (#400). A process of its +//! own, because a signal sent to the test binary would reach every test in it. + +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Output, Stdio}; +use std::time::{Duration, Instant}; + +const EPISODE_BYTES: u64 = 50_000_000; + +/// A feed of one episode, and the episode, sent 4 KB every 50 ms: a download that takes minutes. +fn serve_feed() -> u16 { + let l = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = l.local_addr().unwrap().port(); + std::thread::spawn(move || { + for sock in l.incoming().flatten() { + std::thread::spawn(move || answer(sock, port)); + } + }); + port +} + +fn answer(mut sock: TcpStream, port: u16) { + let mut line = String::new(); + if BufReader::new(&sock).read_line(&mut line).is_err() { + return; + } + if line.contains("/ep.mp3") { + let head = format!("HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nContent-Length: {EPISODE_BYTES}\r\n\r\n"); + // An MP3's frame sync first, so the file is taken for audio. + let mut chunk = vec![0u8; 4096]; + chunk[..2].copy_from_slice(&[0xFF, 0xFB]); + if sock.write_all(head.as_bytes()).is_err() { + return; + } + while sock.write_all(&chunk).is_ok() { + chunk[..2].fill(0); + std::thread::sleep(Duration::from_millis(50)); + } + return; + } + let body = format!( + r#"SlowOneg1"# + ); + let _ = sock.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{body}", body.len()).as_bytes()); +} + +struct Scratch { + root: PathBuf, + web: u16, +} + +impl Scratch { + fn new(feed_port: u16) -> Self { + let root = std::env::temp_dir().join(format!("ipx-daemon-test-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(root.join("data")).unwrap(); + let web = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port(); + std::fs::write( + root.join("config.toml"), + format!( + r#"[general] +download_dir = "{root}/downloads" +socket = "{root}/ipx.sock" +schedule = "every 60m" +fetch_internal = ["127.0.0.1"] + +[torrent] +enabled = false + +[web] +enabled = true +bind = "127.0.0.1:{web}" +token = "" + +[feeds.slow] +url = "http://127.0.0.1:{feed_port}/feed.xml" +"#, + root = root.display() + ), + ) + .unwrap(); + Self { root, web } + } + + fn ipx(&self, args: &[&str]) -> Command { + let mut c = Command::new(env!("CARGO_BIN_EXE_ipx")); + c.args(args) + .env("IPX_CONFIG", self.root.join("config.toml")) + .env("IPX_DATA_DIR", self.root.join("data")) + .env("IPX_ADMIN_PASSWORD", "a test password") + .env("IPX_LOG", "ipx=info") + .stdin(Stdio::null()); + // Nothing of the shell's: no Postgres, no traces, no JSON log, no TypeSafe key. + for v in + ["IPX_DATABASE_URL", "OTEL_EXPORTER_OTLP_ENDPOINT", "IPX_LOG_FORMAT", "TYPESAFE_KEY", "IPX_TYPESAFE_URL"] + { + c.env_remove(v); + } + c + } + + fn run(&self, args: &[&str]) -> Output { + self.ipx(args).output().unwrap() + } + + /// A daemon, its log going to `log`. + fn daemon(&self, log: &str) -> Child { + let log = std::fs::File::create(self.root.join(log)).unwrap(); + self.ipx(&["daemon"]).stdout(Stdio::null()).stderr(log).spawn().unwrap() + } + + fn log(&self, name: &str) -> String { + std::fs::read_to_string(self.root.join(name)).unwrap_or_default() + } + + fn socket(&self) -> PathBuf { + self.root.join("ipx.sock") + } + + /// Bytes of the download under way, in the downloads folder's incomplete files. + fn downloading(&self) -> u64 { + let dir = self.root.join("downloads/.ipx-incomplete"); + std::fs::read_dir(dir).map_or(0, |d| d.flatten().filter_map(|e| e.metadata().ok()).map(|m| m.len()).sum()) + } +} + +fn until(what: &str, limit: Duration, mut ok: impl FnMut() -> bool) { + let started = Instant::now(); + while !ok() { + assert!(started.elapsed() < limit, "{what}: not after {limit:?}"); + std::thread::sleep(Duration::from_millis(100)); + } +} + +fn text(o: &Output) -> String { + format!("{}{}", String::from_utf8_lossy(&o.stdout), String::from_utf8_lossy(&o.stderr)) +} + +/// SIGTERM, as `docker stop` sends; how long it took to exit. +fn terminate(child: &mut Child) -> Duration { + let started = Instant::now(); + assert!(Command::new("kill").args(["-TERM", &child.id().to_string()]).status().unwrap().success()); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success(), "the daemon exited {status}"); + return started.elapsed(); + } + if started.elapsed() > Duration::from_secs(10) { + let _ = child.kill(); + panic!("the daemon ignored SIGTERM for 10 s"); + } + std::thread::sleep(Duration::from_millis(50)); + } +} + +fn get(port: u16, path: &str) -> String { + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + write!(s, "GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n").unwrap(); + let mut out = String::new(); + let _ = s.read_to_string(&mut out); + out +} + +fn token_in(config: &Path) -> String { + let text = std::fs::read_to_string(config).unwrap(); + let line = text.lines().find(|l| l.trim_start().starts_with("token")).expect("a token line"); + line.split('"').nth(1).unwrap_or("").to_owned() +} + +#[test] +fn the_daemon_starts_answers_refuses_a_second_and_stops_on_sigterm_mid_download() { + let s = Scratch::new(serve_feed()); + let mut first = s.daemon("first.log"); + until("the control socket", Duration::from_secs(60), || s.socket().exists() && s.run(&["status"]).status.success()); + + // Through the socket: the daemon's own answer, in words. + let status = s.run(&["status"]); + assert!(text(&status).contains("1 feeds,"), "{}", text(&status)); + let local = s.run(&["--local", "status"]); + assert!(local.status.success() && text(&local).contains("1 feeds,"), "{}", text(&local)); + + // A first start with no token writes one, and it signs in. + let token = token_in(&s.root.join("config.toml")); + assert_eq!(token.len(), 64, "a session-strength token"); + assert!(!s.log("first.log").contains(&token), "the log says where the token is, never what it is"); + let me = get(s.web, &format!("/api/me?token={token}")); + assert!(me.starts_with("HTTP/1.1 200"), "{me}"); + assert!(get(s.web, "/api/me").starts_with("HTTP/1.1 401")); + + // Two daemons on one socket would fight over the database. + let second = s.ipx(&["daemon"]).output().unwrap(); + assert!(!second.status.success()); + assert!(text(&second).contains("a daemon is already listening"), "{}", text(&second)); + assert!(s.run(&["status"]).status.success(), "the first is still serving"); + + // The first pass starts at once and takes the episode, which takes minutes. SIGTERM stops it. + until("the download to start", Duration::from_secs(60), || s.downloading() > 0); + let took = terminate(&mut first); + assert!(took < Duration::from_secs(5), "stopped in {took:?}"); + assert!(!s.socket().exists(), "the socket is removed"); + assert!(s.log("first.log").contains("abandoning the job in progress"), "{}", s.log("first.log")); + + // Never recorded, the abandoned file is still queued, not lost or marked failed. The next start + // takes it at its feed's next check, not at once (#410). + let queued = s.run(&["--local", "status"]); + assert!(text(&queued).contains("1 queued to download, 0 downloaded"), "{}", text(&queued)); + let mut again = s.daemon("again.log"); + until("the restarted daemon", Duration::from_secs(60), || s.log("again.log").contains("daemon started")); + terminate(&mut again); + + // No daemon: the command does the work itself. + let alone = s.run(&["status"]); + assert!(alone.status.success() && text(&alone).contains("1 feeds,"), "{}", text(&alone)); + let _ = std::fs::remove_dir_all(&s.root); +}