tauri-plugin-deeprespond-im 1.0.0-alpha

//! FilesState 的测试：附件的保存与释放、拖放的登记、对本机 HTTP 服务的上传与下载（续传、403、取消、改写元数据盒子）。

use std::collections::HashMap;
use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use super::transfer::{self, Expect, UploadPart};
use super::*;

fn state(root: &Path) -> FilesState {
    let paths = Paths { data: root.join("data"), temp: root.join("temp"), logs: root.join("logs") };
    for d in ["data", "temp", "picked", "dl"] {
        fs::create_dir_all(root.join(d)).unwrap();
    }
    let state = FilesState::new(paths, vec![root.join("picked")], vec![root.join("dl")], None);
    // 测试连本机：不经任何代理
    let _ = state.client.set(transfer::build_client(None, false));
    state
}

#[test]
fn persist_moves_copies_or_references() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = state(r);
    // Blob 落盘的临时文件：移入附件目录
    let tmp = r.join("temp/0123456789abcdef-截图.png");
    fs::write(&tmp, b"png").unwrap();
    let saved = files.persist(tmp.to_str().unwrap(), true, "org#app", "alice", 10).unwrap();
    assert!(saved.owned && saved.size == 3 && saved.mtime > 0);
    assert!(!tmp.exists());
    let outbox = fs::canonicalize(outbox_dir(&r.join("data"), "org#app", "alice")).unwrap();
    let moved = PathBuf::from(&saved.path);
    assert!(moved.starts_with(&outbox) && saved.path.ends_with("-截图.png") && !saved.path.contains("0123456789abcdef"));
    // 再次保存已在附件目录中的：原样交回
    assert_eq!(files.persist(&saved.path, false, "org#app", "alice", 10).unwrap().path, saved.path);
    // 小文件：复制，原文件不动
    let small = r.join("picked/small.pdf");
    fs::write(&small, b"12345").unwrap();
    let copy = files.persist(small.to_str().unwrap(), false, "org#app", "alice", 10).unwrap();
    assert!(copy.owned && small.exists() && Path::new(&copy.path).starts_with(&outbox));
    assert_eq!(fs::read(&copy.path).unwrap(), b"12345");
    // 别处的文件即使标为 temp 也不移动
    let marked = files.persist(small.to_str().unwrap(), true, "org#app", "alice", 10).unwrap();
    assert!(marked.owned && small.exists());
    // 大文件：只记路径，跨重启仍可读
    let outside = r.join("outside");
    fs::create_dir_all(&outside).unwrap();
    let big = outside.join("big.mov");
    fs::write(&big, vec![0u8; 100]).unwrap();
    assert_eq!(files.persist(big.to_str().unwrap(), false, "org#app", "alice", 10).unwrap_err().reason.as_deref(), Some("path_not_allowed"));
    files.allow.register_read(&big).unwrap();
    let reference = files.persist(big.to_str().unwrap(), false, "org#app", "alice", 10).unwrap();
    assert_eq!(reference, Persisted { path: big.to_string_lossy().into_owned(), owned: false, size: 100, mtime: read::mtime_ms(&fs::metadata(&big).unwrap()) });
    let restarted = state(r);
    assert!(restarted.check_read(&big).is_ok());
    restarted.release(big.to_str().unwrap(), false).unwrap();
    assert!(state(r).check_read(&big).is_err());
    assert!(big.exists(), "只记路径的原文件不能被删除");
    // 释放自己的副本：删除；附件目录以外的拒绝
    restarted.release(&copy.path, true).unwrap();
    assert!(!Path::new(&copy.path).exists());
    restarted.release(&copy.path, true).unwrap();
    assert_eq!(restarted.release(small.to_str().unwrap(), true).unwrap_err().reason.as_deref(), Some("path_not_allowed"));
    assert!(small.exists());
}

#[test]
fn drag_drop_registers_files_only() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = state(r);
    let outside = r.join("outside");
    fs::create_dir_all(outside.join("folder")).unwrap();
    fs::write(outside.join("a.txt"), b"a").unwrap();
    let dropped = files.register_dropped(&[outside.join("a.txt"), outside.join("folder"), outside.join("missing")]);
    assert_eq!(dropped, vec![outside.join("a.txt").to_string_lossy().into_owned()]);
    assert!(files.check_read(&outside.join("a.txt")).is_ok());
    assert!(files.check_read(&outside.join("folder")).is_err());
}

