File size: 6,136 Bytes
92ef79b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
#!/usr/bin/env python3
"""
Test script for persistent editing functionality
Tests multiple edit cycles to ensure task plans persist correctly
"""

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

from gradio_llm_interface import GradioLlmInterface
import json

def test_edit_cycle():
    """Test complete edit cycle: generate β†’ edit β†’ update β†’ deploy β†’ edit again"""
    print("Testing Complete Edit Cycle...")
    print("=" * 50)
    
    interface = GradioLlmInterface()
    
    # Step 1: Initial task plan (simulating LLM generation)
    initial_plan = {
        "tasks": [
            {
                "task": "move_soil_1",
                "instruction_function": {
                    "name": "move_soil",
                    "robot_ids": ["robot_excavator_01"],
                    "dependencies": [],
                    "object_keywords": ["soil_pile"]
                }
            }
        ]
    }
    
    state = {'pending_task_plan': initial_plan}
    print("βœ“ Step 1: Initial task plan created")
    
    # Step 2: Open editor with initial plan
    editor_result = interface.show_task_plan_editor(state)
    if not editor_result or len(editor_result) != 4:
        print("βœ— Step 2: Failed to open editor")
        return False
    
    editor_update, dag_btn, validate_btn, status = editor_result
    if "move_soil" in editor_update.get('value', ''):
        print("βœ“ Step 2: Editor opened with correct initial plan")
    else:
        print("βœ— Step 2: Editor does not contain initial plan")
        return False
    
    # Step 3: Edit the plan
    edited_json = """{
  "tasks": [
    {
      "task": "move_soil_1_edited",
      "instruction_function": {
        "name": "move_soil_edited",
        "robot_ids": ["robot_excavator_01", "robot_dump_truck_01"],
        "dependencies": [],
        "object_keywords": ["soil_pile", "edited_keyword"]
      }
    },
    {
      "task": "transport_soil_1",
      "instruction_function": {
        "name": "transport_soil",
        "robot_ids": ["robot_dump_truck_01"],
        "dependencies": ["move_soil_1_edited"],
        "object_keywords": ["destination"]
      }
    }
  ]
}"""
    
    update_result = interface.update_dag_from_editor(edited_json, state)
    if not update_result or len(update_result) != 6:
        print("βœ— Step 3: Failed to update DAG from editor")
        return False
    
    print("βœ“ Step 3: DAG updated with edited plan")
    
    # Step 4: Deploy the plan
    deploy_result = interface.validate_and_deploy_task_plan(state)
    if not deploy_result:
        print("βœ— Step 4: Failed to deploy plan")
        return False
    
    print("βœ“ Step 4: Plan deployed successfully")
    
    # Step 5: Try to edit again (this should show the deployed plan)
    second_editor_result = interface.show_task_plan_editor(state)
    if not second_editor_result or len(second_editor_result) != 4:
        print("βœ— Step 5: Failed to open editor second time")
        return False
    
    second_editor_update, _, _, second_status = second_editor_result
    if "move_soil_1_edited" in second_editor_update.get('value', ''):
        print("βœ“ Step 5: Editor opened with deployed plan (persistent editing working)")
        return True
    else:
        print("βœ— Step 5: Editor lost the deployed plan content")
        print(f"  Editor content: {second_editor_update.get('value', 'No content')[:100]}...")
        return False

def test_empty_state_handling():
    """Test editor behavior with completely empty state"""
    print("\nTesting Empty State Handling...")
    print("=" * 40)
    
    interface = GradioLlmInterface()
    empty_state = {}
    
    result = interface.show_task_plan_editor(empty_state)
    if result and len(result) == 4:
        editor_update, _, _, status = result
        if "example_task_1" in editor_update.get('value', ''):
            print("βœ“ Empty state shows example template")
            return True
        else:
            print("βœ— Empty state does not show proper template")
            return False
    else:
        print("βœ— Failed to handle empty state")
        return False

def test_malformed_state_handling():
    """Test editor behavior with malformed state data"""
    print("\nTesting Malformed State Handling...")
    print("=" * 40)
    
    interface = GradioLlmInterface()
    
    # Test with empty tasks array
    malformed_state = {
        'pending_task_plan': {
            'tasks': []
        }
    }
    
    result = interface.show_task_plan_editor(malformed_state)
    if result and len(result) == 4:
        editor_update, _, _, status = result
        if "example_task_1" in editor_update.get('value', ''):
            print("βœ“ Malformed state (empty tasks) handled correctly")
            return True
        else:
            print("βœ— Malformed state not handled properly")
            return False
    else:
        print("βœ— Failed to handle malformed state")
        return False

def main():
    """Run all persistent editing tests"""
    print("πŸ”„ Persistent Editing Tests")
    print("=" * 50)
    
    tests = [
        test_edit_cycle,
        test_empty_state_handling,
        test_malformed_state_handling
    ]
    
    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"Persistent Editing Tests passed: {passed}/{total}")
    
    if passed == total:
        print("πŸŽ‰ All persistent editing tests passed!")
        print("\nπŸ”„ Persistent Editing Features:")
        print("  βœ“ Task plans persist through edit cycles")
        print("  βœ“ Deployed plans can be re-edited")
        print("  βœ“ State management handles edge cases")
        print("  βœ“ Proper fallback to templates when needed")
        return True
    else:
        print("❌ Some persistent editing tests failed!")
        return False

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