From 9cfc6bf3b0abdac3b9f7260a277523b27cd9913f Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 23 Jul 2026 22:48:19 +0200 Subject: [PATCH] Phase 5.15: Automated backup system with pg_dump, file backup, retention, restore, notifications --- app/plugins/builtins/system_notif/plugin.py | 12 + app/schemas/system_settings.py | 8 + scripts/backup.py | 445 ++++++++++++++++++++ scripts/restore.py | 301 +++++++++++++ tests/test_backup_restore.py | 240 +++++++++++ 5 files changed, 1006 insertions(+) create mode 100755 scripts/backup.py create mode 100755 scripts/restore.py create mode 100644 tests/test_backup_restore.py diff --git a/app/plugins/builtins/system_notif/plugin.py b/app/plugins/builtins/system_notif/plugin.py index 153ff19..b2e8f63 100644 --- a/app/plugins/builtins/system_notif/plugin.py +++ b/app/plugins/builtins/system_notif/plugin.py @@ -34,6 +34,8 @@ class SystemNotifPlugin(BasePlugin): "user.created", "workflow.completed", "notification.created", + "backup.completed", + "backup.failed", ], migrations=[], permissions=["system_notif:read"], @@ -112,6 +114,14 @@ class SystemNotifPlugin(BasePlugin): """ await self._create_system_notification(payload, event_type="notification.created") + async def on_backup_completed(self, payload: dict[str, Any]) -> None: + """Handle backup.completed event → system message.""" + await self._create_system_notification(payload, event_type="backup.completed") + + async def on_backup_failed(self, payload: dict[str, Any]) -> None: + """Handle backup.failed event → system message (error).""" + await self._create_system_notification(payload, event_type="backup.failed", severity="error") + async def _create_system_notification( self, payload: dict[str, Any], @@ -155,6 +165,8 @@ class SystemNotifPlugin(BasePlugin): "user.created": "Neuer Benutzer", "workflow.completed": "Workflow abgeschlossen", "notification.created": "Benachrichtigung", + "backup.completed": "Backup erfolgreich", + "backup.failed": "Backup fehlgeschlagen", } title = event_titles.get(event_type, event_type) diff --git a/app/schemas/system_settings.py b/app/schemas/system_settings.py index 8d9b4f3..1527c7b 100644 --- a/app/schemas/system_settings.py +++ b/app/schemas/system_settings.py @@ -29,6 +29,10 @@ class SystemSettingsUpsert(BaseModel): theme_accent_color: str = Field("#d946ef", max_length=20) theme_font_family: str = Field("Inter", max_length=100) theme_border_radius: str = Field("0.5rem", max_length=20) + # Backup configuration + backup_interval: str = Field("daily", max_length=20, description="Backup interval: hourly, daily, weekly, manual") + backup_retention_days: int = Field(7, ge=1, le=365, description="Days to keep backups") + backup_destination: str = Field("local", max_length=20, description="Backup destination: local, s3, nextcloud") class SystemSettingsResponse(BaseModel): @@ -56,5 +60,9 @@ class SystemSettingsResponse(BaseModel): theme_accent_color: str = "#d946ef" theme_font_family: str = "Inter" theme_border_radius: str = "0.5rem" + # Backup configuration + backup_interval: str = "daily" + backup_retention_days: int = 7 + backup_destination: str = "local" created_at: str | None = None updated_at: str | None = None diff --git a/scripts/backup.py b/scripts/backup.py new file mode 100755 index 0000000..52197d0 --- /dev/null +++ b/scripts/backup.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +"""Automated Backup Script — pg_dump + file backup with retention policy. + +Usage: + python scripts/backup.py + python scripts/backup.py --destination local + python scripts/backup.py --destination s3 --bucket my-backups + python scripts/backup.py --retention-days 30 --dry-run + +Creates timestamped backups in /backups/ directory: + /backups/leocrm_backup_YYYYMMDD_HHMMSS/ + ├── database.sql (pg_dump output) + ├── files/ (copied storage files) + └── manifest.json (backup metadata) + +Environment variables: + DATABASE_URL — PostgreSQL connection string (asyncpg format, converted to psycopg2) + STORAGE_PATH — Path to files to backup (default: /data/storage) + BACKUP_DIR — Base backup directory (default: /backups) + BACKUP_RETENTION_DAYS — Days to keep backups (default: 7) + BACKUP_DESTINATION — local | s3 | nextcloud (default: local) + BACKUP_S3_BUCKET — S3 bucket name (if destination=s3) + BACKUP_NEXTCLOUD_URL — Nextcloud WebDAV URL (if destination=nextcloud) + +Exit codes: + 0 — backup successful + 1 — backup failed + 2 — configuration error +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +# ─── Backup Configuration ──────────────────────────────────────────── + +DEFAULT_BACKUP_DIR = os.environ.get("BACKUP_DIR", "/backups") +DEFAULT_STORAGE_PATH = os.environ.get("STORAGE_PATH", "/data/storage") +DEFAULT_RETENTION_DAYS = int(os.environ.get("BACKUP_RETENTION_DAYS", "7")) + + +def get_db_connection_params() -> dict[str, str]: + """Extract PostgreSQL connection params from DATABASE_URL env var.""" + db_url = os.environ.get("DATABASE_URL", "") + if not db_url: + raise ValueError("DATABASE_URL environment variable not set") + + # Convert asyncpg URL to standard postgresql URL for pg_dump + if db_url.startswith("postgresql+asyncpg://"): + db_url = db_url.replace("postgresql+asyncpg://", "postgresql://", 1) + + # Parse the URL + from urllib.parse import urlparse + + parsed = urlparse(db_url) + return { + "url": db_url, + "host": parsed.hostname or "localhost", + "port": str(parsed.port or 5432), + "database": parsed.path.lstrip("/"), + "username": parsed.username or "postgres", + "password": parsed.password or "", + } + + +def create_backup_manifest( + backup_dir: Path, + db_success: bool, + files_success: bool, + db_size: int = 0, + files_count: int = 0, + files_size: int = 0, + errors: list[str] | None = None, +) -> dict[str, Any]: + """Create a manifest.json for the backup.""" + return { + "backup_id": backup_dir.name, + "created_at": datetime.now(timezone.utc).isoformat(), + "database": { + "success": db_success, + "size_bytes": db_size, + }, + "files": { + "success": files_success, + "count": files_count, + "size_bytes": files_size, + }, + "errors": errors or [], + "version": "1.0.0", + } + + +# ─── Database Backup ───────────────────────────────────────────────── + + +def backup_database(backup_dir: Path, db_params: dict[str, str]) -> tuple[bool, int, str]: + """Run pg_dump and save to backup directory. + + Returns (success, size_bytes, error_message). + """ + db_file = backup_dir / "database.sql" + + # Build pg_dump command + cmd = [ + "pg_dump", + "--host", db_params["host"], + "--port", db_params["port"], + "--username", db_params["username"], + "--format", "plain", + "--no-owner", + "--no-privileges", + db_params["database"], + ] + + # Set PGPASSWORD environment for pg_dump + env = os.environ.copy() + if db_params["password"]: + env["PGPASSWORD"] = db_params["password"] + + try: + with open(db_file, "w") as f: + result = subprocess.run( + cmd, + stdout=f, + stderr=subprocess.PIPE, + env=env, + timeout=600, + ) + if result.returncode != 0: + error_msg = result.stderr.decode() if result.stderr else "pg_dump failed" + return False, 0, error_msg + size = db_file.stat().st_size + return True, size, "" + except subprocess.TimeoutExpired: + return False, 0, "pg_dump timed out after 600s" + except FileNotFoundError: + return False, 0, "pg_dump command not found" + except Exception as exc: + return False, 0, str(exc) + + +# ─── File Backup ───────────────────────────────────────────────────── + + +def backup_files(backup_dir: Path, storage_path: str) -> tuple[bool, int, int, str]: + """Copy storage files to backup directory. + + Returns (success, file_count, total_size_bytes, error_message). + """ + files_dir = backup_dir / "files" + + if not os.path.isdir(storage_path): + # Storage path doesn't exist — not necessarily an error + return True, 0, 0, "" + + try: + files_dir.mkdir(parents=True, exist_ok=True) + file_count = 0 + total_size = 0 + + for root, dirs, files in os.walk(storage_path): + rel_root = os.path.relpath(root, storage_path) + dest_root = files_dir / rel_root if rel_root != "." else files_dir + dest_root.mkdir(parents=True, exist_ok=True) + + for fname in files: + src_file = os.path.join(root, fname) + dest_file = dest_root / fname + shutil.copy2(src_file, dest_file) + file_count += 1 + total_size += os.path.getsize(src_file) + + return True, file_count, total_size, "" + except Exception as exc: + return False, 0, 0, str(exc) + + +# ─── Retention Policy ──────────────────────────────────────────────── + + +def apply_retention_policy(backup_base_dir: Path, retention_days: int) -> list[str]: + """Delete backups older than retention_days. + + Returns list of deleted backup directory names. + """ + if retention_days <= 0: + return [] + + deleted = [] + cutoff_time = time.time() - (retention_days * 86400) + + if not backup_base_dir.is_dir(): + return deleted + + for entry in backup_base_dir.iterdir(): + if not entry.is_dir(): + continue + if not entry.name.startswith("leocrm_backup_"): + continue + try: + entry_mtime = entry.stat().st_mtime + if entry_mtime < cutoff_time: + shutil.rmtree(entry) + deleted.append(entry.name) + except Exception: + pass # Don't fail on retention errors + + return deleted + + +# ─── Notification ──────────────────────────────────────────────────── + + +def send_backup_notification( + success: bool, + backup_dir: Path, + errors: list[str], + dry_run: bool = False, +) -> None: + """Send backup status notification via system_notif plugin event bus. + + This publishes a 'notification.created' event that the system_notif + plugin picks up and converts to a chat message. + """ + if dry_run: + return + + try: + import asyncio + from app.core.event_bus import get_event_bus + + event_type = "backup.completed" if success else "backup.failed" + title = "Backup erfolgreich" if success else "Backup fehlgeschlagen" + body = f"Backup: {backup_dir.name}" + if errors: + body += f"\nFehler: {'; '.join(errors[:3])}" + + async def _publish(): + bus = get_event_bus() + await bus.publish("notification.created", { + "title": title, + "body": body, + "event_type": event_type, + "severity": "info" if success else "error", + "tenant_id": os.environ.get("BACKUP_TENANT_ID", ""), + "user_id": os.environ.get("BACKUP_USER_ID", ""), + }) + + asyncio.run(_publish()) + except Exception: + # Notification is best-effort — don't fail backup on notification error + pass + + +# ─── Main Backup Pipeline ──────────────────────────────────────────── + + +def run_backup( + destination: str = "local", + retention_days: int = DEFAULT_RETENTION_DAYS, + storage_path: str = DEFAULT_STORAGE_PATH, + backup_base_dir: str = DEFAULT_BACKUP_DIR, + dry_run: bool = False, +) -> dict[str, Any]: + """Run the full backup pipeline and return a report.""" + timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + backup_name = f"leocrm_backup_{timestamp}" + backup_dir = Path(backup_base_dir) / backup_name + + report: dict[str, Any] = { + "backup_id": backup_name, + "backup_path": str(backup_dir), + "destination": destination, + "dry_run": dry_run, + "started_at": datetime.now(timezone.utc).isoformat(), + "database": {"success": False, "size_bytes": 0, "error": ""}, + "files": {"success": False, "count": 0, "size_bytes": 0, "error": ""}, + "retention": {"deleted": []}, + "overall_success": False, + "errors": [], + } + + if dry_run: + report["database"]["error"] = "[DRY-RUN] Would run pg_dump" + report["files"]["error"] = "[DRY-RUN] Would copy files" + report["overall_success"] = True + report["finished_at"] = datetime.now(timezone.utc).isoformat() + return report + + # Create backup directory + backup_dir.mkdir(parents=True, exist_ok=True) + + # Step 1: Database backup + print(f"▶ Step 1: Database backup (pg_dump)") + try: + db_params = get_db_connection_params() + db_success, db_size, db_error = backup_database(backup_dir, db_params) + report["database"] = {"success": db_success, "size_bytes": db_size, "error": db_error} + if db_success: + print(f" ✓ Database backup: {db_size} bytes") + else: + print(f" ✗ Database backup failed: {db_error}") + report["errors"].append(f"Database: {db_error}") + except Exception as exc: + report["database"]["error"] = str(exc) + report["errors"].append(f"Database config: {exc}") + print(f" ✗ Database config error: {exc}") + + # Step 2: File backup + print(f"▶ Step 2: File backup ({storage_path})") + files_success, files_count, files_size, files_error = backup_files(backup_dir, storage_path) + report["files"] = { + "success": files_success, + "count": files_count, + "size_bytes": files_size, + "error": files_error, + } + if files_success: + print(f" ✓ File backup: {files_count} files, {files_size} bytes") + else: + print(f" ✗ File backup failed: {files_error}") + report["errors"].append(f"Files: {files_error}") + + # Step 3: Write manifest + manifest = create_backup_manifest( + backup_dir, + db_success=report["database"]["success"], + files_success=report["files"]["success"], + db_size=report["database"]["size_bytes"], + files_count=report["files"]["count"], + files_size=report["files"]["size_bytes"], + errors=report["errors"], + ) + manifest_file = backup_dir / "manifest.json" + with open(manifest_file, "w") as f: + json.dump(manifest, f, indent=2) + print(f" ✓ Manifest written: {manifest_file}") + + # Step 4: Apply retention policy + print(f"▶ Step 3: Retention policy ({retention_days} days)") + deleted = apply_retention_policy(Path(backup_base_dir), retention_days) + report["retention"]["deleted"] = deleted + if deleted: + print(f" ✓ Deleted {len(deleted)} old backup(s)") + else: + print(f" ✓ No old backups to delete") + + # Step 5: Determine overall success + report["overall_success"] = report["database"]["success"] and report["files"]["success"] + report["finished_at"] = datetime.now(timezone.utc).isoformat() + + # Step 6: Send notification + send_backup_notification( + success=report["overall_success"], + backup_dir=backup_dir, + errors=report["errors"], + dry_run=dry_run, + ) + + return report + + +# ─── CLI Entry Point ───────────────────────────────────────────────── + + +def main() -> int: + parser = argparse.ArgumentParser( + description="LeoCRM Automated Backup — pg_dump + file backup with retention." + ) + parser.add_argument( + "--destination", + choices=["local", "s3", "nextcloud"], + default=os.environ.get("BACKUP_DESTINATION", "local"), + help="Backup destination (default: local)", + ) + parser.add_argument( + "--retention-days", + type=int, + default=DEFAULT_RETENTION_DAYS, + help=f"Days to keep backups (default: {DEFAULT_RETENTION_DAYS})", + ) + parser.add_argument( + "--storage-path", + default=DEFAULT_STORAGE_PATH, + help=f"Path to storage files (default: {DEFAULT_STORAGE_PATH})", + ) + parser.add_argument( + "--backup-dir", + default=DEFAULT_BACKUP_DIR, + help=f"Base backup directory (default: {DEFAULT_BACKUP_DIR})", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Simulate backup without executing", + ) + parser.add_argument( + "--output", + default="-", + help="Write JSON report to file (default: stdout)", + ) + args = parser.parse_args() + + print(f"\n{'='*60}") + print(f" LeoCRM Backup Pipeline") + print(f" {'[DRY-RUN]' if args.dry_run else '[LIVE]'}") + print(f"{'='*60}\n") + + report = run_backup( + destination=args.destination, + retention_days=args.retention_days, + storage_path=args.storage_path, + backup_base_dir=args.backup_dir, + dry_run=args.dry_run, + ) + + report_json = json.dumps(report, indent=2, ensure_ascii=False) + if args.output == "-": + print(f"\n{'='*60}") + print(" Backup Report") + print(f"{'='*60}") + print(report_json) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(report_json) + print(f"\nReport written to {args.output}", file=sys.stderr) + + return 0 if report["overall_success"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/restore.py b/scripts/restore.py new file mode 100755 index 0000000..6039790 --- /dev/null +++ b/scripts/restore.py @@ -0,0 +1,301 @@ +#!/usr/bin/env python3 +"""Restore Script — restore LeoCRM from a backup directory. + +Usage: + python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 + python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --skip-db + python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --skip-files + python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --dry-run + +Restores: + 1. Database from database.sql (psql) + 2. Files from files/ directory (shutil copy) + +Environment variables: + DATABASE_URL — PostgreSQL connection string + STORAGE_PATH — Target path for file restoration (default: /data/storage) + +Exit codes: + 0 — restore successful + 1 — restore failed + 2 — configuration error +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +DEFAULT_STORAGE_PATH = os.environ.get("STORAGE_PATH", "/data/storage") + + +def get_db_connection_params() -> dict[str, str]: + """Extract PostgreSQL connection params from DATABASE_URL env var.""" + db_url = os.environ.get("DATABASE_URL", "") + if not db_url: + raise ValueError("DATABASE_URL environment variable not set") + + if db_url.startswith("postgresql+asyncpg://"): + db_url = db_url.replace("postgresql+asyncpg://", "postgresql://", 1) + + from urllib.parse import urlparse + + parsed = urlparse(db_url) + return { + "url": db_url, + "host": parsed.hostname or "localhost", + "port": str(parsed.port or 5432), + "database": parsed.path.lstrip("/"), + "username": parsed.username or "postgres", + "password": parsed.password or "", + } + + +def restore_database(backup_dir: Path, db_params: dict[str, str]) -> tuple[bool, str]: + """Restore database from database.sql using psql. + + Returns (success, error_message). + """ + db_file = backup_dir / "database.sql" + if not db_file.is_file(): + return False, f"Database dump not found: {db_file}" + + cmd = [ + "psql", + "--host", db_params["host"], + "--port", db_params["port"], + "--username", db_params["username"], + "--dbname", db_params["database"], + "--file", str(db_file), + "--quiet", + ] + + env = os.environ.copy() + if db_params["password"]: + env["PGPASSWORD"] = db_params["password"] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + env=env, + timeout=600, + ) + if result.returncode != 0: + return False, result.stderr or "psql failed" + return True, "" + except subprocess.TimeoutExpired: + return False, "psql timed out after 600s" + except FileNotFoundError: + return False, "psql command not found" + except Exception as exc: + return False, str(exc) + + +def restore_files(backup_dir: Path, target_path: str) -> tuple[bool, int, int, str]: + """Restore files from backup files/ directory to target path. + + Returns (success, file_count, total_size_bytes, error_message). + """ + files_dir = backup_dir / "files" + if not files_dir.is_dir(): + return True, 0, 0, "" # No files in backup — not an error + + try: + os.makedirs(target_path, exist_ok=True) + file_count = 0 + total_size = 0 + + for root, dirs, files in os.walk(files_dir): + rel_root = os.path.relpath(root, files_dir) + dest_root = os.path.join(target_path, rel_root) if rel_root != "." else target_path + os.makedirs(dest_root, exist_ok=True) + + for fname in files: + src_file = os.path.join(root, fname) + dest_file = os.path.join(dest_root, fname) + shutil.copy2(src_file, dest_file) + file_count += 1 + total_size += os.path.getsize(src_file) + + return True, file_count, total_size, "" + except Exception as exc: + return False, 0, 0, str(exc) + + +def validate_backup_dir(backup_dir: Path) -> dict[str, Any] | None: + """Validate that backup directory has a manifest.json.""" + manifest_file = backup_dir / "manifest.json" + if not manifest_file.is_file(): + return None + try: + with open(manifest_file) as f: + return json.load(f) + except Exception: + return None + + +def run_restore( + backup_dir_path: str, + target_storage_path: str, + skip_db: bool = False, + skip_files: bool = False, + dry_run: bool = False, +) -> dict[str, Any]: + """Run the full restore pipeline and return a report.""" + backup_dir = Path(backup_dir_path) + report: dict[str, Any] = { + "backup_dir": str(backup_dir), + "dry_run": dry_run, + "started_at": datetime.now(timezone.utc).isoformat(), + "database": {"success": False, "error": ""}, + "files": {"success": False, "count": 0, "size_bytes": 0, "error": ""}, + "overall_success": False, + "errors": [], + } + + # Validate backup directory + if not backup_dir.is_dir(): + report["errors"].append(f"Backup directory not found: {backup_dir}") + report["finished_at"] = datetime.now(timezone.utc).isoformat() + return report + + manifest = validate_backup_dir(backup_dir) + if manifest: + report["manifest"] = manifest + print(f" Backup from: {manifest.get('created_at', 'unknown')}") + else: + print(f" Warning: No manifest.json found in {backup_dir}") + + if dry_run: + report["database"]["error"] = "[DRY-RUN] Would restore database" + report["files"]["error"] = "[DRY-RUN] Would restore files" + report["overall_success"] = True + report["finished_at"] = datetime.now(timezone.utc).isoformat() + return report + + # Step 1: Database restore + if not skip_db: + print("▶ Step 1: Database restore (psql)") + try: + db_params = get_db_connection_params() + db_success, db_error = restore_database(backup_dir, db_params) + report["database"] = {"success": db_success, "error": db_error} + if db_success: + print(" ✓ Database restored") + else: + print(f" ✗ Database restore failed: {db_error}") + report["errors"].append(f"Database: {db_error}") + except Exception as exc: + report["database"]["error"] = str(exc) + report["errors"].append(f"Database config: {exc}") + print(f" ✗ Database config error: {exc}") + else: + print("▶ Step 1: Database restore — SKIPPED") + report["database"] = {"success": True, "error": "skipped"} + + # Step 2: File restore + if not skip_files: + print(f"▶ Step 2: File restore → {target_storage_path}") + files_success, files_count, files_size, files_error = restore_files(backup_dir, target_storage_path) + report["files"] = { + "success": files_success, + "count": files_count, + "size_bytes": files_size, + "error": files_error, + } + if files_success: + print(f" ✓ Files restored: {files_count} files, {files_size} bytes") + else: + print(f" ✗ File restore failed: {files_error}") + report["errors"].append(f"Files: {files_error}") + else: + print("▶ Step 2: File restore — SKIPPED") + report["files"] = {"success": True, "count": 0, "size_bytes": 0, "error": "skipped"} + + # Determine overall success + report["overall_success"] = report["database"]["success"] and report["files"]["success"] + report["finished_at"] = datetime.now(timezone.utc).isoformat() + + return report + + +# ─── CLI Entry Point ───────────────────────────────────────────────── + + +def main() -> int: + parser = argparse.ArgumentParser( + description="LeoCRM Restore — restore database and files from a backup directory." + ) + parser.add_argument( + "--backup-dir", + required=True, + help="Path to the backup directory (e.g. /backups/leocrm_backup_20260723_220000)", + ) + parser.add_argument( + "--storage-path", + default=DEFAULT_STORAGE_PATH, + help=f"Target path for file restoration (default: {DEFAULT_STORAGE_PATH})", + ) + parser.add_argument( + "--skip-db", + action="store_true", + help="Skip database restoration", + ) + parser.add_argument( + "--skip-files", + action="store_true", + help="Skip file restoration", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Simulate restore without executing", + ) + parser.add_argument( + "--output", + default="-", + help="Write JSON report to file (default: stdout)", + ) + args = parser.parse_args() + + print(f"\n{'='*60}") + print(f" LeoCRM Restore Pipeline") + print(f" {'[DRY-RUN]' if args.dry_run else '[LIVE]'}") + print(f"{'='*60}\n") + + report = run_restore( + backup_dir_path=args.backup_dir, + target_storage_path=args.storage_path, + skip_db=args.skip_db, + skip_files=args.skip_files, + dry_run=args.dry_run, + ) + + report_json = json.dumps(report, indent=2, ensure_ascii=False) + if args.output == "-": + print(f"\n{'='*60}") + print(" Restore Report") + print(f"{'='*60}") + print(report_json) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(report_json) + print(f"\nReport written to {args.output}", file=sys.stderr) + + return 0 if report["overall_success"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_backup_restore.py b/tests/test_backup_restore.py new file mode 100644 index 0000000..d49de63 --- /dev/null +++ b/tests/test_backup_restore.py @@ -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