"""
Tests for the MCP File Operations Server
"""
import pytest
from pathlib import Path
from mcp_file_server.server import FileMCPServer
class TestFileMCPServer:
"""Test cases for the FileMCPServer class."""
@pytest.fixture
def server(self):
"""Create a server instance for testing."""
return FileMCPServer()
@pytest.fixture
def temp_docs_dir(self, tmp_path):
"""Create a temporary documents directory."""
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
return docs_dir
def test_validate_path_within_documents(self, server, tmp_path):
"""Test that path validation allows paths within documents directory."""
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
server_module.DOCUMENTS_DIR = docs_dir
try:
# Test valid paths
valid_path = server._validate_path("test.txt")
assert valid_path == docs_dir / "test.txt"
valid_subdir_path = server._validate_path("subdir/file.txt")
assert valid_subdir_path == docs_dir / "subdir" / "file.txt"
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
def test_validate_path_outside_documents(self, server, tmp_path):
"""Test that path validation rejects paths outside documents directory."""
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
server_module.DOCUMENTS_DIR = docs_dir
try:
# Test invalid paths
with pytest.raises(
ValueError, match="outside the allowed documents directory"
):
server._validate_path("../../../etc/passwd")
with pytest.raises(
ValueError, match="outside the allowed documents directory"
):
server._validate_path("..\\..\\..\\windows\\system32\\config\\sam")
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
@pytest.mark.asyncio
class TestAsyncFileOperations:
"""Test cases for async file operations."""
@pytest.fixture
def server(self):
"""Create a server instance for testing."""
return FileMCPServer()
async def test_write_and_read_file(self, server, tmp_path):
"""Test writing and reading a file."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Write a file
write_result = await server.write_file(
{"file_path": "test.txt", "content": "Hello, World!"}
)
assert "Content written to file: test.txt" in write_result.content[0].text
# Read the file
read_result = await server.read_file({"file_path": "test.txt"})
assert "Hello, World!" in read_result.content[0].text
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_list_files(self, server, tmp_path):
"""Test listing files."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Create some test files
(docs_dir / "file1.txt").write_text("content1")
(docs_dir / "file2.py").write_text("content2")
(docs_dir / "subdir").mkdir()
(docs_dir / "subdir" / "file3.md").write_text("content3")
# List files
list_result = await server.list_files({})
assert "📄 file1.txt" in list_result.content[0].text
assert "📄 file2.py" in list_result.content[0].text
assert "📁 subdir/" in list_result.content[0].text
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_create_directory(self, server, tmp_path):
"""Test creating directories."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Create a directory
create_result = await server.create_directory({"dir_path": "new_folder"})
assert "Directory created: new_folder" in create_result.content[0].text
assert (docs_dir / "new_folder").exists()
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_delete_file(self, server, tmp_path):
"""Test deleting files."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Create a test file
test_file = docs_dir / "test_delete.txt"
test_file.write_text("content")
# Delete the file
delete_result = await server.delete_file({"file_path": "test_delete.txt"})
assert "File deleted: test_delete.txt" in delete_result.content[0].text
assert not test_file.exists()
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_delete_directory(self, server, tmp_path):
"""Test deleting directories recursively."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Create a test directory structure
test_dir = docs_dir / "test_folder"
test_dir.mkdir()
(test_dir / "file1.txt").write_text("content1")
(test_dir / "file2.py").write_text("content2")
subdir = test_dir / "subdir"
subdir.mkdir()
(subdir / "file3.md").write_text("content3")
# Delete the directory recursively
delete_result = await server.delete_directory({"dir_path": "test_folder"})
assert "deleted successfully" in delete_result.content[0].text
assert "items removed" in delete_result.content[0].text
assert not test_dir.exists()
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_delete_directory_protection(self, server, tmp_path):
"""Test that the documents directory itself cannot be deleted."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Try to delete the documents directory itself
delete_result = await server.delete_directory({"dir_path": ""})
assert "Cannot delete the documents directory" in delete_result.content[0].text
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_read_pdf(self, server, tmp_path):
"""Test reading PDF files."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Create a simple text file to test non-PDF handling
test_file = docs_dir / "test.txt"
test_file.write_text("This is not a PDF")
# Test that non-PDF files are rejected
result = await server.read_pdf({"file_path": "test.txt"})
assert "File is not a PDF" in result.content[0].text
# Test that non-existent files are handled
result = await server.read_pdf({"file_path": "nonexistent.pdf"})
assert "PDF file not found" in result.content[0].text
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
async def test_get_pdf_info(self, server, tmp_path):
"""Test getting PDF information."""
# Temporarily set the documents directory
import mcp_file_server.server as server_module
original_docs_dir = server_module.DOCUMENTS_DIR
docs_dir = tmp_path / "documents"
docs_dir.mkdir()
server_module.DOCUMENTS_DIR = docs_dir
try:
# Test that non-PDF files are rejected
test_file = docs_dir / "test.txt"
test_file.write_text("This is not a PDF")
result = await server.get_pdf_info({"file_path": "test.txt"})
assert "File is not a PDF" in result.content[0].text
finally:
server_module.DOCUMENTS_DIR = original_docs_dir
if __name__ == "__main__":
pytest.main([__file__])
