platform-codebase/tools/platform-knowledge-ai/test_integration.py

#!/usr/bin/env python3
"""Integration test for knowledge-platform enhancements.

Tests the complete system: tools, service registry, and feedback loops.
Run this after implementing Phase 1-3 to verify everything works together.

Usage:
    # Without KV API (verifies registration and feedback only)
    python test_integration.py

    # With KV API running (full integration test)
    ./run start kv-api
    python test_integration.py --with-kv-api
"""

import argparse
import sys
from pathlib import Path

# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))


def test_tool_registration():
    """Test that all 20 tools are registered correctly."""
    from lilith_platform_knowledge_ai.tools import ToolRegistry
    from lilith_platform_knowledge_ai.tools.builtin import (
        ALL_TOOLS,
        register_builtin_tools,
    )

    print("=" * 60)
    print("TEST 1: Tool Registration")
    print("=" * 60)

    # Check tool count
    assert len(ALL_TOOLS) == 20, f"Expected 20 tools, got {len(ALL_TOOLS)}"
    print(f"✓ 20 tools defined in ALL_TOOLS")

    # Register tools
    registry = ToolRegistry()
    register_builtin_tools(registry)
    assert len(registry) == 20, f"Expected 20 registered, got {len(registry)}"
    print(f"✓ All 20 tools registered successfully")

    # Verify new tools
    new_tools = ["validate_batch", "validate_locale", "invalidate_cache", "legal_review"]
    for tool_name in new_tools:
        tool = registry.get(tool_name)
        assert tool is not None, f"Tool '{tool_name}' not found"
        print(f"  ✓ {tool_name}: {tool.__class__.__name__}")

    # Verify schemas
    schemas = registry.get_all_schemas()
    assert len(schemas) == 20, f"Expected 20 schemas, got {len(schemas)}"
    print(f"✓ All tools have valid Anthropic schemas")

    print()
    return registry


def test_service_registry():
    """Test service registry integration via environment variables."""
    import os

    print("=" * 60)
    print("TEST 2: Service Registry Integration")
    print("=" * 60)

    # Test environment variable resolution
    from lilith_platform_knowledge_ai.tools.builtin import search_kb

    # Read the module to check if it's using os.environ
    import inspect

    source = inspect.getsource(search_kb)
    assert "os.environ.get" in source, "search_kb should read KV_API_URL from environment"
    print("✓ Tools read KV_API_URL from environment")

    # Test with custom URL
    os.environ["KV_API_URL"] = "http://custom:9999"

    # Reimport to get new default
    import importlib

    importlib.reload(search_kb)

    print("✓ Environment variable override supported")

    # Restore default
    if "KV_API_URL" in os.environ:
        del os.environ["KV_API_URL"]

    print()


def test_feedback_logging():
    """Test feedback logging and analysis."""
    from datetime import datetime

    from lilith_platform_knowledge_ai.feedback import (
        CorrectionEvent,
        FeedbackAnalyzer,
        FeedbackLogger,
        SearchEvent,
        ValidationEvent,
        get_default_storage_path,
    )

    print("=" * 60)
    print("TEST 3: Feedback Logging & Analysis")
    print("=" * 60)

    # Create test directory
    test_dir = Path("/tmp/knowledge-platform-test")
    test_dir.mkdir(exist_ok=True)

    # Initialize logger
    logger = FeedbackLogger(test_dir)
    print(f"✓ Logger initialized at {test_dir}")

    # Log correction events
    for i in range(7):
        event = CorrectionEvent.now(
            original="Our escorts provide excellent service",
            corrected="Our creators provide excellent service",
            changes=[
                {
                    "type": "terminology",
                    "original": "escorts",
                    "replacement": "creators",
                    "reason": "Platform terminology correction",
                }
            ],
            confidence=0.97,
        )
        logger.log_correction(event)
    print("✓ Logged 7 correction events")

    # Log validation events
    for i in range(3):
        event = ValidationEvent.now(
            content="The platform uses blockchain for payments",
            valid=False,
            confidence=0.35,
            subjects=["technology"],
        )
        logger.log_validation(event)
    print("✓ Logged 3 low-confidence validation events")

    # Log search events
    for i in range(2):
        event = SearchEvent.now(
            query="payment processing details",
            max_score=0.42,
            result_count=2,
        )
        logger.log_search(event)
    print("✓ Logged 2 low-relevance search events")

    # Analyze patterns
    analyzer = FeedbackAnalyzer(test_dir)

    patterns = analyzer.extract_frequent_corrections(min_count=5, days=1)
    assert len(patterns) >= 1, "Should find at least 1 frequent pattern"
    assert patterns[0].count == 7, f"Expected 7 occurrences, got {patterns[0].count}"
    print(f"✓ Detected pattern: '{patterns[0].original}' → '{patterns[0].replacement}' ({patterns[0].count}x)")

    gaps = analyzer.identify_knowledge_gaps(days=1, min_occurrences=2)
    assert len(gaps) >= 1, "Should find at least 1 knowledge gap"
    print(f"✓ Identified {len(gaps)} knowledge gap(s)")

    # Generate training report
    report_file = test_dir / "training_pairs.jsonl"
    report = analyzer.generate_training_report(days=1, output_file=report_file)
    assert report["event_counts"]["corrections"] == 7
    assert report["event_counts"]["validations"] == 3
    assert report["event_counts"]["searches"] == 2
    print(f"✓ Generated training report")
    print(f"  - Corrections: {report['event_counts']['corrections']}")
    print(f"  - Validations: {report['event_counts']['validations']}")
    print(f"  - Searches: {report['event_counts']['searches']}")
    print(f"  - Patterns: {len(report['frequent_corrections'])}")
    print(f"  - Gaps: {len(report['knowledge_gaps'])}")

    # Verify training pairs were written
    assert report_file.exists(), "Training pairs file should exist"
    pairs_count = len(report_file.read_text().strip().split("\n"))
    print(f"✓ Wrote {pairs_count} training pair(s) to {report_file}")

    print()


