ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: STILL FIXING FUNDAMENTAL TRASH

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-03-05 14:04:17 -05:00
parent 70d18347d7
commit a13a6c5cd0
11 changed files with 113 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/test_spawn_interception_v2.py
View File

@@ -1,16 +1,13 @@
from src import ai_client
from src import events
from src import models
from src import multi_agent_conductor
import pytest
from unittest.mock import MagicMock, patch
from src import ai_client
from src import events
from src import models
from src import multi_agent_conductor
from src.models import Ticket, WorkerContext
from src import events
import time
import threading
from typing import Generator, Any
:
class MockDialog:
def __init__(self, approved: bool, final_payload: dict | None = None) -> None:
self.approved = approved
self.final_payload = final_payload
@@ -22,7 +19,7 @@ from src import multi_agent_conductor
return res
@pytest.fixture
def mock_ai_client() -> None:
def mock_ai_client() -> Generator[MagicMock, None, None]:
with patch("src.ai_client.send") as mock_send:
mock_send.return_value = "Task completed"
yield mock_send
@@ -38,10 +35,10 @@ def test_confirm_spawn_pushed_to_queue() -> None:
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"
@@ -50,10 +47,10 @@ def test_confirm_spawn_pushed_to_queue() -> None:
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]
@@ -63,9 +60,9 @@ def test_confirm_spawn_pushed_to_queue() -> None:
@patch("src.multi_agent_conductor.confirm_spawn")
def test_run_worker_lifecycle_approved(mock_confirm: MagicMock, mock_ai_client: MagicMock, app_instance) -> None:
ticket = models.Ticket(id="T1", description="desc", status="todo", assigned_to="user")
ticket = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
context = WorkerContext(ticket_id="T1", model_name="model", messages=[])
event_queue = app_instance.event_queue
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()
@@ -77,9 +74,9 @@ def test_run_worker_lifecycle_approved(mock_confirm: MagicMock, mock_ai_client:
@patch("src.multi_agent_conductor.confirm_spawn")
def test_run_worker_lifecycle_rejected(mock_confirm: MagicMock, mock_ai_client: MagicMock, app_instance) -> None:
ticket = models.Ticket(id="T1", description="desc", status="todo", assigned_to="user")
ticket = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
context = WorkerContext(ticket_id="T1", model_name="model", messages=[])
event_queue = app_instance.event_queue
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()