Tests for the daemon's start, shutdown and CLI (#400)
The daemon ran as a process in no test but the browser suite's, which never stops it mid-job or starts a second. crates/ipx/tests/daemon.rs runs the built binary on a scratch config and SQLite database, against a feed served from the test whose episode takes minutes: ipx status goes through the socket and with --local does the work itself; a first start with no token writes one, keeps it out of the log, and it signs in; a second daemon on the socket is refused and the first keeps serving; SIGTERM mid-download stops it within 5 s, removes the socket, and leaves the episode queued; it starts again cleanly; and with no daemon, ipx status works on its own. A process of its own, since a signal sent to the test binary would reach every test in it. until_next_scan keeps its floor and ceiling: 10 minutes with nothing to check, 30 s with a feed due now. The abandoned download is taken again only at its feed's next check, not at the next start: #410. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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| {
|
||||
|
||||
218
crates/ipx/tests/daemon.rs
Normal file
218
crates/ipx/tests/daemon.rs
Normal file
@@ -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#"<?xml version="1.0"?><rss version="2.0"><channel><title>Slow</title><item><title>One</title><guid>g1</guid><enclosure url="http://127.0.0.1:{port}/ep.mp3" type="audio/mpeg" length="{EPISODE_BYTES}"/></item></channel></rss>"#
|
||||
);
|
||||
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);
|
||||
}
|
||||
Reference in New Issue
Block a user