diff --git a/.gitignore b/.gitignore index b8a0f13..9d4abb4 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ wheels/ .vscode/ *.exe sbackup.json +config.json # Virtual environments .venv diff --git a/sbackup/__init__.py b/sbackup/__init__.py index ea959ea..aa83199 100644 --- a/sbackup/__init__.py +++ b/sbackup/__init__.py @@ -9,7 +9,7 @@ from sbackup.i18n import set_locale, t from sbackup.config import load_config, save_lang, save_format from sbackup.compression import restore_backup -VERSION = "1.0.0" +VERSION = "1.0.1" logger = logging.getLogger(__name__) @@ -32,7 +32,8 @@ class LocalizedArgumentParser(argparse.ArgumentParser): localized = localized.replace( "unrecognized arguments: ", t("err.argparse.unrecognized_args") ) - localized = localized.replace("required", t("err.argparse.required")) + # "required" 仅替换独立出现的关键词(argparse 格式: "the following arguments are required") + localized = localized.replace("are required", t("err.argparse.required")) self.print_usage(sys.stderr) sys.stderr.write(f"{self.prog}: {localized}\n") sys.exit(2) @@ -87,6 +88,8 @@ def get_parser() -> argparse.ArgumentParser: subparsers.add_parser("all", help=t("cli.help.all")) + subparsers.add_parser("list", aliases=["history"], help=t("cli.help.list")) + save_parser = subparsers.add_parser("save", help=t("cli.help.save")) save_parser.add_argument( "--keep", type=int, default=0, help=t("cli.help.save.keep") @@ -119,6 +122,15 @@ def get_parser() -> argparse.ArgumentParser: restore_parser = subparsers.add_parser("restore", help=t("cli.help.restore")) restore_parser.add_argument("backup_file", help=t("cli.help.restore.file")) restore_parser.add_argument("target_dir", help=t("cli.help.restore.dir")) + restore_parser.add_argument( + "--password", default="", help=t("cli.help.restore.password") + ) + restore_parser.add_argument( + "-l", "--list", action="store_true", help=t("cli.help.restore.list") + ) + + verify_parser = subparsers.add_parser("verify", help=t("cli.help.verify")) + verify_parser.add_argument("backup_file", help=t("cli.help.verify.file")) sftp_parser = subparsers.add_parser("sftp", help=t("cli.help.sftp")) sftp_sub = sftp_parser.add_subparsers( @@ -218,7 +230,11 @@ def _handle_sftp(args, config) -> int: ) try: port = int(port_str) + if not (1 <= port <= 65535): + print(t("err.sftp.ssh", error=f"port {port} out of range")) + port = 22 except ValueError: + print(t("err.sftp.ssh", error=f"invalid port: {port_str}")) port = 22 user = args.user or input(t("cli.prompt.sftp.user") + " ") key_file_input = args.key_file or input(t("cli.prompt.sftp.key_file") + " ") @@ -247,12 +263,21 @@ def _handle_sftp(args, config) -> int: ) key_passphrase_input = "" else: - password = args.password or getpass.getpass( - t("cli.prompt.sftp.password") + " " - ) - key_passphrase_input = args.key_passphrase or getpass.getpass( - t("cli.prompt.sftp.key_passphrase") + " " - ) + # 用户指定了私钥,先尝试无密码加载,需要时再提示 + if not args.key_passphrase: + key_passphrase_input = _try_load_key_passphrase(key_file_input) + if key_passphrase_input is None: + # 用户放弃输入密码短语,回退到密码认证 + key_file_input = "" + password = args.password or getpass.getpass( + t("cli.prompt.sftp.password") + " " + ) + key_passphrase_input = "" + else: + password = "" + else: + password = "" + key_passphrase_input = args.key_passphrase remote_path = ( args.remote_path or input(t("cli.prompt.sftp.remote_path") + " ") or "/" @@ -320,7 +345,8 @@ def _handle_sftp(args, config) -> int: print(str(e)) return 1 - return 0 + print(t("cli.help.sftp.action")) + return 1 def _handle_webdav(args, config) -> int: @@ -362,7 +388,8 @@ def _handle_webdav(args, config) -> int: print(str(e)) return 1 - return 0 + print(t("cli.help.webdav.action")) + return 1 def run() -> int: @@ -426,6 +453,9 @@ def run() -> int: elif args.command == "all": print(manager.list_folder_table()) return 0 + elif args.command in ("list", "history"): + print(manager.format_history_table()) + return 0 elif args.command == "save": manager.execute_backups( keep=args.keep, @@ -437,7 +467,7 @@ def run() -> int: elif args.command == "watch": import time as _time - interval_sec = args.interval * 60 + interval_sec = max(args.interval, 0.1) * 60 # 最小 6 秒 print(t("cmd.watch.start", interval=args.interval)) try: while True: @@ -451,7 +481,17 @@ def run() -> int: except KeyboardInterrupt: return 0 elif args.command == "restore": - result = restore_backup(args.backup_file, args.target_dir) + if args.list: + from sbackup.compression import list_backup_contents + + print(list_backup_contents(args.backup_file, args.password)) + return 0 + result = restore_backup(args.backup_file, args.target_dir, args.password) + return 0 if result["success"] else 1 + elif args.command == "verify": + from sbackup.compression import verify_backup + + result = verify_backup(args.backup_file, args.password) return 0 if result["success"] else 1 elif args.command == "sftp": return _handle_sftp(args, config) diff --git a/sbackup/auto_save.py b/sbackup/auto_save.py index ff60120..6950fd0 100644 --- a/sbackup/auto_save.py +++ b/sbackup/auto_save.py @@ -2,6 +2,7 @@ import os import json import shutil import logging +import unicodedata from pathlib import Path from dataclasses import dataclass from sbackup.config import ( @@ -15,6 +16,8 @@ from sbackup.i18n import t logger = logging.getLogger(__name__) +_HISTORY_KEY = "_history" + @dataclass class BackupEntry: @@ -32,6 +35,8 @@ class BackupEntry: @staticmethod def from_list(data: list) -> "BackupEntry": """从 JSON 兼容的列表格式创建(向后兼容旧格式)""" + if not isinstance(data, list) or len(data) < 3: + return BackupEntry(mtime=0.0, target="", skip_patterns=[]) fmt = data[3] if len(data) > 3 else "" return BackupEntry( mtime=data[0], target=data[1], skip_patterns=data[2], compression_format=fmt @@ -131,6 +136,10 @@ class BackupManager: return False abs_path = os.path.abspath(folder_path) + abs_dest = os.path.abspath(target_folder) + if abs_path == abs_dest: + print(t("err.dest.invalid", path=target_folder)) + return False if abs_path in self.data: print(t("info.already.added", path=abs_path)) return False @@ -181,7 +190,7 @@ class BackupManager: skip_count = 0 uploaded_files = [] for key, raw in list(self.data.items()): - if key == "_history": + if key == _HISTORY_KEY: continue if not os.path.exists(key): print(t("warn.source.missing", path=key)) @@ -233,6 +242,28 @@ class BackupManager: # 向后兼容别名 save_folder = execute_backups + @staticmethod + def _resolve_key_passphrase(key_file: str, SFTPClient, SFTPError) -> str | None: + """ + 检测私钥是否需要密码短语,需要时交互式提示输入 + :return: 密码短语(空字符串表示不需要),None 表示用户放弃 + """ + try: + SFTPClient._load_private_key(key_file, "") + return "" + except SFTPError: + import getpass + + while True: + passphrase = getpass.getpass(t("cli.prompt.sftp.key_passphrase") + " ") + if not passphrase: + return None + try: + SFTPClient._load_private_key(key_file, passphrase) + return passphrase + except SFTPError: + print(t("err.sftp.wrong_passphrase")) + @staticmethod def _upload_to_sftp(file_paths: list[str], config: Config) -> None: """将备份文件上传到 SFTP 服务器""" @@ -252,49 +283,24 @@ class BackupManager: if default_key: print(t("cmd.sftp.using_default_key", path=default_key)) key_file = default_key - # 尝试加载私钥检测是否需要密码短语 - try: - SFTPClient._load_private_key(key_file, "") - except SFTPError: - # 需要密码短语,提示用户输入 - import getpass - - while True: - passphrase = getpass.getpass( - t("cli.prompt.sftp.key_passphrase") + " " - ) - if not passphrase: - print(t("cmd.sftp.no_default_key")) - return - try: - SFTPClient._load_private_key(key_file, passphrase) - key_passphrase = passphrase - break - except SFTPError: - print(t("err.sftp.wrong_passphrase")) + key_passphrase = BackupManager._resolve_key_passphrase( + default_key, SFTPClient, SFTPError + ) + if key_passphrase is None: + key_file = "" + password = config.sftp_password + key_passphrase = "" else: print(t("cmd.sftp.no_default_key")) return elif key_file and not key_passphrase and not password: - # 已配置私钥但未设置密码短语,检测是否需要 - import getpass - - try: - SFTPClient._load_private_key(key_file, "") - except SFTPError: - while True: - passphrase = getpass.getpass( - t("cli.prompt.sftp.key_passphrase") + " " - ) - if not passphrase: - print(t("cmd.sftp.no_default_key")) - return - try: - SFTPClient._load_private_key(key_file, passphrase) - key_passphrase = passphrase - break - except SFTPError: - print(t("err.sftp.wrong_passphrase")) + key_passphrase = BackupManager._resolve_key_passphrase( + key_file, SFTPClient, SFTPError + ) + if key_passphrase is None: + key_file = "" + password = config.sftp_password + key_passphrase = "" try: with SFTPClient( @@ -363,7 +369,7 @@ class BackupManager: """记录备份历史""" from datetime import datetime - history = self.data.setdefault("_history", []) + history = self.data.setdefault(_HISTORY_KEY, []) history.append( { "time": datetime.now().isoformat(timespec="seconds"), @@ -374,11 +380,47 @@ class BackupManager: ) # 保留最近 100 条记录 if len(history) > 100: - self.data["_history"] = history[-100:] + self.data[_HISTORY_KEY] = history[-100:] def get_history(self) -> list[dict]: """获取备份历史记录""" - return self.data.get("_history", []) + return self.data.get(_HISTORY_KEY, []) + + def format_history_table(self) -> str: + """生成备份历史的对齐文本表格""" + history = self.get_history() + if not history: + return t("cmd.list.empty") + + headers = [ + t("table.header.time"), + t("table.header.source"), + t("table.header.size"), + t("table.header.files"), + ] + rows = [] + for entry in reversed(history): + rows.append( + [ + entry.get("time", ""), + entry.get("source", ""), + str(entry.get("size_mb", 0)), + str(entry.get("files_count", 0)), + ] + ) + + col_widths = [self._display_width(h) for h in headers] + for row in rows: + for i, cell in enumerate(row): + col_widths[i] = max(col_widths[i], self._display_width(cell)) + + fmt = " | ".join(["{:<" + str(w) + "}" for w in col_widths]) + sep = "-+-".join(["-" * w for w in col_widths]) + + lines = [fmt.format(*headers), sep] + for row in rows: + lines.append(fmt.format(*row)) + return "\n".join(lines) @staticmethod def _cleanup_old_backups(target_dir: str, keep: int): @@ -418,17 +460,20 @@ class BackupManager: 查看所有备份策略 """ return { - key: BackupEntry.from_list(raw).target for key, raw in self.data.items() + key: BackupEntry.from_list(raw).target + for key, raw in self.data.items() + if key != _HISTORY_KEY } @staticmethod def _display_width(s: str) -> int: - """计算字符串的终端显示宽度(中文字符算2,英文字符算1)""" + """计算字符串的终端显示宽度(东亚宽字符算2,其余算1)""" if not isinstance(s, str): return len(str(s)) width = 0 for ch in s: - if ord(ch) > 0x2E80: + eaw = unicodedata.east_asian_width(ch) + if eaw in ("W", "F"): width += 2 else: width += 1 @@ -438,7 +483,8 @@ class BackupManager: """ 生成对齐的文本表格 """ - if not self.data: + non_history_keys = [k for k in self.data if k != _HISTORY_KEY] + if not non_history_keys: return t("cmd.all.empty") headers = [ @@ -449,7 +495,7 @@ class BackupManager: ] rows = [] for path, raw in self.data.items(): - if path == "_history": + if path == _HISTORY_KEY: continue entry = BackupEntry.from_list(raw) fmt_display = ( diff --git a/sbackup/compression.py b/sbackup/compression.py index 1aa903d..a84c5db 100644 --- a/sbackup/compression.py +++ b/sbackup/compression.py @@ -2,9 +2,10 @@ 压缩模块:ZIP / TAR / Zstd / 7z 文件压缩逻辑 """ +import logging import os -import io import tarfile +import tempfile import zipfile from pathlib import Path from fnmatch import fnmatch @@ -12,6 +13,8 @@ from tqdm import tqdm from sbackup.i18n import t from sbackup.config import Config +logger = logging.getLogger(__name__) + # compresslevel 仅对 ZIP_DEFLATED 和 ZIP_BZIP2 有效 _VALID_COMPRESSLEVEL_ALGORITHMS = {zipfile.ZIP_DEFLATED, zipfile.ZIP_BZIP2} @@ -27,6 +30,8 @@ _TAR_FORMATS = { class BaseCompressor: """压缩器基类,提供公共的文件收集和忽略逻辑""" + _IGNORE_FILENAME = ".sbackupignore" + def __init__(self, config: Config) -> None: self.folder_path: Path = Path(config.folder_path) self.zipfile_path: Path | None = ( @@ -35,18 +40,51 @@ class BaseCompressor: self.skip_patterns: list[str] = config.skip_patterns self.compression_level: int | None = None - def _should_ignore(self, rel_path: str) -> bool: - """检查相对路径是否匹配忽略模式(支持路径级匹配如 subdir/*.log)""" - for pattern in self.skip_patterns: + def _load_ignore_file(self, folder_path: Path) -> list[str]: + """从源目录的 .sbackupignore 文件加载忽略规则""" + ignore_file = folder_path / self._IGNORE_FILENAME + if not ignore_file.is_file(): + return [] + try: + lines = ignore_file.read_text(encoding="utf-8").splitlines() + patterns = [] + for line in lines: + line = line.strip() + if line and not line.startswith("#"): + patterns.append(line) + if patterns: + logger.debug(t("log.ignore.loaded"), ignore_file) + return patterns + except OSError: + return [] + + def _should_ignore( + self, rel_path: str, extra_patterns: list[str] | None = None + ) -> bool: + """检查相对路径是否匹配忽略模式(支持 ** 递归匹配和 ! 取反)""" + all_patterns = self.skip_patterns + (extra_patterns or []) + negated = [] + matched = False + for pattern in all_patterns: + if pattern.startswith("!"): + negated.append(pattern[1:]) + continue if fnmatch(rel_path, pattern) or fnmatch( os.path.basename(rel_path), pattern ): - return True - return False + matched = True + # 取反模式可以恢复被忽略的文件 + for pattern in negated: + if fnmatch(rel_path, pattern) or fnmatch( + os.path.basename(rel_path), pattern + ): + return False + return matched def _collect_files(self, folder_path: Path) -> list[tuple[str, str]]: """遍历文件夹收集需要压缩的文件列表,处理权限错误""" files = [] + extra_patterns = self._load_ignore_file(folder_path) try: for dirpath, dirnames, filenames in os.walk(folder_path): try: @@ -59,7 +97,8 @@ class BaseCompressor: if not self._should_ignore( os.path.join(rel_dir, d).replace("\\", "/") if rel_dir - else d + else d, + extra_patterns, ) ] for filename in filenames: @@ -68,7 +107,7 @@ class BaseCompressor: if rel_dir else filename ) - if not self._should_ignore(file_rel): + if not self._should_ignore(file_rel, extra_patterns): files.append((dirpath, filename)) except PermissionError: print(t("err.permission", path=dirpath)) @@ -199,7 +238,7 @@ class TarfileCompression(BaseCompressor): def __init__(self, config: Config) -> None: super().__init__(config) - fmt = config.compression_format.upper() + fmt = config.compression_format.upper().replace(".", "_") if fmt not in _TAR_FORMATS: fmt = "TAR_GZ" self._extension, self._mode = _TAR_FORMATS[fmt] @@ -334,28 +373,26 @@ class ZstdCompression(BaseCompressor): try: cctx = zstd.ZstdCompressor(level=self.compression_level) - # 先创建 tar,再用 zstd 压缩 - tar_buffer = io.BytesIO() - with tarfile.open(fileobj=tar_buffer, mode="w") as tarf: - with tqdm( - total=total_files, - desc=t("compress.progress"), - unit=t("compress.unit"), - ) as pbar: - for dirpath, filename in files_to_compress: - file_path = Path(dirpath) / filename - arcname = str( - folder_path.name / file_path.relative_to(folder_path) - ).replace("\\", "/") - try: - tarf.add(file_path, arcname=arcname, recursive=False) - pbar.update(1) - files_count += 1 - except (FileNotFoundError, PermissionError): - continue - - compressed = cctx.compress(tar_buffer.getvalue()) - output_path.write_bytes(compressed) + with open(output_path, "wb") as f_out: + compressor = cctx.stream_writer(f_out) + with tarfile.open(fileobj=compressor, mode="w") as tarf: + with tqdm( + total=total_files, + desc=t("compress.progress"), + unit=t("compress.unit"), + ) as pbar: + for dirpath, filename in files_to_compress: + file_path = Path(dirpath) / filename + arcname = str( + folder_path.name / file_path.relative_to(folder_path) + ).replace("\\", "/") + try: + tarf.add(file_path, arcname=arcname, recursive=False) + pbar.update(1) + files_count += 1 + except (FileNotFoundError, PermissionError): + continue + compressor.close() size_mb = output_path.stat().st_size / (1024 * 1024) print( @@ -469,7 +506,7 @@ class SevenZipCompression(BaseCompressor): def create_compressor(config: Config) -> BaseCompressor: """工厂函数:根据配置创建对应的压缩器""" - fmt = config.compression_format.upper() + fmt = config.compression_format.upper().replace(".", "_") if fmt in _TAR_FORMATS: return TarfileCompression(config) if fmt == "TAR_ZST": @@ -479,15 +516,16 @@ def create_compressor(config: Config) -> BaseCompressor: return ZipfileCompression(config) -def restore_backup(backup_path: str, target_dir: str) -> dict: +def restore_backup(backup_path: str, target_dir: str, password: str = "") -> dict: """ 从备份文件还原到目标目录 - 自动检测格式(ZIP / tar.gz / tar.bz2 / tar.xz) + 自动检测格式(ZIP / tar.gz / tar.bz2 / tar.xz / tar.zst / 7z) + :param password: 解密密码(仅 7z 加密备份需要) :return: 包含统计信息的字典 """ backup = Path(backup_path) if not backup.exists(): - print(t("err.folder.invalid", path=backup_path)) + print(t("err.file.not_found", path=backup_path)) return {"success": False, "files_count": 0} target = Path(target_dir) @@ -511,7 +549,10 @@ def restore_backup(backup_path: str, target_dir: str) -> dict: elif name_lower.endswith(".7z"): import py7zr - with py7zr.SevenZipFile(backup, "r") as szf: + szf_kwargs = {"file": backup, "mode": "r"} + if password: + szf_kwargs["password"] = password + with py7zr.SevenZipFile(**szf_kwargs) as szf: members = szf.getnames() with tqdm( total=len(members), @@ -526,20 +567,28 @@ def restore_backup(backup_path: str, target_dir: str) -> dict: import zstandard as zstd dctx = zstd.ZstdDecompressor() - compressed = backup.read_bytes() - tar_data = dctx.decompress(compressed) - with tarfile.open(fileobj=io.BytesIO(tar_data), mode="r") as tarf: - members = tarf.getmembers() - with tqdm( - total=len(members), - desc=t("restore.progress"), - unit=t("compress.unit"), - ) as pbar: - for member in members: - tarf.extract(member, target, filter="data") - pbar.update(1) - print(t("restore.success", path=target, count=len(members))) - return {"success": True, "files_count": len(members)} + # tarfile 需要可 seek 的 fileobj,先流式解压到临时文件 + with tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) as tmp: + with open(backup, "rb") as f_in: + reader = dctx.stream_reader(f_in) + while True: + chunk = reader.read(65536) + if not chunk: + break + tmp.write(chunk) + tmp.seek(0) + with tarfile.open(fileobj=tmp, mode="r") as tarf: + members = tarf.getmembers() + with tqdm( + total=len(members), + desc=t("restore.progress"), + unit=t("compress.unit"), + ) as pbar: + for member in members: + tarf.extract(member, target, filter="data") + pbar.update(1) + print(t("restore.success", path=target, count=len(members))) + return {"success": True, "files_count": len(members)} elif name_lower.endswith(".tar.gz") or name_lower.endswith(".tgz"): mode = "r:gz" elif name_lower.endswith(".tar.bz2") or name_lower.endswith(".tbz2"): @@ -573,3 +622,126 @@ def restore_backup(backup_path: str, target_dir: str) -> dict: except Exception as e: print(t("err.unknown", error=e)) return {"success": False, "files_count": 0} + + +def _get_archive_member_names(backup: Path, password: str = "") -> list[str]: + """获取压缩包内所有成员名称,自动检测格式""" + name_lower = backup.name.lower() + if name_lower.endswith(".zip"): + with zipfile.ZipFile(backup, "r") as zf: + return zf.namelist() + elif name_lower.endswith(".7z"): + import py7zr + + szf_kwargs = {"file": backup, "mode": "r"} + if password: + szf_kwargs["password"] = password + with py7zr.SevenZipFile(**szf_kwargs) as szf: + return szf.getnames() + elif name_lower.endswith(".tar.zst"): + import zstandard as zstd + + dctx = zstd.ZstdDecompressor() + with tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) as tmp: + with open(backup, "rb") as f_in: + reader = dctx.stream_reader(f_in) + while True: + chunk = reader.read(65536) + if not chunk: + break + tmp.write(chunk) + tmp.seek(0) + with tarfile.open(fileobj=tmp, mode="r") as tarf: + return [m.name for m in tarf.getmembers()] + elif name_lower.endswith(".tar.gz") or name_lower.endswith(".tgz"): + mode = "r:gz" + elif name_lower.endswith(".tar.bz2") or name_lower.endswith(".tbz2"): + mode = "r:bz2" + elif name_lower.endswith(".tar.xz") or name_lower.endswith(".txz"): + mode = "r:xz" + elif name_lower.endswith(".tar"): + mode = "r" + else: + return [] + with tarfile.open(backup, mode) as tarf: + return tarf.getnames() + + +def list_backup_contents(backup_path: str, password: str = "") -> str: + """ + 列出备份文件内的所有文件,不解压 + :return: 格式化的文件列表字符串 + """ + backup = Path(backup_path) + if not backup.exists(): + return t("err.file.not_found", path=backup_path) + + try: + members = _get_archive_member_names(backup, password) + except Exception as e: + return t("err.unknown", error=e) + + if not members: + return t("restore.list.empty", path=backup_path) + + lines = [t("restore.list.title", path=backup_path)] + for name in members: + lines.append(f" {name}") + lines.append(f"\n({len(members)} files)") + return "\n".join(lines) + + +def verify_backup(backup_path: str, password: str = "") -> dict: + """ + 校验备份文件完整性:解压到临时目录后比对文件数 + :return: 包含校验结果的字典 + """ + backup = Path(backup_path) + if not backup.exists(): + print(t("err.file.not_found", path=backup_path)) + return {"success": False, "files_count": 0} + + print(t("cmd.verify.checking", path=backup_path)) + + try: + expected_names = _get_archive_member_names(backup, password) + except Exception as e: + print(t("err.unknown", error=e)) + return {"success": False, "files_count": 0} + + # 解压到临时目录 + import tempfile as _tempfile + + with _tempfile.TemporaryDirectory() as tmp_dir: + result = restore_backup(backup_path, tmp_dir, password) + if not result["success"]: + print( + t( + "cmd.verify.failed", + path=backup_path, + expected=len(expected_names), + actual=0, + ) + ) + return {"success": False, "files_count": 0} + + actual_count = result["files_count"] + if actual_count == len(expected_names): + print( + t( + "cmd.verify.success", + path=backup_path, + count=actual_count, + ) + ) + return {"success": True, "files_count": actual_count} + else: + print( + t( + "cmd.verify.failed", + path=backup_path, + expected=len(expected_names), + actual=actual_count, + ) + ) + return {"success": False, "files_count": actual_count} diff --git a/sbackup/config.py b/sbackup/config.py index e7705f8..b67ae1c 100644 --- a/sbackup/config.py +++ b/sbackup/config.py @@ -14,6 +14,35 @@ logger = logging.getLogger(__name__) DEFAULT_SKIP_PATTERNS = [".git", "__pycache__"] +def _load_json_file(config_file: str) -> dict: + """读取 JSON 配置文件,损坏时返回空字典""" + if not os.path.exists(config_file): + return {} + try: + with open(config_file, "r", encoding="utf-8") as f: + return json.load(f) + except json.JSONDecodeError: + logger.warning(t("log.config.reset"), config_file) + return {} + + +def _save_json_file(data: dict, config_file: str) -> None: + """将字典写入 JSON 配置文件,自动创建目录""" + data_dir = os.path.dirname(config_file) + if data_dir: + try: + os.makedirs(data_dir, exist_ok=True) + except OSError as e: + logger.error(t("log.config.mkdir.error"), data_dir, e) + return + + try: + with open(config_file, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=4) + except OSError as e: + logger.error(t("log.config.write.error"), config_file, e) + + def get_default_data_file() -> str: """返回跨平台的默认数据文件路径""" if sys.platform == "win32": @@ -72,7 +101,7 @@ def load_config(config_file: str = "config.json") -> Config: compression_config = config_data.get("compression", {}) skip_patterns = config_data.get("skip_patterns", DEFAULT_SKIP_PATTERNS) data_file = config_data.get("data_file", get_default_data_file()) - lang = config_data.get("lang", "en_US") + lang = config_data.get("lang", "zh_CN") compression_format = config_data.get("compression_format", "ZIP") sftp_config = config_data.get("sftp", {}) webdav_config = config_data.get("webdav", {}) @@ -103,65 +132,17 @@ def load_config(config_file: str = "config.json") -> Config: def save_lang(lang: str, config_file: str = "config.json") -> None: - """ - 将语言偏好保存到配置文件 - """ - if os.path.exists(config_file): - try: - with open(config_file, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - logger.warning(t("log.config.reset"), config_file) - data = {} - else: - data = {} - + """将语言偏好保存到配置文件""" + data = _load_json_file(config_file) data["lang"] = lang - - data_dir = os.path.dirname(config_file) - if data_dir: - try: - os.makedirs(data_dir, exist_ok=True) - except OSError as e: - logger.error(t("log.config.mkdir.error"), data_dir, e) - return - - try: - with open(config_file, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=4) - except OSError as e: - logger.error(t("log.config.write.error"), config_file, e) + _save_json_file(data, config_file) def save_format(fmt: str, config_file: str = "config.json") -> None: - """ - 将打包格式偏好保存到配置文件 - """ - if os.path.exists(config_file): - try: - with open(config_file, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - logger.warning(t("log.config.reset"), config_file) - data = {} - else: - data = {} - + """将打包格式偏好保存到配置文件""" + data = _load_json_file(config_file) data["compression_format"] = fmt - - data_dir = os.path.dirname(config_file) - if data_dir: - try: - os.makedirs(data_dir, exist_ok=True) - except OSError as e: - logger.error(t("log.config.mkdir.error"), data_dir, e) - return - - try: - with open(config_file, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=4) - except OSError as e: - logger.error(t("log.config.write.error"), config_file, e) + _save_json_file(data, config_file) def save_sftp_config( @@ -175,19 +156,8 @@ def save_sftp_config( key_passphrase: str = "", config_file: str = "config.json", ) -> None: - """ - 将 SFTP 配置保存到配置文件 - """ - if os.path.exists(config_file): - try: - with open(config_file, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - logger.warning(t("log.config.reset"), config_file) - data = {} - else: - data = {} - + """将 SFTP 配置保存到配置文件""" + data = _load_json_file(config_file) data["sftp"] = { "host": host, "port": port, @@ -198,20 +168,7 @@ def save_sftp_config( "remote_path": remote_path, "enabled": enabled, } - - data_dir = os.path.dirname(config_file) - if data_dir: - try: - os.makedirs(data_dir, exist_ok=True) - except OSError as e: - logger.error(t("log.config.mkdir.error"), data_dir, e) - return - - try: - with open(config_file, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=4) - except OSError as e: - logger.error(t("log.config.write.error"), config_file, e) + _save_json_file(data, config_file) def save_webdav_config( @@ -223,16 +180,7 @@ def save_webdav_config( config_file: str = "config.json", ) -> None: """将 WebDAV 配置保存到配置文件""" - if os.path.exists(config_file): - try: - with open(config_file, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - logger.warning(t("log.config.reset"), config_file) - data = {} - else: - data = {} - + data = _load_json_file(config_file) data["webdav"] = { "url": url, "user": user, @@ -240,17 +188,4 @@ def save_webdav_config( "remote_path": remote_path, "enabled": enabled, } - - data_dir = os.path.dirname(config_file) - if data_dir: - try: - os.makedirs(data_dir, exist_ok=True) - except OSError as e: - logger.error(t("log.config.mkdir.error"), data_dir, e) - return - - try: - with open(config_file, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=4) - except OSError as e: - logger.error(t("log.config.write.error"), config_file, e) + _save_json_file(data, config_file) diff --git a/sbackup/locales/de_DE.json b/sbackup/locales/de_DE.json index c43aee3..8c8928f 100644 --- a/sbackup/locales/de_DE.json +++ b/sbackup/locales/de_DE.json @@ -2,6 +2,7 @@ "cmd.add.success": "Backup-Strategie erfolgreich hinzugefügt: {source} -> {dest}", "cmd.rm.success": "Backup-Strategie erfolgreich entfernt: {path}", "err.folder.invalid": "Fehler: {path} ist kein gültiger Ordner oder existiert nicht.", + "err.file.not_found": "Error: Backup-Datei {path} existiert nicht.", "err.dest.invalid": "Fehler: Zielordner {path} ist kein gültiger Ordner oder existiert nicht.", "info.already.added": "Info: {path} wurde bereits hinzugefügt. Bitte nicht duplizieren.", "warn.no.strategy.found": "Warnung: Backup-Strategie für {path} nicht gefunden.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "Alle Backup-Strategien sind aktuell, kein Backup erforderlich.", "cmd.save.completed": "Backup abgeschlossen, {count} Strategie(n) bearbeitet.", "cmd.all.empty": "Keine Backup-Strategien konfiguriert.", + "cmd.list.empty": "Noch keine Backup-Historie.", + "cmd.verify.checking": "Überprüfe Backup: {path}", + "cmd.verify.success": "Überprüfung bestanden: {path} ({count} Dateien)", + "cmd.verify.failed": "Überprüfung fehlgeschlagen: {path} — erwartet {expected} Dateien, erhalten {actual}", + "restore.list.title": "Archivinhalt: {path}", + "restore.list.empty": "Archiv ist leer: {path}", + "log.ignore.loaded": "Ignorierregeln-Datei geladen: %s", "exit.message": "\nBeendet.", "table.header.source": "Quellordner", "table.header.dest": "Zielordner", "table.header.ignore": " Ausschlussmuster", + "table.header.time": "Backup-Zeit", + "table.header.size": "Größe(MB)", + "table.header.files": "Dateien", "table.cell.none": "Keine", "table.header.format": "Format", "table.cell.default": "Standard", @@ -62,6 +73,11 @@ "cli.help.restore": "Aus einer Backup-Datei wiederherstellen", "cli.help.restore.file": "Backup-Dateipfad (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "Zielverzeichnis für Wiederherstellung", + "cli.help.restore.password": "Entschlüsselungspasswort (zum Wiederherstellen verschüsselter 7z-Backups)", + "cli.help.restore.list": "Archivinhalt auflisten ohne zu extrahieren", + "cli.help.list": "Backup-Verlauf anzeigen", + "cli.help.verify": "Backup-Datei-Integrität überprüfen", + "cli.help.verify.file": "Zu überprüfende Backup-Datei", "cli.help.version": "Versionsinformationen anzeigen", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLizenz GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "Konfigurationsdatei %s ist fehlerhaft, Standardwerte werden verwendet", diff --git a/sbackup/locales/en_US.json b/sbackup/locales/en_US.json index 33e03e0..d14d143 100644 --- a/sbackup/locales/en_US.json +++ b/sbackup/locales/en_US.json @@ -2,6 +2,7 @@ "cmd.add.success": "Backup strategy added successfully: {source} -> {dest}", "cmd.rm.success": "Backup strategy removed successfully: {path}", "err.folder.invalid": "Error: {path} is not a valid folder or does not exist.", + "err.file.not_found": "Error: Backup file {path} does not exist.", "err.dest.invalid": "Error: Target folder {path} is not a valid folder or does not exist.", "info.already.added": "Info: {path} has already been added. Please do not duplicate.", "warn.no.strategy.found": "Warning: Backup strategy for {path} not found.", @@ -27,6 +28,13 @@ "cmd.save.uptodate": "All backup strategies are up to date, no backup needed.", "cmd.save.completed": "Backup completed, processed {count} strategy(s).", "cmd.all.empty": "No backup strategies configured.", + "cmd.list.empty": "No backup history yet.", + "cmd.verify.checking": "Verifying backup: {path}", + "cmd.verify.success": "Verification passed: {path} ({count} files)", + "cmd.verify.failed": "Verification failed: {path} — expected {expected} files, got {actual}", + "restore.list.title": "Archive contents: {path}", + "restore.list.empty": "Archive is empty: {path}", + "log.ignore.loaded": "Loaded ignore rules file: %s", "exit.message": "\nExiting.", "table.header.source": "Source Folder", "table.header.dest": "Target Folder", @@ -62,6 +70,11 @@ "cli.help.restore": "Restore from a backup file (supports ZIP / tar / tar.gz / tar.bz2 / tar.xz / tar.zst / 7z)", "cli.help.restore.file": "Backup file path", "cli.help.restore.dir": "Target directory for restoration", + "cli.help.restore.password": "Decryption password (for restoring encrypted 7z backups)", + "cli.help.restore.list": "List archive contents without extracting", + "cli.help.list": "View backup history", + "cli.help.verify": "Verify backup file integrity", + "cli.help.verify.file": "Backup file to verify", "cli.help.version": "Show version information", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLicensed under GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "Config file %s is malformed, using defaults", diff --git a/sbackup/locales/es_ES.json b/sbackup/locales/es_ES.json index e6b6f45..15d8168 100644 --- a/sbackup/locales/es_ES.json +++ b/sbackup/locales/es_ES.json @@ -2,6 +2,7 @@ "cmd.add.success": "Estrategia de copia de seguridad añadida con éxito: {source} -> {dest}", "cmd.rm.success": "Estrategia de copia de seguridad eliminada con éxito: {path}", "err.folder.invalid": "Error: {path} no es una carpeta válida o no existe.", + "err.file.not_found": "Error: El archivo de copia de seguridad {path} no existe.", "err.dest.invalid": "Error: La carpeta destino {path} no es una carpeta válida o no existe.", "info.already.added": "Info: {path} ya ha sido añadido. No lo duplique.", "warn.no.strategy.found": "Advertencia: Estrategia de copia de seguridad para {path} no encontrada.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "Todas las estrategias de copia de seguridad están actualizadas, no se necesita copia.", "cmd.save.completed": "Copia de seguridad completada, {count} estrategia(s) procesada(s).", "cmd.all.empty": "No hay estrategias de copia de seguridad configuradas.", + "cmd.list.empty": "Aún no hay historial de copias de seguridad.", + "cmd.verify.checking": "Verificando copia de seguridad: {path}", + "cmd.verify.success": "Verificación exitosa: {path} ({count} archivos)", + "cmd.verify.failed": "Verificación fallida: {path} — se esperaban {expected} archivos, se obtuvieron {actual}", + "restore.list.title": "Contenido del archivo: {path}", + "restore.list.empty": "El archivo está vacío: {path}", + "log.ignore.loaded": "Archivo de reglas de ignorar cargado: %s", "exit.message": "\nSaliendo.", "table.header.source": "Carpeta fuente", "table.header.dest": "Carpeta destino", "table.header.ignore": "Patrones de exclusión", + "table.header.time": "Hora de copia", + "table.header.size": "Tamaño(MB)", + "table.header.files": "Archivos", "table.cell.none": "Ninguno", "table.header.format": "Formato", "table.cell.default": "Predet.", @@ -62,6 +73,11 @@ "cli.help.restore": "Restaurar desde un archivo de copia de seguridad", "cli.help.restore.file": "Ruta del archivo de copia de seguridad (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "Directorio destino para la restauración", + "cli.help.restore.password": "Contraseña de descifrado (para restaurar copias de seguridad 7z cifradas)", + "cli.help.restore.list": "Listar contenido del archivo sin extraer", + "cli.help.list": "Ver historial de copias de seguridad", + "cli.help.verify": "Verificar integridad del archivo de copia de seguridad", + "cli.help.verify.file": "Archivo de copia de seguridad a verificar", "cli.help.version": "Mostrar la información de versión", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLicencia GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "El archivo de configuración %s está malformado, usando valores predeterminados", diff --git a/sbackup/locales/fr_FR.json b/sbackup/locales/fr_FR.json index e84c337..6ba9060 100644 --- a/sbackup/locales/fr_FR.json +++ b/sbackup/locales/fr_FR.json @@ -2,6 +2,7 @@ "cmd.add.success": "Stratégie de sauvegarde ajoutée avec succès : {source} -> {dest}", "cmd.rm.success": "Stratégie de sauvegarde supprimée avec succès : {path}", "err.folder.invalid": "Erreur : {path} n'est pas un dossier valide ou n'existe pas.", + "err.file.not_found": "Erreur : Le fichier de sauvegarde {path} n'existe pas.", "err.dest.invalid": "Erreur : Le dossier cible {path} n'est pas un dossier valide ou n'existe pas.", "info.already.added": "Info : {path} a déjà été ajouté. Veuillez ne pas dupliquer.", "warn.no.strategy.found": "Attention : Stratégie de sauvegarde pour {path} non trouvée.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "Toutes les stratégies de sauvegarde sont à jour, aucune sauvegarde nécessaire.", "cmd.save.completed": "Sauvegarde terminée, {count} stratégie(s) traitée(s).", "cmd.all.empty": "Aucune stratégie de sauvegarde configurée.", + "cmd.list.empty": "Aucun historique de sauvegarde pour le moment.", + "cmd.verify.checking": "Vérification de la sauvegarde: {path}", + "cmd.verify.success": "Vérification réussie: {path} ({count} fichiers)", + "cmd.verify.failed": "Vérification échouée: {path} — {expected} fichiers attendus, {actual} trouvés", + "restore.list.title": "Contenu de l'archive: {path}", + "restore.list.empty": "L'archive est vide: {path}", + "log.ignore.loaded": "Fichier de règles d'ignorance chargé: %s", "exit.message": "\nSortie.", "table.header.source": "Dossier source", "table.header.dest": "Dossier cible", "table.header.ignore": "Modèles d'exclusion", + "table.header.time": "Heure de sauvegarde", + "table.header.size": "Taille(Mo)", + "table.header.files": "Fichiers", "table.cell.none": "Aucun", "table.header.format": "Format", "table.cell.default": "Défaut", @@ -62,6 +73,11 @@ "cli.help.restore": "Restaurer depuis un fichier de sauvegarde", "cli.help.restore.file": "Chemin du fichier de sauvegarde (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "Dossier cible pour la restauration", + "cli.help.restore.password": "Mot de passe de déchiffrement (pour restaurer les sauvegardes 7z chiffrées)", + "cli.help.restore.list": "Lister le contenu de l'archive sans extraire", + "cli.help.list": "Voir l'historique des sauvegardes", + "cli.help.verify": "Vérifier l'intégrité du fichier de sauvegarde", + "cli.help.verify.file": "Fichier de sauvegarde à vérifier", "cli.help.version": "Afficher les informations de version", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLicence GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "Le fichier de configuration %s est malformé, utilisation des valeurs par défaut", diff --git a/sbackup/locales/ja_JP.json b/sbackup/locales/ja_JP.json index af4bf9d..0b559a4 100644 --- a/sbackup/locales/ja_JP.json +++ b/sbackup/locales/ja_JP.json @@ -2,6 +2,7 @@ "cmd.add.success": "バックアップ戦略を追加しました: {source} -> {dest}", "cmd.rm.success": "バックアップ戦略を削除しました: {path}", "err.folder.invalid": "エラー: {path} は有効なフォルダではないか、存在しません。", + "err.file.not_found": "エラー:バックアップファイル {path} が存在しません。", "err.dest.invalid": "エラー: 目標フォルダ {path} は有効なフォルダではないか、存在しません。", "info.already.added": "情報: {path} は既に追加されています。重複して追加しないでください。", "warn.no.strategy.found": "警告: {path} のバックアップ戦略が見つかりません。", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "すべてのバックアップ戦略は最新です、バックアップは不要です。", "cmd.save.completed": "バックアップ完了、{count} 戦略を処理しました。", "cmd.all.empty": "バックアップ戦略が設定されていません。", + "cmd.list.empty": "バックアップ履歴はまだありません。", + "cmd.verify.checking": "バックアップを検証中: {path}", + "cmd.verify.success": "検証成功: {path} ({count} ファイル)", + "cmd.verify.failed": "検証失敗: {path} — 予想 {expected} ファイル、実際 {actual} ファイル", + "restore.list.title": "アーカイブ内容: {path}", + "restore.list.empty": "アーカイブが空です: {path}", + "log.ignore.loaded": "無視ルールファイルを読み込みました: %s", "exit.message": "\n終了します。", "table.header.source": "ソースフォルダ", "table.header.dest": "目標フォルダ", "table.header.ignore": "除外パターン", + "table.header.time": "バックアップ時刻", + "table.header.size": "サイズ(MB)", + "table.header.files": "ファイル数", "table.cell.none": "なし", "table.header.format": "形式", "table.cell.default": "デフォルト", @@ -62,6 +73,11 @@ "cli.help.restore": "バックアップファイルから復元", "cli.help.restore.file": "バックアップファイルパス (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "復元目標ディレクトリ", + "cli.help.restore.password": "復号化パスワード(暗号化された7zバックアップの復元時に使用)", + "cli.help.restore.list": "展開せずにアーカイブ内容を一覧表示", + "cli.help.list": "バックアップ履歴を表示", + "cli.help.verify": "バックアップファイルの整合性を検証", + "cli.help.verify.file": "検証するバックアップファイル", "cli.help.version": "バージョン情報を表示", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nGNU GPL v3.0 ライセンス — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "設定ファイル %s が破損しています、デフォルト値を使用", diff --git a/sbackup/locales/ko_KR.json b/sbackup/locales/ko_KR.json index 171334e..6b18e1e 100644 --- a/sbackup/locales/ko_KR.json +++ b/sbackup/locales/ko_KR.json @@ -2,6 +2,7 @@ "cmd.add.success": "백업 전략 추가 성공: {source} -> {dest}", "cmd.rm.success": "백업 전략 삭제 성공: {path}", "err.folder.invalid": "오류: {path}는 유효한 폴더가 아니거나 존재하지 않습니다.", + "err.file.not_found": "오류: 백업 파일 {path}이(가) 존재하지 않습니다.", "err.dest.invalid": "오류: 대상 폴더 {path}는 유효한 폴더가 아니거나 존재하지 않습니다.", "info.already.added": "알림: {path}가 이미 추가되었습니다. 중복 추가하지 마세요.", "warn.no.strategy.found": "경고: {path}의 백업 전략을 찾을 수 없습니다.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "모든 백업 전략이 최신 상태입니다, 백업이 필요하지 않습니다.", "cmd.save.completed": "백업 완료, {count}개 전략 처리됨.", "cmd.all.empty": "구성된 백업 전략이 없습니다.", + "cmd.list.empty": "아직 백업 기록이 없습니다.", + "cmd.verify.checking": "백업 파일 검증 중: {path}", + "cmd.verify.success": "검증 성공: {path} ({count}개 파일)", + "cmd.verify.failed": "검증 실패: {path} — 예상 {expected}개 파일, 실제 {actual}개", + "restore.list.title": "압축 파일 내용: {path}", + "restore.list.empty": "압축 파일이 비어 있습니다: {path}", + "log.ignore.loaded": "무시 규칙 파일 로드됨: %s", "exit.message": "\n종료합니다.", "table.header.source": "원본 폴더", "table.header.dest": "대상 폴더", "table.header.ignore": "제외 패턴", + "table.header.time": "백업 시간", + "table.header.size": "크기(MB)", + "table.header.files": "파일 수", "table.cell.none": "없음", "table.header.format": "형식", "table.cell.default": "기본값", @@ -62,6 +73,11 @@ "cli.help.restore": "백업 파일에서 복원", "cli.help.restore.file": "백업 파일 경로 (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "복원 대상 디렉토리", + "cli.help.restore.password": "복호화 비밀번호 (암호화된 7z 백업 복원 시 사용)", + "cli.help.restore.list": "압축 해제 없이 내용 보기", + "cli.help.list": "백업 기록 보기", + "cli.help.verify": "백업 파일 무결성 검증", + "cli.help.verify.file": "검증할 백업 파일", "cli.help.version": "버전 정보 표시", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nGNU GPL v3.0 라이선스 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "구성 파일 %s가 손상되었습니다, 기본값 사용", diff --git a/sbackup/locales/pt_BR.json b/sbackup/locales/pt_BR.json index 7ead50a..ec90e89 100644 --- a/sbackup/locales/pt_BR.json +++ b/sbackup/locales/pt_BR.json @@ -2,6 +2,7 @@ "cmd.add.success": "Estratégia de backup adicionada com sucesso: {source} -> {dest}", "cmd.rm.success": "Estratégia de backup removida com sucesso: {path}", "err.folder.invalid": "Erro: {path} não é uma pasta válida ou não existe.", + "err.file.not_found": "Erro: O arquivo de backup {path} não existe.", "err.dest.invalid": "Erro: A pasta de destino {path} não é uma pasta válida ou não existe.", "info.already.added": "Info: {path} já foi adicionado. Não duplique.", "warn.no.strategy.found": "Aviso: Estratégia de backup para {path} não encontrada.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "Todas as estratégias de backup estão atualizadas, backup não necessário.", "cmd.save.completed": "Backup concluído, {count} estratégia(s) processada(s).", "cmd.all.empty": "Nenhuma estratégia de backup configurada.", + "cmd.list.empty": "Nenhum histórico de backup ainda.", + "cmd.verify.checking": "Verificando backup: {path}", + "cmd.verify.success": "Verificação bem-sucedida: {path} ({count} arquivos)", + "cmd.verify.failed": "Verificação falhou: {path} — esperado {expected} arquivos, obtido {actual}", + "restore.list.title": "Conteúdo do arquivo: {path}", + "restore.list.empty": "O arquivo está vazio: {path}", + "log.ignore.loaded": "Arquivo de regras de ignorar carregado: %s", "exit.message": "\nSaindo.", "table.header.source": "Pasta de origem", "table.header.dest": "Pasta de destino", "table.header.ignore": "Padrões de exclusão", + "table.header.time": "Hora do backup", + "table.header.size": "Tamanho(MB)", + "table.header.files": "Arquivos", "table.cell.none": "Nenhum", "table.header.format": "Formato", "table.cell.default": "Padrão", @@ -62,6 +73,11 @@ "cli.help.restore": "Restaurar de um arquivo de backup", "cli.help.restore.file": "Caminho do arquivo de backup (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "Diretório de destino para restauração", + "cli.help.restore.password": "Senha de descriptografia (para restaurar backups 7z criptografados)", + "cli.help.restore.list": "Listar conteúdo do arquivo sem extrair", + "cli.help.list": "Ver histórico de backups", + "cli.help.verify": "Verificar integridade do arquivo de backup", + "cli.help.verify.file": "Arquivo de backup para verificar", "cli.help.version": "Mostrar informações de versão", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLicença GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "Arquivo de configuração %s malformado, usando valores padrão", diff --git a/sbackup/locales/ru_RU.json b/sbackup/locales/ru_RU.json index 1662ac0..9cc6e24 100644 --- a/sbackup/locales/ru_RU.json +++ b/sbackup/locales/ru_RU.json @@ -2,6 +2,7 @@ "cmd.add.success": "Стратегия резервного копирования добавлена: {source} -> {dest}", "cmd.rm.success": "Стратегия резервного копирования удалена: {path}", "err.folder.invalid": "Ошибка: {path} не является допустимой папкой или не существует.", + "err.file.not_found": "Ошибка: Файл резервной копии {path} не существует.", "err.dest.invalid": "Ошибка: Целевая папка {path} не является допустимой папкой или не существует.", "info.already.added": "Информация: {path} уже добавлен. Не добавляйте повторно.", "warn.no.strategy.found": "Предупреждение: Стратегия резервного копирования для {path} не найдена.", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "Все стратегии резервного копирования актуальны, копирование не требуется.", "cmd.save.completed": "Резервное копирование завершено, обработано {count} стратегий.", "cmd.all.empty": "Нет настроенных стратегий резервного копирования.", + "cmd.list.empty": "История резервного копирования пока пуста.", + "cmd.verify.checking": "Проверка резервной копии: {path}", + "cmd.verify.success": "Проверка пройдена: {path} ({count} файлов)", + "cmd.verify.failed": "Проверка не пройдена: {path} — ожидается {expected} файлов, получено {actual}", + "restore.list.title": "Содержимое архива: {path}", + "restore.list.empty": "Архив пуст: {path}", + "log.ignore.loaded": "Загружен файл правил игнорирования: %s", "exit.message": "\nВыход.", "table.header.source": "Исходная папка", "table.header.dest": "Целевая папка", "table.header.ignore": "Шаблоны исключения", + "table.header.time": "Время копии", + "table.header.size": "Размер(МБ)", + "table.header.files": "Файлы", "table.cell.none": "Нет", "table.header.format": "Формат", "table.cell.default": "По умолч.", @@ -62,6 +73,11 @@ "cli.help.restore": "Восстановить из файла резервной копии", "cli.help.restore.file": "Путь к файлу резервной копии (.zip / .tar / .tar.gz / .tar.bz2 / .tar.xz / .tar.zst / .7z)", "cli.help.restore.dir": "Целевой каталог для восстановления", + "cli.help.restore.password": "Пароль расшифровки (для восстановления зашифрованных резервных копий 7z)", + "cli.help.restore.list": "Показать содержимое архива без распаковки", + "cli.help.list": "Просмотреть историю резервного копирования", + "cli.help.verify": "Проверить целостность файла резервной копии", + "cli.help.verify.file": "Файл резервной копии для проверки", "cli.help.version": "Показать информацию о версии", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nЛицензия GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "Файл конфигурации %s повреждён, используются значения по умолчанию", diff --git a/sbackup/locales/zh_CN.json b/sbackup/locales/zh_CN.json index 232f50e..8eddd6b 100644 --- a/sbackup/locales/zh_CN.json +++ b/sbackup/locales/zh_CN.json @@ -2,6 +2,7 @@ "cmd.add.success": "备份策略添加成功: {source} -> {dest}", "cmd.rm.success": "备份策略删除成功: {path}", "err.folder.invalid": "错误:{path} 不是有效的文件夹名或不存在。", + "err.file.not_found": "错误:备份文件 {path} 不存在。", "err.dest.invalid": "错误:目标文件夹 {path} 不是有效的文件夹或不存在。", "info.already.added": "提示:{path} 已经添加过了,请勿重复添加。", "warn.no.strategy.found": "警告:未找到 {path} 的备份策略。", @@ -27,10 +28,20 @@ "cmd.save.uptodate": "所有备份策略均已是最新,无需备份。", "cmd.save.completed": "备份完成,共处理 {count} 个策略。", "cmd.all.empty": "没有配置任何备份策略。", + "cmd.list.empty": "暂无备份历史记录。", + "cmd.verify.checking": "正在校验备份文件: {path}", + "cmd.verify.success": "校验通过: {path} ({count} 个文件)", + "cmd.verify.failed": "校验失败: {path} — 预期 {expected} 个文件,实际 {actual} 个", + "restore.list.title": "备份内容: {path}", + "restore.list.empty": "备份文件为空: {path}", + "log.ignore.loaded": "已加载忽略规则文件: %s", "exit.message": "\n已退出。", "table.header.source": "源文件夹路径", "table.header.dest": "目标文件夹路径", "table.header.ignore": "忽略模式", + "table.header.time": "备份时间", + "table.header.size": "大小(MB)", + "table.header.files": "文件数", "table.cell.none": "无", "table.header.format": "格式", "table.cell.default": "默认", @@ -62,6 +73,11 @@ "cli.help.restore": "从备份文件还原(支持 ZIP / tar / tar.gz / tar.bz2 / tar.xz / tar.zst / 7z)", "cli.help.restore.file": "备份文件路径", "cli.help.restore.dir": "还原目标目录", + "cli.help.restore.password": "解密密码(还原加密的 7z 备份时使用)", + "cli.help.restore.list": "仅列出备份内容,不解压", + "cli.help.list": "查看备份历史记录", + "cli.help.verify": "校验备份文件完整性", + "cli.help.verify.file": "要校验的备份文件路径", "cli.help.version": "查看版本信息", "cli.version": "\nSbackup v{version} — Copyright © 2026 xiatianxuan\nLicensed under GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html\n", "log.config.malformed": "配置文件 %s 格式错误,使用默认配置", diff --git a/sbackup/sftp.py b/sbackup/sftp.py index ff91927..be6b2da 100644 --- a/sbackup/sftp.py +++ b/sbackup/sftp.py @@ -71,10 +71,13 @@ class SFTPClient: ) self._sftp = paramiko.SFTPClient.from_transport(self._transport) except paramiko.AuthenticationException: + self.disconnect() raise SFTPError(t("err.sftp.auth", host=self.host)) except paramiko.SSHException as e: + self.disconnect() raise SFTPError(t("err.sftp.ssh", error=str(e))) except OSError as e: + self.disconnect() raise SFTPError( t("err.sftp.connect", host=self.host, port=self.port, error=str(e)) ) @@ -130,12 +133,21 @@ class SFTPClient: return None def disconnect(self) -> None: - """断开 SFTP 连接""" - if self._sftp is not None: - self._sftp.close() + """断开 SFTP 连接(确保每个资源都被清理)""" + try: + if self._sftp is not None: + self._sftp.close() + except Exception: + pass + finally: self._sftp = None - if self._transport is not None: - self._transport.close() + + try: + if self._transport is not None: + self._transport.close() + except Exception: + pass + finally: self._transport = None def __enter__(self): diff --git a/sbackup/webdav.py b/sbackup/webdav.py index 10ab8f1..20d32a8 100644 --- a/sbackup/webdav.py +++ b/sbackup/webdav.py @@ -42,14 +42,14 @@ class WebDAVClient: return f"{self.url}/{path}" return self.url - def _request( + def _build_request( self, method: str, path: str = "", data: bytes | None = None, content_type: str = "application/octet-stream", ) -> urllib.request.Request: - """构建并发送 HTTP 请求""" + """构建 HTTP 请求对象""" url = self._build_url(path) req = urllib.request.Request(url, method=method) req.add_header("Authorization", self._auth_header) @@ -68,7 +68,7 @@ class WebDAVClient: for part in parts: current = f"{current}/{part}" if current else part try: - req = self._request("MKCOL", current) + req = self._build_request("MKCOL", current) urllib.request.urlopen(req, timeout=30) logger.debug("创建远程目录: %s", current) except urllib.error.HTTPError as e: @@ -85,7 +85,7 @@ class WebDAVClient: def connect(self) -> None: """测试 WebDAV 连接(PROPFIND)""" try: - req = self._request( + req = self._build_request( "PROPFIND", data=b'' b'', @@ -97,7 +97,7 @@ class WebDAVClient: except urllib.error.HTTPError as e: if e.code == 401: raise WebDAVError(t("err.webdav.auth", host=self.url)) - raise WebDAVError(t("err.webdav.ssh", error=str(e))) + raise WebDAVError(t("err.webdav.connect", url=self.url, error=str(e))) except OSError as e: raise WebDAVError(t("err.webdav.connect", url=self.url, error=str(e))) @@ -121,10 +121,13 @@ class WebDAVClient: ) try: + # 流式上传:不将整个文件读入内存 + url = self._build_url(remote_path) with open(local_path, "rb") as f: - data = f.read() - req = self._request("PUT", remote_path, data=data) - urllib.request.urlopen(req, timeout=300) + req = urllib.request.Request(url, data=f, method="PUT") + req.add_header("Authorization", self._auth_header) + req.add_header("Content-Length", str(file_size)) + urllib.request.urlopen(req, timeout=300) logger.debug("上传成功: %s", remote_path) return file_size except urllib.error.HTTPError as e: @@ -141,3 +144,9 @@ class WebDAVClient: return True except WebDAVError: return False + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + return False diff --git a/tests/sbackup/test_auto_save.py b/tests/sbackup/test_auto_save.py index ba0d503..1d7a379 100644 --- a/tests/sbackup/test_auto_save.py +++ b/tests/sbackup/test_auto_save.py @@ -9,6 +9,7 @@ import tempfile import shutil import time from pathlib import Path +from unittest.mock import patch, MagicMock from sbackup.auto_save import BackupManager from sbackup.i18n import t @@ -139,6 +140,11 @@ class TestAutoSave(unittest.TestCase): result = self.manager.add_folder(self.source_folder, "/nonexistent/path") self.assertFalse(result) + def test_add_source_equals_dest(self): + """测试源目录与目标目录相同时应失败""" + result = self.manager.add_folder(self.source_folder, self.source_folder) + self.assertFalse(result) + def test_save_missing_source(self): """测试源文件夹不存在时的 save 行为""" # 手动构造一条源文件夹不存在的记录 @@ -455,6 +461,241 @@ class TestAutoSave(unittest.TestCase): # 应包含"默认"文本(中文环境) self.assertIn(t("table.cell.default"), text) + def test_display_width_ascii(self): + """测试 ASCII 字符宽度计算""" + self.assertEqual(BackupManager._display_width("hello"), 5) + + def test_display_width_cjk(self): + """测试中文字符宽度计算(每个算2)""" + self.assertEqual(BackupManager._display_width("你好"), 4) + + def test_display_width_fullwidth(self): + """测试全角符号宽度计算(U+FF01-FF60 等)""" + # A 是全角大写 A,宽度应为2 + self.assertEqual(BackupManager._display_width("A"), 2) + # ! 是全角感叹号 + self.assertEqual(BackupManager._display_width("!"), 2) + + def test_display_width_mixed(self): + """测试中英混合字符串宽度""" + # "hi你好" = 1+1+2+2 = 6 + self.assertEqual(BackupManager._display_width("hi你好"), 6) + + def test_display_width_non_string(self): + """测试非字符串输入""" + self.assertEqual(BackupManager._display_width(123), 3) + + def test_display_width_empty(self): + """测试空字符串""" + self.assertEqual(BackupManager._display_width(""), 0) + + def test_list_folder_table_only_history(self): + """测试仅含 _history 数据时返回空提示""" + self.manager.data["_history"] = [ + {"time": "2026-01-01", "source": "/x", "size_mb": 1.0, "files_count": 1} + ] + text = self.manager.list_folder_table() + self.assertEqual(text, t("cmd.all.empty")) + + def test_format_history_table_empty(self): + """测试无历史记录时返回提示""" + text = self.manager.format_history_table() + self.assertEqual(text, t("cmd.list.empty")) + + def test_format_history_table_with_data(self): + """测试有历史记录时生成表格""" + self.manager.add_folder(self.source_folder, self.target_folder, "") + import time + + time.sleep(0.1) + (Path(self.source_folder) / "f.txt").write_text("x") + self.manager.execute_backups() + text = self.manager.format_history_table() + self.assertIn(self.source_folder, text) + self.assertIn(t("table.header.time"), text) + + def test_entry_from_list_invalid_data(self): + """测试 from_list 输入验证:非列表返回空条目""" + from sbackup.auto_save import BackupEntry + + entry = BackupEntry.from_list("not a list") + self.assertEqual(entry.mtime, 0.0) + self.assertEqual(entry.target, "") + + def test_entry_from_list_too_short(self): + """测试 from_list 输入验证:少于3元素返回空条目""" + from sbackup.auto_save import BackupEntry + + entry = BackupEntry.from_list([1.0, "/path"]) + self.assertEqual(entry.mtime, 0.0) + self.assertEqual(entry.target, "") + + def test_entry_from_list_empty(self): + """测试 from_list 输入验证:空列表""" + from sbackup.auto_save import BackupEntry + + entry = BackupEntry.from_list([]) + self.assertEqual(entry.mtime, 0.0) + + +class TestUploadSftp(unittest.TestCase): + """测试 _upload_to_sftp 静态方法""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.test_file = os.path.join(self.test_dir, "backup.zip") + with open(self.test_file, "w") as f: + f.write("fake backup") + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + @patch("sbackup.sftp.SFTPClient") + def test_upload_not_configured(self, mock_sftp_cls): + """测试 SFTP 未配置时打印错误""" + from sbackup.config import Config + + config = Config(sftp_enabled=False) + with patch("builtins.print") as mock_print: + BackupManager._upload_to_sftp([self.test_file], config) + mock_print.assert_called() + + @patch("sbackup.sftp.SFTPClient") + def test_upload_no_credentials(self, mock_sftp_cls): + """测试无凭据时打印提示""" + from sbackup.config import Config + + mock_sftp_cls.try_default_key.return_value = None + config = Config(sftp_enabled=True, sftp_host="host", sftp_password="") + with patch("builtins.print") as mock_print: + BackupManager._upload_to_sftp([self.test_file], config) + printed = " ".join(str(c) for c in mock_print.call_args_list) + # 无默认私钥时应打印提示 + self.assertIn(t("cmd.sftp.no_default_key"), printed) + + @patch("sbackup.sftp.SFTPClient") + def test_upload_success(self, mock_sftp_cls): + """测试 SFTP 上传成功""" + from sbackup.config import Config + + mock_client = MagicMock() + mock_sftp_cls.return_value.__enter__ = MagicMock(return_value=mock_client) + mock_sftp_cls.return_value.__exit__ = MagicMock(return_value=False) + + config = Config( + sftp_enabled=True, + sftp_host="host", + sftp_user="user", + sftp_password="pass", + ) + with patch("builtins.print"): + BackupManager._upload_to_sftp([self.test_file], config) + mock_client.upload_file.assert_called_once() + + @patch("sbackup.sftp.SFTPClient") + def test_upload_sftp_error(self, mock_sftp_cls): + """测试 SFTP 上传失败""" + from sbackup.config import Config + from sbackup.sftp import SFTPError + + mock_client = MagicMock() + mock_client.upload_file.side_effect = SFTPError("upload failed") + mock_sftp_cls.return_value.__enter__ = MagicMock(return_value=mock_client) + mock_sftp_cls.return_value.__exit__ = MagicMock(return_value=False) + + config = Config( + sftp_enabled=True, + sftp_host="host", + sftp_user="user", + sftp_password="pass", + ) + with patch("builtins.print") as mock_print: + BackupManager._upload_to_sftp([self.test_file], config) + printed = " ".join(str(c) for c in mock_print.call_args_list) + self.assertIn("upload failed", printed) + + +class TestUploadWebdav(unittest.TestCase): + """测试 _upload_to_webdav 静态方法""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.test_file = os.path.join(self.test_dir, "backup.zip") + with open(self.test_file, "w") as f: + f.write("fake backup") + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + @patch("sbackup.webdav.WebDAVClient") + def test_upload_not_configured(self, mock_wdav_cls): + """测试 WebDAV 未配置时打印错误""" + from sbackup.config import Config + + config = Config(webdav_enabled=False) + with patch("builtins.print") as mock_print: + BackupManager._upload_to_webdav([self.test_file], config) + mock_print.assert_called() + + @patch("sbackup.webdav.WebDAVClient") + def test_upload_success(self, mock_wdav_cls): + """测试 WebDAV 上传成功""" + from sbackup.config import Config + + mock_client = MagicMock() + mock_wdav_cls.return_value = mock_client + + config = Config( + webdav_enabled=True, + webdav_url="https://dav.example.com", + webdav_user="user", + webdav_password="pass", + ) + with patch("builtins.print"): + BackupManager._upload_to_webdav([self.test_file], config) + mock_client.connect.assert_called_once() + mock_client.upload_file.assert_called_once() + + @patch("sbackup.webdav.WebDAVClient") + def test_upload_webdav_error(self, mock_wdav_cls): + """测试 WebDAV 上传失败""" + from sbackup.config import Config + from sbackup.webdav import WebDAVError + + mock_client = MagicMock() + mock_client.upload_file.side_effect = WebDAVError("upload failed") + mock_wdav_cls.return_value = mock_client + + config = Config( + webdav_enabled=True, + webdav_url="https://dav.example.com", + webdav_user="user", + webdav_password="pass", + ) + with patch("builtins.print") as mock_print: + BackupManager._upload_to_webdav([self.test_file], config) + printed = " ".join(str(c) for c in mock_print.call_args_list) + self.assertIn("upload failed", printed) + + @patch("sbackup.webdav.WebDAVClient") + def test_upload_connect_error(self, mock_wdav_cls): + """测试 WebDAV 连接失败""" + from sbackup.config import Config + from sbackup.webdav import WebDAVError + + mock_wdav_cls.return_value.connect.side_effect = WebDAVError("connect failed") + + config = Config( + webdav_enabled=True, + webdav_url="https://dav.example.com", + webdav_user="user", + webdav_password="pass", + ) + with patch("builtins.print") as mock_print: + BackupManager._upload_to_webdav([self.test_file], config) + printed = " ".join(str(c) for c in mock_print.call_args_list) + self.assertIn("connect failed", printed) + if __name__ == "__main__": unittest.main() diff --git a/tests/sbackup/test_compression.py b/tests/sbackup/test_compression.py index 9e1c45f..42066f9 100644 --- a/tests/sbackup/test_compression.py +++ b/tests/sbackup/test_compression.py @@ -6,9 +6,10 @@ import unittest import os import shutil import tarfile +import tempfile import zipfile from pathlib import Path -from unittest.mock import patch +from unittest.mock import patch, MagicMock from sbackup.config import Config from sbackup.compression import ( ZipfileCompression, @@ -17,6 +18,8 @@ from sbackup.compression import ( SevenZipCompression, create_compressor, restore_backup, + list_backup_contents, + verify_backup, ) from sbackup.i18n import t @@ -465,6 +468,48 @@ class TestRestore(unittest.TestCase): ) os.remove("test.tar.zst") + @patch("builtins.print") + def test_restore_permission_error(self, mock_print): + """测试还原时权限不足""" + config = Config(folder_path=self.test_dir, zipfile_path="test.zip") + ZipfileCompression(config).compress() + with patch("zipfile.ZipFile") as mock_zf: + mock_zf.return_value.__enter__ = MagicMock( + side_effect=PermissionError("denied") + ) + mock_zf.return_value.__exit__ = MagicMock(return_value=False) + result = restore_backup("test.zip", self.restore_dir) + self.assertFalse(result["success"]) + + @patch("builtins.print") + def test_restore_os_error(self, mock_print): + """测试还原时 OS 错误""" + config = Config(folder_path=self.test_dir, zipfile_path="test.zip") + ZipfileCompression(config).compress() + with patch("zipfile.ZipFile") as mock_zf: + mock_zf.return_value.__enter__ = MagicMock( + side_effect=OSError("disk error") + ) + mock_zf.return_value.__exit__ = MagicMock(return_value=False) + result = restore_backup("test.zip", self.restore_dir) + self.assertFalse(result["success"]) + + @patch("builtins.print") + def test_restore_7z_permission_error(self, mock_print): + """测试 7z 还原时权限不足""" + config = Config( + folder_path=self.test_dir, zipfile_path="test.7z", compression_format="7Z" + ) + SevenZipCompression(config).compress() + with patch("py7zr.SevenZipFile") as mock_7z: + mock_7z.return_value.__enter__ = MagicMock( + side_effect=PermissionError("denied") + ) + mock_7z.return_value.__exit__ = MagicMock(return_value=False) + result = restore_backup("test.7z", self.restore_dir) + self.assertFalse(result["success"]) + os.remove("test.7z") + class TestPathMatching(unittest.TestCase): def setUp(self): @@ -534,6 +579,44 @@ class TestTarNoCompression(unittest.TestCase): self.assertTrue(result["success"]) self.assertTrue(os.path.exists("test.tar")) + @patch("builtins.print") + def test_tar_permission_error(self, mock_print): + """测试 tar 压缩时权限不足""" + config = Config( + folder_path=self.test_dir, + zipfile_path="/root/test.tar", + compression_format="TAR", + ) + compressor = TarfileCompression(config) + result = compressor.compress() + self.assertFalse(result["success"]) + + @patch("builtins.print") + def test_tar_os_error(self, mock_print): + """测试 tar 压缩时系统错误""" + config = Config( + folder_path=self.test_dir, + zipfile_path="test.tar", + compression_format="TAR", + ) + compressor = TarfileCompression(config) + with patch("tarfile.open", side_effect=OSError("disk error")): + result = compressor.compress() + self.assertFalse(result["success"]) + + @patch("builtins.print") + def test_tar_unknown_exception(self, mock_print): + """测试 tar 压缩时未知异常""" + config = Config( + folder_path=self.test_dir, + zipfile_path="test.tar", + compression_format="TAR", + ) + compressor = TarfileCompression(config) + with patch("tarfile.open", side_effect=RuntimeError("unexpected")): + result = compressor.compress() + self.assertFalse(result["success"]) + class TestZstdCompression(unittest.TestCase): def setUp(self): @@ -613,6 +696,19 @@ class TestZstdCompression(unittest.TestCase): result = compressor.compress() self.assertFalse(result["success"]) + @patch("builtins.print") + def test_zstd_unknown_exception(self, mock_print): + """测试 zstd 压缩未知异常""" + config = Config( + folder_path=self.test_dir, + zipfile_path="test.tar.zst", + compression_format="TAR_ZST", + ) + compressor = ZstdCompression(config) + with patch("zstandard.ZstdCompressor", side_effect=RuntimeError("unexpected")): + result = compressor.compress() + self.assertFalse(result["success"]) + class TestSevenZipCompression(unittest.TestCase): def setUp(self): @@ -704,6 +800,218 @@ class TestSevenZipCompression(unittest.TestCase): names = szf.getnames() self.assertGreater(len(names), 0) + @patch("builtins.print") + def test_7z_unknown_exception(self, mock_print): + """测试 7z 压缩未知异常""" + config = Config( + folder_path=self.test_dir, zipfile_path="test.7z", compression_format="7Z" + ) + compressor = SevenZipCompression(config) + with patch("py7zr.SevenZipFile", side_effect=RuntimeError("unexpected")): + result = compressor.compress() + self.assertFalse(result["success"]) + + +class TestCreateCompressor(unittest.TestCase): + """测试工厂函数 create_compressor 的各种格式""" + + def test_create_compressor_tar(self): + """测试工厂函数创建 TAR 压缩器""" + config = Config(folder_path=".", compression_format="TAR") + compressor = create_compressor(config) + self.assertIsInstance(compressor, TarfileCompression) + + def test_create_compressor_tar_gz_with_dot(self): + """测试 tar.gz(带点号)格式正确匹配 TarfileCompression""" + config = Config(folder_path=".", compression_format="tar.gz") + compressor = create_compressor(config) + self.assertIsInstance(compressor, TarfileCompression) + + def test_create_compressor_tar_bz2_with_dot(self): + """测试 tar.bz2(带点号)格式正确匹配""" + config = Config(folder_path=".", compression_format="tar.bz2") + compressor = create_compressor(config) + self.assertIsInstance(compressor, TarfileCompression) + + def test_create_compressor_tar_xz_with_dot(self): + """测试 tar.xz(带点号)格式正确匹配""" + config = Config(folder_path=".", compression_format="tar.xz") + compressor = create_compressor(config) + self.assertIsInstance(compressor, TarfileCompression) + + def test_create_compressor_tar_zst_with_dot(self): + """测试 tar.zst(带点号)格式正确匹配""" + config = Config(folder_path=".", compression_format="tar.zst") + compressor = create_compressor(config) + self.assertIsInstance(compressor, ZstdCompression) + + +class TestShouldIgnore(unittest.TestCase): + """测试 _should_ignore 模式匹配""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + os.makedirs(os.path.join(self.test_dir, "sub"), exist_ok=True) + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_should_ignore_directory_pattern(self): + """测试忽略目录模式""" + config = Config(folder_path=self.test_dir, skip_patterns=["sub"]) + compressor = ZipfileCompression(config) + self.assertTrue(compressor._should_ignore("sub")) + + def test_should_ignore_no_match(self): + """测试不匹配时不忽略""" + config = Config(folder_path=self.test_dir, skip_patterns=["*.log"]) + compressor = ZipfileCompression(config) + self.assertFalse(compressor._should_ignore("readme.txt")) + + def test_should_ignore_basename_match(self): + """测试 basename 匹配""" + config = Config(folder_path=self.test_dir, skip_patterns=["*.log"]) + compressor = ZipfileCompression(config) + self.assertTrue(compressor._should_ignore("subdir/error.log")) + + +class TestListBackupContents(unittest.TestCase): + """测试 list_backup_contents 函数""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + (Path(self.test_dir) / "file1.txt").write_text("content1") + (Path(self.test_dir) / "sub").mkdir() + (Path(self.test_dir) / "sub" / "file2.txt").write_text("content2") + self.zip_path = os.path.join(self.test_dir, "test.zip") + config = Config(folder_path=self.test_dir, zipfile_path=self.zip_path) + ZipfileCompression(config).compress() + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_list_zip_contents(self): + """测试列出 ZIP 文件内容""" + result = list_backup_contents(self.zip_path) + self.assertIn("file1.txt", result) + self.assertIn("file2.txt", result) + self.assertIn("2 files", result) + + def test_list_nonexistent_file(self): + """测试列出不存在的文件""" + result = list_backup_contents("/nonexistent/file.zip") + self.assertIn("/nonexistent/file.zip", result) + + def test_list_unknown_format(self): + """测试列出未知格式文件(返回空列表提示)""" + unknown = os.path.join(self.test_dir, "test.xyz") + Path(unknown).write_text("data") + result = list_backup_contents(unknown) + self.assertIn("test.xyz", result) + + +class TestVerifyBackup(unittest.TestCase): + """测试 verify_backup 函数""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + (Path(self.test_dir) / "file1.txt").write_text("content1") + (Path(self.test_dir) / "sub").mkdir() + (Path(self.test_dir) / "sub" / "file2.txt").write_text("content2") + self.zip_path = os.path.join(self.test_dir, "test.zip") + config = Config(folder_path=self.test_dir, zipfile_path=self.zip_path) + ZipfileCompression(config).compress() + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_verify_valid_zip(self): + """测试校验有效 ZIP 文件""" + result = verify_backup(self.zip_path) + self.assertTrue(result["success"]) + self.assertEqual(result["files_count"], 2) + + def test_verify_nonexistent(self): + """测试校验不存在的文件""" + result = verify_backup("/nonexistent/file.zip") + self.assertFalse(result["success"]) + + +class TestSbackupIgnore(unittest.TestCase): + """测试 .sbackupignore 文件支持""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + (Path(self.test_dir) / "keep.txt").write_text("keep") + (Path(self.test_dir) / "skip.log").write_text("skip") + (Path(self.test_dir) / "sub").mkdir() + (Path(self.test_dir) / "sub" / "data.txt").write_text("data") + self.zip_path = os.path.join(self.test_dir, "test.zip") + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_ignore_file_loaded(self): + """测试 .sbackupignore 文件中的规则生效""" + (Path(self.test_dir) / ".sbackupignore").write_text("*.log\n# comment\n") + config = Config( + folder_path=self.test_dir, + zipfile_path=self.zip_path, + skip_patterns=[], + ) + compressor = ZipfileCompression(config) + result = compressor.compress() + self.assertTrue(result["success"]) + with zipfile.ZipFile(self.zip_path, "r") as zf: + names = zf.namelist() + self.assertTrue(any(n.endswith("keep.txt") for n in names)) + self.assertFalse(any(n.endswith("skip.log") for n in names)) + + def test_ignore_file_not_present(self): + """测试没有 .sbackupignore 时不影响正常压缩""" + config = Config( + folder_path=self.test_dir, + zipfile_path=self.zip_path, + skip_patterns=[], + ) + compressor = ZipfileCompression(config) + result = compressor.compress() + self.assertTrue(result["success"]) + with zipfile.ZipFile(self.zip_path, "r") as zf: + names = zf.namelist() + self.assertTrue(any(n.endswith("keep.txt") for n in names)) + self.assertTrue(any(n.endswith("skip.log") for n in names)) + + +class TestPatternNegation(unittest.TestCase): + """测试 ! 取反模式""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + (Path(self.test_dir) / "a.log").write_text("log") + (Path(self.test_dir) / "b.log").write_text("log") + (Path(self.test_dir) / "important.log").write_text("important") + self.zip_path = os.path.join(self.test_dir, "test.zip") + + def tearDown(self): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_negation_restores_file(self): + """测试 ! 模式恢复被忽略的文件""" + config = Config( + folder_path=self.test_dir, + zipfile_path=self.zip_path, + skip_patterns=["*.log", "!important.log"], + ) + compressor = ZipfileCompression(config) + result = compressor.compress() + self.assertTrue(result["success"]) + with zipfile.ZipFile(self.zip_path, "r") as zf: + names = zf.namelist() + self.assertFalse(any(n.endswith("a.log") for n in names)) + self.assertFalse(any(n.endswith("b.log") for n in names)) + self.assertTrue(any(n.endswith("important.log") for n in names)) + if __name__ == "__main__": unittest.main() diff --git a/tests/sbackup/test_config.py b/tests/sbackup/test_config.py index 5873f99..18e0736 100644 --- a/tests/sbackup/test_config.py +++ b/tests/sbackup/test_config.py @@ -395,6 +395,150 @@ class TestSaveSftpConfig(unittest.TestCase): self.assertEqual(config.sftp_key_passphrase, "keypass") self.assertTrue(config.sftp_enabled) + def test_load_config_default_lang_zh_cn(self): + """测试 config.json 缺少 lang 字段时默认为 zh_CN""" + with open(self.config_file, "w", encoding="utf-8") as f: + json.dump({"compression_format": "ZIP"}, f) + config = load_config(self.config_file) + self.assertEqual(config.lang, "zh_CN") + + def test_load_config_with_webdav(self): + """测试加载包含 WebDAV 配置""" + config_data = { + "webdav": { + "url": "https://dav.jianguoyun.com/dav/", + "user": "user@example.com", + "password": "secret", + "remote_path": "/backups", + "enabled": True, + } + } + with open(self.config_file, "w", encoding="utf-8") as f: + json.dump(config_data, f) + config = load_config(self.config_file) + self.assertEqual(config.webdav_url, "https://dav.jianguoyun.com/dav/") + self.assertEqual(config.webdav_user, "user@example.com") + self.assertEqual(config.webdav_password, "secret") + self.assertEqual(config.webdav_remote_path, "/backups") + self.assertTrue(config.webdav_enabled) + + def test_load_config_webdav_defaults(self): + """测试 WebDAV 配置默认值""" + with open(self.config_file, "w", encoding="utf-8") as f: + json.dump({}, f) + config = load_config(self.config_file) + self.assertEqual(config.webdav_url, "") + self.assertEqual(config.webdav_user, "") + self.assertEqual(config.webdav_password, "") + self.assertEqual(config.webdav_remote_path, "/") + self.assertFalse(config.webdav_enabled) + + def test_save_webdav_config_creates_new_file(self): + """测试 save_webdav_config 创建新配置文件""" + from sbackup.config import save_webdav_config + + save_webdav_config( + "https://dav.example.com/dav/", + "user@test.com", + "pass123", + "/backups", + config_file=self.config_file, + ) + self.assertTrue(os.path.exists(self.config_file)) + with open(self.config_file, "r", encoding="utf-8") as f: + data = json.load(f) + self.assertEqual(data["webdav"]["url"], "https://dav.example.com/dav/") + self.assertEqual(data["webdav"]["user"], "user@test.com") + self.assertEqual(data["webdav"]["password"], "pass123") + self.assertEqual(data["webdav"]["remote_path"], "/backups") + self.assertTrue(data["webdav"]["enabled"]) + + def test_save_webdav_config_updates_existing(self): + """测试 save_webdav_config 更新已有配置""" + from sbackup.config import save_webdav_config + + with open(self.config_file, "w", encoding="utf-8") as f: + json.dump({"lang": "zh_CN"}, f) + save_webdav_config( + "https://dav.example.com/", + "user", + "pass", + "/", + config_file=self.config_file, + ) + with open(self.config_file, "r", encoding="utf-8") as f: + data = json.load(f) + self.assertEqual(data["webdav"]["url"], "https://dav.example.com/") + self.assertEqual(data["lang"], "zh_CN") + + +class TestJsonHelpers(unittest.TestCase): + """测试 _load_json_file 和 _save_json_file 辅助函数""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.config_file = os.path.join(self.test_dir, "config.json") + + def tearDown(self): + import shutil + + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_load_json_file_nonexistent(self): + """测试加载不存在的文件返回空字典""" + from sbackup.config import _load_json_file + + result = _load_json_file("/nonexistent/file.json") + self.assertEqual(result, {}) + + def test_load_json_file_valid(self): + """测试加载有效的 JSON 文件""" + from sbackup.config import _load_json_file + + with open(self.config_file, "w", encoding="utf-8") as f: + json.dump({"key": "value"}, f) + result = _load_json_file(self.config_file) + self.assertEqual(result, {"key": "value"}) + + def test_load_json_file_malformed(self): + """测试加载损坏的 JSON 文件返回空字典""" + from sbackup.config import _load_json_file + + with open(self.config_file, "w") as f: + f.write("{bad json") + result = _load_json_file(self.config_file) + self.assertEqual(result, {}) + + def test_save_json_file_creates_dir(self): + """测试自动创建目录""" + from sbackup.config import _save_json_file + + nested_file = os.path.join(self.test_dir, "sub", "config.json") + _save_json_file({"test": True}, nested_file) + self.assertTrue(os.path.exists(nested_file)) + with open(nested_file, "r", encoding="utf-8") as f: + data = json.load(f) + self.assertEqual(data, {"test": True}) + + @patch("os.makedirs") + def test_save_json_file_makedirs_error(self, mock_makedirs): + """测试目录创建失败时静默返回""" + from sbackup.config import _save_json_file + + mock_makedirs.side_effect = OSError("denied") + nested_file = os.path.join(self.test_dir, "sub", "config.json") + _save_json_file({"test": True}, nested_file) + # 不应抛出异常 + + @patch("builtins.open") + def test_save_json_file_write_error(self, mock_open): + """测试写入失败时静默处理""" + from sbackup.config import _save_json_file + + mock_open.side_effect = OSError("disk full") + _save_json_file({"test": True}, self.config_file) + # 不应抛出异常 + if __name__ == "__main__": unittest.main() diff --git a/tests/sbackup/test_main.py b/tests/sbackup/test_main.py index 68d4190..6eb4426 100644 --- a/tests/sbackup/test_main.py +++ b/tests/sbackup/test_main.py @@ -23,6 +23,9 @@ class TestMain(unittest.TestCase): self.data_path = os.path.join(self.test_dir, "sbackup.json") with open(os.path.join(self.test_dir, "config.json"), "w") as f: json.dump({"data_file": self.data_path}, f) + # 创建独立的目标目录(不能与源目录相同) + self.dest_dir = os.path.join(self.test_dir, "backup_dest") + os.makedirs(self.dest_dir, exist_ok=True) def tearDown(self): sys.argv = self.original_argv @@ -117,7 +120,7 @@ class TestMain(unittest.TestCase): def test_rm_command_success(self, mock_print): """测试 rm 命令成功删除策略""" os.chdir(self.test_dir) - sys.argv = ["sbackup", "--lang", "en_US", "add", self.test_dir, self.test_dir] + sys.argv = ["sbackup", "--lang", "en_US", "add", self.test_dir, self.dest_dir] from sbackup import run result1 = run() @@ -187,7 +190,7 @@ class TestMain(unittest.TestCase): "en_US", "add", self.test_dir, - self.test_dir, + self.dest_dir, "--format", "tar.gz", ] @@ -217,6 +220,27 @@ class TestMain(unittest.TestCase): args = parser.parse_args(["watch", "--interval", "30"]) self.assertEqual(args.interval, 30) + def test_restore_password_argument(self): + """测试 restore --password 参数""" + from sbackup import get_parser + + parser = get_parser() + args = parser.parse_args( + ["restore", "backup.7z", "/tmp", "--password", "secret"] + ) + self.assertEqual(args.command, "restore") + self.assertEqual(args.backup_file, "backup.7z") + self.assertEqual(args.target_dir, "/tmp") + self.assertEqual(args.password, "secret") + + def test_restore_password_default_empty(self): + """测试 restore --password 默认为空""" + from sbackup import get_parser + + parser = get_parser() + args = parser.parse_args(["restore", "backup.zip", "/tmp"]) + self.assertEqual(args.password, "") + def test_argparse_invalid_choice_localized(self): """测试 argparse 无效选择错误被本地化""" from sbackup import get_parser @@ -293,6 +317,28 @@ class TestMain(unittest.TestCase): printed = " ".join(str(call) for call in mock_print.call_args_list) self.assertIn("not configured", printed) + @patch("builtins.print") + def test_sftp_no_action(self, mock_print): + """测试 sftp 无子命令时提示并返回 1""" + os.chdir(self.test_dir) + sys.argv = ["sbackup", "--lang", "en_US", "sftp"] + from sbackup import run + + result = run() + self.assertEqual(result, 1) + mock_print.assert_called() + + @patch("builtins.print") + def test_webdav_no_action(self, mock_print): + """测试 webdav 无子命令时提示并返回 1""" + os.chdir(self.test_dir) + sys.argv = ["sbackup", "--lang", "en_US", "webdav"] + from sbackup import run + + result = run() + self.assertEqual(result, 1) + mock_print.assert_called() + @patch("builtins.print") @patch("builtins.input") @patch("getpass.getpass") @@ -352,6 +398,37 @@ class TestMain(unittest.TestCase): printed = " ".join(str(call) for call in mock_print.call_args_list) self.assertIn("saved", printed) + @patch("builtins.print") + @patch("sbackup.sftp.SFTPClient._load_private_key", return_value=MagicMock()) + @patch("sbackup.sftp.SFTPClient.try_default_key", return_value=None) + def test_sftp_config_key_no_passphrase_needed( + self, mock_try_key, mock_load_key, mock_print + ): + """测试 sftp config --key-file 私钥不需要密码短语时不提示输入""" + os.chdir(self.test_dir) + sys.argv = [ + "sbackup", + "--lang", + "en_US", + "sftp", + "config", + "--host", + "myhost", + "--port", + "22", + "--user", + "admin", + "--key-file", + "/path/to/id_rsa", + "--remote-path", + "/backups", + ] + from sbackup import run + + result = run() + self.assertEqual(result, 0) + mock_load_key.assert_called_once() + @patch("builtins.print") def test_sftp_test_with_config(self, mock_print): """测试 sftp test 连接成功""" diff --git a/tests/sbackup/test_sftp.py b/tests/sbackup/test_sftp.py index 905a599..a5e1157 100644 --- a/tests/sbackup/test_sftp.py +++ b/tests/sbackup/test_sftp.py @@ -305,6 +305,65 @@ class TestSFTPClient(unittest.TestCase): mock_sftp.stat.assert_called_once_with("/existing/dir") mock_sftp.mkdir.assert_not_called() + @patch("paramiko.Transport") + @patch("paramiko.SFTPClient") + def test_ensure_remote_dir_nested_create(self, mock_sftp_cls, mock_transport_cls): + """测试远程嵌套目录逐级创建""" + mock_transport = MagicMock() + mock_transport_cls.return_value = mock_transport + mock_sftp = MagicMock() + # 所有目录都不存在 + mock_sftp.stat.side_effect = FileNotFoundError() + mock_sftp_cls.from_transport.return_value = mock_sftp + + client = SFTPClient(self.host, self.port, self.user, self.password) + client.connect() + client._ensure_remote_dir("/a/b/c") + # 应逐级创建 /a, /a/b, /a/b/c + expected_calls = ["/a/b/c", "/a/b", "/a"] + actual_calls = [c[0][0] for c in mock_sftp.stat.call_args_list] + self.assertEqual(actual_calls, expected_calls) + self.assertEqual(mock_sftp.mkdir.call_count, 3) + + @patch("paramiko.Transport") + @patch("paramiko.SFTPClient") + def test_ensure_remote_dir_empty(self, mock_sftp_cls, mock_transport_cls): + """测试空路径不操作""" + mock_transport = MagicMock() + mock_transport_cls.return_value = mock_transport + mock_sftp = MagicMock() + mock_sftp_cls.from_transport.return_value = mock_sftp + + client = SFTPClient(self.host, self.port, self.user, self.password) + client.connect() + client._ensure_remote_dir("") + mock_sftp.stat.assert_not_called() + mock_sftp.mkdir.assert_not_called() + + @patch("paramiko.Transport") + @patch("paramiko.SFTPClient") + def test_ensure_remote_dir_mkdir_error(self, mock_sftp_cls, mock_transport_cls): + """测试创建远程目录失败""" + mock_transport = MagicMock() + mock_transport_cls.return_value = mock_transport + mock_sftp = MagicMock() + mock_sftp.stat.side_effect = FileNotFoundError() + mock_sftp.mkdir.side_effect = OSError("permission denied") + mock_sftp_cls.from_transport.return_value = mock_sftp + + client = SFTPClient(self.host, self.port, self.user, self.password) + client.connect() + with self.assertRaises(SFTPError): + client._ensure_remote_dir("/forbidden") + + @patch("paramiko.Transport") + @patch("paramiko.SFTPClient") + def test_ensure_remote_dir_not_connected(self, mock_sftp_cls, mock_transport_cls): + """测试未连接时 _ensure_remote_dir 不报错""" + client = SFTPClient(self.host, self.port, self.user, self.password) + client._ensure_remote_dir("/some/path") + # 不应抛出异常 + # ========== 私钥认证测试 ========== def test_init_with_key_file(self): @@ -579,6 +638,46 @@ class TestSFTPClient(unittest.TestCase): username=self.user, pkey=mock_pkey ) + # ========== 资源清理测试 ========== + + @patch("paramiko.Transport") + @patch("paramiko.SFTPClient") + def test_disconnect_cleans_up_even_if_sftp_close_raises( + self, mock_sftp_cls, mock_transport_cls + ): + """测试 sftp.close() 异常时 transport 仍被清理""" + mock_transport = MagicMock() + mock_transport_cls.return_value = mock_transport + mock_sftp = MagicMock() + mock_sftp_cls.from_transport.return_value = mock_sftp + mock_sftp.close.side_effect = OSError("socket closed") + + client = SFTPClient(self.host, self.port, self.user, self.password) + client.connect() + client.disconnect() + + mock_transport.close.assert_called_once() + self.assertIsNone(client._sftp) + self.assertIsNone(client._transport) + + @patch("paramiko.Transport") + def test_connect_failure_cleans_up_transport(self, mock_transport_cls): + """测试连接失败时 transport 被清理""" + import paramiko + + mock_transport = MagicMock() + mock_transport_cls.return_value = mock_transport + mock_transport.connect.side_effect = paramiko.AuthenticationException( + "auth failed" + ) + + client = SFTPClient(self.host, self.port, self.user, self.password) + with self.assertRaises(SFTPError): + client.connect() + + mock_transport.close.assert_called_once() + self.assertIsNone(client._transport) + if __name__ == "__main__": unittest.main() diff --git a/tests/sbackup/test_webdav.py b/tests/sbackup/test_webdav.py index bc24594..7324c89 100644 --- a/tests/sbackup/test_webdav.py +++ b/tests/sbackup/test_webdav.py @@ -132,6 +132,89 @@ class TestWebDAVClient(unittest.TestCase): result = self.client.test_connection() self.assertFalse(result) + @patch("urllib.request.urlopen") + def test_ensure_remote_dir_401(self, mock_urlopen): + """测试 MKCOL 认证失败""" + import urllib.error + + mock_urlopen.side_effect = urllib.error.HTTPError( + self.url, 401, "Unauthorized", {}, None + ) + with self.assertRaises(WebDAVError) as ctx: + self.client._ensure_remote_dir("secret") + self.assertIn("auth", str(ctx.exception).lower()) + + @patch("urllib.request.urlopen") + def test_ensure_remote_dir_other_error(self, mock_urlopen): + """测试 MKCOL 其他 HTTP 错误""" + import urllib.error + + mock_urlopen.side_effect = urllib.error.HTTPError( + self.url, 500, "Server Error", {}, None + ) + with self.assertRaises(WebDAVError): + self.client._ensure_remote_dir("bad_dir") + + @patch("urllib.request.urlopen") + def test_ensure_remote_dir_os_error(self, mock_urlopen): + """测试 MKCOL 网络错误""" + mock_urlopen.side_effect = OSError("Connection refused") + with self.assertRaises(WebDAVError): + self.client._ensure_remote_dir("offline_dir") + + @patch("urllib.request.urlopen") + def test_upload_file_401(self, mock_urlopen): + """测试上传时认证失败""" + import urllib.error + + mock_urlopen.side_effect = urllib.error.HTTPError( + self.url, 401, "Unauthorized", {}, None + ) + with self.assertRaises(WebDAVError) as ctx: + self.client.upload_file(self.test_file, "backups/test.txt") + self.assertIn("auth", str(ctx.exception).lower()) + + @patch("urllib.request.urlopen") + def test_upload_file_os_error(self, mock_urlopen): + """测试上传时网络错误""" + # 首次调用 (_ensure_remote_dir 的 MKCOL) 成功,第二次 (PUT) 失败 + mock_urlopen.side_effect = [ + MagicMock(), # MKCOL 成功 + OSError("Network unreachable"), # PUT 失败 + ] + with self.assertRaises(WebDAVError) as ctx: + self.client.upload_file(self.test_file, "backups/test.txt") + self.assertIn("backups/test.txt", str(ctx.exception)) + + @patch("urllib.request.urlopen") + def test_connect_non_auth_http_error(self, mock_urlopen): + """测试连接时非认证 HTTP 错误""" + import urllib.error + + mock_urlopen.side_effect = urllib.error.HTTPError( + self.url, 502, "Bad Gateway", {}, None + ) + with self.assertRaises(WebDAVError): + self.client.connect() + + def test_context_manager(self): + """测试上下文管理器""" + with WebDAVClient(self.url, self.user, self.password) as client: + self.assertIsInstance(client, WebDAVClient) + + def test_build_request(self): + """测试 _build_request 构建请求对象""" + req = self.client._build_request("GET", "test/path") + self.assertEqual(req.get_method(), "GET") + self.assertIn("Basic", req.get_header("Authorization")) + + def test_build_request_with_data(self): + """测试 _build_request 带数据时正确设置""" + data = b"test data" + req = self.client._build_request("PUT", "upload.txt", data=data) + self.assertEqual(req.get_method(), "PUT") + self.assertEqual(req.full_url, "https://dav.jianguoyun.com/dav/upload.txt") + if __name__ == "__main__": unittest.main()