MCP Localization Project

#!/usr/bin/env python3 """ MCP-OS Integration Test & Verification System - MCP-OS 아키텍처 통합 테스트 MCP-OS 전체 시스템의 통합 테스트 및 검증 시스템 이 모듈은 MCP-OS 아키텍처의 모든 구성 요소를 통합 테스트합니다: - test_research_first_policy_enforcement: Research-First 정책 100% 강제 시행 검증 - test_local_mcp_modules_integration: 4개 로컬 MCP 모듈 완전 통합 검증 - test_performance_targets_validation: 성능 목표 달성 검증 (91% 압축률, 93.8%+ 성공률) - test_shrimp_backbone_integration: Shrimp Task Manager 백본 시스템 무결점 연동 검증 MCP-OS Integration Test는 완전 로컬화된 MCP 아키텍처의 모든 기능이 올바르게 작동하는지 종합적으로 검증합니다. """ import os import sys import json import time import asyncio import logging import traceback from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass, field from enum import Enum from datetime import datetime, timedelta from pathlib import Path # 로컬 MCP 모듈 임포트 (통합 테스트 대상) try: from mcp_orchestrator import MCPOSLocalOrchestrator from research_first_engine import ResearchFirstPolicyEngine, PolicyViolationType, ResearchRequirementLevel from settings_policy_manager import MCPOSSettingsPolicyManager from shrimp_execution import ExecutionControlManager, ExecutionStatus, ResearchPhase print("✅ 로컬 MCP 모듈 임포트 완료") except ImportError as e: print(f"❌ 모듈 임포트 실패: {e}") sys.exit(1) # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============================================================================= # 통합 테스트 관련 데이터 클래스 (Integration Test Data Classes) # ============================================================================= class TestStatus(Enum): """테스트 상태""" PENDING = "pending" # 대기 중 RUNNING = "running" # 실행 중 PASSED = "passed" # 통과 FAILED = "failed" # 실패 SKIPPED = "skipped" # 건너뜀 ERROR = "error" # 오류 class TestCategory(Enum): """테스트 카테고리""" POLICY_ENFORCEMENT = "policy_enforcement" # 정책 강제 시행 MODULE_INTEGRATION = "module_integration" # 모듈 통합 PERFORMANCE_VALIDATION = "performance_validation" # 성능 검증 BACKBONE_INTEGRATION = "backbone_integration" # 백본 통합 END_TO_END = "end_to_end" # 종단간 테스트 @dataclass class TestResult: """테스트 결과""" test_id: str # 테스트 ID test_name: str # 테스트 이름 category: TestCategory # 테스트 카테고리 status: TestStatus # 테스트 상태 success: bool # 성공 여부 score: float # 점수 (0.0-100.0) execution_time: float # 실행 시간 (초) details: List[str] = field(default_factory=list) # 상세 결과 errors: List[str] = field(default_factory=list) # 오류 목록 metrics: Dict[str, Any] = field(default_factory=dict) # 메트릭 started_at: datetime = field(default_factory=datetime.now) completed_at: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'test_id': self.test_id, 'test_name': self.test_name, 'category': self.category.value, 'status': self.status.value, 'success': self.success, 'score': self.score, 'execution_time': self.execution_time, 'details': self.details, 'errors': self.errors, 'metrics': self.metrics, 'started_at': self.started_at.isoformat(), 'completed_at': self.completed_at.isoformat() if self.completed_at else None } @dataclass class IntegrationTestSuite: """통합 테스트 스위트""" suite_id: str # 스위트 ID suite_name: str # 스위트 이름 test_results: List[TestResult] = field(default_factory=list) # 테스트 결과들 overall_success: bool = False # 전체 성공 여부 overall_score: float = 0.0 # 전체 점수 total_execution_time: float = 0.0 # 총 실행 시간 started_at: datetime = field(default_factory=datetime.now) completed_at: Optional[datetime] = None def calculate_overall_metrics(self): """전체 메트릭 계산""" if not self.test_results: return # 전체 점수 계산 (평균) total_score = sum(result.score for result in self.test_results) self.overall_score = total_score / len(self.test_results) # 전체 성공 여부 (모든 테스트 통과) self.overall_success = all(result.success for result in self.test_results) # 총 실행 시간 self.total_execution_time = sum(result.execution_time for result in self.test_results) # 완료 시간 설정 self.completed_at = datetime.now() # ============================================================================= # MCP-OS Integration Test Manager 메인 클래스 (Main Test Manager Class) # ============================================================================= class MCPOSIntegrationTestManager: """MCP-OS 통합 테스트 관리자 MCP-OS 아키텍처의 모든 구성 요소를 통합 테스트하여 완전한 로컬화 시스템이 올바르게 작동하는지 검증합니다. """ def __init__(self, test_data_dir: str = "test_data"): """통합 테스트 관리자 초기화""" self.test_data_dir = Path(test_data_dir) self.test_data_dir.mkdir(exist_ok=True) # 테스트 대상 시스템 초기화 self.mcp_orchestrator: Optional[MCPOSLocalOrchestrator] = None self.research_engine: Optional[ResearchFirstPolicyEngine] = None self.settings_manager: Optional[MCPOSSettingsPolicyManager] = None self.execution_manager: Optional[ExecutionControlManager] = None # 테스트 결과 저장소 self.test_suites: Dict[str, IntegrationTestSuite] = {} # 성능 목표 설정 (project_plan.md 기준) self.performance_targets = { 'compression_rate': 0.91, # 91% 압축률 'success_rate': 0.938, # 93.8% 성공률 'response_time_ms': 100, # 100ms 이하 응답시간 'policy_enforcement': 1.0 # 100% 정책 강제 시행 } logger.info("MCP-OS 통합 테스트 관리자 초기화 완료") # ========================================================================= # 핵심 통합 테스트 메서드 (Core Integration Test Methods) # ========================================================================= async def test_research_first_policy_enforcement(self) -> TestResult: """Research-First 정책 100% 강제 시행 검증 Research-First Policy Engine이 모든 추측 기반 작업을 완전히 차단하고 정보 수집을 강제하는지 검증합니다. Returns: TestResult: 정책 강제 시행 테스트 결과 """ test_id = "POLICY-ENFORCEMENT-001" logger.info(f"🔒 Research-First 정책 강제 시행 테스트 시작: {test_id}") test_result = TestResult( test_id=test_id, test_name="Research-First Policy 100% Enforcement Test", category=TestCategory.POLICY_ENFORCEMENT, status=TestStatus.RUNNING, success=False, score=0.0, execution_time=0.0 ) start_time = time.time() try: # 1. Research-First Policy Engine 초기화 if not self.research_engine: self.research_engine = ResearchFirstPolicyEngine() test_result.details.append("✅ Research-First Policy Engine 초기화 완료") # 2. 추측 기반 작업 차단 테스트 assumption_test_cases = [ "I think this should work", "Let me assume that the API endpoint is correct", "Probably the configuration is fine", "Maybe we can guess the parameter values", "Without checking, I'll proceed with this approach" ] blocked_count = 0 for i, test_case in enumerate(assumption_test_cases): is_blocked = self.research_engine.block_assumption_based_actions(test_case) if is_blocked: blocked_count += 1 test_result.details.append(f"✅ 추측성 작업 차단됨 ({i+1}): {test_case[:50]}...") else: test_result.errors.append(f"❌ 추측성 작업 미차단 ({i+1}): {test_case[:50]}...") # 3. 정책 강제 적용 테스트 enforcement_test_cases = [ ("정상 작업 컨텍스트", "Based on the documentation, I will implement the feature"), ("추측 작업 컨텍스트", "I assume this will work fine without checking"), ("증거 기반 컨텍스트", "According to the test results, the optimization is needed"), ("추정 작업 컨텍스트", "Probably this configuration should be adequate") ] policy_success_count = 0 for context, action in enforcement_test_cases: allowed, violation = await self.research_engine.enforce_research_first(context, action) # 추측성 언어가 포함된 경우 차단되어야 함 has_assumption = any(keyword in action.lower() for keyword in ['assume', 'probably', 'maybe', 'guess']) if has_assumption and not allowed: policy_success_count += 1 test_result.details.append(f"✅ 정책 강제 적용 성공: {action[:50]}...") elif not has_assumption and allowed: policy_success_count += 1 test_result.details.append(f"✅ 정상 작업 허용: {action[:50]}...") else: test_result.errors.append(f"❌ 정책 강제 적용 실패: {action[:50]}...") # 4. 정보원 검증 테스트 test_sources = [ "https://docs.python.org/3/library/asyncio.html", # 신뢰도 높음 "https://github.com/python/cpython", # 신뢰도 높음 "http://random-blog.com/article", # 신뢰도 낮음 "local_file.py", # 중간 신뢰도 "https://stackoverflow.com/questions/12345" # 중간 신뢰도 ] validation_success, validated_sources = await self.research_engine.validate_information_sources(test_sources) validated_count = len([s for s in validated_sources if s.reliability_score >= 0.7]) test_result.details.append(f"✅ 정보원 검증 완료: {validated_count}/{len(test_sources)} 신뢰 가능") # 5. 종합 점수 계산 assumption_block_rate = blocked_count / len(assumption_test_cases) policy_enforcement_rate = policy_success_count / len(enforcement_test_cases) source_validation_rate = validated_count / len(test_sources) overall_score = (assumption_block_rate * 40 + policy_enforcement_rate * 40 + source_validation_rate * 20) * 100 test_result.score = overall_score test_result.metrics = { 'assumption_block_rate': assumption_block_rate, 'policy_enforcement_rate': policy_enforcement_rate, 'source_validation_rate': source_validation_rate, 'blocked_assumptions': blocked_count, 'total_assumptions': len(assumption_test_cases), 'policy_success': policy_success_count, 'total_policy_tests': len(enforcement_test_cases), 'validated_sources': validated_count, 'total_sources': len(test_sources) } # 6. 성공 여부 판정 (목표: 100% 정책 강제 시행) success_threshold = self.performance_targets['policy_enforcement'] * 100 test_result.success = overall_score >= success_threshold if test_result.success: test_result.status = TestStatus.PASSED test_result.details.append(f"✅ Research-First 정책 강제 시행 테스트 통과 (점수: {overall_score:.1f}%)") else: test_result.status = TestStatus.FAILED test_result.errors.append(f"❌ Research-First 정책 강제 시행 미달 (점수: {overall_score:.1f}% < {success_threshold}%)") logger.info(f"✅ Research-First 정책 강제 시행 테스트 완료: {overall_score:.1f}%") except Exception as e: test_result.status = TestStatus.ERROR test_result.errors.append(f"테스트 실행 중 오류: {e}") test_result.errors.append(traceback.format_exc()) logger.error(f"❌ Research-First 정책 강제 시행 테스트 오류: {e}") finally: test_result.execution_time = time.time() - start_time test_result.completed_at = datetime.now() return test_result async def test_local_mcp_modules_integration(self) -> TestResult: """4개 로컬 MCP 모듈 완전 통합 검증 MCPDebugger, BrowserToolsMCP, ContextCompressor, ResearchFirstPolicyEngine 모듈들이 MCP-OS Orchestrator를 통해 완전히 통합되어 작동하는지 검증합니다. Returns: TestResult: 모듈 통합 테스트 결과 """ test_id = "MODULE-INTEGRATION-002" logger.info(f"🔧 로컬 MCP 모듈 통합 테스트 시작: {test_id}") test_result = TestResult( test_id=test_id, test_name="Local MCP Modules Complete Integration Test", category=TestCategory.MODULE_INTEGRATION, status=TestStatus.RUNNING, success=False, score=0.0, execution_time=0.0 ) start_time = time.time() try: # 1. MCP-OS Orchestrator 초기화 if not self.mcp_orchestrator: self.mcp_orchestrator = MCPOSLocalOrchestrator() test_result.details.append("✅ MCP-OS Orchestrator 초기화 완료") # 2. 로컬 MCP 모듈 로드 테스트 module_load_results = await self.mcp_orchestrator.coordinate_local_mcp_modules() loaded_modules = module_load_results.get('loaded_modules', []) expected_modules = ['MCPDebugger', 'BrowserToolsMCP', 'ContextCompressor', 'ResearchFirstPolicyEngine'] loaded_module_names = [mod['name'] for mod in loaded_modules] successfully_loaded = [name for name in expected_modules if name in loaded_module_names] test_result.details.append(f"✅ 로컬 MCP 모듈 로드 완료: {len(successfully_loaded)}/{len(expected_modules)}") # 3. 통합 워크플로우 관리 테스트 workflow_test_data = { 'workflow_type': 'sequential', 'tasks': [ {'module': 'ResearchFirstPolicyEngine', 'action': 'validate_sources'}, {'module': 'ContextCompressor', 'action': 'compress_content'}, {'module': 'MCPDebugger', 'action': 'create_session'}, {'module': 'BrowserToolsMCP', 'action': 'audit_page'} ] } workflow_result = await self.mcp_orchestrator.manage_integrated_workflow(workflow_test_data) workflow_success = workflow_result.get('success', False) if workflow_success: test_result.details.append("✅ 통합 워크플로우 관리 테스트 통과") else: test_result.errors.append("❌ 통합 워크플로우 관리 테스트 실패") # 4. 로컬 리소스 최적화 테스트 resource_optimization_result = await self.mcp_orchestrator.optimize_local_resource_allocation() performance_improvement = resource_optimization_result.get('performance_improvement_percent', 0) target_improvement = 40 # 40% 성능 향상 목표 if performance_improvement >= target_improvement: test_result.details.append(f"✅ 리소스 최적화 목표 달성: {performance_improvement}% >= {target_improvement}%") else: test_result.errors.append(f"❌ 리소스 최적화 목표 미달: {performance_improvement}% < {target_improvement}%") # 5. 모듈 장애 복구 테스트 failover_test_module = 'MCPDebugger' failover_result = await self.mcp_orchestrator.handle_module_failover(failover_test_module, 'connection_error') failover_success = failover_result.get('recovery_successful', False) if failover_success: test_result.details.append(f"✅ 모듈 장애 복구 테스트 통과: {failover_test_module}") else: test_result.errors.append(f"❌ 모듈 장애 복구 테스트 실패: {failover_test_module}") # 6. 종합 점수 계산 module_load_score = (len(successfully_loaded) / len(expected_modules)) * 25 workflow_score = 25 if workflow_success else 0 optimization_score = min(25, (performance_improvement / target_improvement) * 25) failover_score = 25 if failover_success else 0 overall_score = module_load_score + workflow_score + optimization_score + failover_score test_result.score = overall_score test_result.metrics = { 'loaded_modules_count': len(successfully_loaded), 'expected_modules_count': len(expected_modules), 'workflow_success': workflow_success, 'performance_improvement_percent': performance_improvement, 'failover_success': failover_success, 'module_load_rate': len(successfully_loaded) / len(expected_modules) } # 7. 성공 여부 판정 (목표: 85% 이상) success_threshold = 85.0 test_result.success = overall_score >= success_threshold if test_result.success: test_result.status = TestStatus.PASSED test_result.details.append(f"✅ 로컬 MCP 모듈 통합 테스트 통과 (점수: {overall_score:.1f}%)") else: test_result.status = TestStatus.FAILED test_result.errors.append(f"❌ 로컬 MCP 모듈 통합 미달 (점수: {overall_score:.1f}% < {success_threshold}%)") logger.info(f"✅ 로컬 MCP 모듈 통합 테스트 완료: {overall_score:.1f}%") except Exception as e: test_result.status = TestStatus.ERROR test_result.errors.append(f"테스트 실행 중 오류: {e}") test_result.errors.append(traceback.format_exc()) logger.error(f"❌ 로컬 MCP 모듈 통합 테스트 오류: {e}") finally: test_result.execution_time = time.time() - start_time test_result.completed_at = datetime.now() return test_result