TransferSu/tests/test_transfer.py

"""transfer.py 核心逻辑测试：解析器 / 日志轮转 / 参数解析 / 后端命令构建"""
import re
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import transfer

# ---------- 解析器: rsync ----------

def test_rsync_progress2_line():
    line = "     31,457,280  12%  356.77MB/s    0:00:00 (xfr#1, to-chk=51/53)"
    s = transfer.RsyncParser.feed_progress2(line)
    assert s is not None
    assert s.pct == 12
    assert s.bytes_done == 31457280
    assert s.files_done == 1
    assert s.files_total == 53
    assert s.speed_bps > 0

def test_rsync_progress2_complete():
    line = "   243,401,728 100%  305.39MB/s    0:00:00 (xfr#52, to-chk=0/53)"
    s = transfer.RsyncParser.feed_progress2(line)
    assert s is not None
    assert s.pct == 100
    assert s.files_done == 52
    assert s.files_total == 53

def test_rsync_progress2_unparseable_returns_none():
    assert transfer.RsyncParser.feed_progress2("building file list") is None
    assert transfer.RsyncParser.feed_progress2("") is None

def test_rsync_single_file_line():
    line = "file_1.bin          31,457,280 100%  356.77MB/s    0:00:00"
    f = transfer.RsyncParser.feed_file(line)
    assert f is not None
    assert f.name == "file_1.bin"
    assert f.pct == 100

def test_rsync_error_detection():
    assert transfer.RsyncParser.is_error("rsync: [sender] write error: Broken pipe (32)")
    assert transfer.RsyncParser.is_error("rsync error: some files/attrs were not transferred (code 23)")
    assert not transfer.RsyncParser.is_error("file_1.bin 100%")
    assert not transfer.RsyncParser.is_error("")

# ---------- 解析器: rclone ----------

def test_rclone_file_line():
    line = " *                                    file_7.bin: 98% / 7.023 MiB, 2.715 MiB/s"
    f = transfer.RcloneParser.feed_file(line)
    assert f is not None
    assert f.name == "file_7.bin"
    assert f.pct == 98

def test_rclone_total_line():
    line = "Transferred:     	  232.126 MiB / 232.126 MiB, 100%, 7.997 MiB/s, ETA 0s"
    s = transfer.RcloneParser.feed_total(line)
    assert s is not None
    assert s.pct == 100

def test_rclone_files_count_line():
    line = "Transferred:           52 / 52, 100%"
    s = transfer.RcloneParser.feed_files(line)
    assert s is not None
    assert s.files_done == 52
    assert s.files_total == 52

# ---------- 日志轮转 ----------

def test_logmanager_creates_and_appends(tmp_path):
    lm = transfer.LogManager(tmp_path, keep=20)
    path = lm.open_session("test")
    assert os.path.exists(path)
    lm.write("hello")
    lm.write("world")
    content = open(path).read()
    assert "hello" in content
    assert "world" in content
    assert content.index("hello") < content.index("world")  # 追加顺序正确

def test_logmanager_keeps_only_latest_20(tmp_path):
    lm = transfer.LogManager(tmp_path, keep=20)
    for i in range(25):
        lm.open_session(f"run{i}")
        lm.write(f"data{i}")
    lm.prune()
    files = sorted(f for f in os.listdir(tmp_path) if f.endswith(".log"))
    assert len(files) == 20
    run_nums = [int(re.search(r"run(\d+)", f).group(1)) for f in files]
    assert 5 in run_nums
    assert 4 not in run_nums

def test_logmanager_filenames_sorted_chronologically(tmp_path):
    lm = transfer.LogManager(tmp_path, keep=20)
    p1 = lm.open_session("a")
    p2 = lm.open_session("b")
    assert p1 != p2
    assert p2 > p1  # 时间戳递增

# ---------- 参数解析 ----------

