Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions astrbot/core/backup/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
CommandConfig,
CommandConflict,
ConversationV2,
CronJob,
Persona,
PersonaFolder,
PlatformMessageHistory,
Expand Down Expand Up @@ -43,6 +44,7 @@
MAIN_DB_MODELS: dict[str, type[SQLModel]] = {
"platform_stats": PlatformStat,
"conversations": ConversationV2,
"cron_jobs": CronJob,
"personas": Persona,
"persona_folders": PersonaFolder,
"preferences": Preference,
Expand Down
31 changes: 31 additions & 0 deletions tests/test_backup.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from astrbot.core.db.po import (
ConversationV2,
)
from astrbot.core.db.sqlite import SQLiteDatabase
from astrbot.core.utils.upload import UploadTooLargeError
from astrbot.core.utils.version_comparator import VersionComparator
from astrbot.dashboard.services.backup_service import (
Expand Down Expand Up @@ -1048,6 +1049,36 @@ def test_kb_metadata_models_contain_expected_tables(self):
class TestBackupIntegration:
"""备份集成测试"""

@pytest.mark.asyncio
async def test_cron_jobs_roundtrip(self, tmp_path):
"""Preserve a scheduled task through backup export and import."""
source = SQLiteDatabase(str(tmp_path / "source.db"))
target = SQLiteDatabase(str(tmp_path / "target.db"))
try:
job = await source.create_cron_job(
name="Daily reminder",
job_type="active_agent",
cron_expression="0 9 * * *",
timezone="Asia/Taipei",
payload={"message": "Reminder"},
)
exporter = AstrBotExporter(
main_db=source, config_path=str(tmp_path / "missing-config.json")
)
with patch(
"astrbot.core.backup.exporter.get_backup_directories", return_value={}
):
zip_path = await exporter.export_all(output_dir=str(tmp_path / "backups"))

result = await AstrBotImporter(main_db=target).import_all(zip_path)
assert result.success, result.errors
restored = await target.list_cron_jobs()
assert len(restored) == 1
assert restored[0].model_dump() == job.model_dump()
finally:
await source.engine.dispose()
await target.engine.dispose()

@pytest.mark.asyncio
async def test_export_import_roundtrip(self, tmp_path):
"""测试导出-导入往返"""
Expand Down
Loading