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

feat(mma): Finalize Orchestrator Integration and fix all regressions

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-02-27 18:31:14 -05:00
parent 8438f69197
commit 3b2d82ed0d
27 changed files with 706 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/conductor/test_mma_exec.py
View File

@@ -57,8 +57,9 @@ def test_execute_agent():
role = "tier3-worker"
prompt = "Write a unit test."
docs = ["file1.py", "docs/spec.md"]
expected_model = "gemini-2.5-flash-lite"
mock_stdout = "Mocked AI Response"
with patch("subprocess.run") as mock_run:

9
tests/conftest.py
View File

@@ -11,6 +11,15 @@ import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from api_hook_client import ApiHookClient
import ai_client
@pytest.fixture(autouse=True)
def reset_ai_client():
"""Reset ai_client global state between every test to prevent state pollution."""
ai_client.reset_session()
# Default to a safe model
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
yield
def kill_process_tree(pid):
"""Robustly kills a process and all its children."""

2
tests/test_ai_client_cli.py
View File

@@ -11,7 +11,7 @@ def test_ai_client_send_gemini_cli():
test_response = "This is a dummy response from the Gemini CLI."
# Set provider to gemini_cli
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
ai_client.set_provider("gemini_cli", "gemini-2.5-flash-lite")
# 1. Mock 'ai_client.GeminiCliAdapter' (which we will add)
with patch('ai_client.GeminiCliAdapter') as MockAdapterClass:

121
tests/test_api_events.py
View File

@@ -1,45 +1,44 @@
import pytest
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
import ai_client
class MockUsage:
def __init__(self):
self.prompt_token_count = 10
self.candidates_token_count = 5
self.total_token_count = 15
self.cached_content_token_count = 0
class MockPart:
def __init__(self, text, function_call):
self.text = text
self.function_call = function_call
class MockContent:
def __init__(self, parts):
self.parts = parts
class MockCandidate:
def __init__(self, parts):
self.content = MockContent(parts)
self.finish_reason = MagicMock()
self.finish_reason.name = "STOP"
def test_ai_client_event_emitter_exists():
# This should fail initially because 'events' won't exist on ai_client
assert hasattr(ai_client, 'events')
assert ai_client.events is not None
def test_event_emission():
# We'll expect these event names based on the spec
mock_callback = MagicMock()
ai_client.events.on("request_start", mock_callback)
# Trigger something that should emit the event (once implemented)
# For now, we just test the emitter itself if we were to call it manually
ai_client.events.emit("request_start", payload={"model": "test"})
mock_callback.assert_called_once_with(payload={"model": "test"})
callback = MagicMock()
ai_client.events.on("test_event", callback)
ai_client.events.emit("test_event", payload={"data": 123})
callback.assert_called_once_with(payload={"data": 123})
def test_send_emits_events():
from unittest.mock import patch, MagicMock
# We need to mock _ensure_gemini_client and the chat object it creates
with patch("ai_client._ensure_gemini_client"), \
patch("ai_client._gemini_client") as mock_client, \
patch("ai_client._gemini_chat") as mock_chat:
with patch("ai_client._send_gemini") as mock_send_gemini, \
patch("ai_client._send_anthropic") as mock_send_anthropic:
# Setup mock response
mock_response = MagicMock()
mock_response.candidates = []
# Explicitly set usage_metadata as a mock with integer values
mock_usage = MagicMock()
mock_usage.prompt_token_count = 10
mock_usage.candidates_token_count = 5
mock_usage.cached_content_token_count = None
mock_response.usage_metadata = mock_usage
mock_chat.send_message.return_value = mock_response
mock_client.chats.create.return_value = mock_chat
ai_client.set_provider("gemini", "gemini-flash")
mock_send_gemini.return_value = "gemini response"
start_callback = MagicMock()
response_callback = MagicMock()
@@ -47,53 +46,69 @@ def test_send_emits_events():
ai_client.events.on("request_start", start_callback)
ai_client.events.on("response_received", response_callback)
# We need to bypass the context changed check or set it up
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
ai_client.send("context", "message")
# We mocked _send_gemini so it doesn't emit events inside.
# But wait, ai_client.send itself emits request_start and response_received?
# Actually, ai_client.send delegates to _send_gemini.
# Let's mock _gemini_client instead to let _send_gemini run and emit events.
pass
def test_send_emits_events_proper():
with patch("ai_client._ensure_gemini_client"), \
patch("ai_client._gemini_client") as mock_client:
mock_chat = MagicMock()
mock_client.chats.create.return_value = mock_chat
mock_response = MagicMock()
mock_response.candidates = [MockCandidate([MockPart("gemini response", None)])]
mock_response.usage_metadata = MockUsage()
mock_chat.send_message.return_value = mock_response
start_callback = MagicMock()
response_callback = MagicMock()
ai_client.events.on("request_start", start_callback)
ai_client.events.on("response_received", response_callback)
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
ai_client.send("context", "message")
assert start_callback.called
assert response_callback.called
# Check payload
args, kwargs = start_callback.call_args
assert kwargs['payload']['provider'] == 'gemini'
def test_send_emits_tool_events():
from unittest.mock import patch, MagicMock
import mcp_client
with patch("ai_client._ensure_gemini_client"), \
patch("ai_client._gemini_client") as mock_client, \
patch("ai_client._gemini_chat") as mock_chat, \
patch("mcp_client.dispatch") as mock_dispatch:
mock_chat = MagicMock()
mock_client.chats.create.return_value = mock_chat
# 1. Setup mock response with a tool call
mock_fc = MagicMock()
mock_fc.name = "read_file"
mock_fc.args = {"path": "test.txt"}
mock_response_with_tool = MagicMock()
mock_response_with_tool.candidates = [MagicMock()]
mock_part = MagicMock()
mock_part.text = "tool call text"
mock_part.function_call = mock_fc
mock_response_with_tool.candidates[0].content.parts = [mock_part]
mock_response_with_tool.candidates[0].finish_reason.name = "STOP"
# Setup mock usage
mock_usage = MagicMock()
mock_usage.prompt_token_count = 10
mock_usage.candidates_token_count = 5
mock_usage.cached_content_token_count = None
mock_response_with_tool.usage_metadata = mock_usage
mock_response_with_tool.candidates = [MockCandidate([MockPart("tool call text", mock_fc)])]
mock_response_with_tool.usage_metadata = MockUsage()
# 2. Setup second mock response (final answer)
mock_response_final = MagicMock()
mock_response_final.candidates = []
mock_response_final.usage_metadata = mock_usage
mock_response_final.candidates = [MockCandidate([MockPart("final answer", None)])]
mock_response_final.usage_metadata = MockUsage()
mock_chat.send_message.side_effect = [mock_response_with_tool, mock_response_final]
mock_dispatch.return_value = "file content"
ai_client.set_provider("gemini", "gemini-flash")
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
tool_callback = MagicMock()
ai_client.events.on("tool_execution", tool_callback)