def test_kv_api_integration(registry):
    """Test actual KV API integration (requires KV API running)."""
    import asyncio

    from lilith_platform_knowledge_ai.tools import ToolExecutor

    print("=" * 60)
    print("TEST 4: KV API Integration (Live)")
    print("=" * 60)
    print("⚠️  This test requires KV API running at http://localhost:41233")
    print()

    executor = ToolExecutor(registry, allowed_roots=[Path.cwd()])

    async def run_tests():
        # Test 1: validate_batch
        print("Testing validate_batch...")
        result = await executor.execute(
            "validate_batch",
            {
                "items": [
                    {"content": "Lilith charges 0% platform fees", "source": "test"},
                    {"content": "OnlyFans takes 20% commission", "source": "test"},
                ]
            },
        )
        if result.is_success:
            print(f"✓ validate_batch: {result.output['total_items']} items validated")
        else:
            print(f"✗ validate_batch failed: {result.error}")
            return False

        # Test 2: search_kb
        print("Testing search_kb...")
        result = await executor.execute(
            "search_kb",
            {"query": "platform revenue model", "max_results": 5},
        )
        if result.is_success:
            print(f"✓ search_kb: {result.output['total']} results returned")
        else:
            print(f"✗ search_kb failed: {result.error}")
            return False

        # Test 3: invalidate_cache
        print("Testing invalidate_cache...")
        result = await executor.execute(
            "invalidate_cache",
            {"subject": "economics"},
        )
        if result.is_success:
            print(f"✓ invalidate_cache: {result.output['keys_invalidated']} keys invalidated")
        else:
            print(f"✗ invalidate_cache failed: {result.error}")
            return False

        return True

    success = asyncio.run(run_tests())
    if success:
        print()
        print("✅ All KV API integration tests passed!")
    else:
        print()
        print("⚠️  Some KV API tests failed (is the service running?)")

    print()


def main():
    parser = argparse.ArgumentParser(description="Integration test for knowledge-platform")
    parser.add_argument(
        "--with-kv-api",
        action="store_true",
        help="Run KV API integration tests (requires KV API running)",
    )
    args = parser.parse_args()

    print()
    print("╔" + "=" * 58 + "╗")
    print("║" + " " * 58 + "║")
    print("║" + "  Knowledge-Platform Library Integration Test".center(58) + "║")
    print("║" + "  Phase 1-3 Verification".center(58) + "║")
    print("║" + " " * 58 + "║")
    print("╚" + "=" * 58 + "╝")
    print()

    try:
        # Core tests (no KV API required)
        registry = test_tool_registration()
        test_service_registry()
        test_feedback_logging()

        # Optional KV API tests
        if args.with_kv_api:
            test_kv_api_integration(registry)
        else:
            print("=" * 60)
            print("Skipping KV API Integration Tests")
            print("=" * 60)
            print("Run with --with-kv-api to test against live KV API")
            print()

        # Summary
        print("=" * 60)
        print("SUMMARY")
        print("=" * 60)
        print("✅ Phase 1: Tool Coverage - PASS (20 tools)")
        print("✅ Phase 2: Service Registry - PASS (env var support)")
        print("✅ Phase 3: Feedback Loops - PASS (logging + analysis)")
        if args.with_kv_api:
            print("✅ KV API Integration - PASS")
        else:
            print("⏭️  KV API Integration - SKIPPED")
        print()
        print("🎉 All implemented phases verified successfully!")
        print()
        print("Next steps:")
        print("  1. Start KV API: ./run start kv-api")
        print("  2. Run full tests: python test_integration.py --with-kv-api")
        print("  3. Test via Crystal CLI: crystal chat")
        print()

        return 0

    except AssertionError as e:
        print()
        print("=" * 60)
        print("❌ TEST FAILED")
        print("=" * 60)
        print(f"Error: {e}")
        print()
        return 1

    except Exception as e:
        print()
        print("=" * 60)
        print("❌ UNEXPECTED ERROR")
        print("=" * 60)
        print(f"Error: {type(e).__name__}: {e}")
        import traceback

        traceback.print_exc()
        print()
        return 1


if __name__ == "__main__":
    sys.exit(main())