#!/usr/bin/env python3
"""
Simple test script to verify the RAG MCP server functionality.
This script tests the basic functionality without requiring a full MCP client.
"""
import os
import sys
from unittest.mock import Mock, patch
from rag_service import RagService
from rag_tools import RagTools
def test_rag_service():
"""Test RagService basic functionality"""
print("Testing RagService...")
# Mock the requests module to avoid actual HTTP calls
with patch('rag_service.requests') as mock_requests:
# Mock successful response
mock_response = Mock()
mock_response.json.return_value = [{"id": "1", "name": "test content"}]
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
service = RagService()
result = service.get_all_contents("test", "test_user")
print(f"✓ get_all_contents returned: {result}")
assert isinstance(result, list)
# Test search
mock_response.json.return_value = ["search result 1", "search result 2"]
result = service.search_contents("test query", "test_user")
print(f"✓ search_contents returned: {result}")
assert isinstance(result, list)
def test_rag_tools():
"""Test RagTools functionality"""
print("\nTesting RagTools...")
with patch('rag_tools.RagService') as mock_service_class:
mock_service = Mock()
mock_service.get_all_contents.return_value = [{"id": "1", "name": "test"}]
mock_service.search_contents.return_value = ["result1", "result2"]
mock_service_class.return_value = mock_service
tools = RagTools()
# Test list content names
result = tools.list_content_names("test")
print(f"✓ list_content_names returned: {result}")
assert isinstance(result, list)
# Test search
result = tools.search_organization_contents("test query")
print(f"✓ search_organization_contents returned: {result}")
assert isinstance(result, list)
# Test delete (should return success message)
result = tools.delete_organization_content("test_id")
print(f"✓ delete_organization_content returned: {result}")
assert result == ["Content deleted successfully"]
# Test file upload
result = tools.upload_content_file_about_organization("content", "test.txt", "admin")
print(f"✓ upload_content_file_about_organization returned: {result}")
assert result == ["File uploaded successfully"]
# Test URL upload
result = tools.upload_content_url_about_organization("http://example.com", "admin")
print(f"✓ upload_content_url_about_organization returned: {result}")
assert result == ["URL uploaded successfully"]
def test_environment_variables():
"""Test environment variable handling"""
print("\nTesting environment variables...")
# Test default values
os.environ.pop('USER_ID', None)
os.environ.pop('CONTENT_SERVICE_URL', None)
tools = RagTools()
assert tools.user_id_from_environment == 'invalid'
service = RagService()
assert service.content_service_url == 'http://localhost:8080'
# Test custom values
os.environ['USER_ID'] = 'test_user'
os.environ['CONTENT_SERVICE_URL'] = 'http://test:9000'
tools = RagTools()
assert tools.user_id_from_environment == 'test_user'
service = RagService()
assert service.content_service_url == 'http://test:9000'
print("✓ Environment variables handled correctly")
def main():
"""Run all tests"""
print("Running RAG MCP Server Tests")
print("=" * 40)
try:
test_environment_variables()
test_rag_service()
test_rag_tools()
print("\n" + "=" * 40)
print("✅ All tests passed!")
print("\nThe RAG MCP server components are working correctly.")
print("You can now run the server with: python mcp_server.py")
except Exception as e:
print(f"\n❌ Test failed: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
