"""备份功能单元测试""" import json import os import re import zipfile from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest from astrbot.core.backup import ( BACKUP_MANIFEST_VERSION, KB_METADATA_MODELS, MAIN_DB_MODELS, ImportPreCheckResult, ) from astrbot.core.backup.exporter import AstrBotExporter from astrbot.core.backup.importer import ( DatabaseClearError, PLATFORM_STATS_INVALID_COUNT_WARN_LIMIT, AstrBotImporter, ImportResult, _get_major_version, ) from astrbot.core.config.default import VERSION from astrbot.core.db.po import ( ConversationV2, ) from astrbot.core.utils.version_comparator import VersionComparator from astrbot.dashboard.routes.backup import ( generate_unique_filename, secure_filename, ) @pytest.fixture def temp_backup_dir(tmp_path): """创建临时备份目录""" backup_dir = tmp_path / "backups" backup_dir.mkdir() return backup_dir @pytest.fixture def temp_data_dir(tmp_path): """创建临时数据目录""" data_dir = tmp_path / "data" data_dir.mkdir() # 创建配置文件 config_path = data_dir / "cmd_config.json" config_path.write_text(json.dumps({"test": "config"})) # 创建附件目录 attachments_dir = data_dir / "attachments" attachments_dir.mkdir() return data_dir @pytest.fixture def mock_main_db(): """创建模拟的主数据库""" db = MagicMock() # 模拟异步上下文管理器 session = AsyncMock() db.get_db = MagicMock( return_value=AsyncMock(__aenter__=AsyncMock(return_value=session)) ) return db @pytest.fixture def mock_kb_manager(): """创建模拟的知识库管理器""" kb_manager = MagicMock() kb_manager.kb_insts = {} # 模拟 kb_db kb_db = MagicMock() session = AsyncMock() kb_db.get_db = MagicMock( return_value=AsyncMock(__aenter__=AsyncMock(return_value=session)) ) kb_manager.kb_db = kb_db return kb_manager class TestImportResult: """ImportResult 类测试""" def test_init(self): """测试初始化""" result = ImportResult() assert result.success is True assert result.imported_tables == {} assert result.imported_files == {} assert result.warnings == [] assert result.errors == [] def test_add_warning(self): """测试添加警告""" result = ImportResult() result.add_warning("test warning") assert "test warning" in result.warnings assert result.success is True # 警告不影响成功状态 def test_add_error(self): """测试添加错误""" result = ImportResult() result.add_error("test error") assert "test error" in result.errors assert result.success is False # 错误会导致失败 def test_to_dict(self): """测试转换为字典""" result = ImportResult() result.imported_tables = {"test_table": 10} result.add_warning("warning") d = result.to_dict() assert d["success"] is True assert d["imported_tables"] == {"test_table": 10} assert "warning" in d["warnings"] class TestAstrBotExporter: """AstrBotExporter 类测试""" def test_init(self, mock_main_db, mock_kb_manager, temp_data_dir): """测试初始化""" exporter = AstrBotExporter( main_db=mock_main_db, kb_manager=mock_kb_manager, config_path=str(temp_data_dir / "cmd_config.json"), ) assert exporter.main_db is mock_main_db assert exporter.kb_manager is mock_kb_manager def test_model_to_dict_with_model_dump(self): """测试 _model_to_dict 使用 model_dump 方法""" exporter = AstrBotExporter(main_db=MagicMock()) # 创建一个有 model_dump 方法的模拟对象 mock_record = MagicMock() mock_record.model_dump.return_value = {"id": 1, "name": "test"} result = exporter._model_to_dict(mock_record) assert result == {"id": 1, "name": "test"} def test_model_to_dict_with_datetime(self): """测试 _model_to_dict 处理 datetime 字段""" exporter = AstrBotExporter(main_db=MagicMock()) now = datetime.now() mock_record = MagicMock() mock_record.model_dump.return_value = {"id": 1, "created_at": now} result = exporter._model_to_dict(mock_record) assert result["created_at"] == now.isoformat() def test_add_checksum(self): """测试添加校验和""" exporter = AstrBotExporter(main_db=MagicMock()) exporter._add_checksum("test.json", '{"test": "data"}') assert "test.json" in exporter._checksums assert exporter._checksums["test.json"].startswith("sha256:") def test_generate_manifest(self, mock_main_db, mock_kb_manager): """测试生成清单""" exporter = AstrBotExporter( main_db=mock_main_db, kb_manager=mock_kb_manager, ) main_data = { "platform_stats": [{"id": 1}], "conversations": [], "attachments": [], } kb_meta_data = { "knowledge_bases": [], "kb_documents": [], } dir_stats = { "plugins": {"files": 10, "size": 1024}, "plugin_data": {"files": 5, "size": 512}, } manifest = exporter._generate_manifest(main_data, kb_meta_data, dir_stats) assert manifest["version"] == BACKUP_MANIFEST_VERSION assert manifest["astrbot_version"] == VERSION assert manifest["origin"] == "exported" # 验证备份来源标记 assert "exported_at" in manifest assert "tables" in manifest assert "statistics" in manifest assert "directories" in manifest assert manifest["statistics"]["main_db"]["platform_stats"] == 1 assert manifest["statistics"]["directories"] == dir_stats @pytest.mark.asyncio async def test_export_all_creates_zip( self, mock_main_db, temp_backup_dir, temp_data_dir ): """测试导出创建 ZIP 文件""" # 设置模拟数据库返回空数据 session = AsyncMock() result = MagicMock() result.scalars.return_value.all.return_value = [] session.execute = AsyncMock(return_value=result) mock_main_db.get_db.return_value = AsyncMock( __aenter__=AsyncMock(return_value=session), __aexit__=AsyncMock(return_value=None), ) exporter = AstrBotExporter( main_db=mock_main_db, kb_manager=None, config_path=str(temp_data_dir / "cmd_config.json"), ) zip_path = await exporter.export_all(output_dir=str(temp_backup_dir)) assert os.path.exists(zip_path) assert zip_path.endswith(".zip") assert "astrbot_backup_" in zip_path # 验证 ZIP 文件内容 with zipfile.ZipFile(zip_path, "r") as zf: namelist = zf.namelist() assert "manifest.json" in namelist assert "databases/main_db.json" in namelist assert "config/cmd_config.json" in namelist class TestAstrBotImporter: """AstrBotImporter 类测试""" def test_init(self, mock_main_db, mock_kb_manager, temp_data_dir): """测试初始化""" importer = AstrBotImporter( main_db=mock_main_db, kb_manager=mock_kb_manager, config_path=str(temp_data_dir / "cmd_config.json"), ) assert importer.main_db is mock_main_db assert importer.kb_manager is mock_kb_manager def test_validate_version_match(self): """测试版本匹配验证""" importer = AstrBotImporter(main_db=MagicMock()) manifest = {"astrbot_version": VERSION} # 不应该抛出异常 importer._validate_version(manifest) def test_validate_version_major_diff_rejected(self): """测试主版本不同被拒绝""" importer = AstrBotImporter(main_db=MagicMock()) # 使用一个明显不同的主版本 manifest = {"astrbot_version": "0.0.1"} with pytest.raises(ValueError, match="主版本不兼容"): importer._validate_version(manifest) def test_validate_version_minor_diff_allowed(self): """测试小版本不同被允许（仅警告）""" importer = AstrBotImporter(main_db=MagicMock()) # 获取当前主版本 major_version = _get_major_version(VERSION) # 构造一个同主版本但小版本不同的版本 minor_diff_version = f"{major_version}.999" manifest = {"astrbot_version": minor_diff_version} # 不应该抛出异常 importer._validate_version(manifest) def test_validate_version_missing(self): """测试缺少版本信息""" importer = AstrBotImporter(main_db=MagicMock()) manifest = {} with pytest.raises(ValueError, match="缺少版本信息"): importer._validate_version(manifest) def test_convert_datetime_fields(self): """测试 datetime 字段转换""" importer = AstrBotImporter(main_db=MagicMock()) # 使用 ConversationV2 作为测试模型（它有 created_at 和 updated_at 字段） row = { "conversation_id": "test-123", "platform_id": "test", "user_id": "user1", "created_at": "2024-01-01T12:00:00", "updated_at": "2024-01-01T12:00:00", } result = importer._convert_datetime_fields(row, ConversationV2) # created_at 应该被转换为 datetime 对象 assert isinstance(result["created_at"], datetime) assert isinstance(result["updated_at"], datetime) def test_merge_platform_stats_rows(self): """测试 platform_stats 重复键会在导入前聚合""" importer = AstrBotImporter(main_db=MagicMock()) rows = [ { "id": 1, "timestamp": "2025-12-13T20:00:00Z", "platform_id": "webchat", "platform_type": "unknown", "count": 14, }, { "id": 80, "timestamp": "2025-12-13T20:00:00+00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 3, }, { "id": 81, "timestamp": "2025-12-13T20:00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 2, }, { "id": 2, "timestamp": "2025-12-13T21:00:00", "platform_id": "aiocqhttp", "platform_type": "unknown", "count": 1, }, ] merged_rows = importer._merge_platform_stats_rows(rows) duplicate_count = len(rows) - len(merged_rows) assert duplicate_count == 2 assert len(merged_rows) == 2 webchat_row = next( ( r for r in merged_rows if r.get("timestamp") == "2025-12-13T20:00:00+00:00" and r.get("platform_id") == "webchat" and r.get("platform_type") == "unknown" ), None, ) assert webchat_row is not None assert webchat_row["timestamp"] == "2025-12-13T20:00:00+00:00" assert webchat_row["platform_id"] == "webchat" assert webchat_row["platform_type"] == "unknown" assert webchat_row["count"] == 19 aiocq_row = next( ( r for r in merged_rows if r.get("platform_id") == "aiocqhttp" and r.get("platform_type") == "unknown" ), None, ) assert aiocq_row is not None assert aiocq_row["timestamp"] == "2025-12-13T21:00:00+00:00" def test_merge_platform_stats_rows_normalizes_naive_timestamp_to_utc(self): """测试 platform_stats 合并前会将 naive timestamp 标准化为 UTC 偏移""" importer = AstrBotImporter(main_db=MagicMock()) rows = [ { "timestamp": "2025-12-13T21:00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 1, }, { "timestamp": datetime(2025, 12, 13, 22, 0, 0), "platform_id": "telegram", "platform_type": "unknown", "count": 1, }, ] merged_rows = importer._merge_platform_stats_rows(rows) assert len(merged_rows) == 2 by_platform = {row["platform_id"]: row for row in merged_rows} assert by_platform["webchat"]["timestamp"] == "2025-12-13T21:00:00+00:00" assert by_platform["telegram"]["timestamp"] == "2025-12-13T22:00:00+00:00" def test_merge_platform_stats_rows_warns_on_invalid_count(self): """测试 platform_stats count 非法时会告警并按 0 处理（含上限）""" importer = AstrBotImporter(main_db=MagicMock()) with patch("astrbot.core.backup.importer.logger.warning") as warning_mock: rows = [ { "timestamp": "2025-12-13T20:00:00+00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 5, }, { "timestamp": "2025-12-13T20:00:00Z", "platform_id": "webchat", "platform_type": "unknown", "count": "bad-count", }, ] merged_rows = importer._merge_platform_stats_rows(rows) duplicate_count = len(rows) - len(merged_rows) assert duplicate_count == 1 assert len(merged_rows) == 1 assert merged_rows[0]["count"] == 5 assert warning_mock.call_count == 1 warning_mock.reset_mock() rows_existing_invalid = [ { "timestamp": "2025-12-13T21:00:00+00:00", "platform_id": "webchat", "platform_type": "unknown", "count": "bad-count", }, { "timestamp": "2025-12-13T21:00:00Z", "platform_id": "webchat", "platform_type": "unknown", "count": 7, }, ] merged_rows = importer._merge_platform_stats_rows(rows_existing_invalid) duplicate_count = len(rows_existing_invalid) - len(merged_rows) assert duplicate_count == 1 assert len(merged_rows) == 1 assert merged_rows[0]["count"] == 7 assert warning_mock.call_count == 1 warning_mock.reset_mock() many_invalid_rows = [ { "timestamp": "2025-12-13T22:00:00+00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 1, }, *[ { "timestamp": "2025-12-13T22:00:00Z", "platform_id": "webchat", "platform_type": "unknown", "count": "bad-count", } for _ in range(PLATFORM_STATS_INVALID_COUNT_WARN_LIMIT + 5) ], ] importer._merge_platform_stats_rows(many_invalid_rows) assert ( warning_mock.call_count == PLATFORM_STATS_INVALID_COUNT_WARN_LIMIT + 1 ) assert any( "告警已达到上限" in str(call.args[0]) for call in warning_mock.call_args_list ) warning_mock.reset_mock() single_invalid_row = [ { "timestamp": "2025-12-13T23:00:00+00:00", "platform_id": "telegram", "platform_type": "unknown", "count": "still-bad", }, ] merged_rows = importer._merge_platform_stats_rows(single_invalid_row) duplicate_count = len(single_invalid_row) - len(merged_rows) assert duplicate_count == 0 assert len(merged_rows) == 1 assert merged_rows[0]["count"] == 0 assert warning_mock.call_count == 1 def test_merge_platform_stats_rows_keeps_invalid_timestamps_distinct(self): """测试空/非法 timestamp 不参与聚合，避免误合并""" importer = AstrBotImporter(main_db=MagicMock()) rows = [ { "timestamp": "", "platform_id": "webchat", "platform_type": "unknown", "count": 2, }, { "timestamp": "not-a-datetime", "platform_id": "webchat", "platform_type": "unknown", "count": 3, }, { "timestamp": "not-a-datetime", "platform_id": "webchat", "platform_type": "unknown", "count": 4, }, ] merged_rows = importer._merge_platform_stats_rows(rows) duplicate_count = len(rows) - len(merged_rows) assert duplicate_count == 0 assert len(merged_rows) == 3 assert [row["count"] for row in merged_rows] == [2, 3, 4] def test_merge_platform_stats_rows_keeps_non_string_platform_keys_distinct(self): """测试非字符串 platform_id/platform_type 不参与聚合""" importer = AstrBotImporter(main_db=MagicMock()) rows = [ { "timestamp": "2025-12-13T20:00:00+00:00", "platform_id": None, "platform_type": "unknown", "count": 2, }, { "timestamp": "2025-12-13T20:00:00Z", "platform_id": None, "platform_type": "unknown", "count": 3, }, { "timestamp": "2025-12-13T20:00:00+00:00", "platform_id": "webchat", "platform_type": 1, "count": 4, }, { "timestamp": "2025-12-13T20:00:00Z", "platform_id": "webchat", "platform_type": 1, "count": 5, }, ] merged_rows = importer._merge_platform_stats_rows(rows) duplicate_count = len(rows) - len(merged_rows) assert duplicate_count == 0 assert len(merged_rows) == 4 def test_merge_platform_stats_rows_preserves_input_order(self): """测试 platform_stats 聚合后仍保持输入顺序（按首次出现位置）""" importer = AstrBotImporter(main_db=MagicMock()) rows = [ { "id": 1, "timestamp": "2025-12-13T20:00:00Z", "platform_id": "webchat", "platform_type": "unknown", "count": 2, }, { "id": 2, "timestamp": "", "platform_id": "webchat", "platform_type": "unknown", "count": 3, }, { "id": 3, "timestamp": "2025-12-13T20:00:00+00:00", "platform_id": "webchat", "platform_type": "unknown", "count": 5, }, { "id": 4, "timestamp": "2025-12-13T21:00:00+00:00", "platform_id": "telegram", "platform_type": "unknown", "count": 7, }, ] merged_rows = importer._merge_platform_stats_rows(rows) assert len(merged_rows) == 3 assert [row["id"] for row in merged_rows] == [1, 2, 4] assert merged_rows[0]["count"] == 7 @pytest.mark.asyncio async def test_import_file_not_exists(self, mock_main_db, tmp_path): """测试导入不存在的文件""" importer = AstrBotImporter(main_db=mock_main_db) result = await importer.import_all(str(tmp_path / "nonexistent.zip")) assert result.success is False assert any("不存在" in err for err in result.errors) @pytest.mark.asyncio async def test_import_invalid_zip(self, mock_main_db, tmp_path): """测试导入无效的 ZIP 文件""" # 创建一个无效的文件 invalid_zip = tmp_path / "invalid.zip" invalid_zip.write_text("not a zip file") importer = AstrBotImporter(main_db=mock_main_db) result = await importer.import_all(str(invalid_zip)) assert result.success is False assert any("无效" in err or "ZIP" in err for err in result.errors) @pytest.mark.asyncio async def test_import_missing_manifest(self, mock_main_db, tmp_path): """测试导入缺少 manifest 的 ZIP 文件""" # 创建一个没有 manifest 的 ZIP 文件 zip_path = tmp_path / "no_manifest.zip" with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("test.txt", "test content") importer = AstrBotImporter(main_db=mock_main_db) result = await importer.import_all(str(zip_path)) assert result.success is False assert any("manifest" in err.lower() for err in result.errors) @pytest.mark.asyncio async def test_import_major_version_mismatch(self, mock_main_db, tmp_path): """测试导入主版本不匹配的备份""" # 创建一个主版本不匹配的备份 zip_path = tmp_path / "old_version.zip" manifest = { "version": "1.0", "astrbot_version": "0.0.1", # 主版本不同 "tables": {"main_db": []}, } with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("manifest.json", json.dumps(manifest)) importer = AstrBotImporter(main_db=mock_main_db) result = await importer.import_all(str(zip_path)) assert result.success is False assert any("主版本不兼容" in err for err in result.errors) @pytest.mark.asyncio async def test_import_replace_fails_when_clear_main_db_fails( self, mock_main_db, tmp_path ): """测试 replace 模式下主库清空失败会直接终止导入""" zip_path = tmp_path / "valid_backup.zip" manifest = { "version": "1.1", "astrbot_version": VERSION, "tables": {"platform_stats": 0}, } main_data = {"platform_stats": []} with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("manifest.json", json.dumps(manifest)) zf.writestr("databases/main_db.json", json.dumps(main_data)) importer = AstrBotImporter(main_db=mock_main_db) importer._clear_main_db = AsyncMock( side_effect=DatabaseClearError("清空表 platform_stats 失败: db locked") ) importer._import_main_database = AsyncMock(return_value={}) result = await importer.import_all(str(zip_path), mode="replace") assert result.success is False assert any("清空主数据库失败" in err for err in result.errors) assert any("清空表 platform_stats 失败" in err for err in result.errors) importer._import_main_database.assert_not_awaited() class TestSecureFilename: """安全文件名函数测试""" def test_secure_filename_normal(self): """测试正常文件名""" assert secure_filename("backup.zip") == "backup.zip" assert secure_filename("my_backup_2024.zip") == "my_backup_2024.zip" def test_secure_filename_path_traversal(self): """测试路径遍历攻击""" assert ".." not in secure_filename("../../../etc/passwd") assert "/" not in secure_filename("/etc/passwd") assert "\\" not in secure_filename("..\\..\\windows\\system32") def test_secure_filename_with_path(self): """测试带路径的文件名""" result = secure_filename("/path/to/backup.zip") assert result == "backup.zip" result = secure_filename("C:\\Users\\test\\backup.zip") assert result == "backup.zip" def test_secure_filename_special_chars(self): """测试特殊字符""" result = secure_filename('backup<>:"|?*.zip') # 特殊字符应被替换为下划线 assert "<" not in result assert ">" not in result assert ":" not in result assert '"' not in result assert "|" not in result assert "?" not in result assert "*" not in result def test_secure_filename_hidden_file(self): """测试隐藏文件（前导点）""" result = secure_filename(".hidden_backup.zip") assert not result.startswith(".") def test_secure_filename_empty(self): """测试空文件名""" assert secure_filename("") == "backup" assert secure_filename("...") == "backup" def test_generate_unique_filename(self): """测试生成唯一文件名""" result = generate_unique_filename("backup.zip") # 应包含原文件名和时间戳后缀 assert result.startswith("backup_") assert result.endswith(".zip") # 应包含时间戳格式 YYYYMMDD_HHMMSS assert re.search(r"backup_\d{8}_\d{6}\.zip", result) def test_generate_unique_filename_with_complex_name(self): """测试复杂文件名生成唯一文件名""" result = generate_unique_filename("my_backup_file.zip") # 应在原文件名后添加时间戳 assert result.startswith("my_backup_file_") assert result.endswith(".zip") assert re.search(r"my_backup_file_\d{8}_\d{6}\.zip", result) class TestVersionComparison: """版本比较函数测试 - 使用 VersionComparator""" def test_get_major_version_simple(self): """测试提取简单主版本号""" assert _get_major_version("1.0") == "1.0" assert _get_major_version("2.1") == "2.1" assert _get_major_version("4.9.1") == "4.9" def test_get_major_version_with_prefix(self): """测试带 v 前缀的版本号""" assert _get_major_version("v1.0") == "1.0" assert _get_major_version("V4.9.1") == "4.9" def test_get_major_version_with_prerelease(self): """测试带预发布标签的版本号""" assert _get_major_version("4.9.1-beta") == "4.9" assert _get_major_version("4.9.1-alpha.1") == "4.9" assert _get_major_version("4.9.1+build123") == "4.9" def test_get_major_version_single_part(self): """测试单部分版本号""" assert _get_major_version("1") == "1.0" def test_get_major_version_empty(self): """测试空版本号""" assert _get_major_version("") == "0.0" def test_compare_versions_equal(self): """测试版本相等""" assert VersionComparator.compare_version("1.0", "1.0") == 0 assert VersionComparator.compare_version("1.0.0", "1.0") == 0 assert VersionComparator.compare_version("2.10", "2.10") == 0 def test_compare_versions_less_than(self): """测试版本小于""" assert VersionComparator.compare_version("1.0", "1.1") == -1 assert ( VersionComparator.compare_version("1.9", "1.10") == -1 ) # 关键测试：多位数版本比较 assert VersionComparator.compare_version("1.2", "1.10") == -1 assert VersionComparator.compare_version("1.0", "2.0") == -1 def test_compare_versions_greater_than(self): """测试版本大于""" assert VersionComparator.compare_version("1.1", "1.0") == 1 assert ( VersionComparator.compare_version("1.10", "1.9") == 1 ) # 关键测试：多位数版本比较 assert VersionComparator.compare_version("1.10", "1.2") == 1 assert VersionComparator.compare_version("2.0", "1.0") == 1 def test_compare_versions_different_lengths(self): """测试不同长度版本比较""" assert VersionComparator.compare_version("1.0", "1.0.0") == 0 assert VersionComparator.compare_version("1.0", "1.0.1") == -1 assert VersionComparator.compare_version("1.0.1", "1.0") == 1 def test_compare_versions_prerelease(self): """测试预发布版本比较""" # 预发布版本低于正式版本 assert VersionComparator.compare_version("1.0.0-alpha", "1.0.0") == -1 assert VersionComparator.compare_version("1.0.0", "1.0.0-beta") == 1 # alpha < beta assert VersionComparator.compare_version("1.0.0-alpha", "1.0.0-beta") == -1 class TestImportPreCheckResult: """ImportPreCheckResult 类测试""" def test_init_default_values(self): """测试默认值初始化""" result = ImportPreCheckResult() assert result.valid is False assert result.can_import is False assert result.version_status == "" assert result.backup_version == "" assert result.current_version == VERSION assert result.confirm_message == "" assert result.warnings == [] assert result.error == "" assert result.backup_summary == {} def test_to_dict(self): """测试转换为字典""" result = ImportPreCheckResult( valid=True, can_import=True, version_status="match", backup_version="4.9.0", confirm_message="确认导入？", warnings=["警告1"], backup_summary={"tables": ["table1"]}, ) d = result.to_dict() assert d["valid"] is True assert d["can_import"] is True assert d["version_status"] == "match" assert d["backup_version"] == "4.9.0" assert d["confirm_message"] == "确认导入？" assert "警告1" in d["warnings"] assert d["backup_summary"]["tables"] == ["table1"] class TestPreCheck: """预检查功能测试""" def test_pre_check_file_not_exists(self, mock_main_db): """测试预检查不存在的文件""" importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check("/nonexistent/file.zip") assert result.valid is False assert "不存在" in result.error def test_pre_check_invalid_zip(self, mock_main_db, tmp_path): """测试预检查无效的 ZIP 文件""" invalid_zip = tmp_path / "invalid.zip" invalid_zip.write_text("not a zip file") importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check(str(invalid_zip)) assert result.valid is False assert "ZIP" in result.error or "无效" in result.error def test_pre_check_missing_manifest(self, mock_main_db, tmp_path): """测试预检查缺少 manifest 的 ZIP 文件""" zip_path = tmp_path / "no_manifest.zip" with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("test.txt", "test content") importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check(str(zip_path)) assert result.valid is False assert "manifest" in result.error.lower() def test_pre_check_version_match(self, mock_main_db, tmp_path): """测试预检查版本匹配""" zip_path = tmp_path / "backup.zip" manifest = { "version": "1.1", "astrbot_version": VERSION, "created_at": "2024-01-01T12:00:00", "tables": {"platform_stats": 1}, "has_knowledge_bases": True, "has_config": True, "directories": ["plugins"], } with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("manifest.json", json.dumps(manifest)) importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check(str(zip_path)) assert result.valid is True assert result.can_import is True assert result.version_status == "match" assert result.backup_version == VERSION # confirm_message 现在由前端生成，后端不再生成 assert result.backup_summary["has_knowledge_bases"] is True def test_pre_check_minor_version_diff(self, mock_main_db, tmp_path): """测试预检查小版本差异""" # 构造一个同主版本但小版本不同的版本 major_version = _get_major_version(VERSION) minor_diff_version = f"{major_version}.999" zip_path = tmp_path / "backup.zip" manifest = { "version": "1.1", "astrbot_version": minor_diff_version, "created_at": "2024-01-01T12:00:00", "tables": {}, } with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("manifest.json", json.dumps(manifest)) importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check(str(zip_path)) assert result.valid is True assert result.can_import is True assert result.version_status == "minor_diff" # 版本消息由前端 i18n 生成，后端 warnings 列表不再包含版本相关消息 # warnings 列表保留用于其他非版本相关的警告 def test_pre_check_major_version_diff(self, mock_main_db, tmp_path): """测试预检查主版本差异""" zip_path = tmp_path / "backup.zip" manifest = { "version": "1.1", "astrbot_version": "0.0.1", # 主版本不同 "created_at": "2024-01-01T12:00:00", "tables": {}, } with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("manifest.json", json.dumps(manifest)) importer = AstrBotImporter(main_db=mock_main_db) result = importer.pre_check(str(zip_path)) assert result.valid is True # 文件有效 assert result.can_import is False # 但不能导入 assert result.version_status == "major_diff" # 版本消息由前端 i18n 生成，后端 warnings 列表不再包含版本相关消息 class TestVersionCompatibility: """版本兼容性检查测试""" def test_check_version_compatibility_match(self, mock_main_db): """测试版本完全匹配""" importer = AstrBotImporter(main_db=mock_main_db) result = importer._check_version_compatibility(VERSION) assert result["status"] == "match" assert result["can_import"] is True def test_check_version_compatibility_minor_diff(self, mock_main_db): """测试小版本差异""" major_version = _get_major_version(VERSION) minor_diff_version = f"{major_version}.999" importer = AstrBotImporter(main_db=mock_main_db) result = importer._check_version_compatibility(minor_diff_version) assert result["status"] == "minor_diff" assert result["can_import"] is True def test_check_version_compatibility_major_diff(self, mock_main_db): """测试主版本差异""" importer = AstrBotImporter(main_db=mock_main_db) result = importer._check_version_compatibility("0.0.1") assert result["status"] == "major_diff" assert result["can_import"] is False def test_check_version_compatibility_empty_version(self, mock_main_db): """测试空版本号""" importer = AstrBotImporter(main_db=mock_main_db) result = importer._check_version_compatibility("") assert result["status"] == "major_diff" assert result["can_import"] is False class TestModelMappings: """测试模型映射配置""" def test_main_db_models_not_empty(self): """测试主数据库模型映射非空""" assert len(MAIN_DB_MODELS) > 0 def test_main_db_models_contain_expected_tables(self): """测试主数据库模型映射包含预期的表""" expected_tables = [ "platform_stats", "conversations", "personas", "preferences", "attachments", ] for table in expected_tables: assert table in MAIN_DB_MODELS, f"Missing table: {table}" def test_kb_metadata_models_not_empty(self): """测试知识库元数据模型映射非空""" assert len(KB_METADATA_MODELS) > 0 def test_kb_metadata_models_contain_expected_tables(self): """测试知识库元数据模型映射包含预期的表""" expected_tables = [ "knowledge_bases", "kb_documents", "kb_media", ] for table in expected_tables: assert table in KB_METADATA_MODELS, f"Missing table: {table}" class TestBackupIntegration: """备份集成测试""" @pytest.mark.asyncio async def test_export_import_roundtrip(self, tmp_path): """测试导出-导入往返""" backup_dir = tmp_path / "backups" backup_dir.mkdir() data_dir = tmp_path / "data" data_dir.mkdir() config_path = data_dir / "cmd_config.json" config_path.write_text(json.dumps({"setting": "value"})) attachments_dir = data_dir / "attachments" attachments_dir.mkdir() # 创建模拟数据库 mock_db = MagicMock() session = AsyncMock() result = MagicMock() result.scalars.return_value.all.return_value = [] session.execute = AsyncMock(return_value=result) mock_db.get_db.return_value = AsyncMock( __aenter__=AsyncMock(return_value=session), __aexit__=AsyncMock(return_value=None), ) # 导出 exporter = AstrBotExporter( main_db=mock_db, kb_manager=None, config_path=str(config_path), ) zip_path = await exporter.export_all(output_dir=str(backup_dir)) assert os.path.exists(zip_path) # 验证 ZIP 内容 with zipfile.ZipFile(zip_path, "r") as zf: # 读取 manifest manifest = json.loads(zf.read("manifest.json")) assert manifest["astrbot_version"] == VERSION assert manifest["origin"] == "exported" # 验证备份来源标记 # 读取配置 config = json.loads(zf.read("config/cmd_config.json")) assert config["setting"] == "value" # 读取主数据库 main_db = json.loads(zf.read("databases/main_db.json")) assert "platform_stats" in main_db