test_ai_service.py

"""
test_ai_service.py - Multi-Scenario Test Runner for Factory GenAI Decision Support.

Validates that:
1. Responses follow the exact 5-part user-facing design:
   - WHAT HAPPENED
   - POSSIBLE CAUSES
   - PREDICTION
   - RISK SCORE
   - ACTION REQUIRED (showing ONLY genuinely relevant roles)
2. No internal/debug tags ("Affected Roles", "Action Count") are present.
3. Conversational question answering routes questions to appropriate roles dynamically.
"""

import sys
sys.stdout.reconfigure(encoding='utf-8')

from ai_service import (
    ProductionIntelligenceAIService,
    format_user_facing_response
)

def test_all_scenarios():
    print("=" * 80)
    print("🏭 TESTING FACTORY DECISION-SUPPORT GENAI ACROSS 5 SCENARIOS")
    print("=" * 80)

    service = ProductionIntelligenceAIService()
    
    scenarios = [
        "normal_operation",
        "high_processing_delay",
        "high_error_rate",
        "low_throughput",
        "combined_disruption"
    ]

    results = {}

    for idx, scn in enumerate(scenarios, start=1):
        print(f"\n[{idx}/5] 🔍 SCENARIO TEST: '{scn}'")
        print("-" * 80)
        res = service.analyze_scenario(scn)
        results[scn] = res
        
        # 1. Verify schema keys
        required_keys = ["what_happened", "possible_causes", "prediction", "risk_score", "action_required"]
        for k in required_keys:
            assert k in res, f"Missing key '{k}' in scenario '{scn}'"
            assert res[k], f"Key '{k}' is empty in scenario '{scn}'"
        
        # 2. Verify risk_score structure
        assert isinstance(res["risk_score"], dict), "risk_score must be a dict"
        assert "score" in res["risk_score"], "risk_score missing 'score'"
        assert "severity" in res["risk_score"], "risk_score missing 'severity'"

        # 3. Verify action_required structure
        assert isinstance(res["action_required"], list), "action_required must be a list"
        present_roles = [item["role"] for item in res["action_required"]]
        
        for item in res["action_required"]:
            assert "role" in item, "action_required item missing 'role'"
            assert "actions" in item, "action_required item missing 'actions'"
            assert isinstance(item["actions"], list), "actions must be a list of strings"

        # 4. Render clean user-facing format
        user_facing_text = format_user_facing_response(res)
        assert "WHAT HAPPENED" in user_facing_text
        assert "POSSIBLE CAUSES" in user_facing_text
        assert "PREDICTION" in user_facing_text
        assert "RISK SCORE" in user_facing_text
        assert "ACTION REQUIRED" in user_facing_text
        
        # Assert NO debug tags are present in user-facing text
        assert "Affected Roles" not in user_facing_text
        assert "Action Count" not in user_facing_text

        print(user_facing_text)
        print("-" * 80)

    # 5. Assert Role Filtering Rules
    # low_throughput -> Plant Manager + Supervisor (Maintenance omitted)
    lt_roles = [item["role"] for item in results["low_throughput"]["action_required"]]
    assert "Plant Manager" in lt_roles
    assert "Supervisor" in lt_roles
    assert "Maintenance Engineer" not in lt_roles

    # high_processing_delay -> Maintenance Engineer + Supervisor (Plant Manager omitted)
    hpd_roles = [item["role"] for item in results["high_processing_delay"]["action_required"]]
    assert "Maintenance Engineer" in hpd_roles
    assert "Supervisor" in hpd_roles
    assert "Plant Manager" not in hpd_roles

    # high_error_rate -> Maintenance Engineer + Supervisor (Plant Manager omitted)
    her_roles = [item["role"] for item in results["high_error_rate"]["action_required"]]
    assert "Maintenance Engineer" in her_roles
    assert "Supervisor" in her_roles
    assert "Plant Manager" not in her_roles

    # combined_disruption -> All 3 roles present
    cd_roles = [item["role"] for item in results["combined_disruption"]["action_required"]]
    assert set(cd_roles) == {"Maintenance Engineer", "Supervisor", "Plant Manager"}

    # normal_operation -> Only Supervisor
    norm_roles = [item["role"] for item in results["normal_operation"]["action_required"]]
    assert "Maintenance Engineer" not in norm_roles
    assert "Plant Manager" not in norm_roles

    # 6. Conversational Q&A Testing
    print("\n" + "=" * 80)
    print("💬 TESTING CONVERSATIONAL FACTORY ASSISTANT ROUTING")
    print("=" * 80)
    
    # Test on low_throughput scenario (material starvation)
    lt_context = service.load_context()["scenarios"]["low_throughput"]
    
    # Maintenance question when maintenance is NOT needed
    maint_ans = service.chat_assistant(lt_context, [], "What should maintenance do?")
    print("\nQ: What should maintenance do? (during material starvation)")
    print(f"A: {maint_ans}")
    assert "No immediate Maintenance" in maint_ans

    # Plant Manager question when management IS needed
    pm_ans = service.chat_assistant(lt_context, [], "What should the plant manager do?")
    print("\nQ: What should the plant manager do? (during material starvation)")
    print(f"A: {pm_ans}")
    assert "Plant Manager" in pm_ans

    print("\n" + "=" * 80)
    print("✅ ALL USER-FACING DESIGN & CONVERSATIONAL ASSERTIONS PASSED")
    print("=" * 80 + "\n")

if __name__ == "__main__":
    test_all_scenarios()