chore: sync local assets for ddd doc steward

2026-08-16 20:38:04 +00:00 · 2025-12-21 03:56:58 +08:00
parent c09cc6cca4
commit c3d4fcb2ef
136 changed files with 48796 additions and 1 deletions
@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""
+Test MCP Integration with Unified Scraping
+
+Tests that the MCP server correctly handles unified configs.
+"""
+
+import sys
+import os
+import json
+import tempfile
+import asyncio
+import pytest
+from pathlib import Path
+
+# WORKAROUND for shadowing issue: Temporarily change to /tmp to import external mcp
+# This avoids any local mcp/ directory being in the import path
+_original_dir = os.getcwd()
+MCP_AVAILABLE = False
+try:
+    os.chdir('/tmp')  # Change away from project directory
+    from mcp.types import TextContent
+    MCP_AVAILABLE = True
+except ImportError:
+    pass
+finally:
+    os.chdir(_original_dir)  # Restore original directory
+
+# Configure pytest to only use asyncio backend (not trio)
+pytestmark = pytest.mark.anyio
+
+if MCP_AVAILABLE:
+    from skill_seekers.mcp.server import validate_config_tool, scrape_docs_tool
+else:
+    validate_config_tool = None
+    scrape_docs_tool = None
+
+
+@pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP package not installed")
+async def test_mcp_validate_unified_config():
+    """Test that MCP can validate unified configs"""
+    print("\n✓ Testing MCP validate_config_tool with unified config...")
+
+    # Use existing unified config
+    config_path = "configs/react_unified.json"
+
+    if not Path(config_path).exists():
+        print(f"  ⚠️  Skipping: {config_path} not found")
+        return
+
+    args = {"config_path": config_path}
+    result = await validate_config_tool(args)
+
+    # Check result
+    text = result[0].text
+    assert "✅" in text, f"Expected success, got: {text}"
+    assert "Unified" in text, f"Expected unified format detected, got: {text}"
+    assert "Sources:" in text, f"Expected sources count, got: {text}"
+
+    print("  ✅ MCP correctly validates unified config")
+
+
+@pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP package not installed")
+async def test_mcp_validate_legacy_config():
+    """Test that MCP can validate legacy configs"""
+    print("\n✓ Testing MCP validate_config_tool with legacy config...")
+
+    # Use existing legacy config
+    config_path = "configs/react.json"
+
+    if not Path(config_path).exists():
+        print(f"  ⚠️  Skipping: {config_path} not found")
+        return
+
+    args = {"config_path": config_path}
+    result = await validate_config_tool(args)
+
+    # Check result
+    text = result[0].text
+    assert "✅" in text, f"Expected success, got: {text}"
+    assert "Legacy" in text, f"Expected legacy format detected, got: {text}"
+
+    print("  ✅ MCP correctly validates legacy config")
+
+
+@pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP package not installed")
+async def test_mcp_scrape_docs_detection():
+    """Test that MCP scrape_docs correctly detects format"""
+    print("\n✓ Testing MCP scrape_docs format detection...")
+
+    # Create temporary unified config
+    unified_config = {
+        "name": "test_mcp_unified",
+        "description": "Test unified via MCP",
+        "merge_mode": "rule-based",
+        "sources": [
+            {
+                "type": "documentation",
+                "base_url": "https://example.com",
+                "extract_api": True,
+                "max_pages": 5
+            }
+        ]
+    }
+
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
+        json.dump(unified_config, f)
+        unified_config_path = f.name
+
+    # Create temporary legacy config
+    legacy_config = {
+        "name": "test_mcp_legacy",
+        "description": "Test legacy via MCP",
+        "base_url": "https://example.com",
+        "max_pages": 5
+    }
+
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
+        json.dump(legacy_config, f)
+        legacy_config_path = f.name
+
+    try:
+        # Test unified detection
+        with open(unified_config_path, 'r') as f:
+            config = json.load(f)
+
+        is_unified = 'sources' in config and isinstance(config['sources'], list)
+        assert is_unified, "Should detect unified format"
+        print("  ✅ Unified format detected correctly")
+
+        # Test legacy detection
+        with open(legacy_config_path, 'r') as f:
+            config = json.load(f)
+
+        is_unified = 'sources' in config and isinstance(config['sources'], list)
+        assert not is_unified, "Should detect legacy format"
+        print("  ✅ Legacy format detected correctly")
+
+    finally:
+        # Cleanup
+        Path(unified_config_path).unlink(missing_ok=True)
+        Path(legacy_config_path).unlink(missing_ok=True)
+
+
+@pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP package not installed")
+async def test_mcp_merge_mode_override():
+    """Test that MCP can override merge mode"""
+    print("\n✓ Testing MCP merge_mode override...")
+
+    # Create unified config
+    config = {
+        "name": "test_merge_override",
+        "description": "Test merge mode override",
+        "merge_mode": "rule-based",
+        "sources": [
+            {"type": "documentation", "base_url": "https://example.com"}
+        ]
+    }
+
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
+        json.dump(config, f)
+        config_path = f.name
+
+    try:
+        # Test that we can override merge_mode in args
+        args = {
+            "config_path": config_path,
+            "merge_mode": "claude-enhanced"  # Override
+        }
+
+        # Check that args has merge_mode
+        assert args.get("merge_mode") == "claude-enhanced"
+        print("  ✅ Merge mode override supported")
+
+    finally:
+        Path(config_path).unlink(missing_ok=True)
+
+
+# Run all tests
+async def run_all_tests():
+    print("=" * 60)
+    print("MCP Unified Scraping Integration Tests")
+    print("=" * 60)
+
+    try:
+        await test_mcp_validate_unified_config()
+        await test_mcp_validate_legacy_config()
+        await test_mcp_scrape_docs_detection()
+        await test_mcp_merge_mode_override()
+
+        print("\n" + "=" * 60)
+        print("✅ All MCP integration tests passed!")
+        print("=" * 60)
+
+    except AssertionError as e:
+        print(f"\n❌ Test failed: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"\n❌ Unexpected error: {e}")
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)
+
+
+if __name__ == '__main__':
+    asyncio.run(run_all_tests())