From 1253b470a55ad3258479c1bf47694cf84f5702bf Mon Sep 17 00:00:00 2001 From: rays Date: Sat, 7 Nov 2026 04:19:59 +0000 Subject: [PATCH] Tests for torrents: a real swarm, the guard and the scan's deferral (#397) torrent.rs's fetch, and scan.rs's spawn_torrent and torrent_one, ran in no test. A test now makes a swarm on 127.0.0.1: a seeding librqbit session and a web server that serves the .torrent and answers as its tracker, naming the seed. A detached torrent then downloads through ipx's own session into the enclosure's own folder, the row is recorded done with that folder as its path, the file is byte for byte the seed's, and the slot is free again. The .torrent is refused from an internal address by the guarded client unless fetch_internal allows it. A scan hands a .torrent enclosure to the torrent worker, counted as a torrent and left 'downloading' while it waits for a slot, and skips it, still counted, with torrents off. A test's torrent session joins no public DHT and sends no LAN multicast (cfg!(test) in Torrents::new); production's is unchanged. Co-Authored-By: Claude Opus 5.5 --- crates/ipx/src/scan.rs | 33 ++++++++++ crates/ipx/src/torrent.rs | 133 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+) diff --git a/crates/ipx/src/scan.rs b/crates/ipx/src/scan.rs index 5f7d472..f58c00f 100644 --- a/crates/ipx/src/scan.rs +++ b/crates/ipx/src/scan.rs @@ -1060,6 +1060,39 @@ mod tests { assert_eq!(ctx.db.feed_summary("iconic").await.unwrap().image, Some(icon)); } + #[tokio::test] + async fn a_torrent_enclosure_goes_to_the_torrent_worker_not_the_file_downloader() { + let body = b"TOneg1\ + "; + let scan = |ctx: Arc| async move { + let got = feed::Fetched::Body { bytes: body.to_vec(), etag: None, last_modified: None }; + let cfg = config::Feed { url: "http://127.0.0.1:9/feed".into(), ..feed() }; + match scan_one(&ctx, "t", &cfg, &db::HttpState::default(), true, Some((Ok((got, None)), None))) + .await + .unwrap() + { + Outcome::Feed(s) => (s.downloaded, s.torrents), + _ => panic!("a feed"), + } + }; + let state = |ctx: Arc| async move { ctx.db.strings_for_test("SELECT state FROM enclosures").await }; + + // Detached, as the daemon runs them: no slot free, so it waits its turn as 'downloading'. + let mut ctx = test_ctx(config::Config::default()).await; + ctx.detach_torrents = true; + ctx.torrent_slots = Arc::new(tokio::sync::Semaphore::new(0)); + let ctx = Arc::new(ctx); + assert_eq!(scan(ctx.clone()).await, (0, 1), "counted as a torrent, not a download"); + assert_eq!(state(ctx).await, ["downloading"], "so the next scan does not queue it again"); + + // Turned off, it is skipped and still counted. + let mut cfg = config::Config::default(); + cfg.torrent.enabled = false; + let ctx = Arc::new(test_ctx(cfg).await); + assert_eq!(scan(ctx.clone()).await, (0, 1)); + assert_eq!(state(ctx).await, ["skipped"]); + } + #[tokio::test] async fn nothing_from_a_feed_that_never_posted_is_an_error_and_from_one_that_has_is_not() { let base = empty_or_feed_server().await; diff --git a/crates/ipx/src/torrent.rs b/crates/ipx/src/torrent.rs index f44ad40..880e16c 100644 --- a/crates/ipx/src/torrent.rs +++ b/crates/ipx/src/torrent.rs @@ -29,6 +29,9 @@ impl Torrents { let opts = SessionOptions { listen: Some(ListenerOptions { listen_addr, ..Default::default() }), client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()), + // A test's swarm is two sessions and a tracker on 127.0.0.1, not the public DHT or the LAN. + dht: if cfg!(test) { None } else { Some(Default::default()) }, + disable_local_service_discovery: cfg!(test), ..Default::default() }; @@ -184,6 +187,136 @@ pub fn ratio(uploaded: u64, total: u64) -> f64 { mod tests { use super::*; + fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port() + } + + /// A swarm on 127.0.0.1: a seeding session holding `file`, and a web server answering for the + /// .torrent (`/ep.torrent`) and as its tracker, naming the seed as the one peer. The server's + /// address, and the seeding session, which stops seeding when dropped. + async fn swarm(file: &Path) -> (SocketAddr, Arc) { + let web = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let at = web.local_addr().unwrap(); + let seed_port = free_port(); + let made = librqbit::create_torrent( + file, + librqbit::CreateTorrentOptions { + name: None, + trackers: vec![format!("http://{at}/announce")], + piece_length: Some(16 * 1024), + }, + &librqbit::spawn_utils::BlockingSpawner::new(2), + ) + .await + .unwrap(); + let torrent = made.as_bytes().unwrap(); + // Compact peers: the address's four bytes, then the port's two. + let mut peers = b"d8:intervali60e5:peers6:".to_vec(); + peers.extend_from_slice(&[127, 0, 0, 1]); + peers.extend_from_slice(&seed_port.to_be_bytes()); + peers.push(b'e'); + let app = axum::Router::new() + .route("/ep.torrent", axum::routing::get(move || std::future::ready(torrent.clone()))) + .route("/announce", axum::routing::get(move || std::future::ready(peers.clone()))); + tokio::spawn(async move { axum::serve(web, app).await }); + + let seed = Session::new_with_opts( + file.parent().unwrap().to_owned(), + SessionOptions { + dht: None, + disable_local_service_discovery: true, + listen: Some(ListenerOptions { + listen_addr: format!("127.0.0.1:{seed_port}").parse().unwrap(), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .unwrap(); + seed.add_torrent( + AddTorrent::from_bytes(made.as_bytes().unwrap()), + Some(AddTorrentOptions { + output_folder: Some(file.parent().unwrap().display().to_string()), + overwrite: true, + ..Default::default() + }), + ) + .await + .unwrap(); + (at, seed) + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_detached_torrent_lands_in_its_own_folder_is_recorded_and_frees_its_slot() { + let root = std::env::temp_dir().join(format!("ipx-torrent-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let (seeded, downloads) = (root.join("seed"), root.join("downloads")); + std::fs::create_dir_all(&seeded).unwrap(); + let episode: Vec = (0..200_000u32).map(|i| (i * 7 % 251) as u8).collect(); + std::fs::write(seeded.join("episode.mp3"), &episode).unwrap(); + let (at, _seed) = swarm(&seeded.join("episode.mp3")).await; + + let mut cfg = Config::default(); + cfg.general.download_dir = downloads.clone(); + cfg.torrent.enabled = true; + let port = free_port(); + cfg.torrent.port_range = format!("{port}-{port}"); + cfg.torrent.seed_ratio = 0.0; + let mut ctx = crate::test_util::test_ctx(cfg.clone()).await; + ctx.detach_torrents = true; + let ctx = Arc::new(ctx); + let url = format!("http://{at}/ep.torrent"); + ctx.db + .exec_for_test(&format!( + "INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES (9, 'f', 'g', '{url}', 'downloading')" + )) + .await + .unwrap(); + + crate::scan::spawn_torrent(&ctx, "f".into(), 9, url.clone(), downloads.join("Show")); + let row = tokio::time::timeout(Duration::from_secs(90), async { + loop { + let row = ctx.db.enclosure(9).await.unwrap().unwrap(); + if row.state != "downloading" { + return row; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + }) + .await + .expect("the torrent finished within 90 s"); + let folder = own_folder(&downloads.join("Show"), 9); + assert_eq!((row.state.as_str(), row.last_error.as_deref()), ("done", None)); + assert_eq!(row.path.as_deref(), Some(folder.display().to_string().as_str()), "the folder is ours (#318)"); + assert_eq!(std::fs::read(folder.join("episode.mp3")).unwrap(), episode); + // The slot is let go once the row is written; give the task its moment. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(ctx.torrent_slots.available_permits(), 2, "the slot is free again"); + let _ = std::fs::remove_dir_all(&root); + } + + #[tokio::test] + async fn a_torrent_file_from_an_internal_address_is_refused_unless_allowed() { + let app = axum::Router::new().route("/ep.torrent", axum::routing::get(|| async { "d4:infode" })); + let web = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let at = web.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(web, app).await }); + let guarded = |allowed: &[&str]| { + let allowed = + crate::feed::guard::Allowed::parse(&allowed.iter().map(|s| s.to_string()).collect::>()).unwrap(); + reqwest::Client::builder().connector_layer(crate::feed::guard::Guard::new(allowed)).build().unwrap() + }; + let url = format!("http://{at}/ep.torrent"); + let err = torrent_file(&guarded(&[]), &url).await.unwrap_err(); + assert!(format!("{err:#}").contains("fetching the .torrent"), "{err:#}"); + assert_eq!( + torrent_file(&guarded(&["127.0.0.1"]), &url).await.unwrap(), + b"d4:infode", + "fetch_internal allows it" + ); + } + #[test] fn ratio_handles_the_empty_torrent() { assert_eq!(ratio(0, 100), 0.0);