305 lines
11 KiB
Python
305 lines
11 KiB
Python
import os
|
|
import pytest
|
|
import platform
|
|
import ctypes
|
|
import tempfile
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
from app.core.database import init_db, AsyncSessionLocal
|
|
from app.models.models import Client, BackupJob, BackupFile
|
|
from app.api.jobs import purge_orphans
|
|
from app.schemas.schemas import PurgeOrphansRequest
|
|
from app.core.config import settings
|
|
from agent.scanner import DirectoryScanner
|
|
from agent.service import is_job_due
|
|
|
|
def set_windows_attributes(filepath: Path, hidden: bool = False, system: bool = False, readonly: bool = False):
|
|
if platform.system() != "Windows":
|
|
return
|
|
attrs = 0
|
|
if readonly:
|
|
attrs |= 0x01
|
|
if hidden:
|
|
attrs |= 0x02
|
|
if system:
|
|
attrs |= 0x04
|
|
# Set attributes
|
|
ctypes.windll.kernel32.SetFileAttributesW(str(filepath), attrs)
|
|
|
|
def test_directory_scanner_filtering():
|
|
"""
|
|
Tests DirectoryScanner filtering logic:
|
|
- Verifies inclusion patterns (e.g. *.txt).
|
|
- Verifies exclude patterns (e.g. *.tmp, log_dir).
|
|
- Verifies Windows-specific attributes skipping (Hidden, System, Read-only).
|
|
"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root_path = Path(tmpdir)
|
|
|
|
# Create normal files
|
|
normal_txt = root_path / "normal.txt"
|
|
normal_txt.write_text("Hello World")
|
|
|
|
# Create excluded extension file
|
|
temp_file = root_path / "temp.tmp"
|
|
temp_file.write_text("Temp Data")
|
|
|
|
# Create file in excluded folder
|
|
log_dir = root_path / "logs"
|
|
log_dir.mkdir()
|
|
log_file = log_dir / "app.log"
|
|
log_file.write_text("Log info")
|
|
|
|
# Create hidden/system/readonly files (using ctypes helper on Windows)
|
|
hidden_txt = root_path / "hidden.txt"
|
|
hidden_txt.write_text("Hidden file content")
|
|
set_windows_attributes(hidden_txt, hidden=True)
|
|
|
|
system_txt = root_path / "system.txt"
|
|
system_txt.write_text("System file content")
|
|
set_windows_attributes(system_txt, system=True)
|
|
|
|
readonly_txt = root_path / "readonly.txt"
|
|
readonly_txt.write_text("Readonly file content")
|
|
set_windows_attributes(readonly_txt, readonly=True)
|
|
|
|
# Scanner 1: Inclusions (*.txt) and Exclusions (*.tmp, logs)
|
|
scanner = DirectoryScanner(
|
|
source_path=str(root_path),
|
|
file_patterns="*.txt",
|
|
exclude_patterns="*.tmp, logs",
|
|
skip_hidden=False,
|
|
skip_system=False,
|
|
skip_readonly=False,
|
|
min_stable_seconds=0
|
|
)
|
|
files = scanner.scan()
|
|
file_names = {f.name for f in files}
|
|
|
|
assert "normal.txt" in file_names
|
|
assert "temp.tmp" not in file_names
|
|
assert "app.log" not in file_names
|
|
|
|
# If running on Windows, verify attributes filtering
|
|
if platform.system() == "Windows":
|
|
assert "hidden.txt" in file_names
|
|
assert "system.txt" in file_names
|
|
assert "readonly.txt" in file_names
|
|
|
|
# Scanner 2: Skip hidden files
|
|
scanner_hidden = DirectoryScanner(
|
|
source_path=str(root_path),
|
|
file_patterns="*.txt",
|
|
exclude_patterns="*.tmp, logs",
|
|
skip_hidden=True,
|
|
min_stable_seconds=0
|
|
)
|
|
files_hidden = scanner_hidden.scan()
|
|
names_hidden = {f.name for f in files_hidden}
|
|
assert "normal.txt" in names_hidden
|
|
assert "hidden.txt" not in names_hidden
|
|
|
|
# Scanner 3: Skip system files
|
|
scanner_sys = DirectoryScanner(
|
|
source_path=str(root_path),
|
|
file_patterns="*.txt",
|
|
exclude_patterns="*.tmp, logs",
|
|
skip_system=True,
|
|
min_stable_seconds=0
|
|
)
|
|
files_sys = scanner_sys.scan()
|
|
names_sys = {f.name for f in files_sys}
|
|
assert "normal.txt" in names_sys
|
|
assert "system.txt" not in names_sys
|
|
|
|
# Scanner 4: Skip read-only files
|
|
scanner_ro = DirectoryScanner(
|
|
source_path=str(root_path),
|
|
file_patterns="*.txt",
|
|
exclude_patterns="*.tmp, logs",
|
|
skip_readonly=True,
|
|
min_stable_seconds=0
|
|
)
|
|
files_ro = scanner_ro.scan()
|
|
names_ro = {f.name for f in files_ro}
|
|
assert "normal.txt" in names_ro
|
|
assert "readonly.txt" not in names_ro
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_replica_mode_purge_orphans():
|
|
"""
|
|
Tests the purge orphans logic:
|
|
- Enforces replica mode deletes files on server that no longer exist on client.
|
|
- Decrements total client storage usage bytes accordingly.
|
|
"""
|
|
await init_db()
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
# 1. Setup client and job with sync_deletions enabled
|
|
client = Client(
|
|
client_code="CLIENT-M-01",
|
|
name="Mirroring Test Machine",
|
|
status="ONLINE",
|
|
storage_used_bytes=3000,
|
|
storage_quota_bytes=100000
|
|
)
|
|
db.add(client)
|
|
await db.flush()
|
|
|
|
job = BackupJob(
|
|
job_code="JOB-M-01",
|
|
client_id=client.id,
|
|
name="Replica Test Job",
|
|
source_path="C:\\MirrorSource",
|
|
sync_deletions=True
|
|
)
|
|
db.add(job)
|
|
await db.flush()
|
|
|
|
# 2. Add three synthetic backup files to backend DB and storage
|
|
files_data = [
|
|
("file1.txt", "clients/CLIENT-M-01/JOB-M-01/file1.txt", 1000),
|
|
("file2.txt", "clients/CLIENT-M-01/JOB-M-01/file2.txt", 1000),
|
|
("file3.txt", "clients/CLIENT-M-01/JOB-M-01/file3.txt", 1000),
|
|
]
|
|
|
|
db_files = []
|
|
for name, rel_path, size in files_data:
|
|
# Create physical file in settings.STORAGE_ROOT
|
|
full_path = Path(settings.STORAGE_ROOT) / rel_path
|
|
full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
full_path.write_text("mock file contents")
|
|
|
|
bf = BackupFile(
|
|
client_id=client.id,
|
|
job_id=job.id,
|
|
filename=name,
|
|
relative_path=rel_path,
|
|
client_relative_path=name,
|
|
file_size=size,
|
|
sha256="test-sha256",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc)
|
|
)
|
|
db.add(bf)
|
|
db_files.append(bf)
|
|
|
|
await db.commit()
|
|
|
|
# Verify initial storage state
|
|
assert client.storage_used_bytes == 3000
|
|
|
|
# 3. Simulate client backup run reporting only file1.txt and file3.txt are active
|
|
# file2.txt has been deleted by client local environment, so it is an orphan
|
|
payload = PurgeOrphansRequest(active_relative_paths=["file1.txt", "file3.txt"])
|
|
|
|
# Invoke purge_orphans API method directly
|
|
result = await purge_orphans(
|
|
job_id=job.id,
|
|
payload=payload,
|
|
db=db,
|
|
current_client=client
|
|
)
|
|
|
|
# 4. Verify results
|
|
assert result["purged_count"] == 1
|
|
assert result["freed_bytes"] == 1000
|
|
|
|
# Verify file2.txt DB record is no longer active
|
|
await db.refresh(db_files[1])
|
|
assert db_files[1].is_active == False
|
|
|
|
# Verify file2.txt physical file was deleted
|
|
deleted_file_path = Path(settings.STORAGE_ROOT) / db_files[1].relative_path
|
|
assert not deleted_file_path.exists(), "Orphan physical file must be deleted from storage root"
|
|
|
|
# Verify client storage usage was decremented by 1000 bytes
|
|
assert client.storage_used_bytes == 2000
|
|
|
|
# Verify file1.txt and file3.txt remain untouched and active
|
|
await db.refresh(db_files[0])
|
|
await db.refresh(db_files[2])
|
|
assert db_files[0].is_active == True
|
|
assert db_files[2].is_active == True
|
|
assert (Path(settings.STORAGE_ROOT) / db_files[0].relative_path).exists()
|
|
|
|
def test_is_job_due_advanced():
|
|
# 1. Traditional intervals
|
|
# Last run was 50 mins ago (3000s)
|
|
last_run_iso = (datetime.now(timezone.utc) - timedelta(minutes=50)).isoformat()
|
|
assert is_job_due("hourly", last_run_iso) == False
|
|
|
|
# Last run was 70 mins ago (4200s)
|
|
last_run_iso = (datetime.now(timezone.utc) - timedelta(minutes=70)).isoformat()
|
|
assert is_job_due("hourly", last_run_iso) == True
|
|
|
|
# 2. Calendar schedule match
|
|
# Let's test a simple day:
|
|
# If the scheduled time occurred in the window, it is due.
|
|
five_mins_ago = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
day_name = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"][five_mins_ago.weekday()]
|
|
time_str = five_mins_ago.strftime("%H:%M")
|
|
|
|
sched_str = f"{day_name} {time_str}"
|
|
last_run = (five_mins_ago - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
|
|
assert is_job_due(sched_str, last_run) == True
|
|
|
|
# 3. Cron schedule match
|
|
# Cron format: minute hour day month weekday (0-6 starting Sunday)
|
|
cron_weekday = (five_mins_ago.weekday() + 1) % 7
|
|
cron_str = f"{five_mins_ago.minute} {five_mins_ago.hour} * * {cron_weekday}"
|
|
assert is_job_due(cron_str, last_run) == True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_job_runs_api_telemetry():
|
|
"""
|
|
Tests the creation, execution tracking, and completion endpoints of job runs.
|
|
"""
|
|
async with AsyncSessionLocal() as db:
|
|
# 1. Setup client and job
|
|
client = Client(
|
|
client_code="CLIENT-R-01",
|
|
name="Runs Test Machine",
|
|
status="ONLINE",
|
|
storage_used_bytes=0,
|
|
storage_quota_bytes=100000
|
|
)
|
|
db.add(client)
|
|
await db.flush()
|
|
|
|
job = BackupJob(
|
|
job_code="JOB-R-01",
|
|
client_id=client.id,
|
|
name="Runs Test Job",
|
|
source_path="C:\\RunsSource"
|
|
)
|
|
db.add(job)
|
|
await db.commit()
|
|
|
|
# 2. Simulate runs endpoints inside backend
|
|
from app.api.jobs import start_job_run, complete_job_run
|
|
from app.schemas.schemas import JobRunStartRequest, JobRunCompleteRequest
|
|
|
|
start_req = JobRunStartRequest(client_id=client.id)
|
|
run_res = await start_job_run(job_id=job.id, payload=start_req, db=db, current_client=client)
|
|
assert run_res.status == "RUNNING"
|
|
assert run_res.job_id == job.id
|
|
|
|
# Complete the run
|
|
complete_req = JobRunCompleteRequest(
|
|
status="SUCCESS",
|
|
files_scanned=10,
|
|
files_copied=3,
|
|
files_skipped=7,
|
|
errors_count=0,
|
|
bytes_transferred=50000
|
|
)
|
|
completed_res = await complete_job_run(job_id=job.id, run_id=run_res.id, payload=complete_req, db=db, current_client=client)
|
|
assert completed_res.status == "SUCCESS"
|
|
assert completed_res.files_scanned == 10
|
|
assert completed_res.files_copied == 3
|
|
assert completed_res.files_skipped == 7
|
|
assert completed_res.errors_count == 0
|
|
assert completed_res.bytes_transferred == 50000
|