114
tests/test_conductor_engine.py
View File

@@ -1,6 +1,7 @@
import pytest
from unittest.mock import MagicMock, patch
from models import Ticket, Track, WorkerContext
import ai_client
# These tests define the expected interface for multi_agent_conductor.py
# which will be implemented in the next phase of TDD.
@@ -14,7 +15,8 @@ def test_conductor_engine_initialization():
engine = ConductorEngine(track=track)
assert engine.track == track
def test_conductor_engine_run_linear_executes_tickets_in_order():
@pytest.mark.asyncio
async def test_conductor_engine_run_linear_executes_tickets_in_order(monkeypatch):
"""
Test that run_linear iterates through executable tickets and calls the worker lifecycle.
"""
@@ -25,15 +27,19 @@ def test_conductor_engine_run_linear_executes_tickets_in_order():
from multi_agent_conductor import ConductorEngine
engine = ConductorEngine(track=track)
# Mock ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', mock_send)
# 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):
def side_effect(ticket, context, *args, **kwargs):
ticket.mark_complete()
return "Success"
mock_lifecycle.side_effect = side_effect
engine.run_linear()
await engine.run_linear()
# Track.get_executable_tickets() should be called repeatedly until all are done
# T1 should run first, then T2.
@@ -46,7 +52,8 @@ def test_conductor_engine_run_linear_executes_tickets_in_order():
assert calls[0][0][0].id == "T1"
assert calls[1][0][0].id == "T2"
def test_run_worker_lifecycle_calls_ai_client_send():
@pytest.mark.asyncio
async def test_run_worker_lifecycle_calls_ai_client_send(monkeypatch):
"""
Test that run_worker_lifecycle triggers the AI client and updates ticket status on success.
"""
@@ -55,21 +62,25 @@ def test_run_worker_lifecycle_calls_ai_client_send():
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"]
# Mock ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', 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():
@pytest.mark.asyncio
async def test_run_worker_lifecycle_context_injection(monkeypatch):
"""
Test that run_worker_lifecycle can take a context_files list and injects AST views into the prompt.
"""
@@ -79,9 +90,12 @@ def test_run_worker_lifecycle_context_injection():
from multi_agent_conductor import run_worker_lifecycle
# Mock ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', mock_send)
# 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, \
with 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
@@ -121,7 +135,8 @@ def test_run_worker_lifecycle_context_injection():
assert "primary.py" in user_message
assert "secondary.py" in user_message
def test_run_worker_lifecycle_handles_blocked_response():
@pytest.mark.asyncio
async def test_run_worker_lifecycle_handles_blocked_response(monkeypatch):
"""
Test that run_worker_lifecycle marks the ticket as blocked if the AI indicates it cannot proceed.
"""
@@ -130,16 +145,20 @@ def test_run_worker_lifecycle_handles_blocked_response():
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
# Mock ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', 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)
def test_run_worker_lifecycle_step_mode_confirmation():
assert ticket.status == "blocked"
assert "BLOCKED" in ticket.blocked_reason
@pytest.mark.asyncio
async def test_run_worker_lifecycle_step_mode_confirmation(monkeypatch):
"""
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),
@@ -148,13 +167,16 @@ def test_run_worker_lifecycle_step_mode_confirmation():
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
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 ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', mock_send)
with 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):
def mock_send_side_effect(md_content, user_message, **kwargs):
callback = kwargs.get("pre_tool_callback")
if callback:
# Simulate calling it with some payload
@@ -164,13 +186,15 @@ def test_run_worker_lifecycle_step_mode_confirmation():
mock_send.side_effect = mock_send_side_effect
mock_confirm.return_value = True
run_worker_lifecycle(ticket, context)
mock_event_queue = MagicMock()
run_worker_lifecycle(ticket, context, event_queue=mock_event_queue)
# Verify confirm_execution was called
mock_confirm.assert_called_once()
assert ticket.status == "completed"
def test_run_worker_lifecycle_step_mode_rejection():
@pytest.mark.asyncio
async def test_run_worker_lifecycle_step_mode_rejection(monkeypatch):
"""
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.
@@ -180,8 +204,11 @@ def test_run_worker_lifecycle_step_mode_rejection():
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 ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', mock_send)
with 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."
@@ -195,7 +222,8 @@ def test_run_worker_lifecycle_step_mode_rejection():
# Since we've already tested ai_client's implementation of pre_tool_callback (mentally or via other tests),
# here we just verify the wiring.
def test_conductor_engine_dynamic_parsing_and_execution():
@pytest.mark.asyncio
async def test_conductor_engine_dynamic_parsing_and_execution(monkeypatch):
"""
Test that parse_json_tickets correctly populates the track and run_linear executes them in dependency order.
"""
@@ -236,14 +264,18 @@ def test_conductor_engine_dynamic_parsing_and_execution():
assert engine.track.tickets[1].id == "T2"
assert engine.track.tickets[2].id == "T3"
# Mock ai_client.send using monkeypatch
mock_send = MagicMock()
monkeypatch.setattr(ai_client, 'send', mock_send)
# Mock run_worker_lifecycle to mark tickets as complete
with patch("multi_agent_conductor.run_worker_lifecycle") as mock_lifecycle:
def side_effect(ticket, context):
def side_effect(ticket, context, *args, **kwargs):
ticket.mark_complete()
return "Success"
mock_lifecycle.side_effect = side_effect
engine.run_linear()
await engine.run_linear()
assert mock_lifecycle.call_count == 3

