37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from ea_chatbot.graph.nodes.executor import executor_node
|
|
from ea_chatbot.graph.state import AgentState
|
|
|
|
def test_executor_with_vfs():
|
|
"""Verify that the executor node provides VFS access to the code."""
|
|
state = AgentState(
|
|
messages=[],
|
|
question="test",
|
|
analysis={},
|
|
next_action="test",
|
|
iterations=0,
|
|
checklist=[],
|
|
current_step=0,
|
|
vfs={},
|
|
plots=[],
|
|
dfs={}
|
|
)
|
|
|
|
# Code that uses the 'vfs' helper
|
|
code = """
|
|
vfs.write("output.txt", "Execution Result", metadata={"type": "text"})
|
|
print("VFS Write Complete")
|
|
"""
|
|
state["code"] = code
|
|
|
|
result = executor_node(state)
|
|
|
|
# Check if the execution was successful
|
|
assert result["error"] is None
|
|
assert "VFS Write Complete" in result["code_output"]
|
|
|
|
# Verify that the VFS state was updated
|
|
# Note: executor_node returns a dict of updates, which should include the updated 'vfs'
|
|
assert "vfs" in result
|
|
assert "output.txt" in result["vfs"]
|
|
assert result["vfs"]["output.txt"]["content"] == "Execution Result"
|