197 lines
8.4 KiB
Python
197 lines
8.4 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from models import Ticket, Track, WorkerContext
|
|
|
|
# These tests define the expected interface for multi_agent_conductor.py
|
|
# which will be implemented in the next phase of TDD.
|
|
|
|
def test_conductor_engine_initialization():
|
|
"""
|
|
Test that ConductorEngine can be initialized with a Track.
|
|
"""
|
|
track = Track(id="test_track", description="Test Track")
|
|
from multi_agent_conductor import ConductorEngine
|
|
engine = ConductorEngine(track=track)
|
|
assert engine.track == track
|
|
|
|
def test_conductor_engine_run_linear_executes_tickets_in_order():
|
|
"""
|
|
Test that run_linear iterates through executable tickets and calls the worker lifecycle.
|
|
"""
|
|
ticket1 = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1")
|
|
ticket2 = Ticket(id="T2", description="Task 2", status="todo", assigned_to="worker2", depends_on=["T1"])
|
|
track = Track(id="track1", description="Track 1", tickets=[ticket1, ticket2])
|
|
|
|
from multi_agent_conductor import ConductorEngine
|
|
engine = ConductorEngine(track=track)
|
|
|
|
# We mock run_worker_lifecycle as it is expected to be in the same module
|
|
with patch("multi_agent_conductor.run_worker_lifecycle") as mock_lifecycle:
|
|
# Mocking lifecycle to mark ticket as complete so dependencies can be resolved
|
|
def side_effect(ticket, context):
|
|
ticket.mark_complete()
|
|
return "Success"
|
|
mock_lifecycle.side_effect = side_effect
|
|
|
|
engine.run_linear()
|
|
|
|
# Track.get_executable_tickets() should be called repeatedly until all are done
|
|
# T1 should run first, then T2.
|
|
assert mock_lifecycle.call_count == 2
|
|
assert ticket1.status == "completed"
|
|
assert ticket2.status == "completed"
|
|
|
|
# Verify sequence: T1 before T2
|
|
calls = mock_lifecycle.call_args_list
|
|
assert calls[0][0][0].id == "T1"
|
|
assert calls[1][0][0].id == "T2"
|
|
|
|
def test_run_worker_lifecycle_calls_ai_client_send():
|
|
"""
|
|
Test that run_worker_lifecycle triggers the AI client and updates ticket status on success.
|
|
"""
|
|
ticket = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1")
|
|
context = WorkerContext(ticket_id="T1", model_name="test-model", messages=[])
|
|
|
|
from multi_agent_conductor import run_worker_lifecycle
|
|
|
|
with patch("ai_client.send") as mock_send:
|
|
mock_send.return_value = "Task complete. I have updated the file."
|
|
|
|
result = run_worker_lifecycle(ticket, context)
|
|
|
|
assert result == "Task complete. I have updated the file."
|
|
assert ticket.status == "completed"
|
|
mock_send.assert_called_once()
|
|
|
|
# Check if description was passed to send()
|
|
args, kwargs = mock_send.call_args
|
|
# user_message is passed as a keyword argument
|
|
assert ticket.description in kwargs["user_message"]
|
|
|
|
def test_run_worker_lifecycle_context_injection():
|
|
"""
|
|
Test that run_worker_lifecycle can take a context_files list and injects AST views into the prompt.
|
|
"""
|
|
ticket = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1")
|
|
context = WorkerContext(ticket_id="T1", model_name="test-model", messages=[])
|
|
context_files = ["primary.py", "secondary.py"]
|
|
|
|
from multi_agent_conductor import run_worker_lifecycle
|
|
|
|
# We mock ASTParser which is expected to be imported in multi_agent_conductor
|
|
with patch("ai_client.send") as mock_send, \
|
|
patch("multi_agent_conductor.ASTParser") as mock_ast_parser_class, \
|
|
patch("builtins.open", new_callable=MagicMock) as mock_open:
|
|
|
|
# Setup open mock to return different content for different files
|
|
file_contents = {
|
|
"primary.py": "def primary(): pass",
|
|
"secondary.py": "def secondary(): pass"
|
|
}
|
|
|
|
def mock_open_side_effect(file, *args, **kwargs):
|
|
content = file_contents.get(file, "")
|
|
mock_file = MagicMock()
|
|
mock_file.read.return_value = content
|
|
mock_file.__enter__.return_value = mock_file
|
|
return mock_file
|
|
|
|
mock_open.side_effect = mock_open_side_effect
|
|
|
|
# Setup ASTParser mock
|
|
mock_ast_parser = mock_ast_parser_class.return_value
|
|
mock_ast_parser.get_curated_view.return_value = "CURATED VIEW"
|
|
mock_ast_parser.get_skeleton.return_value = "SKELETON VIEW"
|
|
|
|
mock_send.return_value = "Success"
|
|
|
|
run_worker_lifecycle(ticket, context, context_files=context_files)
|
|
|
|
# Verify ASTParser calls:
|
|
# First file (primary) should get curated view, others (secondary) get skeleton
|
|
mock_ast_parser.get_curated_view.assert_called_once_with("def primary(): pass")
|
|
mock_ast_parser.get_skeleton.assert_called_once_with("def secondary(): pass")
|
|
|
|
# Verify user_message contains the views
|
|
_, kwargs = mock_send.call_args
|
|
user_message = kwargs["user_message"]
|
|
assert "CURATED VIEW" in user_message
|
|
assert "SKELETON VIEW" in user_message
|
|
assert "primary.py" in user_message
|
|
assert "secondary.py" in user_message
|
|
|
|
def test_run_worker_lifecycle_handles_blocked_response():
|
|
"""
|
|
Test that run_worker_lifecycle marks the ticket as blocked if the AI indicates it cannot proceed.
|
|
"""
|
|
ticket = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1")
|
|
context = WorkerContext(ticket_id="T1", model_name="test-model", messages=[])
|
|
|
|
from multi_agent_conductor import run_worker_lifecycle
|
|
|
|
with patch("ai_client.send") as mock_send:
|
|
# Simulate a response indicating a block
|
|
mock_send.return_value = "I am BLOCKED because I don't have enough information."
|
|
|
|
run_worker_lifecycle(ticket, context)
|
|
|
|
assert ticket.status == "blocked"
|
|
assert "BLOCKED" in ticket.blocked_reason
|
|
|
|
def test_run_worker_lifecycle_step_mode_confirmation():
|
|
"""
|
|
Test that run_worker_lifecycle passes confirm_execution to ai_client.send when step_mode is True.
|
|
Verify that if confirm_execution is called (simulated by mocking ai_client.send to call its callback),
|
|
the flow works as expected.
|
|
"""
|
|
ticket = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1", step_mode=True)
|
|
context = WorkerContext(ticket_id="T1", model_name="test-model", messages=[])
|
|
|
|
from multi_agent_conductor import run_worker_lifecycle, confirm_execution
|
|
|
|
with patch("ai_client.send") as mock_send, \
|
|
patch("multi_agent_conductor.confirm_execution") as mock_confirm:
|
|
|
|
# We simulate ai_client.send by making it call the pre_tool_callback it received
|
|
def mock_send_side_effect(*args, **kwargs):
|
|
callback = kwargs.get("pre_tool_callback")
|
|
if callback:
|
|
# Simulate calling it with some payload
|
|
callback('{"tool": "read_file", "args": {"path": "test.txt"}}')
|
|
return "Success"
|
|
|
|
mock_send.side_effect = mock_send_side_effect
|
|
mock_confirm.return_value = True
|
|
|
|
run_worker_lifecycle(ticket, context)
|
|
|
|
# Verify confirm_execution was called
|
|
mock_confirm.assert_called_once()
|
|
assert ticket.status == "completed"
|
|
|
|
def test_run_worker_lifecycle_step_mode_rejection():
|
|
"""
|
|
Verify that if confirm_execution returns False, the logic (in ai_client, which we simulate here)
|
|
would prevent execution. In run_worker_lifecycle, we just check if it's passed.
|
|
"""
|
|
ticket = Ticket(id="T1", description="Task 1", status="todo", assigned_to="worker1", step_mode=True)
|
|
context = WorkerContext(ticket_id="T1", model_name="test-model", messages=[])
|
|
|
|
from multi_agent_conductor import run_worker_lifecycle
|
|
|
|
with patch("ai_client.send") as mock_send, \
|
|
patch("multi_agent_conductor.confirm_execution") as mock_confirm:
|
|
|
|
mock_confirm.return_value = False
|
|
mock_send.return_value = "Task failed because tool execution was rejected."
|
|
|
|
run_worker_lifecycle(ticket, context)
|
|
|
|
# Verify it was passed to send
|
|
args, kwargs = mock_send.call_args
|
|
assert kwargs["pre_tool_callback"] is not None
|
|
|
|
# Since we've already tested ai_client's implementation of pre_tool_callback (mentally or via other tests),
|
|
# here we just verify the wiring.
|