/// 测试用的 HTTP 服务收到的请求。
#[derive(Clone, Debug, Default)]
struct Recorded {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

type Handler = dyn Fn(&Recorded, &mut TcpStream) + Send + Sync;

/// 极简的 HTTP/1.1 服务：每个连接一个线程，处理一个请求后关闭。
fn serve(handler: Arc<Handler>) -> (String, Arc<Mutex<Vec<Recorded>>>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let log: Arc<Mutex<Vec<Recorded>>> = Arc::default();
    let seen = log.clone();
    std::thread::spawn(move || {
        for stream in listener.incoming().flatten() {
            let (handler, seen) = (handler.clone(), seen.clone());
            std::thread::spawn(move || {
                let mut stream = stream;
                let Some(req) = read_request(&mut stream) else { return };
                seen.lock().unwrap().push(req.clone());
                handler(&req, &mut stream);
            });
        }
    });
    (base, log)
}

fn read_request(stream: &mut TcpStream) -> Option<Recorded> {
    let mut reader = BufReader::new(stream.try_clone().ok()?);
    let mut line = String::new();
    reader.read_line(&mut line).ok()?;
    let mut parts = line.split_whitespace();
    let mut req = Recorded { method: parts.next()?.to_string(), path: parts.next()?.to_string(), ..Default::default() };
    loop {
        let mut h = String::new();
        reader.read_line(&mut h).ok()?;
        let h = h.trim_end();
        if h.is_empty() {
            break;
        }
        let (k, v) = h.split_once(':')?;
        req.headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string());
    }
    if let Some(len) = req.headers.get("content-length").and_then(|v| v.parse::<usize>().ok()) {
        req.body = vec![0u8; len];
        reader.read_exact(&mut req.body).ok()?;
    }
    Some(req)
}

fn respond(stream: &mut TcpStream, status: &str, headers: &[(&str, String)], body: &[u8]) {
    let mut head = format!("HTTP/1.1 {status}\r\nConnection: close\r\n");
    for (k, v) in headers {
        head.push_str(&format!("{k}: {v}\r\n"));
    }
    head.push_str("\r\n");
    let _ = stream.write_all(head.as_bytes());
    let _ = stream.write_all(body);
}

fn content() -> Vec<u8> {
    (0..300 * 1024u32).map(|i| (i % 251) as u8).collect()
}

/// /file 支持 Range 和 If-Range；/cut 只发一部分就断开；/expired 返回 403；/slow 发出一块后停住。
fn file_server() -> (String, Arc<Mutex<Vec<Recorded>>>) {
    let data = Arc::new(content());
    serve(Arc::new(move |req: &Recorded, stream: &mut TcpStream| {
        let etag = ("ETag", "\"v1\"".to_string());
        let route = req.path.split('?').next().unwrap();
        let route = if route.starts_with("/file") { "/file" } else { route };
        match route {
            "/file" | "/changed" => {
                let range = req.headers.get("range").and_then(|r| r.strip_prefix("bytes=")).and_then(|r| r.trim_end_matches('-').parse::<usize>().ok());
                let if_range_ok = req.headers.get("if-range").is_none_or(|v| v == "\"v1\"" && req.path.starts_with("/file"));
                match range {
                    Some(start) if if_range_ok => {
                        let body = &data[start..];
                        let cr = format!("bytes {start}-{}/{}", data.len() - 1, data.len());
                        respond(stream, "206 Partial Content", &[("Content-Length", body.len().to_string()), ("Content-Range", cr), etag], body);
                    }
                    _ => respond(stream, "200 OK", &[("Content-Length", data.len().to_string()), etag], &data),
                }
            }
            "/cut" => {
                let head = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nETag: \"v1\"\r\nConnection: close\r\n\r\n", data.len());
                let _ = stream.write_all(head.as_bytes());
                let _ = stream.write_all(&data[..100 * 1024]);
                let _ = stream.flush();
                let _ = stream.shutdown(std::net::Shutdown::Both);
            }
            "/expired" => respond(stream, "403 Forbidden", &[("Content-Length", "0".into())], b""),
            "/missing" => respond(stream, "404 Not Found", &[("Content-Length", "0".into())], b""),
            "/slow" => {
                let head = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", data.len());
                let _ = stream.write_all(head.as_bytes());
                let _ = stream.write_all(&data[..64 * 1024]);
                std::thread::sleep(std::time::Duration::from_secs(5));
            }
            _ => respond(stream, "200 OK", &[("Content-Length", "0".into())], b""),
        }
    }))
}

