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:
2026-11-07 04:24:54 +00:00
parent 1253b470a5
commit 2db42b5fae
2 changed files with 228 additions and 0 deletions

View File

@@ -358,6 +358,16 @@ mod tests {
use super::*; use super::*;
use crate::test_util::{feed, test_ctx}; 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] #[tokio::test]
async fn a_reload_takes_the_catalogue_another_ipx_wrote() { async fn a_reload_takes_the_catalogue_another_ipx_wrote() {
let with = |id: &str| { let with = |id: &str| {

218
crates/ipx/tests/daemon.rs Normal file
View 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);
}