test_safety_workflow.py

#!/usr/bin/env python3
"""
Test script for safety confirmation workflow
"""

import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from gradio_llm_interface import GradioLlmInterface
import json

def test_safety_workflow():
    """Test the safety confirmation workflow"""
    print("Testing Safety Confirmation Workflow...")
    print("=" * 50)
    
    # Create mock task data
    mock_task_data = {
        "tasks": [
            {
                "task": "excavate_soil_from_pile",
                "instruction_function": {
                    "name": "excavate_soil_from_pile",
                    "robot_ids": ["robot_excavator_01"],
                    "dependencies": [],
                    "object_keywords": ["soil_pile"]
                }
            },
            {
                "task": "transport_soil_to_pit",
                "instruction_function": {
                    "name": "transport_soil_to_pit", 
                    "robot_ids": ["robot_dump_truck_01"],
                    "dependencies": ["excavate_soil_from_pile"],
                    "object_keywords": ["pit"]
                }
            }
        ]
    }
    
    # Initialize interface
    interface = GradioLlmInterface()
    
    # Create mock state with pending task plan
    mock_state = {
        'pending_task_plan': mock_task_data,
        'history': []
    }
    
    print("✓ Mock state created with pending task plan")
    print(f"  Tasks: {len(mock_task_data['tasks'])}")
    
    # Test validation and deployment
    try:
        result = interface.validate_and_deploy_task_plan(mock_state)
        
        if result:
            confirmation_msg, approved_image_path, button_update, updated_state = result
            
            print("\n📋 Validation Results:")
            print(f"  Confirmation Message: {confirmation_msg[:100]}...")
            print(f"  Image Generated: {'✓' if approved_image_path else '✗'}")
            print(f"  Button Hidden: {'✓' if not button_update.get('visible', True) else '✗'}")
            print(f"  State Updated: {'✓' if updated_state.get('pending_task_plan') is None else '✗'}")
            
            return True
        else:
            print("✗ No result returned from validation function")
            return False
            
    except Exception as e:
        print(f"✗ Error during validation: {e}")
        return False

def test_empty_state():
    """Test with empty state (no pending task plan)"""
    print("\nTesting Empty State Handling...")
    print("=" * 30)
    
    interface = GradioLlmInterface()
    empty_state = {'history': []}
    
    try:
        result = interface.validate_and_deploy_task_plan(empty_state)
        
        if result:
            warning_msg, image_path, button_update, state = result
            
            if "No Task Plan to Deploy" in warning_msg:
                print("✓ Empty state handled correctly")
                return True
            else:
                print("✗ Unexpected warning message")
                return False
        else:
            print("✗ No result returned for empty state")
            return False
            
    except Exception as e:
        print(f"✗ Error handling empty state: {e}")
        return False

def main():
    """Run safety workflow tests"""
    print("🔒 Safety Confirmation Workflow Tests")
    print("=" * 50)
    
    tests = [
        test_safety_workflow,
        test_empty_state
    ]
    
    passed = 0
    total = len(tests)
    
    for test in tests:
        try:
            if test():
                passed += 1
        except Exception as e:
            print(f"✗ Test failed with exception: {e}")
    
    print("\n" + "=" * 50)
    print(f"Safety Tests passed: {passed}/{total}")
    
    if passed == total:
        print("🎉 All safety workflow tests passed!")
        print("\n🔐 Safety Features:")
        print("  ✓ Task plan approval workflow")
        print("  ✓ Visual confirmation before deployment") 
        print("  ✓ Error handling and validation")
        print("  ✓ State management for pending plans")
        return True
    else:
        print("❌ Some safety tests failed!")
        return False

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)