"""Integration test for the PDF Redaction MCP Server."""
import asyncio
import sys
from pathlib import Path
# Add the src directory to the path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
# Import the server to get access to tool functions
import redact_mcp.server as server_module
class MockContext:
"""Mock context for testing."""
async def info(self, msg):
print(f"[INFO] {msg}")
async def warning(self, msg):
print(f"[WARNING] {msg}")
async def error(self, msg):
print(f"[ERROR] {msg}")
async def test_redaction():
"""Test the PDF redaction workflow."""
# Path to test PDF
test_pdf = Path(__file__).parent / "test_document.pdf"
if not test_pdf.exists():
print(f"Error: Test PDF not found at {test_pdf}")
print("Please run create_test_pdf.py first")
return
ctx = MockContext()
print("=" * 70)
print("PDF Redaction MCP Server - Integration Test")
print("=" * 70)
try:
# Get the actual function implementations from the decorated tools
# FunctionTool has a .fn attribute that holds the original function
load_pdf_func = server_module.load_pdf.fn
redact_text_func = server_module.redact_text.fn
save_redacted_pdf_func = server_module.save_redacted_pdf.fn
list_loaded_pdfs_func = server_module.list_loaded_pdfs.fn
close_pdf_func = server_module.close_pdf.fn
# Step 1: Load the PDF
print("\n1. Loading test PDF...")
content = await load_pdf_func(str(test_pdf), ctx)
print(f" Loaded {len(server_module._loaded_pdfs)} PDF(s)")
print(f" Content preview: {content[:200]}...")
# Step 2: List loaded PDFs
print("\n2. Listing loaded PDFs...")
loaded = await list_loaded_pdfs_func(ctx)
print(f" {loaded}")
# Step 3: Redact SSN
print("\n3. Redacting Social Security Number...")
result = await redact_text_func(str(test_pdf), "123-45-6789", (0, 0, 0), ctx)
print(f" {result}")
# Step 4: Redact "confidential"
print("\n4. Redacting word 'confidential'...")
result = await redact_text_func(str(test_pdf), "confidential", (0, 0, 0), ctx)
print(f" {result}")
# Step 5: Redact account number
print("\n5. Redacting account number...")
result = await redact_text_func(str(test_pdf), "ACCT-9876543210", (0, 0, 0), ctx)
print(f" {result}")
# Step 6: Redact credit card on page 2
print("\n6. Redacting credit card number...")
result = await redact_text_func(str(test_pdf), "4111-1111-1111-1111", (0, 0, 0), ctx)
print(f" {result}")
# Step 7: Save the redacted PDF
print("\n7. Saving redacted PDF...")
result = await save_redacted_pdf_func(str(test_pdf), None, ctx)
print(f" {result}")
# Step 8: Close the PDF
print("\n8. Closing PDF...")
result = await close_pdf_func(str(test_pdf), ctx)
print(f" {result}")
print("\n" + "=" * 70)
print("Integration test completed successfully!")
print("=" * 70)
# Check that the redacted file exists
redacted_path = test_pdf.parent / "test_document_redacted.pdf"
if redacted_path.exists():
print(f"\n✓ Redacted PDF created: {redacted_path}")
print(f" File size: {redacted_path.stat().st_size} bytes")
else:
print(f"\n✗ Redacted PDF not found: {redacted_path}")
except Exception as e:
print(f"\n✗ Test failed with error: {e}")
import traceback
traceback.print_exc()
return False
return True
if __name__ == "__main__":
success = asyncio.run(test_redaction())
sys.exit(0 if success else 1)