alex
7676438466
Add await db.refresh() calls with relationship attributes before accessing test_job.executions and execution.backups in async test context. Fixes lazy loading errors in: - test_execute_full_backup - test_execute_incremental_without_full
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
import pytest
|
|
import pytest_asyncio
|
|
import tempfile
|
|
import os
|
|
from pathlib import Path
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
from app.models import Source, Job, JobExecution
|
|
from backup.engine import BackupEngine
|
|
|
|
@pytest_asyncio.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_asyncio.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"
|
|
|
|
# Refresh test_job to load executions relationship
|
|
await db.refresh(test_job, ["executions"])
|
|
|
|
# Verify backup was created
|
|
assert len(test_job.executions) == 1
|
|
|
|
# Refresh execution to load backups relationship
|
|
await db.refresh(test_job.executions[0], ["backups"])
|
|
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"
|
|
|
|
# Refresh execution to load backups relationship
|
|
await db.refresh(execution, ["backups"])
|
|
backup = execution.backups[0]
|
|
assert backup.type == "full"
|
|
assert backup.parent_backup_id is None
|