Phase 5.15: Automated backup system with pg_dump, file backup, retention, restore, notifications
This commit is contained in:
Agent Zero
@@ -0,0 +1,240 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user