539 lines
17 KiB
Python
539 lines
17 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.
|
|
|
|
"""
|
|
Unit tests for AsyncVault (from filevault 2.0.0)
|
|
Tests async content-addressable and seed-based storage
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
from neopig.filevault import AsyncVault, create_async_vault, content_hash
|
|
|
|
|
|
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)
|
|
|
|
def test_async_vault_initialization(self):
|
|
"""Test AsyncVault initializes correctly"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=5)
|
|
|
|
assert vault.vaultpath == Path(self.vault_path)
|
|
assert vault.depth == 5
|
|
assert vault.salt == b"neopig"
|
|
|
|
def test_sync_methods_work(self):
|
|
"""Test that sync methods (filename generation) work without await"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
|
|
# These should work synchronously
|
|
filename = vault.create_filename("test_seed", ".json", absolute=True)
|
|
assert filename.startswith(self.vault_path)
|
|
assert filename.endswith(".json")
|
|
|
|
random_filename = vault.create_random_filename(".txt")
|
|
assert random_filename.endswith(".txt")
|
|
|
|
def test_factory_function(self):
|
|
"""Test create_async_vault factory function"""
|
|
vault = create_async_vault(vaultpath=self.vault_path, salt="test")
|
|
assert isinstance(vault, AsyncVault)
|
|
assert vault.salt == b"test"
|
|
|
|
|
|
class TestAsyncContentAddressable:
|
|
"""Test async content-addressable storage"""
|
|
|
|
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_get(self):
|
|
"""Test async store and get"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=3)
|
|
await vault.init()
|
|
|
|
data = b"Hello, Async World!"
|
|
h = content_hash(data)
|
|
|
|
path = await vault.store(h, data, ".txt")
|
|
assert path.exists()
|
|
|
|
retrieved = await vault.get(h)
|
|
assert retrieved == data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exists(self):
|
|
"""Test async exists check"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=3)
|
|
await vault.init()
|
|
|
|
data = b"unique data"
|
|
h = content_hash(data)
|
|
|
|
assert await vault.exists(h) is False
|
|
await vault.store(h, data, ".bin")
|
|
assert await vault.exists(h) is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete(self):
|
|
"""Test async delete"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=3)
|
|
await vault.init()
|
|
|
|
data = b"delete me"
|
|
h = content_hash(data)
|
|
|
|
await vault.store(h, data, ".txt")
|
|
assert await vault.exists(h)
|
|
|
|
result = await vault.delete(h)
|
|
assert result is True
|
|
assert await vault.exists(h) is False
|
|
|
|
|
|
class TestAsyncJSONOperations:
|
|
"""Test async JSON 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_async_write_json(self):
|
|
"""Test async JSON write"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("async_json", ".json", absolute=True)
|
|
test_data = {"key": "value", "number": 42}
|
|
|
|
await vault.write_json(test_file, test_data)
|
|
|
|
assert os.path.exists(test_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_read_json(self):
|
|
"""Test async JSON read"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("async_read", ".json", absolute=True)
|
|
test_data = {"async": True, "data": [1, 2, 3]}
|
|
|
|
await vault.write_json(test_file, test_data)
|
|
read_data = await vault.read_json(test_file)
|
|
|
|
assert read_data == test_data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_read_json_with_default(self):
|
|
"""Test async JSON read with default"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("nonexistent", ".json", absolute=True)
|
|
|
|
result = await vault.read_json(test_file, default={"default": True})
|
|
assert result == {"default": True}
|
|
|
|
|
|
class TestAsyncTextOperations:
|
|
"""Test async text file 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_async_write_text(self):
|
|
"""Test async text file write"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("async_text", ".txt", absolute=True)
|
|
test_content = "Async content\nLine 2"
|
|
|
|
await vault.write_text(test_file, test_content)
|
|
|
|
assert os.path.exists(test_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_read_text(self):
|
|
"""Test async text file read"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("async_read_text", ".txt", absolute=True)
|
|
test_content = "Test async read"
|
|
|
|
await vault.write_text(test_file, test_content)
|
|
read_content = await vault.read_text(test_file)
|
|
|
|
assert read_content == test_content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_read_text_with_default(self):
|
|
"""Test async text read with default"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("nonexistent", ".txt", absolute=True)
|
|
|
|
result = await vault.read_text(test_file, default="default content")
|
|
assert result == "default content"
|
|
|
|
|
|
class TestAsyncBytesOperations:
|
|
"""Test async bytes 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_async_write_and_read_bytes(self):
|
|
"""Test async bytes write and read"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("bytes_test", ".bin", absolute=True)
|
|
test_data = b"\x00\x01\x02\xff\xfe\xfd"
|
|
|
|
await vault.write_bytes(test_file, test_data)
|
|
read_data = await vault.read_bytes(test_file)
|
|
|
|
assert read_data == test_data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_read_bytes_nonexistent(self):
|
|
"""Test async read bytes returns None for non-existent"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
result = await vault.read_bytes("/nonexistent/path")
|
|
assert result is None
|
|
|
|
|
|
class TestAsyncFileOperations:
|
|
"""Test async file existence and removal"""
|
|
|
|
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_async_file_exists(self):
|
|
"""Test async file existence check"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("exists_test", ".txt", absolute=True)
|
|
|
|
# File doesn't exist yet
|
|
exists_before = await vault.file_exists(test_file)
|
|
assert exists_before is False
|
|
|
|
# Create file
|
|
await vault.write_text(test_file, "test")
|
|
|
|
# Now it exists
|
|
exists_after = await vault.file_exists(test_file)
|
|
assert exists_after is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_file_exists_with_cache(self):
|
|
"""Test async file_exists uses cache when available"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("cached", ".txt", absolute=True)
|
|
|
|
# Manually set cache
|
|
vault._cache_set(test_file, True)
|
|
|
|
# Should return from cache without I/O
|
|
exists = await vault.file_exists(test_file)
|
|
assert exists is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_remove(self):
|
|
"""Test async file removal"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("to_remove", ".txt", absolute=True)
|
|
|
|
await vault.write_text(test_file, "test")
|
|
assert os.path.exists(test_file)
|
|
|
|
result = await vault.remove(test_file)
|
|
assert result is True
|
|
assert not os.path.exists(test_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_remove_nonexistent(self):
|
|
"""Test async remove of non-existent file"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("nonexistent", ".txt", absolute=True)
|
|
|
|
result = await vault.remove(test_file)
|
|
assert result is False
|
|
|
|
|
|
class TestAsyncCache:
|
|
"""Test async cache 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)
|
|
|
|
def test_cache_disabled_by_default(self):
|
|
"""Test cache disabled by default"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
assert vault._cache is None
|
|
|
|
def test_cache_enabled(self):
|
|
"""Test cache can be enabled"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
assert vault._cache == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_updates_cache(self):
|
|
"""Test write updates cache"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
await vault.init()
|
|
|
|
test_file = vault.create_filename("cache_test", ".txt", absolute=True)
|
|
await vault.write_text(test_file, "content")
|
|
|
|
assert vault._cache[test_file] is True
|
|
|
|
def test_clear_cache(self):
|
|
"""Test clearing cache"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault._cache["path1"] = True
|
|
vault._cache["path2"] = False
|
|
|
|
vault.clear_cache()
|
|
|
|
assert vault._cache == {}
|
|
|
|
|
|
class TestAsyncConcurrency:
|
|
"""Test async concurrent 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_concurrent_writes(self):
|
|
"""Test multiple concurrent async writes"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=5)
|
|
await vault.init()
|
|
|
|
async def write_item(i):
|
|
filename = vault.create_filename(f"item_{i}", ".json", absolute=True)
|
|
await vault.write_json(filename, {"index": i})
|
|
return i
|
|
|
|
# Run 10 concurrent writes
|
|
results = await asyncio.gather(*[write_item(i) for i in range(10)])
|
|
|
|
assert len(results) == 10
|
|
assert set(results) == set(range(10))
|
|
|
|
# Verify all files exist
|
|
for i in range(10):
|
|
filename = vault.create_filename(f"item_{i}", ".json", absolute=True)
|
|
data = await vault.read_json(filename)
|
|
assert data["index"] == i
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_reads(self):
|
|
"""Test multiple concurrent async reads"""
|
|
vault = AsyncVault(vaultpath=self.vault_path)
|
|
await vault.init()
|
|
|
|
# Write files first
|
|
for i in range(5):
|
|
filename = vault.create_filename(f"read_{i}", ".json", absolute=True)
|
|
await vault.write_json(filename, {"value": i * 10})
|
|
|
|
async def read_item(i):
|
|
filename = vault.create_filename(f"read_{i}", ".json", absolute=True)
|
|
return await vault.read_json(filename)
|
|
|
|
# Concurrent reads
|
|
results = await asyncio.gather(*[read_item(i) for i in range(5)])
|
|
|
|
assert len(results) == 5
|
|
for i, result in enumerate(results):
|
|
assert result["value"] == i * 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_store(self):
|
|
"""Test concurrent content-addressable storage"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, depth=3)
|
|
await vault.init()
|
|
|
|
async def store_item(i):
|
|
data = f"item_{i}".encode()
|
|
h = content_hash(data)
|
|
await vault.store(h, data, ".bin")
|
|
return h
|
|
|
|
hashes = await asyncio.gather(*[store_item(i) for i in range(10)])
|
|
|
|
# Verify all stored
|
|
for i, h in enumerate(hashes):
|
|
data = await vault.get(h)
|
|
assert data == f"item_{i}".encode()
|
|
|
|
|
|
class TestAsyncIntegration:
|
|
"""Integration tests for async workflows"""
|
|
|
|
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_full_async_workflow(self):
|
|
"""Test complete async CRUD workflow"""
|
|
vault = AsyncVault(
|
|
vaultpath=self.vault_path,
|
|
depth=5,
|
|
enable_memory_cache=True
|
|
)
|
|
await vault.init()
|
|
|
|
key = "workflow_key"
|
|
filename = vault.create_filename(key, ".json", absolute=True)
|
|
|
|
# Create
|
|
await vault.write_json(filename, {"version": 1})
|
|
assert await vault.file_exists(filename)
|
|
|
|
# Read
|
|
data = await vault.read_json(filename)
|
|
assert data["version"] == 1
|
|
|
|
# Update
|
|
data["version"] = 2
|
|
data["updated"] = True
|
|
await vault.write_json(filename, data)
|
|
|
|
# Verify
|
|
updated = await vault.read_json(filename)
|
|
assert updated["version"] == 2
|
|
assert updated["updated"] is True
|
|
|
|
# Delete
|
|
await vault.remove(filename)
|
|
assert not await vault.file_exists(filename)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_9_deep_structure(self):
|
|
"""Test async with 9-deep structure (neopig default)"""
|
|
vault = AsyncVault(
|
|
vaultpath=self.vault_path,
|
|
depth=9,
|
|
salt="test_salt"
|
|
)
|
|
await vault.init()
|
|
|
|
# Simulate domain-based storage
|
|
domains = ["example.com", "test.org", "sample.net"]
|
|
|
|
for domain in domains:
|
|
filename = vault.create_filename(domain, ".json", absolute=True)
|
|
await vault.write_json(filename, {"domain": domain, "crawled": True})
|
|
|
|
# Verify all
|
|
for domain in domains:
|
|
filename = vault.create_filename(domain, ".json", absolute=True)
|
|
data = await vault.read_json(filename)
|
|
assert data["domain"] == domain
|
|
assert data["crawled"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_stats(self):
|
|
"""Test async stats"""
|
|
vault = AsyncVault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
await vault.init()
|
|
|
|
# Store some content
|
|
data = b"test data for stats"
|
|
h = content_hash(data)
|
|
await vault.store(h, data, ".txt")
|
|
|
|
stats = await vault.stats()
|
|
assert stats["count"] == 1
|
|
assert stats["total_size"] == len(data)
|
|
assert stats["cache_enabled"] is True
|