FusionAGI/tests/test_interfaces.py

"""Tests for interface layer: admin panel, multi-modal UI, voice, conversation."""

import pytest

from fusionagi.core import EventBus, StateManager, Orchestrator
from fusionagi.interfaces.admin_panel import AdminControlPanel, SystemStatus, AgentConfig
from fusionagi.interfaces.voice import VoiceLibrary, VoiceProfile, VoiceInterface
from fusionagi.interfaces.conversation import (
    ConversationTuner,
    ConversationStyle,
    ConversationManager,
    ConversationTurn,
)
from fusionagi.interfaces.multimodal_ui import MultiModalUI
from fusionagi.interfaces.base import ModalityType, InterfaceMessage


def test_voice_library() -> None:
    """Test voice library management."""
    library = VoiceLibrary()
    
    # Add voice
    voice = VoiceProfile(
        name="Test Voice",
        language="en-US",
        gender="neutral",
        style="professional",
    )
    voice_id = library.add_voice(voice)
    assert voice_id == voice.id
    
    # Get voice
    retrieved = library.get_voice(voice_id)
    assert retrieved is not None
    assert retrieved.name == "Test Voice"
    
    # List voices
    voices = library.list_voices()
    assert len(voices) == 1
    
    # Set default
    assert library.set_default_voice(voice_id)
    default = library.get_default_voice()
    assert default is not None
    assert default.id == voice_id
    
    # Update voice
    assert library.update_voice(voice_id, {"pitch": 1.2})
    updated = library.get_voice(voice_id)
    assert updated is not None
    assert updated.pitch == 1.2
    
    # Remove voice
    assert library.remove_voice(voice_id)
    assert library.get_voice(voice_id) is None


def test_voice_interface() -> None:
    """Test voice interface capabilities."""
    library = VoiceLibrary()
    voice = VoiceProfile(name="Test", language="en-US")
    library.add_voice(voice)
    
    interface = VoiceInterface(voice_library=library)
    
    # Check capabilities
    caps = interface.capabilities()
    assert ModalityType.VOICE in caps.supported_modalities
    assert caps.supports_streaming
    assert caps.supports_interruption
    
    # Set active voice
    assert interface.set_active_voice(voice.id)


def test_conversation_tuner() -> None:
    """Test conversation style tuning."""
    tuner = ConversationTuner()
    
    # Register style
    style = ConversationStyle(
        formality="formal",
        verbosity="detailed",
        empathy_level=0.5,
        technical_depth=0.9,
    )
    tuner.register_style("technical", style)
    
    # Get style
    retrieved = tuner.get_style("technical")
    assert retrieved is not None
    assert retrieved.formality == "formal"
    
    # List styles
    styles = tuner.list_styles()
    assert "technical" in styles
    
    # Tune for context
    tuned = tuner.tune_for_context(domain="technical")
    assert tuned.technical_depth >= 0.8  # Should be high for technical domain


def test_conversation_manager() -> None:
    """Test conversation management."""
    manager = ConversationManager()
    
    # Create session
    session_id = manager.create_session(user_id="test_user", language="en")
    assert session_id is not None
    
    # Get session
    session = manager.get_session(session_id)
    assert session is not None
    assert session.user_id == "test_user"
    
    # Add turns
    turn1 = ConversationTurn(
        session_id=session_id,
        speaker="user",
        content="Hello",
    )
    manager.add_turn(turn1)
    
    turn2 = ConversationTurn(
        session_id=session_id,
        speaker="agent",
        content="Hi there!",
    )
    manager.add_turn(turn2)
    
    # Get history
    history = manager.get_history(session_id)
    assert len(history) == 2
    assert history[0].speaker == "user"
    assert history[1].speaker == "agent"
    
    # Get context summary
    summary = manager.get_context_summary(session_id)
    assert summary["session_id"] == session_id
    assert summary["turn_count"] == 2
    
    # End session
    assert manager.end_session(session_id)
    assert manager.get_session(session_id) is None


def test_admin_control_panel() -> None:
    """Test admin control panel."""
    bus = EventBus()
    state = StateManager()
    orch = Orchestrator(event_bus=bus, state_manager=state)
    
    admin = AdminControlPanel(
        orchestrator=orch,
        event_bus=bus,
        state_manager=state,
    )
    
    # Voice management
    voice = VoiceProfile(name="Admin Voice", language="en-US")
    voice_id = admin.add_voice_profile(voice)
    assert voice_id is not None
    
    voices = admin.list_voices()
    assert len(voices) == 1
    
    # Conversation style management
    style = ConversationStyle(formality="neutral")
    admin.register_conversation_style("default", style)
    
    styles = admin.list_conversation_styles()
    assert "default" in styles
    
    # Agent configuration
    config = AgentConfig(
        agent_id="test_agent",
        agent_type="executor",
        enabled=True,
    )
    admin.configure_agent(config)
    
    retrieved_config = admin.get_agent_config("test_agent")
    assert retrieved_config is not None
    assert retrieved_config.agent_id == "test_agent"
    
    # System status
    status = admin.get_system_status()
    assert isinstance(status, SystemStatus)
    assert status.status in ("healthy", "degraded", "offline")
    
    # Task statistics
    stats = admin.get_task_statistics()
    assert "total_tasks" in stats
    assert "by_state" in stats
    
    # Configuration export/import
    config_data = admin.export_configuration()
    assert "voices" in config_data
    assert "conversation_styles" in config_data
    
    assert admin.import_configuration(config_data)


def test_multimodal_ui() -> None:
    """Test multi-modal UI."""
    bus = EventBus()
    state = StateManager()
    orch = Orchestrator(event_bus=bus, state_manager=state)
    
    conv_manager = ConversationManager()
    voice_interface = VoiceInterface()
    ui = MultiModalUI(
        orchestrator=orch,
        conversation_manager=conv_manager,
        voice_interface=voice_interface,
    )
    
    # Create session
    session_id = ui.create_session(
        user_id="test_user",
        preferred_modalities=[ModalityType.TEXT],
    )
    assert session_id is not None
    
    # Get session
    session = ui.get_session(session_id)
    assert session is not None
    assert session.user_id == "test_user"
    assert ModalityType.TEXT in session.active_modalities
    
    # Enable/disable modalities (voice interface is registered)
    assert ui.enable_modality(session_id, ModalityType.VOICE)
    session = ui.get_session(session_id)
    assert ModalityType.VOICE in session.active_modalities
    
    assert ui.disable_modality(session_id, ModalityType.VOICE)
    session = ui.get_session(session_id)
    assert ModalityType.VOICE not in session.active_modalities
    
    # Get statistics
    stats = ui.get_session_statistics(session_id)
    assert stats["session_id"] == session_id
    assert stats["user_id"] == "test_user"
    
    # End session
    assert ui.end_session(session_id)
    assert ui.get_session(session_id) is None


def test_multimodal_ui_sync() -> None:
    """Test multi-modal UI synchronous operations."""
    bus = EventBus()
    state = StateManager()
    orch = Orchestrator(event_bus=bus, state_manager=state)
    
    conv_manager = ConversationManager()
    ui = MultiModalUI(
        orchestrator=orch,
        conversation_manager=conv_manager,
    )
    
    session_id = ui.create_session(user_id="test_user")
    
    # Test that session was created
    assert session_id is not None
    session = ui.get_session(session_id)
    assert session is not None
    
    # Test available modalities
    available = ui.get_available_modalities()
    assert isinstance(available, list)
    
    ui.end_session(session_id)


if __name__ == "__main__":
    pytest.main([__file__, "-v"])