fn no_progress(_: u64, _: Option<u64>) {}

#[tokio::test]
async fn download_resumes_and_renames() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = state(r);
    let (base, log) = file_server();
    let dl = r.join("dl");
    let dest = dl.to_str().unwrap();

    // 第一次：中途断开，.part 留下
    let err = files.download("t1", "main", 1, &format!("{base}/cut"), Some(dest), true, Some("报告.pdf"), &no_progress).await.unwrap_err();
    assert_eq!(err.code, "network_error");
    assert!(!err.message.contains("127.0.0.1"), "错误信息不带地址：{}", err.message);
    let part = fs::canonicalize(&dl).unwrap().join("报告.pdf.part");
    assert_eq!(fs::metadata(&part).unwrap().len(), 100 * 1024);

    // 重新换取地址后：按 Range 续传（If-Range 带上 ETag）
    type Reports = Arc<Mutex<Vec<(u64, Option<u64>)>>>;
    let reported: Reports = Arc::default();
    let sink = reported.clone();
    let progress = move |loaded: u64, total: Option<u64>| sink.lock().unwrap().push((loaded, total));
    let (path, size) = files.download("t1", "main", 1, &format!("{base}/file?sig=2"), Some(dest), true, Some("报告.pdf"), &progress).await.unwrap();
    assert_eq!(size, 300 * 1024);
    assert_eq!(fs::read(&path).unwrap(), content());
    assert!(path.ends_with("报告.pdf") && !part.exists());
    let last = log.lock().unwrap().last().cloned().unwrap();
    assert_eq!(last.headers.get("range").map(String::as_str), Some("bytes=102400-"));
    assert_eq!(last.headers.get("if-range").map(String::as_str), Some("\"v1\""));
    let reported = reported.lock().unwrap().clone();
    assert_eq!(reported.first(), Some(&(100 * 1024, Some(300 * 1024))));
    assert_eq!(reported.last(), Some(&(300 * 1024, Some(300 * 1024))));
    assert!(reported.windows(2).all(|w| w[1].0 - w[0].0 >= 64 * 1024 || w[1].0 == 300 * 1024));
    // 下载的文件可以打开、定位
    assert!(files.check_read(&path).is_ok());

    // 重名：改为“名称 (1).扩展名”
    let (second, _) = files.download("t2", "main", 1, &format!("{base}/file"), Some(dest), true, Some("报告.pdf"), &no_progress).await.unwrap();
    assert!(second.ends_with("报告 (1).pdf"));
    // overwrite：覆盖同名文件
    let (third, _) = files.download("t3", "main", 1, &format!("{base}/file"), Some(dest), false, Some("报告.pdf"), &no_progress).await.unwrap();
    assert_eq!(third, path);
    // 文件名清理：不会写到目标目录以外
    let (evil, _) = files.download("t4", "main", 1, &format!("{base}/file"), Some(dest), true, Some("..\\..\\evil.exe"), &no_progress).await.unwrap();
    assert_eq!(evil, fs::canonicalize(&dl).unwrap().join("evil.exe"));
    // 没有 name 时取地址中的文件名；完整路径作为目标
    let (named, _) = files.download("t5", "main", 1, &format!("{base}/file/x.bin?sig=1"), Some(dest), true, None, &no_progress).await.unwrap();
    assert!(named.ends_with("x.bin"));
    let full = dl.join("sub/指定.dat");
    let (exact, _) = files.download("t6", "main", 1, &format!("{base}/file"), Some(full.to_str().unwrap()), true, Some("忽略.bin"), &no_progress).await.unwrap();
    assert!(exact.ends_with("sub/指定.dat"));
}

