import os import pytest import platform import ctypes import tempfile from pathlib import Path from datetime import datetime, timezone 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 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()