ed
16c6705b80
Phase 1.3 migrated run_worker_lifecycle to send_result(). The mock_ai_client fixture in test_spawn_interception_v2.py mocked src.ai_client.send and returned a string. The test_run_worker_lifecycle_approved test asserts on the call_args (user_message + md_content), which still works with the new mock name. Changes: - Replace patch(src.ai_client.send) with patch(src.ai_client.send_result) - Rename mock_send to mock_send_result - Wrap mock return_value with Result(data="Task completed") - Add "from src.result_types import Result" import All 3 tests in test_spawn_interception_v2.py pass.
87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from src import multi_agent_conductor
|
|
from src.models import Ticket, WorkerContext
|
|
from src import events
|
|
from src.result_types import Result
|
|
import threading
|
|
from typing import Generator
|
|
|
|
class MockDialog:
|
|
def __init__(self, approved: bool, final_payload: dict | None = None) -> None:
|
|
self.approved = approved
|
|
self.final_payload = final_payload
|
|
|
|
def wait(self) -> dict:
|
|
res = {'approved': self.approved, 'abort': False}
|
|
if self.final_payload:
|
|
res.update(self.final_payload)
|
|
return res
|
|
|
|
@pytest.fixture
|
|
def mock_ai_client() -> Generator[MagicMock, None, None]:
|
|
with patch("src.ai_client.send_result") as mock_send_result:
|
|
mock_send_result.return_value = Result(data="Task completed")
|
|
yield mock_send_result
|
|
|
|
def test_confirm_spawn_pushed_to_queue() -> None:
|
|
event_queue = events.SyncEventQueue()
|
|
ticket_id = "T1"
|
|
role = "Tier 3 Worker"
|
|
prompt = "Original Prompt"
|
|
context_md = "Original Context"
|
|
|
|
results = []
|
|
def run_confirm():
|
|
res = multi_agent_conductor.confirm_spawn(role, prompt, context_md, event_queue, ticket_id)
|
|
results.append(res)
|
|
|
|
t = threading.Thread(target=run_confirm)
|
|
t.start()
|
|
|
|
# Wait for the event to appear in the queue
|
|
event_name, payload = event_queue.get()
|
|
assert event_name == "mma_spawn_approval"
|
|
assert payload["ticket_id"] == ticket_id
|
|
assert payload["role"] == role
|
|
assert payload["prompt"] == prompt
|
|
assert payload["context_md"] == context_md
|
|
assert "dialog_container" in payload
|
|
|
|
# Simulate GUI injecting a dialog
|
|
payload["dialog_container"][0] = MockDialog(True, {"prompt": "Modified Prompt", "context_md": "Modified Context"})
|
|
|
|
t.join(timeout=5)
|
|
assert not t.is_alive()
|
|
approved, final_prompt, final_context = results[0]
|
|
assert approved is True
|
|
assert final_prompt == "Modified Prompt"
|
|
assert final_context == "Modified Context"
|
|
|
|
@patch("src.multi_agent_conductor.confirm_spawn")
|
|
def test_run_worker_lifecycle_approved(mock_confirm: MagicMock, mock_ai_client: MagicMock, app_instance) -> None:
|
|
ticket = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
|
|
context = WorkerContext(ticket_id="T1", model_name="model", messages=[])
|
|
event_queue = app_instance.controller.event_queue
|
|
mock_confirm.return_value = (True, "Modified Prompt", "Modified Context")
|
|
multi_agent_conductor.run_worker_lifecycle(ticket, context, event_queue=event_queue)
|
|
mock_confirm.assert_called_once()
|
|
# Check that ai_client.send was called with modified values
|
|
args, kwargs = mock_ai_client.call_args
|
|
assert kwargs["user_message"] == "Modified Prompt"
|
|
assert kwargs["md_content"] == "Modified Context"
|
|
assert ticket.status == "completed"
|
|
|
|
@patch("src.multi_agent_conductor.confirm_spawn")
|
|
def test_run_worker_lifecycle_rejected(mock_confirm: MagicMock, mock_ai_client: MagicMock, app_instance) -> None:
|
|
ticket = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
|
|
context = WorkerContext(ticket_id="T1", model_name="model", messages=[])
|
|
event_queue = app_instance.controller.event_queue
|
|
mock_confirm.return_value = (False, "Original Prompt", "Original Context")
|
|
result = multi_agent_conductor.run_worker_lifecycle(ticket, context, event_queue=event_queue)
|
|
mock_confirm.assert_called_once()
|
|
mock_ai_client.assert_not_called()
|
|
assert ticket.status == "blocked"
|
|
assert "Spawn rejected by user" in ticket.blocked_reason
|
|
assert "BLOCKED" in result
|