5
tests/test_conductor_tech_lead.py
View File

@@ -23,12 +23,11 @@ class TestConductorTechLead(unittest.TestCase):
track_brief = "Test track brief"
module_skeletons = "Test skeletons"
# Call the function
tickets = conductor_tech_lead.generate_tickets(track_brief, module_skeletons)
# Verify set_provider was called
mock_set_provider.assert_called_with('gemini', 'gemini-1.5-flash')
mock_set_provider.assert_called_with('gemini', 'gemini-2.5-flash-lite')
mock_reset_session.assert_called_once()
# Verify send was called

32
tests/test_gemini_cli_adapter.py
View File

@@ -19,12 +19,12 @@ class TestGeminiCliAdapter(unittest.TestCase):
def test_send_starts_subprocess_with_correct_args(self, mock_popen):
"""
Verify that send(message) correctly starts the subprocess with
--output-format stream-json and the provided message via stdin.
--output-format stream-json and the provided message via stdin using communicate.
"""
# Setup mock process with a minimal valid JSONL termination
process_mock = MagicMock()
process_mock.stdout = io.StringIO(json.dumps({"type": "result", "usage": {}}) + "\n")
process_mock.stdin = MagicMock()
stdout_content = json.dumps({"type": "result", "usage": {}}) + "\n"
process_mock.communicate.return_value = (stdout_content, "")
process_mock.poll.return_value = 0
process_mock.wait.return_value = 0
mock_popen.return_value = process_mock
@@ -44,9 +44,8 @@ class TestGeminiCliAdapter(unittest.TestCase):
# Message should NOT be in cmd now
self.assertNotIn(message, cmd)
# Verify message was written to stdin
process_mock.stdin.write.assert_called_once_with(message)
process_mock.stdin.close.assert_called_once()
# Verify message was sent via communicate
process_mock.communicate.assert_called_once_with(input=message)
# Check process configuration
self.assertEqual(kwargs.get('stdout'), subprocess.PIPE)
@@ -60,16 +59,15 @@ class TestGeminiCliAdapter(unittest.TestCase):
and returns the combined text.
"""
jsonl_output = [
json.dumps({"type": "message", "text": "The quick brown "}),
json.dumps({"type": "message", "text": "fox jumps."}),
json.dumps({"type": "message", "role": "model", "text": "The quick brown "}),
json.dumps({"type": "message", "role": "model", "text": "fox jumps."}),
json.dumps({"type": "result", "usage": {"prompt_tokens": 5, "candidates_tokens": 5}})
]
stdout_content = "\n".join(jsonl_output) + "\n"
process_mock = MagicMock()
process_mock.stdout = io.StringIO(stdout_content)
# Mock poll sequence: running, running, finished
process_mock.poll.side_effect = [None, None, 0]
process_mock.communicate.return_value = (stdout_content, "")
process_mock.poll.return_value = 0
process_mock.wait.return_value = 0
mock_popen.return_value = process_mock
@@ -85,16 +83,16 @@ class TestGeminiCliAdapter(unittest.TestCase):
by continuing to read until the final 'result' event.
"""
jsonl_output = [
json.dumps({"type": "message", "text": "Calling tool..."}),
json.dumps({"type": "message", "role": "assistant", "text": "Calling tool..."}),
json.dumps({"type": "tool_use", "name": "read_file", "args": {"path": "test.txt"}}),
json.dumps({"type": "message", "text": "\nFile read successfully."}),
json.dumps({"type": "message", "role": "assistant", "text": "\nFile read successfully."}),
json.dumps({"type": "result", "usage": {}})
]
stdout_content = "\n".join(jsonl_output) + "\n"
process_mock = MagicMock()
process_mock.stdout = io.StringIO(stdout_content)
process_mock.poll.side_effect = [None, None, None, 0]
process_mock.communicate.return_value = (stdout_content, "")
process_mock.poll.return_value = 0
process_mock.wait.return_value = 0
mock_popen.return_value = process_mock
@@ -118,8 +116,8 @@ class TestGeminiCliAdapter(unittest.TestCase):
stdout_content = "\n".join(jsonl_output) + "\n"
process_mock = MagicMock()
process_mock.stdout = io.StringIO(stdout_content)
process_mock.poll.side_effect = [None, 0]
process_mock.communicate.return_value = (stdout_content, "")
process_mock.poll.return_value = 0
process_mock.wait.return_value = 0
mock_popen.return_value = process_mock