#[tokio::test]
async fn download_errors() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = state(r);
    let (base, _) = file_server();
    let dl = r.join("dl");
    let dest = dl.to_str().unwrap();
    // 403：地址过期，交回由核心重新换取
    let err = files.download("e1", "main", 1, &format!("{base}/expired"), Some(dest), true, Some("a.bin"), &no_progress).await.unwrap_err();
    assert_eq!(err.code, "url_expired");
    let err = files.download("e2", "main", 1, &format!("{base}/missing"), Some(dest), true, Some("a.bin"), &no_progress).await.unwrap_err();
    assert_eq!(err.code, "not_found");
    // 不允许写入的位置
    let outside = r.join("outside");
    fs::create_dir_all(&outside).unwrap();
    let err = files.download("e3", "main", 1, &format!("{base}/file"), Some(outside.to_str().unwrap()), true, Some("a.bin"), &no_progress).await.unwrap_err();
    assert_eq!(err.reason.as_deref(), Some("path_not_allowed"));
    // 服务端的文件变了（If-Range 不符）：从头下载，内容完整
    let _ = files.download("e4", "main", 1, &format!("{base}/cut"), Some(dest), true, Some("c.bin"), &no_progress).await.unwrap_err();
    let (path, size) = files.download("e4", "main", 1, &format!("{base}/changed"), Some(dest), true, Some("c.bin"), &no_progress).await.unwrap();
    assert_eq!((size, fs::read(&path).unwrap()), (300 * 1024, content()));
    // 别的运行留下的同名 .part 不续传
    fs::write(dl.join("stale.bin.part"), vec![9u8; 1000]).unwrap();
    let (path, _) = files.download("e5", "main", 1, &format!("{base}/file"), Some(dest), true, Some("stale.bin"), &no_progress).await.unwrap();
    assert_eq!(fs::read(&path).unwrap(), content());
}

#[tokio::test]
async fn download_cancel_removes_part() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = Arc::new(state(r));
    let (base, _) = file_server();
    let dest = r.join("dl");
    let canceller = files.clone();
    let progress = move |loaded: u64, _: Option<u64>| {
        if loaded > 0 {
            canceller.cancel("c1");
        }
    };
    let err = files.download("c1", "main", 1, &format!("{base}/slow"), Some(dest.to_str().unwrap()), true, Some("s.bin"), &progress).await.unwrap_err();
    assert_eq!(err.code, "aborted");
    assert!(!dest.join("s.bin.part").exists() && !dest.join("s.bin").exists());
    // 开始之前到达的取消
    files.cancel("c2");
    let err = files.download("c2", "main", 1, &format!("{base}/file"), Some(dest.to_str().unwrap()), true, Some("t.bin"), &no_progress).await.unwrap_err();
    assert_eq!(err.code, "aborted");
}

fn upload_server() -> (String, Arc<Mutex<Vec<Recorded>>>) {
    serve(Arc::new(|req: &Recorded, stream: &mut TcpStream| {
        let status = if req.path.starts_with("/denied") { "403 Forbidden" } else { "200 OK" };
        respond(stream, status, &[("Content-Length", "0".into()), ("ETag", "\"part\"".into())], b"");
    }))
}

fn part(url: String, offset: u64, length: u64, expect: Option<Expect>, patches: Vec<u64>) -> UploadPart {
    let headers = HashMap::from([("Content-Type".to_string(), "video/mp4".to_string()), ("Content-Length".to_string(), "1".to_string())]);
    UploadPart { path: PathBuf::new(), offset, length, url, method: "PUT".into(), headers, expect, patches }
}

#[tokio::test]
async fn upload_streams_ranges_with_patches() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = state(r);
    let (base, log) = upload_server();
    let (mut bytes, patches) = super::video::tests::synthetic_mp4();
    bytes.resize(200 * 1024, 0x55);
    let path = r.join("picked/movie.mp4");
    fs::write(&path, &bytes).unwrap();
    let meta = video::prepare_video(&path).unwrap();
    assert_eq!(meta.patches.len(), patches.len());
    let p = path.to_str().unwrap();
    let progress: Arc<Mutex<Vec<u64>>> = Arc::default();
    let sink = progress.clone();
    let report: Arc<dyn Fn(u64) + Send + Sync> = Arc::new(move |loaded| sink.lock().unwrap().push(loaded));
    let size = bytes.len() as u64;
    let status = files.upload_part("u1", ENGINE, 1, p, part(format!("{base}/up?sig=1"), 0, size, None, meta.patches.clone()), report).await.unwrap();
    assert_eq!(status, 200);
    let req = log.lock().unwrap().last().cloned().unwrap();
    assert_eq!(req.method, "PUT");
    // 对象存储要求明确的长度，不能是分块编码；调用方给的 Content-Length 被忽略
    assert_eq!(req.headers.get("content-length").map(String::as_str), Some(size.to_string().as_str()));
    assert!(!req.headers.contains_key("transfer-encoding"));
    assert_eq!(req.headers.get("content-type").map(String::as_str), Some("video/mp4"));
    let mut expected = bytes.clone();
    for &at in &meta.patches {
        expected[at as usize..at as usize + 4].copy_from_slice(b"free");
    }
    assert_eq!(req.body, expected);
    assert_eq!(fs::read(&path).unwrap(), bytes, "原文件不能被改动");
    assert_eq!(progress.lock().unwrap().as_slice(), &[65536, 131072, 196608, size]);

    // 中间一片：偏移相对文件
    let status = files.upload_part("u2", ENGINE, 1, p, part(format!("{base}/up"), 10, 1000, None, meta.patches.clone()), Arc::new(|_| {})).await.unwrap();
    assert_eq!(status, 200);
    assert_eq!(log.lock().unwrap().last().unwrap().body, expected[10..1010].to_vec());
    // 非 2xx 的状态码原样交回，由 JavaScript 判断
    let status = files.upload_part("u3", ENGINE, 1, p, part(format!("{base}/denied"), 0, 10, None, vec![]), Arc::new(|_| {})).await.unwrap();
    assert_eq!(status, 403);
}

