diff --git a/backend/backup/engine.py b/backend/backup/engine.py new file mode 100644 index 0000000..1babed3 --- /dev/null +++ b/backend/backup/engine.py @@ -0,0 +1,130 @@ +import os +import hashlib +import shutil +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select +from app.models import Job, JobExecution, Backup +from backup.adapters import get_adapter + +class BackupEngine: + def __init__(self, db: AsyncSession): + self.db = db + + async def execute_job(self, job_id: int, triggered_by: str = "manual") -> JobExecution: + # Create execution record + execution = JobExecution( + job_id=job_id, + status="pending", + triggered_by=triggered_by + ) + self.db.add(execution) + await self.db.commit() + await self.db.refresh(execution) + + try: + # Load job with source + result = await self.db.execute( + select(Job).where(Job.id == job_id) + ) + job = result.scalar_one() + + # Update status to running + execution.status = "running" + execution.started_at = datetime.now(timezone.utc) + await self.db.commit() + + # Determine strategy + strategy = job.strategy + parent_backup_id = None + + if strategy == "incremental": + # Find last successful full backup + result = await self.db.execute( + select(Backup) + .join(JobExecution) + .where( + JobExecution.job_id == job_id, + JobExecution.status == "success", + Backup.type == "full" + ) + .order_by(Backup.created_at.desc()) + ) + last_full = result.scalar_one_or_none() + + if last_full: + parent_backup_id = last_full.id + else: + # No full backup exists, do full instead + strategy = "full" + + # Create backup directory + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H%M%S") + backup_dir = Path(job.destination_path) / str(job_id) / f"{timestamp}_{strategy}" + backup_dir.mkdir(parents=True, exist_ok=True) + + # Get adapter and connect + adapter = get_adapter(job.source.type, job.source.config) + await adapter.connect() + + try: + # Copy files + total_processed = 0 + total_backed_up = 0 + + source_path = Path(job.source.config.get("path", ".")) + + for item in source_path.rglob("*"): + if item.is_file(): + rel_path = item.relative_to(source_path) + dest_path = backup_dir / "data" / rel_path + dest_path.parent.mkdir(parents=True, exist_ok=True) + + # Copy file + shutil.copy2(item, dest_path) + + size = item.stat().st_size + total_processed += size + total_backed_up += size + + # Calculate checksum + checksum = await self._calculate_checksum(backup_dir) + + # Create backup record + backup = Backup( + execution_id=execution.id, + storage_path=str(backup_dir), + size_bytes=total_backed_up, + checksum=checksum, + type=strategy, + parent_backup_id=parent_backup_id + ) + self.db.add(backup) + + # Update execution + execution.status = "success" + execution.completed_at = datetime.now(timezone.utc) + execution.bytes_processed = total_processed + execution.bytes_backed_up = total_backed_up + + finally: + await adapter.disconnect() + + except Exception as e: + execution.status = "failed" + execution.completed_at = datetime.now(timezone.utc) + execution.error_message = str(e) + + await self.db.commit() + return execution + + async def _calculate_checksum(self, path: Path) -> str: + hasher = hashlib.sha256() + for item in sorted(path.rglob("*")): + if item.is_file(): + with open(item, "rb") as f: + while chunk := f.read(8192): + hasher.update(chunk) + return hasher.hexdigest() diff --git a/backend/tests/test_engine.py b/backend/tests/test_engine.py new file mode 100644 index 0000000..678e02b --- /dev/null +++ b/backend/tests/test_engine.py @@ -0,0 +1,71 @@ +import pytest +import tempfile +import os +from pathlib import Path +from sqlalchemy.ext.asyncio import AsyncSession +from app.models import Source, Job, JobExecution +from backup.engine import BackupEngine + +@pytest.fixture +async def test_source(db: AsyncSession): + with tempfile.TemporaryDirectory() as tmpdir: + # Create test files + (Path(tmpdir) / "test.txt").write_text("Hello, World!") + (Path(tmpdir) / "subdir").mkdir() + (Path(tmpdir) / "subdir" / "nested.txt").write_text("Nested content") + + source = Source( + name="Test Source", + type="local", + config={"path": tmpdir} + ) + db.add(source) + await db.commit() + await db.refresh(source) + yield source + +@pytest.fixture +async def test_job(db: AsyncSession, test_source): + with tempfile.TemporaryDirectory() as tmpdir: + job = Job( + name="Test Job", + source_id=test_source.id, + strategy="full", + destination_path=tmpdir + ) + db.add(job) + await db.commit() + await db.refresh(job) + yield job + +@pytest.mark.asyncio +async def test_execute_full_backup(db: AsyncSession, test_job): + engine = BackupEngine(db) + execution = await engine.execute_job(test_job.id, triggered_by="manual") + + assert execution.status == "success" + assert execution.bytes_processed > 0 + assert execution.bytes_backed_up > 0 + assert execution.triggered_by == "manual" + + # Verify backup was created + assert len(test_job.executions) == 1 + backup = test_job.executions[0].backups[0] + assert backup.type == "full" + assert backup.checksum is not None + assert os.path.exists(backup.storage_path) + +@pytest.mark.asyncio +async def test_execute_incremental_without_full(db: AsyncSession, test_job): + # Set job to incremental but no full backup exists + test_job.strategy = "incremental" + await db.commit() + + engine = BackupEngine(db) + execution = await engine.execute_job(test_job.id) + + # Should fall back to full backup + assert execution.status == "success" + backup = execution.backups[0] + assert backup.type == "full" + assert backup.parent_backup_id is None