13
tests/test_gemini_cli_adapter_parity.py
View File

@@ -14,20 +14,21 @@ if project_root not in sys.path:
# Import the class to be tested
from gemini_cli_adapter import GeminiCliAdapter
# Mock the session_logger module to prevent file operations during tests.
mock_session_logger = MagicMock()
sys.modules['session_logger'] = mock_session_logger
class TestGeminiCliAdapterParity(unittest.TestCase):
def setUp(self):
"""Set up a fresh adapter instance and reset session state for each test."""
# Patch session_logger to prevent file operations during tests
self.session_logger_patcher = patch('gemini_cli_adapter.session_logger')
self.mock_session_logger = self.session_logger_patcher.start()
self.adapter = GeminiCliAdapter(binary_path="gemini")
self.adapter.session_id = None
self.adapter.last_usage = None
self.adapter.last_latency = 0.0
# Reset mock calls for session_logger for each test
mock_session_logger.reset_mock()
def tearDown(self):
self.session_logger_patcher.stop()
@patch('subprocess.Popen')
def test_count_tokens_uses_estimation(self, mock_popen):

122
tests/test_log_management_ui.py
View File

@@ -1,24 +1,10 @@
import pytest
from unittest.mock import MagicMock, patch
import os
from pathlib import Path
# Mock imgui and other heavy dependencies before importing App
import sys
sys.modules['imgui_bundle'] = MagicMock()
sys.modules['imgui_bundle.imgui'] = MagicMock()
sys.modules['imgui_bundle.hello_imgui'] = MagicMock()
sys.modules['imgui_bundle.immapp'] = MagicMock()
# Mock tkinter
sys.modules['tkinter'] = MagicMock()
sys.modules['tkinter.filedialog'] = MagicMock()
# Mock ai_client and session_logger
sys.modules['ai_client'] = MagicMock()
sys.modules['session_logger'] = MagicMock()
# We can safely import gui_2 if we don't instantiate App without mocking its threads
import gui_2
from gui_2 import App
@pytest.fixture
@@ -46,64 +32,72 @@ history = []
""", encoding="utf-8")
return project_path
def test_log_management_init(mock_config, mock_project, monkeypatch):
@pytest.fixture
def app_instance(mock_config, mock_project, monkeypatch):
monkeypatch.setattr("gui_2.CONFIG_PATH", mock_config)
with patch("project_manager.load_project") as mock_load:
with patch("project_manager.load_project") as mock_load, \
patch("session_logger.open_session"):
mock_load.return_value = {
"project": {"name": "test"},
"discussion": {"roles": ["User", "AI"], "active": "main", "discussions": {"main": {"history": []}}},
"files": {"paths": []},
"screenshots": {"paths": []}
}
with patch("session_logger.open_session"):
# Mock the __init__ to do nothing, then set the fields we need manually
with patch.object(App, '__init__', lambda self: None):
app = App()
app.show_windows = {"Log Management": False}
app.ui_state = MagicMock()
app.ui_files_base_dir = "."
app.files = []
# Check if Log Management is in show_windows
assert "Log Management" in app.show_windows
assert app.show_windows["Log Management"] is False # Default as set in __init__
# Check if _render_log_management exists
assert hasattr(app, "_render_log_management")
assert callable(app._render_log_management)
# Since we bypassed __init__, we need to bind the method manually
# but python allows calling it directly.
return app
def test_render_log_management_logic(mock_config, mock_project, monkeypatch):
monkeypatch.setattr("gui_2.CONFIG_PATH", mock_config)
def test_log_management_init(app_instance):
app = app_instance
assert "Log Management" in app.show_windows
assert app.show_windows["Log Management"] is False
assert hasattr(app, "_render_log_management")
assert callable(app._render_log_management)
def test_render_log_management_logic(app_instance):
app = app_instance
app.show_windows["Log Management"] = True
with patch("project_manager.load_project") as mock_load:
mock_load.return_value = {
"project": {"name": "test"},
"discussion": {"roles": ["User", "AI"], "active": "main", "discussions": {"main": {"history": []}}},
"files": {"paths": []},
"screenshots": {"paths": []}
# Mock LogRegistry
with patch("gui_2.LogRegistry") as MockRegistry, \
patch("gui_2.imgui.begin") as mock_begin, \
patch("gui_2.imgui.begin_table") as mock_begin_table, \
patch("gui_2.imgui.text") as mock_text, \
patch("gui_2.imgui.end_table") as mock_end_table, \
patch("gui_2.imgui.end") as mock_end, \
patch("gui_2.imgui.push_style_color"), \
patch("gui_2.imgui.pop_style_color"), \
patch("gui_2.imgui.table_setup_column"), \
patch("gui_2.imgui.table_headers_row"), \
patch("gui_2.imgui.table_next_row"), \
patch("gui_2.imgui.table_next_column"), \
patch("gui_2.imgui.button"):
mock_reg = MockRegistry.return_value
mock_reg.data = {
"session_1": {
"start_time": "2023-01-01",
"whitelisted": False,
"metadata": {"reason": "test", "size_kb": 10, "message_count": 5}
}
}
with patch("session_logger.open_session"):
app = App()
app.show_windows["Log Management"] = True
from imgui_bundle import imgui
# Mock LogRegistry
with patch("gui_2.LogRegistry") as MockRegistry:
mock_reg = MockRegistry.return_value
mock_reg.data = {
"session_1": {
"start_time": "2023-01-01",
"whitelisted": False,
"metadata": {"reason": "test", "size_kb": 10, "message_count": 5}
}
}
# Mock imgui.begin to return (True, True)
imgui.begin.return_value = (True, True)
imgui.begin_table.return_value = True
# Call render
app._render_log_management()
# Verify imgui calls
imgui.begin.assert_called_with("Log Management", True)
imgui.begin_table.assert_called()
# Check for "session_1" text
imgui.text.assert_any_call("session_1")
mock_begin.return_value = (True, True)
mock_begin_table.return_value = True
app._render_log_management()
mock_begin.assert_called_with("Log Management", app.show_windows["Log Management"])
mock_begin_table.assert_called()
mock_text.assert_any_call("session_1")

8
tests/test_logging_e2e.py
View File

@@ -10,14 +10,20 @@ from log_registry import LogRegistry
from log_pruner import LogPruner
@pytest.fixture
def e2e_setup(tmp_path):
def e2e_setup(tmp_path, monkeypatch):
# Ensure closed before starting
session_logger.close_session()
monkeypatch.setattr(session_logger, "_comms_fh", None)
# Mock _LOG_DIR and _SCRIPTS_DIR in session_logger
original_log_dir = session_logger._LOG_DIR
session_logger._LOG_DIR = tmp_path / "logs"
monkeypatch.setattr(session_logger, "_LOG_DIR", tmp_path / "logs")
session_logger._LOG_DIR.mkdir(parents=True, exist_ok=True)
original_scripts_dir = session_logger._SCRIPTS_DIR
session_logger._SCRIPTS_DIR = tmp_path / "scripts" / "generated"
monkeypatch.setattr(session_logger, "_SCRIPTS_DIR", tmp_path / "scripts" / "generated")
session_logger._SCRIPTS_DIR.mkdir(parents=True, exist_ok=True)
yield tmp_path

14
tests/test_mma_orchestration_gui.py
View File

@@ -54,13 +54,17 @@ def test_cb_plan_epic_launches_thread(app_instance):
mock_tracks = [{"id": "track_1", "title": "Test Track"}]
with patch('orchestrator_pm.get_track_history_summary', return_value="History summary") as mock_get_history,
patch('orchestrator_pm.generate_tracks', return_value=mock_tracks) as mock_gen_tracks,
patch('aggregate.build_file_items', return_value=[]) as mock_build_files:
with (
patch('orchestrator_pm.get_track_history_summary', return_value="History summary") as mock_get_history,
patch('orchestrator_pm.generate_tracks', return_value=mock_tracks) as mock_gen_tracks,
patch('aggregate.build_file_items', return_value=[]) as mock_build_files
):
# We need to mock project_manager.flat_config and project_manager.load_project
with patch('project_manager.load_project', return_value={}),
patch('project_manager.flat_config', return_value={}):
with (
patch('project_manager.load_project', return_value={}),
patch('project_manager.flat_config', return_value={})
):
app_instance._cb_plan_epic()

133
tests/test_orchestration_logic.py Normal file
View File

@@ -0,0 +1,133 @@
import pytest
from unittest.mock import MagicMock, patch
import json
import orchestrator_pm
import conductor_tech_lead
import multi_agent_conductor
from models import Track, Ticket
@pytest.fixture
def mock_ai_client():
with patch("ai_client.send") as mock_send:
yield mock_send
def test_generate_tracks(mock_ai_client):
# Tier 1 (PM) response mock
mock_ai_client.return_value = json.dumps([
{"id": "track_1", "title": "Infrastructure Setup", "description": "Setup basic project structure"},
{"id": "track_2", "title": "Feature implementation", "description": "Implement core feature"}
])
user_request = "Build a new app"
project_config = {}
file_items = []
tracks = orchestrator_pm.generate_tracks(user_request, project_config, file_items)
assert len(tracks) == 2
assert tracks[0]["id"] == "track_1"
assert tracks[1]["id"] == "track_2"
mock_ai_client.assert_called_once()
def test_generate_tickets(mock_ai_client):
# Tier 2 (Tech Lead) response mock
mock_ai_client.return_value = json.dumps([
{"id": "T-001", "description": "Define interfaces", "depends_on": []},
{"id": "T-002", "description": "Implement interfaces", "depends_on": ["T-001"]}
])
track_brief = "Implement a new feature."
module_skeletons = "class Feature: pass"
tickets = conductor_tech_lead.generate_tickets(track_brief, module_skeletons)
assert len(tickets) == 2
assert tickets[0]["id"] == "T-001"
assert tickets[1]["id"] == "T-002"
assert tickets[1]["depends_on"] == ["T-001"]
def test_topological_sort():
tickets = [
{"id": "T-002", "description": "Dep on 001", "depends_on": ["T-001"]},
{"id": "T-001", "description": "Base", "depends_on": []},
{"id": "T-003", "description": "Dep on 002", "depends_on": ["T-002"]}
]
sorted_tickets = conductor_tech_lead.topological_sort(tickets)
assert sorted_tickets[0]["id"] == "T-001"
assert sorted_tickets[1]["id"] == "T-002"
assert sorted_tickets[2]["id"] == "T-003"
def test_topological_sort_circular():
tickets = [
{"id": "T-001", "depends_on": ["T-002"]},
{"id": "T-002", "depends_on": ["T-001"]}
]
with pytest.raises(ValueError, match="Circular dependency detected"):
conductor_tech_lead.topological_sort(tickets)
def test_track_executable_tickets():
t1 = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
t2 = Ticket(id="T2", description="desc", status="todo", assigned_to="user", depends_on=["T1"])
track = Track(id="track_1", description="desc", tickets=[t1, t2])
executable = track.get_executable_tickets()
assert len(executable) == 1
assert executable[0].id == "T1"
# Complete T1
t1.status = "completed"
executable = track.get_executable_tickets()
assert len(executable) == 1
assert executable[0].id == "T2"
@pytest.mark.asyncio
async def test_conductor_engine_run_linear():
t1 = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
t2 = Ticket(id="T2", description="desc", status="todo", assigned_to="user", depends_on=["T1"])
track = Track(id="track_1", description="desc", tickets=[t1, t2])
engine = multi_agent_conductor.ConductorEngine(track)
with patch("multi_agent_conductor.run_worker_lifecycle") as mock_worker:
# Mock worker to complete tickets
def complete_ticket(ticket, context, **kwargs):
ticket.status = "completed"
mock_worker.side_effect = complete_ticket
await engine.run_linear()
assert t1.status == "completed"
assert t2.status == "completed"
assert mock_worker.call_count == 2
def test_conductor_engine_parse_json_tickets():
track = Track(id="track_1", description="desc")
engine = multi_agent_conductor.ConductorEngine(track)
json_data = json.dumps([
{"id": "T1", "description": "desc 1", "depends_on": []},
{"id": "T2", "description": "desc 2", "depends_on": ["T1"]}
])
engine.parse_json_tickets(json_data)
assert len(track.tickets) == 2
assert track.tickets[0].id == "T1"
assert track.tickets[1].id == "T2"
assert track.tickets[1].depends_on == ["T1"]
def test_run_worker_lifecycle_blocked(mock_ai_client):
ticket = Ticket(id="T1", description="desc", status="todo", assigned_to="user")
context = multi_agent_conductor.WorkerContext(ticket_id="T1", model_name="model", messages=[])
mock_ai_client.return_value = "BLOCKED because of missing info"
multi_agent_conductor.run_worker_lifecycle(ticket, context)
assert ticket.status == "blocked"
assert ticket.blocked_reason == "BLOCKED because of missing info"

19
tests/test_orchestrator_pm.py
View File

@@ -40,30 +40,29 @@ class TestOrchestratorPM(unittest.TestCase):
mock_send.assert_called_once()
args, kwargs = mock_send.call_args
self.assertEqual(kwargs['md_content'], "")
self.assertEqual(kwargs['system_prompt'], expected_system_prompt)
# Cannot check system_prompt via mock_send kwargs anymore as it's set globally
# But we can verify user_message was passed
self.assertIn(user_request, kwargs['user_message'])
self.assertIn("REPO_MAP_CONTENT", kwargs['user_message'])
self.assertEqual(kwargs['model_name'], "gemini-1.5-pro")
# Verify result
self.assertEqual(result, mock_response_data)
# Verify result
self.assertEqual(result[0]['id'], mock_response_data[0]['id'])
@patch('summarize.build_summary_markdown')
@patch('ai_client.send')
def test_generate_tracks_markdown_wrapped(self, mock_send, mock_summarize):
mock_summarize.return_value = "REPO_MAP"
mock_response_data = [{"id": "track_1"}]
expected_result = [{"id": "track_1", "title": "Untitled Track"}]
# Wrapped in ```json ... ```
mock_send.return_value = f"Here is the plan:\n```json\n{json.dumps(mock_response_data)}\n```\nHope this helps."
result = orchestrator_pm.generate_tracks("req", {}, [])
self.assertEqual(result, mock_response_data)
self.assertEqual(result, expected_result)
# Wrapped in ``` ... ```
mock_send.return_value = f"```\n{json.dumps(mock_response_data)}\n```"
result = orchestrator_pm.generate_tracks("req", {}, [])
self.assertEqual(result, mock_response_data)
self.assertEqual(result, expected_result)
@patch('summarize.build_summary_markdown')
@patch('ai_client.send')

10
tests/test_process_pending_gui_tasks.py
View File

@@ -40,13 +40,11 @@ def test_redundant_calls_in_process_pending_gui_tasks(app_instance):
# ai_client.reset_session()
# ai_client.set_provider(value, self.current_model)
# _process_pending_gui_tasks ALSO calls:
# ai_client.set_provider(self.current_provider, self.current_model)
# ai_client.reset_session()
# _process_pending_gui_tasks NO LONGER calls it redundantly:
# Total should be 2 calls for each if redundant.
assert mock_set_provider.call_count == 2
assert mock_reset_session.call_count == 2
# Total should be 1 call for each.
assert mock_set_provider.call_count == 1
assert mock_reset_session.call_count == 1
def test_gcli_path_updates_adapter(app_instance):
# Setup

8
tests/test_session_logging.py
View File

@@ -8,15 +8,21 @@ import session_logger
import tomllib
@pytest.fixture
def temp_logs(tmp_path):
def temp_logs(tmp_path, monkeypatch):
# Ensure closed before starting
session_logger.close_session()
monkeypatch.setattr(session_logger, "_comms_fh", None)
# Mock _LOG_DIR in session_logger
original_log_dir = session_logger._LOG_DIR
session_logger._LOG_DIR = tmp_path / "logs"
monkeypatch.setattr(session_logger, "_LOG_DIR", tmp_path / "logs")
session_logger._LOG_DIR.mkdir(parents=True, exist_ok=True)
# Mock _SCRIPTS_DIR
original_scripts_dir = session_logger._SCRIPTS_DIR
session_logger._SCRIPTS_DIR = tmp_path / "scripts" / "generated"
monkeypatch.setattr(session_logger, "_SCRIPTS_DIR", tmp_path / "scripts" / "generated")
session_logger._SCRIPTS_DIR.mkdir(parents=True, exist_ok=True)
yield tmp_path / "logs"

95
tests/visual_orchestration_verification.py Normal file
View File

@@ -0,0 +1,95 @@
import pytest
import time
import sys
import os
from pathlib import Path
# Ensure project root is in path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from api_hook_client import ApiHookClient
@pytest.mark.integration
def test_mma_epic_lifecycle(live_gui):
"""
Integration test for the full MMA Epic lifecycle.
1. Start App.
2. Trigger 'New Epic' request.
3. Verify Tier 1 generates tracks.
4. Trigger 'Start Track' for one of the tracks.
5. Verify Tier 2 generates tickets.
6. Verify execution loop starts.
"""
client = ApiHookClient()
assert client.wait_for_server(timeout=15), "API hook server failed to start."
print("[Test] Initializing MMA Epic lifecycle test...")
# 0. Setup: Ensure we have a project and are in a clean state
client.click("btn_reset")
time.sleep(1)
# 1. Set Epic input
epic_text = "Improve the logging system to include timestamps in all tool calls."
print(f"[Test] Setting Epic input: {epic_text}")
client.set_value("mma_epic_input", epic_text)
# 2. Trigger 'New Epic' (Plan Epic)
print("[Test] Clicking 'Plan Epic (Tier 1)'...")
client.click("btn_mma_plan_epic")
# 3. Verify that Tier 1 generates tracks
print("[Test] Polling for Tier 1 tracks...")
tracks_generated = False
for i in range(120):
status = client.get_value("ai_status")
# Check if the proposal modal is shown or status changed
if status and "Epic tracks generated" in str(status):
tracks_generated = True
print(f"[Test] Tracks generated after {i}s")
break
time.sleep(1)
assert tracks_generated, "Tier 1 failed to generate tracks within 60 seconds."
# 4. Trigger 'Start Track' for the first track
print("[Test] Triggering 'Start Track' for track index 0...")
client.click("btn_mma_start_track", user_data={"index": 0})
# 5. Verify that Tier 2 generates tickets and starts execution
print("[Test] Polling for Tier 2 ticket generation and execution start...")
execution_started = False
for i in range(60):
mma_status = client.get_mma_status()
status_str = mma_status.get("mma_status", "idle")
active_tier = mma_status.get("active_tier", "")
if status_str == "running" or "Tier 3" in str(active_tier):
execution_started = True
print(f"[Test] Execution started (Status: {status_str}, Tier: {active_tier}) after {i}s")
break
current_ai_status = client.get_value("ai_status")
if i % 5 == 0:
print(f" ... still waiting. Current AI Status: {current_ai_status}")
time.sleep(1)
assert execution_started, "Tier 2 failed to generate tickets or execution failed to start within 60 seconds."
# 6. Final verification of MMA state
final_mma = client.get_mma_status()
print(f"[Test] Final MMA Status: {final_mma.get('mma_status')}")
print(f"[Test] Active Tier: {final_mma.get('active_tier')}")
print(f"[Test] Ticket Count: {len(final_mma.get('active_tickets', []))}")
assert final_mma.get("mma_status") in ["running", "done", "blocked"]
assert len(final_mma.get("active_tickets", [])) > 0
print("[Test] MMA Epic lifecycle verification successful!")
if __name__ == "__main__":
# If run directly, try to use pytest
import subprocess
# Using sys.executable to ensure we use the same environment
subprocess.run([sys.executable, "-m", "pytest", "-v", __file__])