def test_parse_args_with_positional():
    args = transfer.parse_args(["src", "dst"])
    assert args.source == "src"
    assert args.dest == "dst"
    assert args.backend == "rsync"
    assert args.keep_logs == 20
    assert args.mode == "tui"  # 未显式指定 cli 时默认 TUI

def test_parse_args_explicit_cli():
    args = transfer.parse_args(["--mode", "cli", "src", "dst"])
    assert args.mode == "cli"

def test_parse_args_no_positional_goes_tui():
    args = transfer.parse_args([])
    assert args.mode == "tui"
    assert args.source is None

def test_parse_args_full_options():
    args = transfer.parse_args([
        "--backend", "rclone", "--bwlimit", "30M", "--resume",
        "--checksum", "--mode", "cli", "--keep-logs", "5",
        "--logdir", "/tmp/x", "src", "dst",
    ])
    assert args.backend == "rclone"
    assert args.bwlimit == "30M"
    assert args.resume is True
    assert args.checksum is True
    assert args.keep_logs == 5
    assert args.logdir == "/tmp/x"

# ---------- 后端命令构建 ----------

def test_rsync_backend_cmd():
    cmd = transfer.RsyncBackend.build_cmd("a/", "b/", bwlimit="30000", resume=True, checksum=False)
    assert cmd[0] == "rsync"
    joined = " ".join(cmd)
    assert "--info=progress2" in joined
    assert "--bwlimit=30000" in joined
    assert "--partial" in joined
    assert "--append" in joined
    assert "-a" in joined

def test_rsync_backend_checksum_flag():
    cmd = transfer.RsyncBackend.build_cmd("a/", "b/", bwlimit="", resume=False, checksum=True)
    assert "--checksum" in " ".join(cmd)

def test_rclone_backend_cmd():
    cmd = transfer.RcloneBackend.build_cmd("a/", "b/", bwlimit="8M", transfers=4)
    assert cmd[0] == "rclone"
    joined = " ".join(cmd)
    assert "--bwlimit" in joined
    assert "--transfers" in joined
    assert "--progress" in joined

def test_tar_pv_backend_cmd():
    cmd = transfer.TarPvBackend.build_cmd("a/", "b/")
    assert "tar" in cmd
    assert "pv" in cmd
    assert "|" in cmd  # 管道串接

# ---------- 冲突策略: 参数 ----------

def test_policy_default_overwrite():
    args = transfer.parse_args(["src", "dst"])
    assert args.policy == "overwrite"

def test_policy_skip_flag():
    args = transfer.parse_args(["--policy", "skip", "src", "dst"])
    assert args.policy == "skip"

def test_policy_ask_flag():
    args = transfer.parse_args(["--policy", "ask", "src", "dst"])
    assert args.policy == "ask"

def test_policy_synonyms():
    assert transfer.parse_args(["--overwrite", "s", "d"]).policy == "overwrite"
    assert transfer.parse_args(["--skip", "s", "d"]).policy == "skip"
    assert transfer.parse_args(["--ask", "s", "d"]).policy == "ask"

# ---------- 冲突策略: 扫描解析 ----------

def test_scan_rsync_new_file():
    conflicts = transfer.ConflictScanner.parse_rsync_line(">f+++++++++|4|new.txt")
    assert conflicts == ("new.txt", "new")

def test_scan_rsync_update():
    conflicts = transfer.ConflictScanner.parse_rsync_line(">f.st......|123|f.txt")
    assert conflicts == ("f.txt", "update")

def test_scan_rsync_metadata_only_update():
    conflicts = transfer.ConflictScanner.parse_rsync_line(">f..t......|0|same.txt")
    assert conflicts == ("same.txt", "update")

def test_scan_rsync_ignore_non_file_lines():
    assert transfer.ConflictScanner.parse_rsync_line("cd+++++++++|0|dir/") is None
    assert transfer.ConflictScanner.parse_rsync_line("*deleting   |0|old.txt") is None
    assert transfer.ConflictScanner.parse_rsync_line("") is None

