865 lines
28 KiB
Python
865 lines
28 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 FileVault 2.0.0
|
|
Tests content-addressable and seed-based storage with hex pair directories
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path to import modules
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
from neopig.filevault import (
|
|
Vault,
|
|
create_vault,
|
|
ensure_bytes,
|
|
content_hash,
|
|
hash_to_path,
|
|
__version__,
|
|
)
|
|
|
|
|
|
class TestFileVaultBasics:
|
|
"""Test basic Vault functionality"""
|
|
|
|
def setup_method(self):
|
|
"""Setup test environment"""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
|
|
def teardown_method(self):
|
|
"""Cleanup test environment"""
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_vault_initialization_default_params(self):
|
|
"""Test Vault initialization with default parameters"""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
|
|
assert vault.vaultpath == Path(self.vault_path)
|
|
assert vault.depth == 9
|
|
assert vault.salt == b"neopig"
|
|
assert vault.enable_memory_cache is False
|
|
|
|
def test_vault_initialization_custom_params(self):
|
|
"""Test Vault initialization with custom parameters"""
|
|
custom_salt = "test_salt_123"
|
|
custom_depth = 5
|
|
|
|
vault = Vault(vaultpath=self.vault_path, depth=custom_depth, salt=custom_salt)
|
|
|
|
assert vault.vaultpath == Path(self.vault_path)
|
|
assert vault.depth == custom_depth
|
|
assert vault.salt == b"test_salt_123"
|
|
|
|
def test_vault_init_creates_directory(self):
|
|
"""Test that init() creates the base directory"""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
assert os.path.exists(self.vault_path)
|
|
|
|
def test_vault_directory_already_exists(self):
|
|
"""Test vault initialization when directory already exists"""
|
|
os.makedirs(self.vault_path)
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
vault.init()
|
|
assert os.path.exists(self.vault_path)
|
|
|
|
|
|
class TestContentAddressableStorage:
|
|
"""Test content-addressable storage (by hash)"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path, depth=3)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_store_and_get(self):
|
|
"""Test storing and retrieving data by hash"""
|
|
data = b"Hello, World!"
|
|
h = content_hash(data)
|
|
|
|
path = self.vault.store(h, data, ".txt")
|
|
assert path.exists()
|
|
|
|
retrieved = self.vault.get(h)
|
|
assert retrieved == data
|
|
|
|
def test_store_creates_directory_structure(self):
|
|
"""Test that store creates hex pair directories"""
|
|
data = b"test data"
|
|
h = content_hash(data) # e.g., "abcdef123..."
|
|
|
|
path = self.vault.store(h, data, ".txt")
|
|
|
|
# Should have depth levels of 2-char hex dirs
|
|
relative = path.relative_to(self.vault.vaultpath)
|
|
parts = list(relative.parts)
|
|
|
|
# depth=3 means 3 directory levels + filename
|
|
assert len(parts) == 4
|
|
# Each dir should be 2 hex chars
|
|
for i in range(3):
|
|
assert len(parts[i]) == 2
|
|
|
|
def test_exists(self):
|
|
"""Test checking if content exists"""
|
|
data = b"unique data"
|
|
h = content_hash(data)
|
|
|
|
assert self.vault.exists(h) is False
|
|
self.vault.store(h, data, ".bin")
|
|
assert self.vault.exists(h) is True
|
|
|
|
def test_get_path(self):
|
|
"""Test getting path for stored content"""
|
|
data = b"path test"
|
|
h = content_hash(data)
|
|
|
|
assert self.vault.get_path(h) is None
|
|
stored_path = self.vault.store(h, data, ".txt")
|
|
retrieved_path = self.vault.get_path(h)
|
|
assert retrieved_path == stored_path
|
|
|
|
def test_delete(self):
|
|
"""Test deleting content"""
|
|
data = b"delete me"
|
|
h = content_hash(data)
|
|
|
|
self.vault.store(h, data, ".txt")
|
|
assert self.vault.exists(h)
|
|
|
|
result = self.vault.delete(h)
|
|
assert result is True
|
|
assert self.vault.exists(h) is False
|
|
|
|
def test_delete_nonexistent(self):
|
|
"""Test deleting non-existent content"""
|
|
result = self.vault.delete("nonexistent" * 2)
|
|
assert result is False
|
|
|
|
def test_get_nonexistent(self):
|
|
"""Test getting non-existent content"""
|
|
result = self.vault.get("nonexistent" * 2)
|
|
assert result is None
|
|
|
|
|
|
class TestSeedBasedStorage:
|
|
"""Test seed-based deterministic paths"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path, depth=3, salt="test_salt")
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_create_filename_deterministic(self):
|
|
"""Test deterministic filename creation"""
|
|
seed = "test_seed_123"
|
|
|
|
filename1 = self.vault.create_filename(seed)
|
|
filename2 = self.vault.create_filename(seed)
|
|
|
|
assert filename1 == filename2
|
|
|
|
def test_create_filename_different_seeds(self):
|
|
"""Test different seeds produce different filenames"""
|
|
filename1 = self.vault.create_filename("seed1")
|
|
filename2 = self.vault.create_filename("seed2")
|
|
|
|
assert filename1 != filename2
|
|
|
|
def test_create_filename_with_extension(self):
|
|
"""Test filename creation with extensions"""
|
|
seed = "test_seed"
|
|
|
|
# Test extension with dot
|
|
filename_with_dot = self.vault.create_filename(seed, ".json")
|
|
assert filename_with_dot.endswith(".json")
|
|
|
|
# Test extension without dot
|
|
filename_without_dot = self.vault.create_filename(seed, "json")
|
|
assert filename_without_dot.endswith(".json")
|
|
|
|
def test_create_filename_relative_vs_absolute(self):
|
|
"""Test relative vs absolute path generation"""
|
|
seed = "test_seed"
|
|
|
|
relative_path = self.vault.create_filename(seed, ".json", absolute=False)
|
|
absolute_path = self.vault.create_filename(seed, ".json", absolute=True)
|
|
|
|
# Relative path should not start with vault path
|
|
assert not relative_path.startswith(self.vault_path)
|
|
|
|
# Absolute path should start with vault path
|
|
assert absolute_path.startswith(self.vault_path)
|
|
|
|
def test_create_random_filename_uniqueness(self):
|
|
"""Test random filename generation produces unique results"""
|
|
filenames = set()
|
|
for i in range(10):
|
|
filename = self.vault.create_random_filename(".txt")
|
|
filenames.add(filename)
|
|
|
|
assert len(filenames) == 10
|
|
|
|
def test_salt_affects_output(self):
|
|
"""Test that salt affects filename generation"""
|
|
seed = "constant_seed"
|
|
salts = ["salt1", "salt2", "different"]
|
|
filenames = []
|
|
|
|
for salt in salts:
|
|
vault = Vault(vaultpath=self.vault_path, salt=salt)
|
|
filenames.append(vault.create_filename(seed))
|
|
|
|
assert len(set(filenames)) == len(filenames)
|
|
|
|
|
|
class TestDirectoryStructure:
|
|
"""Test hex pair directory structure"""
|
|
|
|
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_depth_9_structure(self):
|
|
"""Test 9-deep directory structure (neopig default)"""
|
|
vault = Vault(vaultpath=self.vault_path, depth=9)
|
|
|
|
filename = vault.create_filename("test_seed", absolute=True)
|
|
relative_part = filename.replace(self.vault_path + "/", "")
|
|
path_parts = relative_part.split("/")
|
|
|
|
# 9 directories + filename
|
|
assert len(path_parts) == 10
|
|
|
|
# All directory parts should be 2 hex chars
|
|
for i in range(9):
|
|
assert len(path_parts[i]) == 2
|
|
assert all(c in "0123456789abcdef" for c in path_parts[i])
|
|
|
|
def test_depth_3_structure(self):
|
|
"""Test 3-deep directory structure"""
|
|
vault = Vault(vaultpath=self.vault_path, depth=3)
|
|
|
|
filename = vault.create_filename("test", absolute=True)
|
|
relative_part = filename.replace(self.vault_path + "/", "")
|
|
path_parts = relative_part.split("/")
|
|
|
|
assert len(path_parts) == 4
|
|
|
|
def test_hash_to_path_function(self):
|
|
"""Test the hash_to_path utility function"""
|
|
h = "abcdef1234567890" * 4 # 64 char hash
|
|
|
|
path = hash_to_path(h, depth=3, ext=".txt")
|
|
parts = list(path.parts)
|
|
|
|
# ab/cd/ef/hash.txt
|
|
assert parts[0] == "ab"
|
|
assert parts[1] == "cd"
|
|
assert parts[2] == "ef"
|
|
assert parts[3] == f"{h}.txt"
|
|
|
|
|
|
class TestTextFileOperations:
|
|
"""Test text file read/write operations"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_write_and_read_text(self):
|
|
"""Test basic text file write and read"""
|
|
test_file = self.vault.create_filename("text_test", ".txt", absolute=True)
|
|
test_content = "Hello, World!\nThis is a test."
|
|
|
|
self.vault.write_text(test_file, test_content)
|
|
read_content = self.vault.read_text(test_file)
|
|
|
|
assert read_content == test_content
|
|
|
|
def test_write_text_creates_directories(self):
|
|
"""Test that write_text creates necessary directories"""
|
|
vault = Vault(vaultpath=self.vault_path, depth=5)
|
|
|
|
test_file = vault.create_filename("deep_test", ".txt", absolute=True)
|
|
vault.write_text(test_file, "Deep content")
|
|
|
|
assert os.path.exists(test_file)
|
|
|
|
def test_read_text_with_default(self):
|
|
"""Test read_text returns default for non-existent file"""
|
|
non_existent = self.vault.create_filename("nonexistent", ".txt", absolute=True)
|
|
|
|
result = self.vault.read_text(non_existent, default="default_content")
|
|
assert result == "default_content"
|
|
|
|
def test_read_text_returns_none_without_default(self):
|
|
"""Test read_text returns None for non-existent file"""
|
|
non_existent = self.vault.create_filename("nonexistent", ".txt", absolute=True)
|
|
|
|
result = self.vault.read_text(non_existent)
|
|
assert result is None
|
|
|
|
def test_write_text_unicode(self):
|
|
"""Test write/read with unicode content"""
|
|
test_file = self.vault.create_filename("unicode_test", ".txt", absolute=True)
|
|
test_content = "Unicode: café, naïve, 日本語"
|
|
|
|
self.vault.write_text(test_file, test_content)
|
|
read_content = self.vault.read_text(test_file)
|
|
|
|
assert read_content == test_content
|
|
|
|
def test_backward_compat_aliases(self):
|
|
"""Test backward compatibility aliases"""
|
|
test_file = self.vault.create_filename("alias_test", ".txt", absolute=True)
|
|
|
|
self.vault.write_text_file(test_file, "content")
|
|
result = self.vault.read_text_file(test_file)
|
|
|
|
assert result == "content"
|
|
|
|
|
|
class TestBytesOperations:
|
|
"""Test bytes read/write operations"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_write_and_read_bytes(self):
|
|
"""Test basic bytes write and read"""
|
|
test_file = self.vault.create_filename("bytes_test", ".bin", absolute=True)
|
|
test_data = b"\x00\x01\x02\xff\xfe\xfd"
|
|
|
|
self.vault.write_bytes(test_file, test_data)
|
|
read_data = self.vault.read_bytes(test_file)
|
|
|
|
assert read_data == test_data
|
|
|
|
def test_read_bytes_nonexistent(self):
|
|
"""Test read_bytes returns None for non-existent file"""
|
|
result = self.vault.read_bytes("/nonexistent/path")
|
|
assert result is None
|
|
|
|
|
|
class TestJSONOperations:
|
|
"""Test JSON file operations"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_write_and_read_json(self):
|
|
"""Test atomic JSON write functionality"""
|
|
test_file = self.vault.create_filename("test_json", ".json", absolute=True)
|
|
test_data = {"key": "value", "number": 42, "nested": {"a": 1}}
|
|
|
|
self.vault.write_json(test_file, test_data)
|
|
|
|
data = self.vault.read_json(test_file)
|
|
assert data == test_data
|
|
|
|
def test_write_json_with_list(self):
|
|
"""Test atomic JSON write with list data"""
|
|
test_file = self.vault.create_filename("test_list", ".json", absolute=True)
|
|
test_data = [1, 2, 3, {"key": "value"}]
|
|
|
|
self.vault.write_json(test_file, test_data)
|
|
|
|
data = self.vault.read_json(test_file)
|
|
assert data == test_data
|
|
|
|
def test_write_json_custom_indent(self):
|
|
"""Test atomic JSON write with custom indent"""
|
|
test_file = self.vault.create_filename("indented", ".json", absolute=True)
|
|
test_data = {"key": "value"}
|
|
|
|
self.vault.write_json(test_file, test_data, indent=4)
|
|
|
|
with open(test_file, "r") as f:
|
|
content = f.read()
|
|
assert " " in content
|
|
|
|
def test_read_json_with_default(self):
|
|
"""Test JSON read with default value"""
|
|
non_existent = self.vault.create_filename("nonexistent", ".json", absolute=True)
|
|
|
|
result = self.vault.read_json(non_existent, default={"default": True})
|
|
assert result == {"default": True}
|
|
|
|
def test_read_json_invalid_returns_default(self):
|
|
"""Test read_json returns default for invalid JSON"""
|
|
test_file = self.vault.create_filename("invalid", ".json", absolute=True)
|
|
os.makedirs(os.path.dirname(test_file), exist_ok=True)
|
|
|
|
with open(test_file, "w") as f:
|
|
f.write("not valid json {{{")
|
|
|
|
result = self.vault.read_json(test_file, default={"fallback": True})
|
|
assert result == {"fallback": True}
|
|
|
|
def test_backward_compat_aliases(self):
|
|
"""Test backward compatibility aliases"""
|
|
test_file = self.vault.create_filename("compat", ".json", absolute=True)
|
|
|
|
self.vault.atomic_write_json(test_file, {"test": 1})
|
|
result = self.vault.safe_read_json(test_file)
|
|
|
|
assert result == {"test": 1}
|
|
|
|
|
|
class TestMemoryCache:
|
|
"""Test memory 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 that memory cache is disabled by default"""
|
|
vault = Vault(vaultpath=self.vault_path)
|
|
|
|
assert vault.enable_memory_cache is False
|
|
assert vault._cache is None
|
|
|
|
def test_cache_enabled(self):
|
|
"""Test enabling memory cache"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
|
|
assert vault.enable_memory_cache is True
|
|
assert vault._cache == {}
|
|
|
|
def test_file_exists_uses_cache(self):
|
|
"""Test that file_exists uses memory cache"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault.init()
|
|
|
|
test_file = vault.create_filename("cache_test", ".txt", absolute=True)
|
|
os.makedirs(os.path.dirname(test_file), exist_ok=True)
|
|
with open(test_file, "w") as f:
|
|
f.write("test")
|
|
|
|
# First call caches result
|
|
assert vault.file_exists(test_file) is True
|
|
assert test_file in vault._cache
|
|
|
|
# Delete file, cache still returns True
|
|
os.remove(test_file)
|
|
assert vault.file_exists(test_file) is True
|
|
|
|
def test_write_updates_cache(self):
|
|
"""Test that writes update the cache"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault.init()
|
|
|
|
test_file = vault.create_filename("write_cache", ".txt", absolute=True)
|
|
vault.write_text(test_file, "content")
|
|
|
|
assert vault._cache[test_file] is True
|
|
|
|
def test_remove_updates_cache(self):
|
|
"""Test that remove updates the cache"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault.init()
|
|
|
|
test_file = vault.create_filename("remove_cache", ".txt", absolute=True)
|
|
vault.write_text(test_file, "content")
|
|
vault.remove(test_file)
|
|
|
|
assert vault._cache[test_file] is False
|
|
|
|
def test_clear_cache(self):
|
|
"""Test clearing the cache"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault._cache["path1"] = True
|
|
vault._cache["path2"] = False
|
|
|
|
vault.clear_cache()
|
|
|
|
assert vault._cache == {}
|
|
|
|
def test_mark_file_exists(self):
|
|
"""Test mark_file_exists (backward compat)"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
|
|
vault.mark_file_exists("/path", True)
|
|
assert vault._cache["/path"] is True
|
|
|
|
|
|
class TestFileRemoval:
|
|
"""Test file removal operations"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_remove_file(self):
|
|
"""Test basic file removal"""
|
|
test_file = self.vault.create_filename("to_remove", ".txt", absolute=True)
|
|
os.makedirs(os.path.dirname(test_file), exist_ok=True)
|
|
with open(test_file, "w") as f:
|
|
f.write("test")
|
|
|
|
result = self.vault.remove(test_file)
|
|
|
|
assert result is True
|
|
assert not os.path.exists(test_file)
|
|
|
|
def test_remove_nonexistent(self):
|
|
"""Test removing non-existent file returns False"""
|
|
test_file = self.vault.create_filename("nonexistent", ".txt", absolute=True)
|
|
|
|
result = self.vault.remove(test_file)
|
|
assert result is False
|
|
|
|
def test_remove_file_alias(self):
|
|
"""Test remove_file backward compat alias"""
|
|
test_file = self.vault.create_filename("alias_remove", ".txt", absolute=True)
|
|
os.makedirs(os.path.dirname(test_file), exist_ok=True)
|
|
with open(test_file, "w") as f:
|
|
f.write("test")
|
|
|
|
result = self.vault.remove_file(test_file)
|
|
assert result is True
|
|
|
|
|
|
class TestFileLocking:
|
|
"""Test file locking functionality"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_file_lock_write(self):
|
|
"""Test file_lock creates directories for write operations"""
|
|
test_file = self.vault.create_filename("lock_test", ".txt", absolute=True)
|
|
|
|
with self.vault.file_lock(test_file, "w") as f:
|
|
f.write("locked write")
|
|
|
|
assert os.path.exists(test_file)
|
|
|
|
def test_file_lock_read(self):
|
|
"""Test file_lock for reading existing file"""
|
|
test_file = self.vault.create_filename("read_lock", ".txt", absolute=True)
|
|
os.makedirs(os.path.dirname(test_file), exist_ok=True)
|
|
with open(test_file, "w") as f:
|
|
f.write("content")
|
|
|
|
with self.vault.file_lock(test_file, "r") as f:
|
|
content = f.read()
|
|
|
|
assert content == "content"
|
|
|
|
def test_file_lock_append(self):
|
|
"""Test file_lock in append mode"""
|
|
test_file = self.vault.create_filename("append_lock", ".txt", absolute=True)
|
|
|
|
with self.vault.file_lock(test_file, "w") as f:
|
|
f.write("line1\n")
|
|
|
|
with self.vault.file_lock(test_file, "a") as f:
|
|
f.write("line2\n")
|
|
|
|
with open(test_file, "r") as f:
|
|
content = f.read()
|
|
|
|
assert content == "line1\nline2\n"
|
|
|
|
|
|
class TestVersionAndCompatibility:
|
|
"""Test version info and backwards compatibility"""
|
|
|
|
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_version_info(self):
|
|
"""Test that version information is available"""
|
|
from neopig import filevault
|
|
|
|
assert hasattr(filevault, "__version__")
|
|
assert filevault.__version__ == "2.0.0"
|
|
assert hasattr(filevault, "__author__")
|
|
|
|
def test_factory_function(self):
|
|
"""Test create_vault factory function"""
|
|
vault = create_vault(vaultpath=self.vault_path, salt="factory_test")
|
|
assert isinstance(vault, Vault)
|
|
assert vault.salt == b"factory_test"
|
|
|
|
def test_ensure_bytes(self):
|
|
"""Test ensure_bytes function"""
|
|
result = ensure_bytes("test_string")
|
|
assert isinstance(result, bytes)
|
|
assert result == b"test_string"
|
|
|
|
byte_input = b"test_bytes"
|
|
result2 = ensure_bytes(byte_input)
|
|
assert result2 == byte_input
|
|
|
|
def test_content_hash(self):
|
|
"""Test content_hash function"""
|
|
data = b"test data"
|
|
h = content_hash(data)
|
|
|
|
assert len(h) == 32 # MD5 hex
|
|
assert all(c in "0123456789abcdef" for c in h)
|
|
|
|
|
|
class TestStatistics:
|
|
"""Test vault statistics"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_stats_empty(self):
|
|
"""Test stats on empty vault"""
|
|
stats = self.vault.stats()
|
|
|
|
assert stats["path"] == self.vault_path
|
|
assert stats["depth"] == 9
|
|
assert stats["count"] == 0
|
|
assert stats["total_size"] == 0
|
|
assert stats["cache_enabled"] is True
|
|
assert stats["cache_entries"] == 0
|
|
|
|
def test_stats_with_files(self):
|
|
"""Test stats with files"""
|
|
data = b"test data for stats"
|
|
h = content_hash(data)
|
|
self.vault.store(h, data, ".txt")
|
|
|
|
stats = self.vault.stats()
|
|
assert stats["count"] == 1
|
|
assert stats["total_size"] == len(data)
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration tests for complete 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)
|
|
|
|
def test_full_json_workflow(self):
|
|
"""Test complete JSON workflow: create, read, update, delete"""
|
|
vault = Vault(vaultpath=self.vault_path, enable_memory_cache=True)
|
|
vault.init()
|
|
|
|
test_file = vault.create_filename("workflow_test", ".json", absolute=True)
|
|
|
|
# Create
|
|
vault.write_json(test_file, {"version": 1, "data": "initial"})
|
|
assert vault.file_exists(test_file)
|
|
|
|
# Read
|
|
data = vault.read_json(test_file)
|
|
assert data["version"] == 1
|
|
|
|
# Update
|
|
data["version"] = 2
|
|
vault.write_json(test_file, data)
|
|
|
|
# Verify
|
|
updated = vault.read_json(test_file)
|
|
assert updated["version"] == 2
|
|
|
|
# Delete
|
|
vault.remove(test_file)
|
|
assert not vault.file_exists(test_file)
|
|
|
|
def test_content_addressable_workflow(self):
|
|
"""Test content-addressable storage workflow"""
|
|
vault = Vault(vaultpath=self.vault_path, depth=5)
|
|
vault.init()
|
|
|
|
# Store multiple items
|
|
items = [b"item1", b"item2", b"item3"]
|
|
hashes = []
|
|
|
|
for item in items:
|
|
h = content_hash(item)
|
|
hashes.append(h)
|
|
vault.store(h, item, ".bin")
|
|
|
|
# Retrieve all
|
|
for i, h in enumerate(hashes):
|
|
data = vault.get(h)
|
|
assert data == items[i]
|
|
|
|
# Delete one
|
|
vault.delete(hashes[1])
|
|
assert vault.exists(hashes[0])
|
|
assert not vault.exists(hashes[1])
|
|
assert vault.exists(hashes[2])
|
|
|
|
def test_deterministic_across_sessions(self):
|
|
"""Test filenames are deterministic across vault instances"""
|
|
seed = "consistent_seed"
|
|
salt = "consistent_salt"
|
|
|
|
vault1 = Vault(vaultpath=self.vault_path, salt=salt)
|
|
filename1 = vault1.create_filename(seed)
|
|
|
|
vault2 = Vault(vaultpath=self.vault_path, salt=salt)
|
|
filename2 = vault2.create_filename(seed)
|
|
|
|
assert filename1 == filename2
|
|
|
|
def test_multiple_vaults_same_path(self):
|
|
"""Test multiple vault instances on same path with different salts"""
|
|
vault1 = Vault(vaultpath=self.vault_path, salt="vault1")
|
|
vault1.init()
|
|
vault2 = Vault(vaultpath=self.vault_path, salt="vault2")
|
|
|
|
# Same seed, different salts = different files
|
|
file1 = vault1.create_filename("shared_seed", ".json", absolute=True)
|
|
file2 = vault2.create_filename("shared_seed", ".json", absolute=True)
|
|
|
|
assert file1 != file2
|
|
|
|
vault1.write_json(file1, {"vault": 1})
|
|
vault2.write_json(file2, {"vault": 2})
|
|
|
|
assert vault1.read_json(file1) == {"vault": 1}
|
|
assert vault2.read_json(file2) == {"vault": 2}
|
|
|
|
|
|
class TestSecurityAndEdgeCases:
|
|
"""Test security and edge cases"""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_path = os.path.join(self.temp_dir, "test_vault")
|
|
self.vault = Vault(vaultpath=self.vault_path)
|
|
self.vault.init()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_path_traversal_prevented(self):
|
|
"""Test that path traversal attacks are prevented"""
|
|
malicious_seeds = [
|
|
"../../../etc/passwd",
|
|
"..\\..\\windows\\system32",
|
|
"/absolute/path/attack",
|
|
"seed/with/slashes",
|
|
]
|
|
|
|
for seed in malicious_seeds:
|
|
filename = self.vault.create_filename(seed)
|
|
assert "../" not in filename
|
|
assert "etc/passwd" not in filename
|
|
|
|
def test_handles_unicode_seed(self):
|
|
"""Test vault handles Unicode characters in seed"""
|
|
unicode_seed = "test_unicode_日本語_emoji"
|
|
|
|
filename = self.vault.create_filename(unicode_seed)
|
|
assert isinstance(filename, str)
|
|
assert len(filename) > 0
|
|
|
|
def test_handles_empty_seed(self):
|
|
"""Test vault handles empty seed"""
|
|
filename = self.vault.create_filename("")
|
|
assert isinstance(filename, str)
|
|
assert len(filename) > 0
|
|
|
|
def test_hash_length_consistency(self):
|
|
"""Test that generated hashes have consistent length"""
|
|
seeds = ["short", "much_longer_seed_text", "", "x" * 1000]
|
|
|
|
for seed in seeds:
|
|
filename = self.vault.create_filename(seed)
|
|
hash_part = filename.split("/")[-1]
|
|
# SHA256 = 64 hex chars
|
|
assert len(hash_part) == 64
|