diff --git a/package.json b/package.json index 94b50f0..a18177d 100644 --- a/package.json +++ b/package.json @@ -35,6 +35,7 @@ "check:updates": "node scripts/check-updates.js", "smoke:torrent": "node scripts/smoke-torrent.js", "smoke:torrent:failure-paths": "node scripts/smoke-torrent.js --failure-paths", + "test:torrent:rpc": "cd src-tauri && cargo test torrent_probe::tests::http_rpc_harness -- --nocapture", "verify:macos-signing": "node scripts/verify-macos-signing.js", "preview": "vite preview", "tauri": "tauri", diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 68aacde..2961c57 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3267,7 +3267,30 @@ pub(crate) async fn rpc_call( .await .map_err(|e| e.to_string())?; - let json: serde_json::Value = res.json().await.map_err(|e| e.to_string())?; + let http_status = res.status(); + let body = res.text().await.map_err(|e| e.to_string())?; + let json: serde_json::Value = serde_json::from_str(&body).map_err(|error| { + if http_status.is_success() { + error.to_string() + } else { + format!( + "HTTP {} {}: invalid JSON-RPC response: {error}", + http_status.as_u16(), + http_status.canonical_reason().unwrap_or("response error") + ) + } + })?; + if !http_status.is_success() { + let detail = json + .get("error") + .map(serde_json::Value::to_string) + .unwrap_or_else(|| "aria2 returned no JSON-RPC error".to_string()); + return Err(format!( + "HTTP {} {}: {detail}", + http_status.as_u16(), + http_status.canonical_reason().unwrap_or("response error") + )); + } if let Some(error) = json.get("error") { return Err(error.to_string()); } diff --git a/src-tauri/src/torrent_probe.rs b/src-tauri/src/torrent_probe.rs index 6360ac0..85fee99 100644 --- a/src-tauri/src/torrent_probe.rs +++ b/src-tauri/src/torrent_probe.rs @@ -298,9 +298,273 @@ async fn wait_for_stopped(client: &C, gid: &str) -> Result<(), Str #[cfg(test)] mod tests { use super::*; + use axum::{ + body::Body, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, + routing::post, + Json, Router, + }; + use std::collections::HashMap; use std::collections::VecDeque; + use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; + use tokio::sync::{oneshot, Notify}; + + enum ScriptedReply { + Result(Value), + RpcError(String), + HttpError(StatusCode, String), + Malformed(String), + Delay(Duration, Box), + Hang, + } + + struct RecordedCall { + method: String, + params: Value, + } + + struct ScriptedRpcState { + secret: String, + scripts: Mutex>>, + calls: Mutex>, + call_notification: Notify, + } + + struct ScriptedRpcServer { + address: SocketAddr, + state: Arc, + shutdown: Option>, + task: Option>, + } + + impl ScriptedRpcServer { + async fn start(scripts: impl IntoIterator)>) -> Self { + let secret = "torrent-probe-test-secret".to_string(); + let state = Arc::new(ScriptedRpcState { + secret: secret.clone(), + scripts: Mutex::new( + scripts + .into_iter() + .map(|(method, replies)| (method, replies.into_iter().collect())) + .collect(), + ), + calls: Mutex::new(Vec::new()), + call_notification: Notify::new(), + }); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("scripted RPC listener should bind"); + let address = listener + .local_addr() + .expect("scripted RPC listener should have an address"); + let app = Router::new() + .route("/jsonrpc", post(scripted_rpc_handler)) + .with_state(Arc::clone(&state)); + let (shutdown, shutdown_signal) = oneshot::channel(); + let task = tokio::spawn(async move { + let _ = axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_signal.await; + }) + .await; + }); + Self { + address, + state, + shutdown: Some(shutdown), + task: Some(task), + } + } + + fn client(&self) -> Arc { + Arc::new(crate::Aria2RpcClient { + port: self.address.port(), + secret: self.state.secret.clone(), + }) + } + + fn calls(&self) -> Vec<(String, Value)> { + self.state + .calls + .lock() + .expect("scripted RPC call log lock should work") + .iter() + .map(|call| (call.method.clone(), call.params.clone())) + .collect() + } + + async fn wait_for_method(&self, method: &str, occurrence: usize) { + loop { + let count = self + .state + .calls + .lock() + .expect("scripted RPC call log lock should work") + .iter() + .filter(|call| call.method == method) + .count(); + if count >= occurrence { + return; + } + self.state.call_notification.notified().await; + } + } + + async fn shutdown(mut self) { + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + if let Some(task) = self.task.take() { + task.await.expect("scripted RPC server should stop cleanly"); + } + } + + async fn terminate(mut self) { + if let Some(task) = self.task.take() { + task.abort(); + let _ = task.await; + } + self.shutdown.take(); + } + } + + impl Drop for ScriptedRpcServer { + fn drop(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + } + } + + async fn scripted_rpc_handler( + State(state): State>, + Json(request): Json, + ) -> Response { + let id = request.get("id").cloned().unwrap_or(Value::Null); + let method = request + .get("method") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + let params = request.get("params").cloned().unwrap_or(Value::Null); + state + .calls + .lock() + .expect("scripted RPC call log lock should work") + .push(RecordedCall { + method: method.clone(), + params: params.clone(), + }); + state.call_notification.notify_waiters(); + + let token = params + .as_array() + .and_then(|params| params.first()) + .and_then(Value::as_str); + let expected_token = format!("token:{}", state.secret); + if token != Some(expected_token.as_str()) { + return rpc_error_response( + id, + StatusCode::UNAUTHORIZED, + "invalid RPC token".to_string(), + ); + } + + let reply = state + .scripts + .lock() + .expect("scripted RPC scripts lock should work") + .get_mut(&method) + .and_then(VecDeque::pop_front) + .unwrap_or_else(|| { + ScriptedReply::RpcError(format!("unexpected scripted RPC method {method}")) + }); + scripted_reply_response(id, reply).await + } + + async fn scripted_reply_response(id: Value, mut reply: ScriptedReply) -> Response { + loop { + match reply { + ScriptedReply::Delay(delay, next) => { + tokio::time::sleep(delay).await; + reply = *next; + } + ScriptedReply::Hang => return std::future::pending::().await, + ScriptedReply::Result(result) => { + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "result": result, + })) + .into_response(); + } + ScriptedReply::RpcError(message) => { + return rpc_error_response(id, StatusCode::OK, message); + } + ScriptedReply::HttpError(status, message) => { + return rpc_error_response(id, status, message); + } + ScriptedReply::Malformed(body) => { + return (StatusCode::OK, Body::from(body)).into_response(); + } + } + } + } + + fn rpc_error_response(id: Value, status: StatusCode, message: String) -> Response { + ( + status, + Json(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": 1, "message": message }, + })), + ) + .into_response() + } + + fn scripts( + entries: impl IntoIterator)>, + ) -> Vec<(String, Vec)> { + entries + .into_iter() + .map(|(method, replies)| (method.to_string(), replies)) + .collect() + } + + async fn probe_fixture() -> (tempfile::TempDir, PathBuf, PathBuf) { + let temporary = tempfile::tempdir().expect("temporary probe storage should exist"); + let probe_dir = temporary.path().join("probe"); + tokio::fs::create_dir(&probe_dir) + .await + .expect("probe directory should be created"); + let metadata_path = probe_dir.join("metadata.torrent"); + tokio::fs::write(&metadata_path, b"torrent metadata") + .await + .expect("metadata fixture should be writable"); + (temporary, probe_dir, metadata_path) + } + + async fn wait_for_path(path: &Path, should_exist: bool) { + let deadline = Instant::now() + Duration::from_secs(3); + while Instant::now() < deadline { + if path.exists() == should_exist { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!( + path.exists(), + should_exist, + "path state did not become {}: {}", + should_exist, + path.display() + ); + } struct FakeRpc { statuses: Mutex>>, @@ -638,4 +902,491 @@ mod tests { .await .expect("cancellation cleanup should remove the probe directory"); } + + fn status_reply(status: &str) -> ScriptedReply { + ScriptedReply::Result(json!({ "status": status })) + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_exercises_production_client_and_status_order() { + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ + status_reply("active"), + status_reply("waiting"), + status_reply("paused"), + status_reply("complete"), + status_reply("removed"), + ], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::Result(json!("gid-1"))], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let mut options = Map::new(); + options.insert("bt-metadata-only".to_string(), json!("true")); + options.insert("bt-save-metadata".to_string(), json!("true")); + let bytes = run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + options, + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect("HTTP-backed metadata probe should resolve"); + + assert_eq!(bytes, b"torrent metadata"); + let calls = server.calls(); + assert_eq!( + calls + .iter() + .map(|(method, _)| method.as_str()) + .collect::>(), + vec![ + "aria2.addUri", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.forceRemove", + "aria2.tellStatus", + ] + ); + assert!(calls.iter().all(|(_, params)| { + params + .as_array() + .and_then(|params| params.first()) + .and_then(Value::as_str) + == Some("token:torrent-probe-test-secret") + })); + let add_params = calls[0] + .1 + .as_array() + .expect("addUri params should be an array"); + assert_eq!( + add_params[1], + json!(["magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567"]) + ); + let options = add_params[2] + .as_object() + .expect("addUri options should be an object"); + assert_eq!( + options.get("bt-metadata-only"), + Some(&json!("true")), + "recorded addUri params: {add_params:?}" + ); + assert_eq!( + options.get("bt-save-metadata"), + Some(&json!("true")), + "recorded addUri params: {add_params:?}" + ); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("successful probe fixture should be removable"); + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_reports_add_uri_failure_without_fabricating_cleanup() { + let server = ScriptedRpcServer::start(scripts([( + "aria2.addUri", + vec![ScriptedReply::HttpError( + StatusCode::SERVICE_UNAVAILABLE, + "daemon is starting".to_string(), + )], + )])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let error = run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("addUri failure should fail metadata resolution"); + assert!(matches!( + error, + ProbeFailure::Metadata(message) if message.contains("could not start") + )); + assert_eq!( + server + .calls() + .iter() + .map(|(method, _)| method.as_str()) + .collect::>(), + vec!["aria2.addUri"] + ); + assert!(probe_dir.exists()); + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_covers_terminal_malformed_and_missing_gid_statuses() { + for (initial_status, expected_message) in [ + ( + ScriptedReply::Result(json!({ + "status": "error", + "errorCode": "19", + "errorMessage": "tracker rejected metadata", + })), + "could not resolve", + ), + (status_reply("removed"), "could not resolve"), + ( + ScriptedReply::Result(json!({ "unexpected": "shape" })), + "invalid metadata probe status", + ), + (ScriptedReply::Malformed("{".to_string()), "status failed"), + ] { + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![initial_status, status_reply("removed")], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::Result(json!("gid-1"))], + ), + ])) + .await; + let (_temporary, _probe_dir, metadata_path) = probe_fixture().await; + let error = run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("terminal or malformed status should fail the probe"); + assert!(matches!( + error, + ProbeFailure::Metadata(message) if message.contains(expected_message) + )); + server.shutdown().await; + } + + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ScriptedReply::RpcError("gid gid-1 not found".to_string())], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::RpcError("gid gid-1 not found".to_string())], + ), + ])) + .await; + let (_temporary, _probe_dir, metadata_path) = probe_fixture().await; + let error = run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("a missing GID should fail metadata resolution"); + assert!(matches!( + error, + ProbeFailure::Metadata(message) if message.contains("removed") + )); + assert_eq!( + server + .calls() + .iter() + .map(|(method, _)| method.as_str()) + .collect::>(), + vec!["aria2.addUri", "aria2.tellStatus", "aria2.forceRemove"] + ); + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_retries_polling_and_force_remove_outages() { + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ + ScriptedReply::HttpError( + StatusCode::SERVICE_UNAVAILABLE, + "temporary polling outage".to_string(), + ), + status_reply("complete"), + status_reply("removed"), + ], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::Result(json!("gid-1"))], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect("a transient polling outage should be retried"); + assert_eq!( + server + .calls() + .iter() + .map(|(method, _)| method.as_str()) + .collect::>(), + vec![ + "aria2.addUri", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.forceRemove", + "aria2.tellStatus", + ] + ); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("successful probe fixture should be removable"); + server.shutdown().await; + + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![status_reply("complete"), status_reply("complete")], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::HttpError( + StatusCode::SERVICE_UNAVAILABLE, + "temporary cleanup outage".to_string(), + )], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect("a completion race after forceRemove should be accepted"); + assert_eq!( + server + .calls() + .iter() + .map(|(method, _)| method.as_str()) + .collect::>(), + vec![ + "aria2.addUri", + "aria2.tellStatus", + "aria2.forceRemove", + "aria2.tellStatus", + ] + ); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("successful probe fixture should be removable"); + server.shutdown().await; + + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ + status_reply("complete"), + status_reply("active"), + status_reply("active"), + status_reply("removed"), + ], + ), + ( + "aria2.forceRemove", + vec![ + ScriptedReply::HttpError( + StatusCode::SERVICE_UNAVAILABLE, + "temporary cleanup outage".to_string(), + ), + ScriptedReply::RpcError("connection reset by peer".to_string()), + ScriptedReply::Result(json!("gid-1")), + ], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let error = run_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("cleanup must report an unverified active GID"); + assert!(matches!(error, ProbeFailure::Cleanup(_))); + server.wait_for_method("aria2.forceRemove", 3).await; + wait_for_path(&probe_dir, false).await; + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_covers_cancellation_before_and_after_gid_assignment() { + let server = ScriptedRpcServer::start(scripts([( + "aria2.addUri", + vec![ScriptedReply::Delay( + Duration::from_millis(250), + Box::new(ScriptedReply::Result(json!("gid-1"))), + )], + )])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let probe_path = metadata_path.clone(); + let client = server.client(); + let task = tokio::spawn(async move { + run_metadata_probe( + client, + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &probe_path, + Duration::from_secs(60), + Duration::from_secs(60), + ) + .await + }); + server.wait_for_method("aria2.addUri", 1).await; + task.abort(); + let _ = task.await; + wait_for_path(&probe_dir, false).await; + server.shutdown().await; + + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ScriptedReply::Hang, status_reply("removed")], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::Result(json!("gid-1"))], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let probe_path = metadata_path.clone(); + let client = server.client(); + let task = tokio::spawn(async move { + run_metadata_probe( + client, + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &probe_path, + Duration::from_secs(60), + Duration::from_secs(60), + ) + .await + }); + server.wait_for_method("aria2.tellStatus", 1).await; + task.abort(); + let _ = task.await; + server.wait_for_method("aria2.forceRemove", 1).await; + wait_for_path(&probe_dir, false).await; + server.terminate().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_rpc_harness_retains_probe_when_daemon_shuts_down_during_resolution_or_cleanup() { + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ScriptedReply::Delay( + Duration::from_millis(250), + Box::new(status_reply("active")), + )], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let probe_path = metadata_path.clone(); + let client = server.client(); + let task = tokio::spawn(async move { + run_metadata_probe( + client, + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &probe_path, + Duration::from_millis(40), + Duration::from_millis(5), + ) + .await + }); + server.wait_for_method("aria2.tellStatus", 1).await; + server.terminate().await; + let result = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("probe should finish after daemon shutdown") + .expect("probe task should not panic") + .expect_err("daemon shutdown should not report metadata success"); + assert!(matches!(result, ProbeFailure::Cleanup(_))); + assert!(probe_dir.exists()); + + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ("aria2.tellStatus", vec![status_reply("complete")]), + ( + "aria2.forceRemove", + vec![ScriptedReply::Delay( + Duration::from_secs(1), + Box::new(ScriptedReply::Result(json!("gid-1"))), + )], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let probe_path = metadata_path.clone(); + let client = server.client(); + let task = tokio::spawn(async move { + run_metadata_probe( + client, + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &probe_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + }); + server.wait_for_method("aria2.forceRemove", 1).await; + server.terminate().await; + let result = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("cleanup should finish after daemon shutdown") + .expect("cleanup task should not panic") + .expect_err("daemon shutdown should report cleanup uncertainty"); + assert!(matches!(result, ProbeFailure::Cleanup(_))); + assert!(probe_dir.exists()); + } } diff --git a/src-tauri/tests/README.md b/src-tauri/tests/README.md index 0d1ca99..aadfdf4 100644 --- a/src-tauri/tests/README.md +++ b/src-tauri/tests/README.md @@ -36,6 +36,14 @@ npm run smoke:torrent Use `node scripts/smoke-torrent.js --binary /path/to/aria2c` when validating a packaged or target-specific Aria2 binary. +Run the HTTP-boundary Torrent probe harness with a controllable local JSON-RPC +server. It drives the production Aria2 RPC client through scripted status, +outage, race, cancellation, and daemon-termination cases: + +```sh +npm run test:torrent:rpc +``` + Run the deterministic unavailable-tracker and Aria2-daemon-exit checks with: ```sh