430 lines
13 KiB
Python
430 lines
13 KiB
Python
# This is free software for the public good of a permacomputer hosted at
|
|
# permacomputer.com, an always-on computer by the people, for the people.
|
|
# One which is durable, easy to repair, & distributed like tap water
|
|
# for machine learning intelligence.
|
|
#
|
|
# The permacomputer is community-owned infrastructure optimized around
|
|
# four values:
|
|
#
|
|
# TRUTH First principles, math & science, open source code freely distributed
|
|
# FREEDOM Voluntary partnerships, freedom from tyranny & corporate control
|
|
# HARMONY Minimal waste, self-renewing systems with diverse thriving connections
|
|
# LOVE Be yourself without hurting others, cooperation through natural law
|
|
#
|
|
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
|
|
# Code is seeds to sprout on any abandoned technology.
|
|
|
|
"""
|
|
Tests for filevault module (AsyncVault).
|
|
|
|
Tests async content-addressable storage with MD5 hashing.
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
import hashlib
|
|
|
|
from neopig.filevault import AsyncVault, Vault, content_hash, hash_to_path
|
|
|
|
|
|
class TestAsyncVaultBasics:
|
|
"""Test basic AsyncVault 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 AsyncVault initializes correctly."""
|
|
vault = AsyncVault(vaultpath=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_double_init(self):
|
|
"""Test that double initialization is safe."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
await vault.init() # Should not raise
|
|
|
|
assert vault._initialized is True
|
|
|
|
|
|
class TestAsyncVaultStorage:
|
|
"""Test AsyncVault 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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Hello, World! This is test image data."
|
|
md5_hash = content_hash(data)
|
|
|
|
# Store
|
|
path = await vault.store(md5_hash, data, ext="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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"PNG image data here"
|
|
md5_hash = content_hash(data)
|
|
|
|
path = await vault.store(md5_hash, data, ext="png")
|
|
assert str(path).endswith(".png")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_without_extension(self):
|
|
"""Test storing without file extension."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Binary data without extension"
|
|
md5_hash = content_hash(data)
|
|
|
|
path = await vault.store(md5_hash, data)
|
|
assert md5_hash in str(path)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exists_true(self):
|
|
"""Test exists returns True for stored data."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for exists check"
|
|
md5_hash = content_hash(data)
|
|
|
|
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 = AsyncVault(vaultpath=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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
result = await vault.get(fake_hash)
|
|
assert result is None
|
|
|
|
|
|
class TestAsyncVaultDeduplication:
|
|
"""Test AsyncVault 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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Duplicate test data"
|
|
md5_hash = content_hash(data)
|
|
|
|
path1 = await vault.store(md5_hash, data, ext="jpg")
|
|
path2 = await vault.store(md5_hash, data, ext="jpg")
|
|
|
|
assert path1 == path2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_data_different_path(self):
|
|
"""Test that different data goes to different paths."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data1 = b"First unique data"
|
|
data2 = b"Second unique data"
|
|
hash1 = content_hash(data1)
|
|
hash2 = content_hash(data2)
|
|
|
|
path1 = await vault.store(hash1, data1)
|
|
path2 = await vault.store(hash2, data2)
|
|
|
|
assert path1 != path2
|
|
|
|
|
|
class TestAsyncVaultDeletion:
|
|
"""Test AsyncVault 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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Data to be deleted"
|
|
md5_hash = content_hash(data)
|
|
|
|
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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
fake_hash = "0" * 32
|
|
result = await vault.delete(fake_hash)
|
|
assert result is False
|
|
|
|
|
|
class TestAsyncVaultPath:
|
|
"""Test AsyncVault 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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for path"
|
|
md5_hash = content_hash(data)
|
|
|
|
await vault.store(md5_hash, data, ext="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 = AsyncVault(vaultpath=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_pairs(self):
|
|
"""Test that path uses 2-char hex pairs as subdirectories."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
data = b"Test data for subdir check"
|
|
md5_hash = content_hash(data)
|
|
# First two chars form first directory
|
|
prefix = md5_hash[:2]
|
|
|
|
path = await vault.store(md5_hash, data)
|
|
path_str = str(path)
|
|
|
|
# Path should contain the prefix as a subdirectory
|
|
assert f"/{prefix}/" in path_str or f"\\{prefix}\\" in path_str
|
|
|
|
|
|
class TestAsyncVaultStats:
|
|
"""Test AsyncVault 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 = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
stats = await vault.stats()
|
|
assert stats['count'] == 0
|
|
assert stats['total_size'] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_with_files(self):
|
|
"""Test stats with files in vault."""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
# Store some files
|
|
for i in range(3):
|
|
data = f"Test data {i}".encode()
|
|
md5_hash = content_hash(data)
|
|
await vault.store(md5_hash, data)
|
|
|
|
stats = await vault.stats()
|
|
assert stats['count'] == 3
|
|
assert stats['total_size'] > 0
|
|
|
|
|
|
class TestSyncVault:
|
|
"""Test synchronous Vault."""
|
|
|
|
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)
|
|
|
|
def test_sync_store_and_get(self):
|
|
"""Test sync vault store and get."""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
|
|
data = b"Sync test data"
|
|
md5_hash = content_hash(data)
|
|
|
|
path = vault.store(md5_hash, data, ext="bin")
|
|
assert path.exists()
|
|
|
|
retrieved = vault.get(md5_hash)
|
|
assert retrieved == data
|
|
|
|
def test_sync_exists(self):
|
|
"""Test sync vault exists."""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
|
|
data = b"Exists test"
|
|
md5_hash = content_hash(data)
|
|
|
|
assert vault.exists(md5_hash) is False
|
|
vault.store(md5_hash, data)
|
|
assert vault.exists(md5_hash) is True
|
|
|
|
def test_sync_delete(self):
|
|
"""Test sync vault delete."""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
|
|
data = b"Delete test"
|
|
md5_hash = content_hash(data)
|
|
|
|
vault.store(md5_hash, data)
|
|
assert vault.delete(md5_hash) is True
|
|
assert vault.exists(md5_hash) is False
|
|
|
|
def test_sync_json_operations(self):
|
|
"""Test sync vault JSON operations."""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
|
|
path = vault.create_filename("test-key", ext="json")
|
|
data = {"key": "value", "number": 42}
|
|
|
|
vault.write_json(path, data)
|
|
loaded = vault.read_json(path)
|
|
|
|
assert loaded == data
|
|
|
|
def test_sync_text_operations(self):
|
|
"""Test sync vault text operations."""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
|
|
path = vault.create_filename("text-key", ext="txt")
|
|
content = "Hello, World!"
|
|
|
|
vault.write_text(path, content)
|
|
loaded = vault.read_text(path)
|
|
|
|
assert loaded == content
|
|
|
|
|
|
class TestHashToPath:
|
|
"""Test hash_to_path utility function."""
|
|
|
|
def test_basic_hash(self):
|
|
"""Test basic hash to path conversion."""
|
|
h = "abcdef1234567890abcdef1234567890"
|
|
path = hash_to_path(h, depth=3)
|
|
assert str(path) == "ab/cd/ef/abcdef1234567890abcdef1234567890"
|
|
|
|
def test_with_extension(self):
|
|
"""Test hash to path with extension."""
|
|
h = "abcdef1234567890abcdef1234567890"
|
|
path = hash_to_path(h, depth=3, ext="jpg")
|
|
assert str(path) == "ab/cd/ef/abcdef1234567890abcdef1234567890.jpg"
|
|
|
|
def test_depth_9(self):
|
|
"""Test default depth of 9."""
|
|
h = "abcdef1234567890abcdef1234567890"
|
|
path = hash_to_path(h, depth=9)
|
|
expected = "ab/cd/ef/12/34/56/78/90/ab/abcdef1234567890abcdef1234567890"
|
|
assert str(path) == expected
|
|
|
|
|
|
class TestContentHash:
|
|
"""Test content_hash utility function."""
|
|
|
|
def test_content_hash(self):
|
|
"""Test content hash generation."""
|
|
data = b"Hello, World!"
|
|
h = content_hash(data)
|
|
assert h == hashlib.md5(data).hexdigest()
|
|
assert len(h) == 32
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|