"""备份功能（契约1.21）：引擎链式备份、配置存储、保留策略与 HTTP 接口。 引擎级测试不依赖数据库（数据库导出以伪实现替代），始终执行；接口级测试沿用 B2 惯例，提供 DMS_TEST_* 环境变量时执行，否则安全跳过。 """ from __future__ import annotations import json import time from pathlib import Path import pytest CONFIG_URL = "/api/v1/backup/config" JOBS_URL = "/api/v1/backup/jobs" CURRENT_URL = "/api/v1/backup/jobs/current" CANCEL_URL = "/api/v1/backup/jobs/current/cancel" PROGRESS_KEYS = { "backupId", "type", "trigger", "phase", "bytesTotal", "bytesCopied", "filesTotal", "filesCopied", "pushPercent", "error", "startedAt", "finishedAt", } HISTORY_ENTRY_KEYS = { "id", "type", "parent", "started_at", "finished_at", "database_file", "database_size", "transferred_files", "transferred_bytes", "storage_file_count", "storage_total_bytes", } # database_url 仅做字符串解析（伪造导出后不连库），任意含 user@host 的串即可 _FAKE_DATABASE_URL = "dm+dmPython://DMS_TESTER:secret@localhost:5236/DMS_TESTER" def _fake_dump(monkeypatch) -> None: from dms.services import sql_dump def fake_export_dump(database_url, output_directory, schema): output_directory.mkdir(parents=True, exist_ok=True) dump_file = output_directory / sql_dump.DUMP_FILE_NAME dump_file.write_text("-- FAKE-DM-DATA-DUMP\n", encoding="utf-8") return dump_file monkeypatch.setattr(sql_dump, "export_dump", fake_export_dump) @pytest.fixture() def engine_env(monkeypatch, tmp_path: Path): """独立引擎（存储/配置/env/备份根全在 tmp_path），不依赖数据库。""" from dms.services.backup_service import BackupConfig, BackupEngine engine = BackupEngine( storage_root=tmp_path / "storage", business_timezone="Asia/Shanghai", database_url=_FAKE_DATABASE_URL, backup_config_path=tmp_path / "dms-backup-config.json", env_path=tmp_path / ".env", ) (tmp_path / ".env").write_text("DMS_BACKUP_TEST_ENV=1\n", encoding="utf-8") backup_root = tmp_path / "backup-root" BackupConfig(local_path=str(backup_root)).save(engine.backup_config_path) _fake_dump(monkeypatch) engine.backup_root = backup_root # type: ignore[attr-defined] return engine @pytest.fixture() def backup_env(b2_app, monkeypatch, tmp_path: Path): """接口测试用：替换应用引擎（备份根/配置/环境文件均指向 tmp_path）。""" from dms.services.backup_service import BackupEngine b2_app.extensions["dms_backup_engine"].stop_scheduler() original = b2_app.extensions["dms_backup_engine"] engine = BackupEngine( storage_root=original.storage_root, business_timezone=b2_app.config["DMS_BUSINESS_TIMEZONE"], database_url=b2_app.config["SQLALCHEMY_DATABASE_URI"], backup_config_path=tmp_path / "dms-backup-config.json", env_path=tmp_path / ".env", ) b2_app.extensions["dms_backup_engine"] = engine (tmp_path / ".env").write_text("DMS_BACKUP_TEST_ENV=1\n", encoding="utf-8") _fake_dump(monkeypatch) b2_app.config["BACKUP_TEST_ROOT"] = tmp_path / "backup-root" return engine def _auth(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} def _save_local_config(client, token, backup_root: Path, **overrides): payload = { "target_mode": "LOCAL", "local_path": str(backup_root), "rsync_url": None, "rsync_password": None, "schedule_enabled": False, "schedule_time": "02:00", "full_interval_days": 7, "retain_full_count": 2, } payload.update(overrides) response = client.put(CONFIG_URL, json=payload, headers=_auth(token)) assert response.status_code == 200 return response.get_json()["data"] def _wait_finished(engine, timeout_seconds: float = 15.0) -> dict: deadline = time.monotonic() + timeout_seconds while engine.is_running or engine.progress_snapshot()["phase"] is None: if time.monotonic() > deadline: pytest.fail("备份任务超时未结束") time.sleep(0.05) return engine.progress_snapshot() def _write_storage_file(engine, relative: str, content: bytes) -> None: path = engine.storage_root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(content) # ================================================================ 引擎级（无数据库，始终执行） def test_engine_progress_snapshot_initial_shape(engine_env): snapshot = engine_env.progress_snapshot() assert set(snapshot) == PROGRESS_KEYS assert snapshot["backupId"] is None assert snapshot["phase"] is None assert snapshot["bytesTotal"] == 0 assert snapshot["pushPercent"] is None def test_engine_full_and_incremental_chain(engine_env): engine = engine_env _write_storage_file(engine, "original/a.pdf", b"AAA") _write_storage_file(engine, "original/sub/b.docx", b"BBBB") _write_storage_file(engine, "preview/c.jpg", b"C") # 全量 engine.start_async("FULL", "MANUAL") full_snapshot = _wait_finished(engine) assert full_snapshot["phase"] == "SUCCESS" assert full_snapshot["error"] is None full_id = full_snapshot["backupId"] assert full_id.endswith("-F") assert full_snapshot["trigger"] == "MANUAL" assert full_snapshot["filesCopied"] == 3 assert full_snapshot["bytesCopied"] == 8 assert full_snapshot["pushPercent"] == 100 # 目标目录布局：manifest / 数据脚本 / env / 存储文件 / 索引 full_dir = engine.backup_root / full_id assert (full_dir / "manifest.json").is_file() assert (full_dir / "database" / "dms-data.sql").is_file() assert (full_dir / "env" / ".env").is_file() assert (full_dir / "storage" / "original" / "a.pdf").is_file() assert (full_dir / "storage" / "original" / "sub" / "b.docx").is_file() assert (full_dir / "storage" / "preview" / "c.jpg").is_file() assert (engine.backup_root / "index.json").is_file() full_manifest = json.loads( (full_dir / "manifest.json").read_text(encoding="utf-8") ) assert full_manifest["type"] == "FULL" assert full_manifest["parent"] is None assert set(full_manifest["files"]) == { "original/a.pdf", "original/sub/b.docx", "preview/c.jpg", } assert full_manifest["transferred_files"] == 3 # 增量：新增一个文件 _write_storage_file(engine, "original/new.txt", b"NEW-DATA") engine.start_async("INCREMENTAL", "SCHEDULE") incremental_snapshot = _wait_finished(engine) incremental_id = incremental_snapshot["backupId"] assert incremental_id.endswith("-I") assert incremental_snapshot["phase"] == "SUCCESS" assert incremental_snapshot["filesCopied"] == 1 assert incremental_snapshot["trigger"] == "SCHEDULE" incremental_manifest = json.loads( (engine.backup_root / incremental_id / "manifest.json").read_text( encoding="utf-8" ) ) assert incremental_manifest["parent"] == full_id # 增量目录仅落差异文件 assert ( engine.backup_root / incremental_id / "storage" / "original" / "new.txt" ).is_file() assert not ( engine.backup_root / incremental_id / "storage" / "original" / "a.pdf" ).exists() # 增量 manifest.files 为链尾累计清单 assert set(incremental_manifest["files"]) == { "original/a.pdf", "original/sub/b.docx", "preview/c.jpg", "original/new.txt", } # 历史与定时类型决策：当日已有全量 → 距今未超间隔 → INCREMENTAL config = engine.load_config() history = engine.list_history(config) assert [entry["id"] for entry in history] == [full_id, incremental_id] assert history[1]["parent"] == full_id assert history[1]["storage_file_count"] == 4 assert engine.decide_scheduled_type(config) == "INCREMENTAL" # 距最近全量超过间隔 → FULL（用三天前的全量独占索引，避免与今日真实全量混淆） config.full_interval_days = 1 old_full_id = "dms-bak-" + time.strftime( "%Y%m%d", time.localtime(time.time() - 3 * 86400) ) + "-F" (engine.backup_root / "index.json").write_text( json.dumps([{"id": old_full_id, "type": "FULL"}]), encoding="utf-8" ) assert engine.decide_scheduled_type(config) == "FULL" def test_engine_incremental_without_history_upgrades_to_full(engine_env): engine = engine_env _write_storage_file(engine, "original/only.txt", b"X") engine.start_async("INCREMENTAL", "MANUAL") snapshot = _wait_finished(engine) assert snapshot["phase"] == "SUCCESS" assert snapshot["type"] == "FULL" assert snapshot["backupId"].endswith("-F") def test_engine_fails_when_database_export_fails(engine_env, monkeypatch): from dms.common.errors import BackupFailedError from dms.services import sql_dump def failing_export(database_url, output_directory, schema): raise BackupFailedError("数据库导出失败（测试模拟）") monkeypatch.setattr(sql_dump, "export_dump", failing_export) engine = engine_env _write_storage_file(engine, "original/a.pdf", b"AAA") engine.start_async("FULL", "MANUAL") snapshot = _wait_finished(engine) assert snapshot["phase"] == "FAILED" assert "数据库导出失败" in snapshot["error"] # 失败备份不进历史（目标端无 manifest 即不可见），暂存目录已清理 assert engine.list_history(engine.load_config()) == [] assert not (engine.storage_root / "backup-staging").exists() def test_engine_completion_callback_success_and_failure(engine_env, monkeypatch): """终态回调（BACKUP_CREATE 审计结果来源）：恰调一次、先于 running 复位。""" from dms.common.errors import BackupFailedError from dms.services import sql_dump engine = engine_env _write_storage_file(engine, "original/a.pdf", b"AAA") # 成功：outcome 携带最终类型/触发方式/备份ID，回调内任务仍视为运行中 outcomes = [] running_inside = [] engine.start_async( "FULL", "MANUAL", on_complete=lambda outcome: ( outcomes.append(outcome), running_inside.append(engine.is_running), ), ) snapshot = _wait_finished(engine) assert snapshot["phase"] == "SUCCESS" assert len(outcomes) == 1 success = outcomes[0] assert success.success is True assert success.type == "FULL" assert success.trigger == "MANUAL" assert success.backup_id == snapshot["backupId"] assert success.error is None assert running_inside == [True] # 失败：outcome 携带失败原因与备份ID（导出失败发生于备份目录建立后） def failing_export(database_url, output_directory, schema): raise BackupFailedError("数据库导出失败（测试模拟）") monkeypatch.setattr(sql_dump, "export_dump", failing_export) failed_outcomes = [] engine.start_async( "INCREMENTAL", "SCHEDULE", on_complete=failed_outcomes.append ) failed_snapshot = _wait_finished(engine) assert failed_snapshot["phase"] == "FAILED" assert len(failed_outcomes) == 1 failure = failed_outcomes[0] assert failure.success is False assert failure.type == "INCREMENTAL" assert failure.trigger == "SCHEDULE" assert failure.backup_id == failed_snapshot["backupId"] assert "数据库导出失败" in failure.error # 早期失败（目标不可用，备份目录建立前）：backup_id 为 None config_path = engine.backup_config_path from dms.services.backup_service import BackupConfig blocking = engine.storage_root / "not-a-directory" blocking.write_text("x", encoding="utf-8") BackupConfig(local_path=str(blocking)).save(config_path) early_outcomes = [] engine.start_async("FULL", "MANUAL", on_complete=early_outcomes.append) _wait_finished(engine) assert len(early_outcomes) == 1 assert early_outcomes[0].success is False assert early_outcomes[0].backup_id is None assert early_outcomes[0].error def test_engine_running_mutually_exclusive(engine_env): engine = engine_env from dms.common.errors import BackupRunningError engine._running.set() with pytest.raises(BackupRunningError): engine.start_async("FULL", "MANUAL") engine._running.clear() # 取消空闲任务为空操作，不抛异常 engine.cancel() def test_retention_cleanup_removes_oldest_chains_and_residue(engine_env): from dms.services.backup_service import LocalBackupTarget target = LocalBackupTarget(engine_env.backup_root.parent / "retention-root") ids = { "full1": "dms-bak-20260101-000000-F", "inc1": "dms-bak-20260102-000000-I", "full2": "dms-bak-20260108-000000-F", "inc2": "dms-bak-20260109-000000-I", "full3": "dms-bak-20260115-000000-F", "residue": "dms-bak-20260105-000000-F", } for backup_id in ids.values(): directory = target.root / backup_id directory.mkdir(parents=True) (directory / "manifest.json").write_text("{}", encoding="utf-8") index = [ {"id": ids["full1"], "type": "FULL"}, {"id": ids["inc1"], "type": "INCREMENTAL"}, {"id": ids["full2"], "type": "FULL"}, {"id": ids["inc2"], "type": "INCREMENTAL"}, {"id": ids["full3"], "type": "FULL"}, ] engine_env._cleanup_retention_policy(target, index, retain_full_count=2) # 最旧链（full1+inc1）与残留目录被清理，剩余两条链 assert not (target.root / ids["full1"]).exists() assert not (target.root / ids["inc1"]).exists() assert not (target.root / ids["residue"]).exists() assert [entry["id"] for entry in index] == [ ids["full2"], ids["inc2"], ids["full3"], ] # 按备份 ID 字典序（= 时间序） assert sorted(target.list_backup_directories()) == [ ids["full2"], ids["inc2"], ids["full3"], ] def test_backup_config_json_roundtrip_and_validation(tmp_path: Path): from dms.common.errors import BackupConfigInvalidError from dms.services.backup_service import BackupConfig config = BackupConfig( local_path=str(tmp_path / "backup-root"), rsync_password="secret-value", schedule_enabled=True, ) config.save(tmp_path / "dms-backup-config.json") # 磁盘 JSON 为 snake_case 全字段（含明文密码，双后端共享） disk = json.loads( (tmp_path / "dms-backup-config.json").read_text(encoding="utf-8") ) assert disk == { "target_mode": "LOCAL", "local_path": str(tmp_path / "backup-root"), "rsync_url": "", "rsync_password": "secret-value", "schedule_enabled": True, "schedule_time": "02:00", "full_interval_days": 7, "retain_full_count": 2, } loaded = BackupConfig.load(tmp_path / "dms-backup-config.json") assert loaded.rsync_password == "secret-value" assert loaded.schedule_enabled is True # 回显视图永不回明文密码 view = loaded.public_view() assert "rsync_password" not in view assert view["rsync_password_set"] is True # 损坏文件回退默认配置 (tmp_path / "broken.json").write_text("{", encoding="utf-8") assert BackupConfig.load(tmp_path / "broken.json") == BackupConfig() assert BackupConfig.load(tmp_path / "missing.json") == BackupConfig() # 校验：模式/路径/时刻/间隔/保留数 with pytest.raises(BackupConfigInvalidError): BackupConfig(target_mode="FTP", local_path="x").validate() with pytest.raises(BackupConfigInvalidError): BackupConfig().validate() # LOCAL 缺 local_path rsync = BackupConfig(target_mode="RSYNC", rsync_url="ftp://bad/x") with pytest.raises(BackupConfigInvalidError): rsync.validate() ok_rsync = BackupConfig(target_mode="RSYNC", rsync_url="rsync://u@h:873/mod/") ok_rsync.validate() with pytest.raises(BackupConfigInvalidError): BackupConfig(local_path="x", schedule_time="24:99").validate() with pytest.raises(BackupConfigInvalidError): BackupConfig(local_path="x", full_interval_days=0).validate() with pytest.raises(BackupConfigInvalidError): BackupConfig(local_path="x", retain_full_count=0).validate() # ================================================================ 接口级（B2 门控） def test_backup_endpoints_require_admin(b2_client, backup_env, token_for): missing = b2_client.get(CONFIG_URL) invalid = b2_client.get( CONFIG_URL, headers={"Authorization": "Bearer invalid-token"} ) assert missing.status_code == 401 assert missing.get_json()["code"] == "TOKEN_INVALID" assert invalid.status_code == 401 assert invalid.get_json()["code"] == "TOKEN_INVALID" user_headers = _auth(token_for("user")) checks = ( ("get", CONFIG_URL, None), ("put", CONFIG_URL, {}), ("post", JOBS_URL, {"type": "FULL"}), ("get", CURRENT_URL, None), ("post", CANCEL_URL, None), ("get", JOBS_URL, None), ) for method, url, body in checks: response = getattr(b2_client, method)( url, json=body, headers=user_headers ) assert response.status_code == 403 assert response.get_json()["code"] == "FORBIDDEN" def test_config_default_view_and_roundtrip_with_password_masking( b2_app, b2_client, backup_env, login_token ): engine = backup_env response = b2_client.get(CONFIG_URL, headers=_auth(login_token)) assert response.status_code == 200 default = response.get_json()["data"] assert default == { "target_mode": "LOCAL", "local_path": "", "rsync_url": "", "rsync_password_set": False, "schedule_enabled": False, "schedule_time": "02:00", "full_interval_days": 7, "retain_full_count": 2, } saved = _save_local_config( b2_client, login_token, b2_app.config["BACKUP_TEST_ROOT"], rsync_password="secret-value", ) assert saved["rsync_password_set"] is True assert "rsync_password" not in saved assert saved["local_path"] == str(b2_app.config["BACKUP_TEST_ROOT"]) # 空密码 = 保持不变 _save_local_config(b2_client, login_token, b2_app.config["BACKUP_TEST_ROOT"]) disk = json.loads( engine.backup_config_path.read_text(encoding="utf-8") ) assert disk["rsync_password"] == "secret-value" def test_config_validation_rejects_invalid_payload(b2_client, backup_env, login_token): headers = _auth(login_token) incomplete = b2_client.put( CONFIG_URL, json={"target_mode": "LOCAL"}, headers=headers, ) assert incomplete.status_code == 400 assert incomplete.get_json()["code"] == "INVALID_ARGUMENT" base = { "target_mode": "LOCAL", "local_path": str(backup_env.backup_config_path.parent / "backup-root"), "rsync_url": None, "rsync_password": None, "schedule_enabled": False, "schedule_time": "02:00", "full_interval_days": 7, "retain_full_count": 2, } for field, value in ( ("target_mode", "FTP"), ("schedule_time", "24:99"), ("full_interval_days", 0), ("retain_full_count", 0), ): payload = dict(base, **{field: value}) response = b2_client.put(CONFIG_URL, json=payload, headers=headers) assert response.status_code == 400 assert response.get_json()["code"] == "BACKUP_CONFIG_INVALID" def test_api_full_and_incremental_with_history_and_audit( b2_app, b2_client, backup_env, login_token ): from dms.extensions import db from dms.models import AuditLog engine = backup_env backup_root: Path = b2_app.config["BACKUP_TEST_ROOT"] _save_local_config(b2_client, login_token, backup_root) _write_storage_file(engine, "original/a.pdf", b"AAA") # 触发：202 受理 + 初始快照（12 业务字段） started = b2_client.post( JOBS_URL, json={"type": "FULL"}, headers=_auth(login_token) ) assert started.status_code == 202 assert set(started.get_json()["data"]) == PROGRESS_KEYS full_snapshot = _wait_finished(engine) full_id = full_snapshot["backupId"] assert full_snapshot["phase"] == "SUCCESS" assert full_id.endswith("-F") # 当前进度：12 业务字段 + running current = b2_client.get(CURRENT_URL, headers=_auth(login_token)) assert current.status_code == 200 current_data = current.get_json()["data"] assert set(current_data) == PROGRESS_KEYS | {"running"} assert current_data["running"] is False assert current_data["backupId"] == full_id # 历史：字段形状与链尾累计口径 history = b2_client.get(JOBS_URL, headers=_auth(login_token)) assert history.status_code == 200 entries = history.get_json()["data"]["entries"] assert [entry["id"] for entry in entries] == [full_id] assert set(entries[0]) == HISTORY_ENTRY_KEYS assert entries[0]["storage_file_count"] == 1 # 触发审计：记一条 BACKUP_CREATE（任务终态落库，成功） with b2_app.app_context(): rows = db.session.execute( db.select(AuditLog).where(AuditLog.action_type == "BACKUP_CREATE") ).scalars().all() assert len(rows) == 1 assert rows[0].target_type == "BACKUP" assert rows[0].operation_result == "SUCCESS" assert rows[0].failure_reason is None assert rows[0].target_name == "FULL" assert rows[0].operation_detail["backupId"] == full_id def test_api_failed_backup_audit_records_failure( b2_app, b2_client, backup_env, login_token, monkeypatch ): """备份任务失败：审计落 FAILURE 且携带失败原因（修复“失败显示成功”）。""" from dms.common.errors import BackupFailedError from dms.extensions import db from dms.models import AuditLog from dms.services import sql_dump def failing_export(database_url, output_directory, schema): raise BackupFailedError("数据库导出失败（测试模拟）") monkeypatch.setattr(sql_dump, "export_dump", failing_export) engine = backup_env _save_local_config(b2_client, login_token, b2_app.config["BACKUP_TEST_ROOT"]) _write_storage_file(engine, "original/a.pdf", b"AAA") started = b2_client.post( JOBS_URL, json={"type": "FULL"}, headers=_auth(login_token) ) assert started.status_code == 202 snapshot = _wait_finished(engine) assert snapshot["phase"] == "FAILED" with b2_app.app_context(): rows = db.session.execute( db.select(AuditLog).where(AuditLog.action_type == "BACKUP_CREATE") ).scalars().all() assert len(rows) == 1 row = rows[0] assert row.operation_result == "FAILURE" assert "数据库导出失败" in row.failure_reason assert row.target_name == "FULL" assert row.operation_detail["backupId"] == snapshot["backupId"] assert row.operation_detail["trigger"] == "MANUAL" def test_api_invalid_type_rejected(b2_client, backup_env, login_token): response = b2_client.post( JOBS_URL, json={"type": "PARTIAL"}, headers=_auth(login_token) ) assert response.status_code == 400 assert response.get_json()["code"] == "INVALID_ARGUMENT" def test_api_start_conflict_while_running(b2_client, backup_env, login_token): engine = backup_env engine._running.set() try: response = b2_client.post( JOBS_URL, json={"type": "FULL"}, headers=_auth(login_token) ) assert response.status_code == 409 assert response.get_json()["code"] == "BACKUP_RUNNING" finally: engine._running.clear() def test_api_cancel_without_running_job(b2_client, backup_env, login_token): response = b2_client.post(CANCEL_URL, headers=_auth(login_token)) assert response.status_code == 200 assert response.get_json()["data"] == {"canceled": False}