# ---------- 冲突策略: 决策 ----------

def test_apply_policy_overwrite():
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    skipped = transfer.apply_policy(conflicts, "overwrite", {})
    assert skipped == []

def test_apply_policy_skip():
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    skipped = transfer.apply_policy(conflicts, "skip", {})
    assert set(skipped) == {"a.txt", "b.txt"}

def test_apply_policy_ask_mixed_answers():
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    answers = {"a.txt": "y", "b.txt": "n"}
    skipped = transfer.apply_policy(conflicts, "ask", answers)
    assert skipped == ["b.txt"]  # 仅跳过选 n 的

def test_apply_policy_ask_all_overwrite():
    conflicts = [("a.txt", "update")]
    skipped = transfer.apply_policy(conflicts, "ask", {"*": "a"})
    assert skipped == []

def test_apply_policy_ask_all_skip():
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    skipped = transfer.apply_policy(conflicts, "ask", {"*": "s"})
    assert set(skipped) == {"a.txt", "b.txt"}

# ---------- 后端 exclude 集成 ----------

def test_rsync_backend_exclude_from():
    cmd = transfer.RsyncBackend.build_cmd("a/", "b/", exclude_from="/tmp/excl.txt")
    assert "--exclude-from=/tmp/excl.txt" in cmd

def test_rclone_backend_skip_policy():
    cmd = transfer.RcloneBackend.build_cmd("a/", "b/", skip_existing=True)
    assert "--ignore-existing" in " ".join(cmd)

def test_rsync_backend_no_append_when_overwriting():
    """覆盖场景禁用 --append (追加模式会阻止重写)"""
    cmd = transfer.RsyncBackend.build_cmd("a/", "b/", resume=True, append=False)
    assert "--append" not in cmd
    assert "--partial" in cmd

def test_apply_policy_global_skip_keeps_explicit_yes():
    """先选 y 再选 s: 已确认覆盖的保留, 其余跳过"""
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    answers = {"a.txt": "y", "*": "s"}
    skipped = transfer.apply_policy(conflicts, "ask", answers)
    assert skipped == ["b.txt"]

def test_apply_policy_global_overwrite_keeps_explicit_no():
    """先选 n 再选 a: 已确认跳过的保留, 其余覆盖"""
    conflicts = [("a.txt", "update"), ("b.txt", "update")]
    answers = {"a.txt": "n", "*": "a"}
    skipped = transfer.apply_policy(conflicts, "ask", answers)
    assert skipped == ["a.txt"]

# ---------- 速度统计 ----------

def test_speed_stats_basic():
    h = [10 * 1024 * 1024, 20 * 1024 * 1024, 15 * 1024 * 1024]
    cur, peak, avg = transfer.speed_stats(h)
    assert cur == 15 * 1024 * 1024
    assert peak == 20 * 1024 * 1024
    assert avg == 15 * 1024 * 1024

def test_speed_stats_empty():
    assert transfer.speed_stats([]) == (0.0, 0.0, 0.0)

def test_speed_stats_single():
    cur, peak, avg = transfer.speed_stats([8 * 1024 * 1024])
    assert (cur, peak, avg) == (8 * 1024 * 1024, 8 * 1024 * 1024, 8 * 1024 * 1024)

# ---------- 目录大小轮询 (rsync 输出缓冲替代方案) ----------

def test_dir_size_returns_bytes(tmp_path):
    """du 目录返回字节数"""
    (tmp_path / "a.bin").write_bytes(b"x" * 1000)
    (tmp_path / "sub").mkdir()
    (tmp_path / "sub" / "b.bin").write_bytes(b"y" * 500)
    assert transfer.dir_size(str(tmp_path)) == 1500

def test_dir_size_missing_returns_zero():
    assert transfer.dir_size("/nonexistent/path/xyz") == 0