const ENGINE: &str = crate::hub::ENGINE_LABEL;

#[tokio::test]
async fn upload_checks_attachment_and_cancels() {
    let dir = tempfile::tempdir().unwrap();
    let r = dir.path();
    let files = Arc::new(state(r));
    let (base, log) = upload_server();
    let path = r.join("picked/big.bin");
    fs::write(&path, vec![1u8; 300 * 1024]).unwrap();
    let p = path.to_str().unwrap();
    let meta = fs::metadata(&path).unwrap();
    let mtime = read::mtime_ms(&meta) as f64;
    // 大小和修改时间都对：照常上传
    let ok = Some(Expect { size: meta.len(), mtime });
    assert_eq!(files.upload_part("a1", ENGINE, 1, p, part(format!("{base}/up"), 0, 100, ok, vec![]), Arc::new(|_| {})).await.unwrap(), 200);
    let before = log.lock().unwrap().len();
    // 修改时间变了、大小变了：attachment_changed，不发请求
    for expect in [Expect { size: meta.len(), mtime: mtime - 2000.0 }, Expect { size: meta.len() + 1, mtime }] {
        let err = files.upload_part("a2", ENGINE, 1, p, part(format!("{base}/up"), 0, 100, Some(expect), vec![]), Arc::new(|_| {})).await.unwrap_err();
        assert_eq!((err.code.as_str(), err.reason.as_deref()), ("send_abandoned", Some("attachment_changed")));
    }
    assert_eq!(log.lock().unwrap().len(), before);
    // 超出文件的范围、不允许的路径
    assert!(files.upload_part("a3", ENGINE, 1, p, part(format!("{base}/up"), 300 * 1024, 1, None, vec![]), Arc::new(|_| {})).await.is_err());
    let err = files.upload_part("a4", ENGINE, 1, "/etc/hostname", part(format!("{base}/up"), 0, 1, None, vec![]), Arc::new(|_| {})).await.unwrap_err();
    assert_eq!(err.reason.as_deref(), Some("path_not_allowed"));
    // 进度到达后取消
    let canceller = files.clone();
    let report: Arc<dyn Fn(u64) + Send + Sync> = Arc::new(move |_| canceller.cancel("a5"));
    let err = files.upload_part("a5", ENGINE, 1, p, part(format!("{base}/up"), 0, 300 * 1024, None, vec![]), report).await.unwrap_err();
    assert_eq!(err.code, "aborted");
    assert!(files.tasks.lock().unwrap().is_empty());
}

#[test]
fn release_instance_cancels_uploads_only() {
    let dir = tempfile::tempdir().unwrap();
    let files = state(dir.path());
    let upload = files.start_task("up", ENGINE, 1, true).unwrap();
    let other = files.start_task("up2", ENGINE, 2, true).unwrap();
    let download = files.start_task("down", ENGINE, 1, false).unwrap();
    assert_eq!(files.start_task("up", ENGINE, 1, true).unwrap_err().reason.as_deref(), Some("task_running"));
    files.release_instance(ENGINE, 1);
    // 引擎消失：它发起的上传取消（新的引擎会续传），下载照常进行
    assert!(upload.is_cancelled());
    assert!(!other.is_cancelled() && !download.is_cancelled());
}