test_core_functionality.py•8.43 kB
#!/usr/bin/env python3
"""Test script to validate core functionality without MCP dependencies."""
import asyncio
import tempfile
from pathlib import Path
from src.file_system_mcp_server.config import Config, ConfigManager
from src.file_system_mcp_server.safety import SafetyManager
from src.file_system_mcp_server.file_operations import FileOperationsManager
async def test_core_functionality():
"""Test core functionality without MCP server."""
print("🧪 Testing File System MCP Server Core Functionality...")
# Create temporary workspace
with tempfile.TemporaryDirectory() as temp_dir:
workspace = Path(temp_dir)
print(f"📁 Using workspace: {workspace}")
# Create test configuration
config = Config(
backup_directory=str(workspace / "backups"),
max_file_size=1024 * 1024, # 1MB
max_recursion_depth=5,
protected_paths=["/etc", "/usr"],
enable_backups=True,
log_level="INFO"
)
# Initialize components
safety_manager = SafetyManager(config)
file_ops = FileOperationsManager(config, safety_manager)
print("✅ Components initialized successfully")
# Test file operations
test_file = workspace / "test.txt"
test_content = "Hello, MCP Server!"
# Test write
print("📝 Testing file write...")
write_result = await file_ops.write_file(str(test_file), test_content)
assert write_result.success, f"Write failed: {write_result.error}"
print(f"✅ File written: {write_result.path}")
# Verify file exists
assert test_file.exists(), "File should exist after write"
assert test_file.read_text() == test_content, "File content should match"
# Test read
print("📖 Testing file read...")
read_result = await file_ops.read_file(str(test_file))
assert read_result.success, f"Read failed: {read_result.error}"
assert read_result.content == test_content, "Content mismatch"
print(f"✅ File read successfully: {len(read_result.content)} bytes")
# Test update
print("🔄 Testing file update...")
new_content = "Updated content!"
update_result = await file_ops.update_file(str(test_file), new_content)
assert update_result.success, f"Update failed: {update_result.error}"
print(f"✅ File updated with backup: {update_result.backup_created}")
# Verify update worked
assert test_file.read_text() == new_content, "File should be updated"
# Test directory listing
print("📋 Testing directory listing...")
list_result = await file_ops.list_directory(str(workspace))
assert list_result.success, f"List failed: {list_result.error}"
print(f"✅ Directory listed: {list_result.total_entries} entries")
# Test file info
print("ℹ️ Testing file info...")
info_result = await file_ops.get_file_info(str(test_file))
assert info_result.success, f"Info failed: {info_result.error}"
assert info_result.exists, "File should exist"
print(f"✅ File info retrieved: {info_result.metadata.size} bytes")
# Test safety features
print("🔒 Testing safety features...")
# Test protected path
protected_result = await file_ops.read_file("/etc/passwd")
assert not protected_result.success, "Protected path should be blocked"
assert protected_result.error.error_type == "SecurityError"
print("✅ Protected path blocked successfully")
# Test backup creation
backup_info = safety_manager.get_backup_info(str(test_file))
assert len(backup_info) > 0, "Backup should exist"
print(f"✅ Backup system working: {len(backup_info)} backups")
# Test configuration management
print("⚙️ Testing configuration...")
config_file = workspace / "test_config.json"
ConfigManager.save_config(config, config_file)
loaded_config = ConfigManager.load_config(str(config_file))
assert loaded_config.backup_directory == config.backup_directory
print("✅ Configuration save/load successful")
# Test file deletion
print("🗑️ Testing file deletion...")
delete_result = await file_ops.delete_file(str(test_file))
assert delete_result.success, f"Delete failed: {delete_result.error}"
assert not test_file.exists(), "File should be deleted"
print(f"✅ File deleted with backup: {delete_result.backup_location}")
# Test pattern matching
print("🔍 Testing pattern matching...")
# Create some test files
(workspace / "test1.py").write_text("python file")
(workspace / "test2.txt").write_text("text file")
(workspace / "test3.py").write_text("another python file")
pattern_result = await file_ops.list_directory(str(workspace), pattern="*.py")
assert pattern_result.success, "Pattern listing should succeed"
py_files = [entry for entry in pattern_result.entries if entry.name.endswith(".py")]
assert len(py_files) == 2, f"Should find 2 .py files, found {len(py_files)}"
print("✅ Pattern matching working")
# Test recursive listing
print("📂 Testing recursive listing...")
subdir = workspace / "subdir"
subdir.mkdir()
(subdir / "nested.txt").write_text("nested file")
recursive_result = await file_ops.list_directory(str(workspace), recursive=True)
assert recursive_result.success, "Recursive listing should succeed"
nested_files = [entry for entry in recursive_result.entries if "nested.txt" in entry.name]
assert len(nested_files) == 1, "Should find nested file"
print("✅ Recursive listing working")
print("\n🎉 All core functionality tests passed!")
print("\n📋 Summary:")
print(f" • File operations: ✅ Write, Read, Update, Delete, List, Info")
print(f" • Safety features: ✅ Backups, Path validation, Protected paths")
print(f" • Advanced features: ✅ Pattern matching, Recursive listing")
print(f" • Configuration: ✅ Save, Load, Validate")
print(f" • Error handling: ✅ Security errors, File not found")
print(f" • Workspace: {workspace}")
return True
def test_cli_functionality():
"""Test CLI functionality."""
print("\n🖥️ Testing CLI functionality...")
# Test configuration creation
with tempfile.TemporaryDirectory() as temp_dir:
config_file = Path(temp_dir) / "test_config.json"
# Create example config
from src.file_system_mcp_server.cli_utils import create_example_config
create_example_config(str(config_file))
assert config_file.exists(), "Config file should be created"
print("✅ CLI configuration creation successful")
# Test config validation
from src.file_system_mcp_server.cli_utils import validate_config
result = validate_config(str(config_file))
assert result, "Config validation should succeed"
print("✅ CLI configuration validation successful")
return True
if __name__ == "__main__":
print("🚀 File System MCP Server - Core Functionality Test")
print("=" * 60)
try:
# Test async functionality
asyncio.run(test_core_functionality())
# Test CLI functionality
test_cli_functionality()
print("\n" + "=" * 60)
print("🎊 Core functionality test completed successfully!")
print("\nThe File System MCP Server core is working correctly.")
print("\nTo use with MCP, install the mcp library:")
print(" pip install mcp")
print("\nThen you can run:")
print(" python -m file_system_mcp_server.main --help")
print(" python -m file_system_mcp_server.main --create-example-config config.json")
except Exception as e:
print(f"\n❌ Test failed: {e}")
import traceback
traceback.print_exc()
exit(1)