Agent Zero
241 lines
8.0 KiB
Python
241 lines
8.0 KiB
Python
"""Tests for backup and restore scripts (scripts/backup.py, scripts/restore.py)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
# Add scripts dir to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
|
|
|
|
|
|
def test_get_db_connection_params_from_asyncpg_url():
|
|
"""get_db_connection_params should convert asyncpg URL to standard postgresql URL."""
|
|
from backup import get_db_connection_params
|
|
|
|
with patch.dict(os.environ, {
|
|
"DATABASE_URL": "postgresql+asyncpg://user:pass@localhost:5432/leocrm"
|
|
}):
|
|
params = get_db_connection_params()
|
|
assert params["host"] == "localhost"
|
|
assert params["port"] == "5432"
|
|
assert params["database"] == "leocrm"
|
|
assert params["username"] == "user"
|
|
assert params["password"] == "pass"
|
|
|
|
|
|
def test_get_db_connection_params_missing_env():
|
|
"""get_db_connection_params should raise ValueError when DATABASE_URL is missing."""
|
|
from backup import get_db_connection_params
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(ValueError, match="DATABASE_URL"):
|
|
get_db_connection_params()
|
|
|
|
|
|
def test_create_backup_manifest_structure():
|
|
"""create_backup_manifest should return a valid manifest dict."""
|
|
from backup import create_backup_manifest
|
|
|
|
manifest = create_backup_manifest(
|
|
Path("/tmp/test_backup"),
|
|
db_success=True,
|
|
files_success=True,
|
|
db_size=1024,
|
|
files_count=5,
|
|
files_size=2048,
|
|
)
|
|
assert manifest["backup_id"] == "test_backup"
|
|
assert manifest["database"]["success"] is True
|
|
assert manifest["database"]["size_bytes"] == 1024
|
|
assert manifest["files"]["count"] == 5
|
|
assert manifest["version"] == "1.0.0"
|
|
|
|
|
|
def test_backup_files_nonexistent_path():
|
|
"""backup_files should return success when storage path doesn't exist."""
|
|
from backup import backup_files
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
success, count, size, error = backup_files(Path(tmpdir), "/nonexistent/path")
|
|
assert success is True
|
|
assert count == 0
|
|
assert size == 0
|
|
|
|
|
|
def test_backup_files_copies_files():
|
|
"""backup_files should copy files from storage to backup directory."""
|
|
from backup import backup_files
|
|
|
|
with tempfile.TemporaryDirectory() as storage_dir:
|
|
# Create test files
|
|
(Path(storage_dir) / "file1.txt").write_text("hello")
|
|
(Path(storage_dir) / "subdir").mkdir()
|
|
(Path(storage_dir) / "subdir" / "file2.txt").write_text("world")
|
|
|
|
with tempfile.TemporaryDirectory() as backup_dir:
|
|
success, count, size, error = backup_files(Path(backup_dir), storage_dir)
|
|
|
|
assert success is True
|
|
assert count == 2
|
|
assert size == 10 # "hello" (5) + "world" (5)
|
|
|
|
|
|
def test_apply_retention_policy_deletes_old_backups():
|
|
"""apply_retention_policy should delete backups older than retention_days."""
|
|
import time
|
|
from backup import apply_retention_policy
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
base = Path(tmpdir)
|
|
# Create old backup (modify mtime to past)
|
|
old_backup = base / "leocrm_backup_20200101_000000"
|
|
old_backup.mkdir()
|
|
(old_backup / "database.sql").write_text("old")
|
|
old_time = time.time() - (30 * 86400) # 30 days ago
|
|
os.utime(old_backup, (old_time, old_time))
|
|
|
|
# Create recent backup
|
|
recent_backup = base / "leocrm_backup_20260723_220000"
|
|
recent_backup.mkdir()
|
|
(recent_backup / "database.sql").write_text("recent")
|
|
|
|
deleted = apply_retention_policy(base, retention_days=7)
|
|
|
|
assert len(deleted) == 1
|
|
assert "leocrm_backup_20200101_000000" in deleted[0]
|
|
assert recent_backup.exists()
|
|
|
|
|
|
def test_apply_retention_policy_zero_days_keeps_all():
|
|
"""apply_retention_policy with 0 days should keep all backups."""
|
|
from backup import apply_retention_policy
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
base = Path(tmpdir)
|
|
(base / "leocrm_backup_20200101_000000").mkdir()
|
|
deleted = apply_retention_policy(base, retention_days=0)
|
|
assert len(deleted) == 0
|
|
|
|
|
|
def test_run_backup_dry_run():
|
|
"""run_backup in dry-run mode should return success without executing."""
|
|
from backup import run_backup
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
report = run_backup(
|
|
destination="local",
|
|
retention_days=7,
|
|
storage_path="/tmp",
|
|
backup_base_dir=tmpdir,
|
|
dry_run=True,
|
|
)
|
|
assert report["dry_run"] is True
|
|
assert report["overall_success"] is True
|
|
assert "DRY-RUN" in report["database"]["error"]
|
|
|
|
|
|
def test_restore_files_copies_files():
|
|
"""restore_files should copy files from backup to target path."""
|
|
from restore import restore_files
|
|
|
|
with tempfile.TemporaryDirectory() as backup_dir:
|
|
# Create backup files
|
|
files_dir = Path(backup_dir) / "files"
|
|
files_dir.mkdir()
|
|
(files_dir / "file1.txt").write_text("hello")
|
|
(files_dir / "subdir").mkdir()
|
|
(files_dir / "subdir" / "file2.txt").write_text("world")
|
|
|
|
with tempfile.TemporaryDirectory() as target_dir:
|
|
success, count, size, error = restore_files(Path(backup_dir), target_dir)
|
|
|
|
assert success is True
|
|
assert count == 2
|
|
assert size == 10
|
|
|
|
|
|
def test_restore_files_no_files_dir():
|
|
"""restore_files should return success when no files/ directory exists."""
|
|
from restore import restore_files
|
|
|
|
with tempfile.TemporaryDirectory() as backup_dir:
|
|
with tempfile.TemporaryDirectory() as target_dir:
|
|
success, count, size, error = restore_files(Path(backup_dir), target_dir)
|
|
|
|
assert success is True
|
|
assert count == 0
|
|
|
|
|
|
def test_validate_backup_dir_with_manifest():
|
|
"""validate_backup_dir should return manifest dict when manifest.json exists."""
|
|
from restore import validate_backup_dir
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
manifest = {"backup_id": "test", "version": "1.0.0"}
|
|
(Path(tmpdir) / "manifest.json").write_text(json.dumps(manifest))
|
|
result = validate_backup_dir(Path(tmpdir))
|
|
assert result is not None
|
|
assert result["backup_id"] == "test"
|
|
|
|
|
|
def test_validate_backup_dir_without_manifest():
|
|
"""validate_backup_dir should return None when manifest.json doesn't exist."""
|
|
from restore import validate_backup_dir
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
result = validate_backup_dir(Path(tmpdir))
|
|
assert result is None
|
|
|
|
|
|
def test_run_restore_nonexistent_dir():
|
|
"""run_restore should report error when backup directory doesn't exist."""
|
|
from restore import run_restore
|
|
|
|
report = run_restore(
|
|
backup_dir_path="/nonexistent/backup/dir",
|
|
target_storage_path="/tmp",
|
|
dry_run=False,
|
|
)
|
|
assert report["overall_success"] is False
|
|
assert any("not found" in e for e in report["errors"])
|
|
|
|
|
|
def test_run_restore_dry_run():
|
|
"""run_restore in dry-run mode should return success without executing."""
|
|
from restore import run_restore
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
report = run_restore(
|
|
backup_dir_path=tmpdir,
|
|
target_storage_path="/tmp",
|
|
dry_run=True,
|
|
)
|
|
assert report["dry_run"] is True
|
|
assert report["overall_success"] is True
|
|
|
|
|
|
def test_system_notif_has_backup_events():
|
|
"""system_notif plugin should register backup.completed and backup.failed events."""
|
|
from app.plugins.builtins.system_notif.plugin import SystemNotifPlugin
|
|
|
|
events = SystemNotifPlugin.manifest.events
|
|
assert "backup.completed" in events
|
|
assert "backup.failed" in events
|
|
|
|
|
|
def test_system_settings_has_backup_config():
|
|
"""SystemSettingsUpsert should have backup configuration fields."""
|
|
from app.schemas.system_settings import SystemSettingsUpsert
|
|
|
|
fields = SystemSettingsUpsert.model_fields
|
|
assert "backup_interval" in fields
|
|
assert "backup_retention_days" in fields
|
|
assert "backup_destination" in fields
|