Spaces:
Running
Running
File size: 14,905 Bytes
0646b18 |
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 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 |
"""
System test for variable creation order validation.
Tests that variables are created and stored in the correct order
when multiple variables are created in sequence during code execution.
"""
import pytest
from cuga.backend.cuga_graph.nodes.cuga_lite.cuga_agent_base import eval_with_tools_async
from cuga.backend.cuga_graph.state.agent_state import AgentState
@pytest.mark.asyncio
class TestVariableCreationOrder:
async def test_variable_creation_order_preserved(self):
"""
Test that variables created in sequence maintain their creation order.
This test validates that when multiple variables are created in code,
they are stored and displayed in the order they were created, not by
timestamp (which can be unreliable for variables created in the same millisecond).
"""
# Create a sample AgentState
state = AgentState(
input="test task",
url="",
)
# Code that creates multiple variables in sequence
# Order: var1, var2, var3, var4
code = """var1 = "first"
var2 = ["second", "item"]
var3 = {"third": "value"}
var4 = var1 + " -> " + str(var2)
print(f"Created: {var1}, {var2}, {var3}, {var4}")
"""
# Empty locals dict
_locals = {}
# Execute the code
output, new_vars = await eval_with_tools_async(
code=code,
_locals=_locals,
state=state,
thread_id=None,
apps_list=None,
)
# Verify all variables were created
expected_vars = ["var1", "var2", "var3", "var4"]
for var_name in expected_vars:
assert var_name in new_vars, f"{var_name} should be in new_vars"
assert state.variables_manager.get_variable(var_name) is not None, (
f"{var_name} should be in variables_manager"
)
# Validate creation order in new_vars dictionary
new_var_names = list(new_vars.keys())
assert new_var_names == expected_vars, f"Expected order {expected_vars}, got {new_var_names}"
# Validate creation order in variables_manager
all_var_names = state.variables_manager.get_variable_names()
indices = {var: all_var_names.index(var) for var in expected_vars}
# Verify each variable comes before the next one
for i in range(len(expected_vars) - 1):
current_var = expected_vars[i]
next_var = expected_vars[i + 1]
assert indices[current_var] < indices[next_var], (
f"{current_var} (index {indices[current_var]}) should come before {next_var} (index {indices[next_var]})"
)
# Validate order in summary (should match creation order)
summary = state.variables_manager.get_variables_summary()
positions = {var: summary.find(f"## {var}") for var in expected_vars}
# Verify each variable appears before the next one in summary
for i in range(len(expected_vars) - 1):
current_var = expected_vars[i]
next_var = expected_vars[i + 1]
assert positions[current_var] < positions[next_var], (
f"{current_var} should appear before {next_var} in summary"
)
async def test_variable_creation_order_with_existing_vars(self):
"""
Test that new variables maintain creation order even when existing variables are present.
"""
# Create a sample AgentState
state = AgentState(
input="test task",
url="",
)
# Add existing variables
state.variables_manager.add_variable("existing1", name="existing_var1")
state.variables_manager.add_variable("existing2", name="existing_var2")
# Code that creates new variables in sequence
code = """new_var1 = "first_new"
new_var2 = ["second_new"]
new_var3 = new_var1 + " -> " + str(new_var2)
print(new_var3)
"""
# Pass existing variables in locals
_locals = {"existing_var1": "existing1", "existing_var2": "existing2"}
# Execute the code
output, new_vars = await eval_with_tools_async(
code=code,
_locals=_locals,
state=state,
thread_id=None,
apps_list=None,
)
# Verify new variables were created
expected_new_vars = ["new_var1", "new_var2", "new_var3"]
for var_name in expected_new_vars:
assert var_name in new_vars, f"{var_name} should be in new_vars"
# Validate creation order in new_vars
new_var_names = list(new_vars.keys())
assert new_var_names == expected_new_vars, f"Expected order {expected_new_vars}, got {new_var_names}"
# Validate that new variables are added after existing ones
all_var_names = state.variables_manager.get_variable_names()
existing_indices = {
"existing_var1": all_var_names.index("existing_var1"),
"existing_var2": all_var_names.index("existing_var2"),
}
new_indices = {var: all_var_names.index(var) for var in expected_new_vars}
# All new variables should come after existing ones
for existing_var, existing_idx in existing_indices.items():
for new_var, new_idx in new_indices.items():
assert new_idx > existing_idx, (
f"New variable {new_var} (index {new_idx}) should come after existing {existing_var} (index {existing_idx})"
)
# Validate order among new variables themselves
for i in range(len(expected_new_vars) - 1):
current_var = expected_new_vars[i]
next_var = expected_new_vars[i + 1]
assert new_indices[current_var] < new_indices[next_var], (
f"{current_var} should come before {next_var}"
)
async def test_variable_creation_order_multiple_sequential_creations(self):
"""
Test that variables created in multiple sequential code executions
maintain their relative creation order.
"""
# Create a sample AgentState
state = AgentState(
input="test task",
url="",
)
# First code execution: creates var1 and var2
code1 = """var1 = "first"
var2 = "second"
print("Created var1 and var2")
"""
_locals1 = {}
output1, new_vars1 = await eval_with_tools_async(
code=code1,
_locals=_locals1,
state=state,
thread_id=None,
apps_list=None,
)
# Second code execution: creates var3 and var4
code2 = """var3 = "third"
var4 = "fourth"
print("Created var3 and var4")
"""
_locals2 = {}
output2, new_vars2 = await eval_with_tools_async(
code=code2,
_locals=_locals2,
state=state,
thread_id=None,
apps_list=None,
)
# Verify all variables exist
all_vars = state.variables_manager.get_variable_names()
assert "var1" in all_vars
assert "var2" in all_vars
assert "var3" in all_vars
assert "var4" in all_vars
# Validate order: var1, var2 should come before var3, var4
var1_idx = all_vars.index("var1")
var2_idx = all_vars.index("var2")
var3_idx = all_vars.index("var3")
var4_idx = all_vars.index("var4")
assert var1_idx < var2_idx < var3_idx < var4_idx, (
f"Expected order: var1 ({var1_idx}) < var2 ({var2_idx}) < var3 ({var3_idx}) < var4 ({var4_idx})"
)
# Validate order in summary
summary = state.variables_manager.get_variables_summary()
var1_pos = summary.find("## var1")
var2_pos = summary.find("## var2")
var3_pos = summary.find("## var3")
var4_pos = summary.find("## var4")
assert var1_pos < var2_pos < var3_pos < var4_pos, (
"Variables should appear in order: var1, var2, var3, var4"
)
async def test_results_is_last_variable_with_missing_dependencies(self):
"""
Test that validates 'results' is the last variable in the summary when code
uses missing variables and functions (with dummy implementations).
This test validates:
1. 'results' should be the LAST variable in the summary (final output)
2. Missing variable 'contacts_content' is provided in _locals
3. Missing functions are provided as dummy implementations in _locals
4. Variables appearing in the last print statement are moved to the end
"""
# Create a sample AgentState
state = AgentState(
input="test task",
url="",
)
# Dummy async function implementations
async def dummy_crm_get_contacts_contacts_get(email: str):
"""Dummy function for crm_get_contacts_contacts_get."""
return {"items": [{"first_name": "John", "last_name": "Doe", "account_id": 123}]}
async def dummy_crm_get_account_accounts_account_id_get(account_id: int):
"""Dummy function for crm_get_account_accounts_account_id_get."""
return {"id": account_id, "name": "Test Account"}
# Code that creates variables in sequence, with 'results' as the last one
code = """import json
# Parse the email list from the provided variable
email_lines = contacts_content.get("result", "").splitlines()
emails = [email.strip() for email in email_lines if email.strip()]
results = []
for email in emails:
# Query contacts by email
contact_response = await crm_get_contacts_contacts_get(email=email)
contact_items = contact_response.get("items", [])
if contact_items:
contact = contact_items[0] # assume first match is the desired one
contact_name = f"{contact.get('first_name', '')} {contact.get('last_name', '')}".strip()
account_id = contact.get("account_id")
# Retrieve associated account details if account_id is present
if account_id is not None:
account = await crm_get_account_accounts_account_id_get(account_id=account_id)
else:
account = None
results.append({
"email": email,
"contact_name": contact_name,
"account": account
})
else:
# No contact found for this email
results.append({
"email": email,
"contact_name": None,
"account": None,
"note": "Contact not found"
})
# Print the aggregated results
print(json.dumps(results, indent=2))
"""
# Provide missing variable and functions in _locals
_locals = {
"contacts_content": {"result": "[email protected]\[email protected]"},
"crm_get_contacts_contacts_get": dummy_crm_get_contacts_contacts_get,
"crm_get_account_accounts_account_id_get": dummy_crm_get_account_accounts_account_id_get,
}
# Execute the code
output, new_vars = await eval_with_tools_async(
code=code,
_locals=_locals,
state=state,
thread_id=None,
apps_list=None,
)
# Key variables that should be created (excluding loop variables)
key_vars = ["email_lines", "emails", "results"]
# Verify all key variables were created
for var_name in key_vars:
assert var_name in new_vars, f"{var_name} should be in new_vars"
assert state.variables_manager.get_variable(var_name) is not None, (
f"{var_name} should be in variables_manager"
)
# Validate creation order in new_vars dictionary
new_var_names = list(new_vars.keys())
# Verify key variables exist and 'results' is the last key variable
key_var_indices = {var: new_var_names.index(var) for var in key_vars if var in new_var_names}
# Validate that 'results' comes after 'email_lines' and 'emails'
assert key_var_indices["email_lines"] < key_var_indices["emails"], (
"'email_lines' should come before 'emails'"
)
assert key_var_indices["emails"] < key_var_indices["results"], "'emails' should come before 'results'"
# Validate that 'results' is the last meaningful variable (may have loop vars after, but results should be last key var)
# Find the last key variable in the list
last_key_var_idx = max(key_var_indices.values())
last_key_var = [var for var, idx in key_var_indices.items() if idx == last_key_var_idx][0]
assert last_key_var == "results", f"'results' should be the last key variable, but got {last_key_var}"
# Validate creation order in variables_manager
all_var_names = state.variables_manager.get_variable_names()
key_var_indices_in_manager = {var: all_var_names.index(var) for var in key_vars}
# Verify each key variable comes before the next one
assert key_var_indices_in_manager["email_lines"] < key_var_indices_in_manager["emails"], (
"'email_lines' should come before 'emails' in variables_manager"
)
assert key_var_indices_in_manager["emails"] < key_var_indices_in_manager["results"], (
"'emails' should come before 'results' in variables_manager"
)
# Validate order in summary (should match creation order)
summary = state.variables_manager.get_variables_summary()
key_var_positions = {var: summary.find(f"## {var}") for var in key_vars}
# Get all variable names from summary to find the last one
import re
all_vars_in_summary = re.findall(r'## (\w+)', summary)
# Verify each key variable appears before the next one in summary
assert key_var_positions["email_lines"] < key_var_positions["emails"], (
"'email_lines' should appear before 'emails' in summary"
)
assert key_var_positions["emails"] < key_var_positions["results"], (
"'emails' should appear before 'results' in summary"
)
# CRITICAL: Verify 'results' is the LAST variable in the summary
# This validates that 'results' (the final output) appears after all other variables
# including loop variables like 'email', 'contact_response', etc.
assert all_vars_in_summary[-1] == "results", (
f"'results' should be the LAST variable in summary, but got '{all_vars_in_summary[-1]}'. "
f"All variables in order: {all_vars_in_summary}"
)
# Also verify 'results' appears after all other key variables in summary
results_pos = key_var_positions["results"]
for var in ["email_lines", "emails"]:
assert key_var_positions[var] < results_pos, f"{var} should appear before 'results' in summary"
|