mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
202 lines
6.8 KiB
Python
202 lines
6.8 KiB
Python
"""Tests for incremental scanning module."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestIncremental:
|
|
"""Test Incremental class."""
|
|
|
|
def test_save_checkpoint(self, temp_dir):
|
|
"""Test saving checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
scan_path = str(temp_dir)
|
|
processed_files = {
|
|
"file1.txt": {"hash": "abc123", "size": 100},
|
|
"file2.txt": {"hash": "def456", "size": 200}
|
|
}
|
|
metadata = {"version": "1.0"}
|
|
|
|
Incremental.save_checkpoint(scan_path, processed_files, metadata)
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
|
|
assert checkpoint_file.exists()
|
|
|
|
with open(checkpoint_file) as f:
|
|
data = json.load(f)
|
|
|
|
assert data["scan_path"] == scan_path
|
|
assert "file1.txt" in data["processed_files"]
|
|
assert "file2.txt" in data["processed_files"]
|
|
assert data["metadata"]["version"] == "1.0"
|
|
assert "timestamp" in data
|
|
|
|
def test_save_checkpoint_no_metadata(self, temp_dir):
|
|
"""Test saving checkpoint without metadata."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
scan_path = str(temp_dir)
|
|
processed_files = {"file1.txt": {"hash": "abc123"}}
|
|
|
|
Incremental.save_checkpoint(scan_path, processed_files)
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
|
|
assert checkpoint_file.exists()
|
|
|
|
with open(checkpoint_file) as f:
|
|
data = json.load(f)
|
|
|
|
assert data["metadata"] == {}
|
|
|
|
def test_load_checkpoint(self, temp_dir):
|
|
"""Test loading checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
# Create checkpoint file manually
|
|
checkpoint_data = {
|
|
"scan_path": str(temp_dir),
|
|
"processed_files": {"file1.txt": {"hash": "abc123"}},
|
|
"timestamp": "2025-01-01T00:00:00",
|
|
"metadata": {}
|
|
}
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
with open(checkpoint_file, 'w') as f:
|
|
json.dump(checkpoint_data, f)
|
|
|
|
result = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert result is not None
|
|
assert result["scan_path"] == str(temp_dir)
|
|
assert "file1.txt" in result["processed_files"]
|
|
|
|
def test_load_checkpoint_no_file(self, temp_dir):
|
|
"""Test loading checkpoint when no file exists."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
result = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert result is None
|
|
|
|
def test_load_checkpoint_invalid_json(self, temp_dir):
|
|
"""Test loading checkpoint with invalid JSON."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
checkpoint_file.write_text("not valid json")
|
|
|
|
result = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert result is None
|
|
|
|
def test_load_checkpoint_corrupted(self, temp_dir):
|
|
"""Test loading corrupted checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
checkpoint_file.write_text("")
|
|
|
|
result = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert result is None
|
|
|
|
def test_get_remaining_files_no_checkpoint(self, temp_dir):
|
|
"""Test getting remaining files when no checkpoint exists."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
all_files = ["file1.txt", "file2.txt", "file3.txt"]
|
|
|
|
result = Incremental.get_remaining_files(str(temp_dir), all_files)
|
|
|
|
assert result == all_files
|
|
|
|
def test_get_remaining_files_with_checkpoint(self, temp_dir):
|
|
"""Test getting remaining files with existing checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
# Create checkpoint with some processed files
|
|
checkpoint_data = {
|
|
"scan_path": str(temp_dir),
|
|
"processed_files": {
|
|
"file1.txt": {"hash": "abc123"},
|
|
"file2.txt": {"hash": "def456"}
|
|
},
|
|
"timestamp": "2025-01-01T00:00:00",
|
|
"metadata": {}
|
|
}
|
|
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
with open(checkpoint_file, 'w') as f:
|
|
json.dump(checkpoint_data, f)
|
|
|
|
all_files = ["file1.txt", "file2.txt", "file3.txt", "file4.txt"]
|
|
|
|
result = Incremental.get_remaining_files(str(temp_dir), all_files)
|
|
|
|
assert "file1.txt" not in result
|
|
assert "file2.txt" not in result
|
|
assert "file3.txt" in result
|
|
assert "file4.txt" in result
|
|
|
|
def test_update_checkpoint(self, temp_dir):
|
|
"""Test updating checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
# First save a checkpoint
|
|
initial_files = {"file1.txt": {"hash": "abc123"}}
|
|
Incremental.save_checkpoint(str(temp_dir), initial_files)
|
|
|
|
# Update with new files
|
|
new_files = {"file2.txt": {"hash": "def456"}}
|
|
Incremental.update_checkpoint(str(temp_dir), new_files)
|
|
|
|
# Check that both are present
|
|
checkpoint = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert "file1.txt" in checkpoint["processed_files"]
|
|
assert "file2.txt" in checkpoint["processed_files"]
|
|
|
|
def test_update_checkpoint_no_existing(self, temp_dir):
|
|
"""Test updating checkpoint when none exists."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
new_files = {"file1.txt": {"hash": "abc123"}}
|
|
Incremental.update_checkpoint(str(temp_dir), new_files)
|
|
|
|
checkpoint = Incremental.load_checkpoint(str(temp_dir))
|
|
|
|
assert checkpoint is not None
|
|
assert "file1.txt" in checkpoint["processed_files"]
|
|
|
|
def test_cleanup_checkpoint(self, temp_dir):
|
|
"""Test cleaning up checkpoint."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
# Create checkpoint file
|
|
checkpoint_file = temp_dir / ".nodupe_checkpoint.json"
|
|
checkpoint_file.write_text("{}")
|
|
|
|
result = Incremental.cleanup_checkpoint(str(temp_dir))
|
|
|
|
assert result is True
|
|
assert not checkpoint_file.exists()
|
|
|
|
def test_cleanup_checkpoint_no_file(self, temp_dir):
|
|
"""Test cleaning up checkpoint when no file exists."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
result = Incremental.cleanup_checkpoint(str(temp_dir))
|
|
|
|
assert result is False
|
|
|
|
def test_checkpoint_file_name(self):
|
|
"""Test checkpoint file name constant."""
|
|
from nodupe.tools.scanner_engine.incremental import Incremental
|
|
|
|
assert Incremental.CHECKPOINT_FILE == ".nodupe_checkpoint.json"
|