Russell Ballestrini
1847a2fee6
Tests for: - async_web_fetcher: URL handling, media extraction, crawl modes, scoring - domain_vault: VaultManager, HTML/Media/Linkpeek vaults - screenshot: ScreenshotCapture configuration and availability - storage: ImageVault with MD5 deduplication 185 tests total, all passing.
330 lines
9.8 KiB
Python
330 lines
9.8 KiB
Python
"""
|
|
Tests for storage module (ImageVault).
|
|
|
|
Tests async content-addressable storage with MD5 hashing.
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
import hashlib
|
|
|
|
from storage import ImageVault
|
|
|
|
|
|
class TestImageVaultBasics:
|
|
"""Test basic ImageVault functionality."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vault_initialization(self):
|
|
"""Test ImageVault initializes correctly."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
assert vault._initialized is True
|
|
assert os.path.exists(self.vault_path)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vault_creates_subdirectories(self):
|
|
"""Test that vault creates hash bucket subdirectories."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
# Should create 256 subdirectories (00-ff)
|
|
subdirs = os.listdir(self.vault_path)
|
|
assert len(subdirs) == 256
|
|
|
|
# Check some specific ones
|
|
assert "00" in subdirs
|
|
assert "ff" in subdirs
|
|
assert "a5" in subdirs
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vault_double_init(self):
|
|
"""Test that double initialization is safe."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
await vault.init() # Should not raise
|
|
|
|
assert vault._initialized is True
|
|
|
|
|
|
class TestImageVaultStorage:
|
|
"""Test ImageVault storage operations."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_and_retrieve(self):
|
|
"""Test storing and retrieving data."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
# Create test data
|
|
data = b"Hello, World! This is test image data."
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
# Store
|
|
path = await vault.store(md5_hash, data, extension="jpg")
|
|
assert path is not None
|
|
|
|
# Retrieve
|
|
retrieved = await vault.get(md5_hash)
|
|
assert retrieved == data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_with_extension(self):
|
|
"""Test storing with file extension."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"PNG image data here"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
path = await vault.store(md5_hash, data, extension="png")
|
|
assert path.endswith(".png")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_without_extension(self):
|
|
"""Test storing without file extension."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Binary data without extension"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
path = await vault.store(md5_hash, data)
|
|
# Should just be the hash
|
|
assert md5_hash in path
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exists_true(self):
|
|
"""Test exists returns True for stored data."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for exists check"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
await vault.store(md5_hash, data)
|
|
assert await vault.exists(md5_hash) is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exists_false(self):
|
|
"""Test exists returns False for missing data."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
assert await vault.exists(fake_hash) is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_missing_returns_none(self):
|
|
"""Test get returns None for missing data."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
result = await vault.get(fake_hash)
|
|
assert result is None
|
|
|
|
|
|
class TestImageVaultDeduplication:
|
|
"""Test ImageVault deduplication behavior."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_data_same_path(self):
|
|
"""Test that same data stored twice goes to same path."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Duplicate test data"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
path1 = await vault.store(md5_hash, data, extension="jpg")
|
|
path2 = await vault.store(md5_hash, data, extension="jpg")
|
|
|
|
# Should be the same path
|
|
assert path1 == path2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_data_different_path(self):
|
|
"""Test that different data goes to different paths."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data1 = b"First unique data"
|
|
data2 = b"Second unique data"
|
|
hash1 = hashlib.md5(data1).hexdigest()
|
|
hash2 = hashlib.md5(data2).hexdigest()
|
|
|
|
path1 = await vault.store(hash1, data1)
|
|
path2 = await vault.store(hash2, data2)
|
|
|
|
assert path1 != path2
|
|
|
|
|
|
class TestImageVaultDeletion:
|
|
"""Test ImageVault deletion operations."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_existing(self):
|
|
"""Test deleting existing file."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Data to be deleted"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
await vault.store(md5_hash, data)
|
|
assert await vault.exists(md5_hash) is True
|
|
|
|
result = await vault.delete(md5_hash)
|
|
assert result is True
|
|
assert await vault.exists(md5_hash) is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_missing(self):
|
|
"""Test deleting non-existent file."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
result = await vault.delete(fake_hash)
|
|
assert result is False
|
|
|
|
|
|
class TestImageVaultPath:
|
|
"""Test ImageVault path operations."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_path_existing(self):
|
|
"""Test getting path for existing file."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for path"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
|
|
await vault.store(md5_hash, data, extension="bin")
|
|
path = await vault.get_path(md5_hash)
|
|
|
|
assert path is not None
|
|
assert path.exists()
|
|
assert md5_hash in str(path)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_path_missing(self):
|
|
"""Test getting path for missing file."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
path = await vault.get_path(fake_hash)
|
|
assert path is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_uses_hash_prefix(self):
|
|
"""Test that path uses first 2 chars as subdirectory."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for subdir check"
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
prefix = md5_hash[:2]
|
|
|
|
path = await vault.store(md5_hash, data)
|
|
|
|
# Path should contain the prefix subdirectory
|
|
assert f"/{prefix}/" in path or f"\\{prefix}\\" in path
|
|
|
|
|
|
class TestImageVaultStats:
|
|
"""Test ImageVault statistics."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_empty_vault(self):
|
|
"""Test stats on empty vault."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
stats = await vault.stats()
|
|
assert stats['count'] == 0
|
|
assert stats['total_size_bytes'] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_with_files(self):
|
|
"""Test stats with files in vault."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
# Store some files
|
|
for i in range(3):
|
|
data = f"Test data {i}".encode()
|
|
md5_hash = hashlib.md5(data).hexdigest()
|
|
await vault.store(md5_hash, data)
|
|
|
|
stats = await vault.stats()
|
|
assert stats['count'] == 3
|
|
assert stats['total_size_bytes'] > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_backend_type(self):
|
|
"""Test stats reports correct backend."""
|
|
vault = ImageVault(vault_path=self.vault_path)
|
|
await vault.init()
|
|
|
|
stats = await vault.stats()
|
|
# Should be 'directory' when filevault is not available
|
|
assert 'backend' in stats
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|