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 <noreply@anthropic.com>

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"<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title><item><title>One</title><guid>g1</guid>\
+            <enclosure url=\"http://127.0.0.1:9/ep.torrent\" type=\"application/x-bittorrent\" length=\"10\"/></item></channel></rss>";
+        let scan = |ctx: Arc<Ctx>| 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<Ctx>| 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;

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<Session>) {
+        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<u8> = (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::<Vec<_>>()).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);