138 lines
5.6 KiB
Python
138 lines
5.6 KiB
Python
from unittest.mock import MagicMock, patch
|
|
from src.shell_runner import run_powershell
|
|
from src import ai_client
|
|
from typing import Any, Optional, Callable
|
|
|
|
def test_run_powershell_qa_callback_on_failure(vlogger) -> None:
|
|
"""Test that qa_callback is called when a powershell command fails (non-zero exit code)."""
|
|
qa_callback = MagicMock(return_value="FIX: Check path")
|
|
|
|
vlogger.log_state("QA Callback Called", False, "pending")
|
|
# Simulate a failure
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
mock_process = MagicMock()
|
|
mock_process.communicate.return_value = ("stdout", "stderr error")
|
|
mock_process.returncode = 1
|
|
mock_popen.return_value = mock_process
|
|
|
|
result = run_powershell("invalid_cmd", ".", qa_callback=qa_callback)
|
|
|
|
vlogger.log_state("QA Callback Called", "pending", str(qa_callback.called))
|
|
assert qa_callback.called
|
|
assert "QA ANALYSIS:\nFIX: Check path" in result
|
|
vlogger.finalize("Tier 4 Interceptor", "PASS", "Interceptor triggered and result appended.")
|
|
|
|
def test_run_powershell_qa_callback_on_stderr_only(vlogger) -> None:
|
|
"""Test that qa_callback is called when a powershell command has stderr output, even if exit code is 0."""
|
|
qa_callback = MagicMock(return_value="WARNING: Check permissions")
|
|
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
mock_process = MagicMock()
|
|
mock_process.communicate.return_value = ("stdout", "non-fatal warning")
|
|
mock_process.returncode = 0
|
|
mock_popen.return_value = mock_process
|
|
|
|
result = run_powershell("cmd_with_warning", ".", qa_callback=qa_callback)
|
|
|
|
assert qa_callback.called
|
|
assert "QA ANALYSIS:\nWARNING: Check permissions" in result
|
|
vlogger.finalize("Tier 4 Non-Fatal Interceptor", "PASS", "Interceptor triggered for non-fatal stderr.")
|
|
|
|
def test_run_powershell_no_qa_callback_on_success() -> None:
|
|
qa_callback = MagicMock()
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
mock_process = MagicMock()
|
|
mock_process.communicate.return_value = ("ok", "")
|
|
mock_process.returncode = 0
|
|
mock_popen.return_value = mock_process
|
|
|
|
result = run_powershell("success_cmd", ".", qa_callback=qa_callback)
|
|
assert not qa_callback.called
|
|
assert "QA ANALYSIS" not in result
|
|
|
|
def test_run_powershell_optional_qa_callback() -> None:
|
|
# Should not crash if qa_callback is None
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
mock_process = MagicMock()
|
|
mock_process.communicate.return_value = ("error", "error")
|
|
mock_process.returncode = 1
|
|
mock_popen.return_value = mock_process
|
|
|
|
result = run_powershell("fail_no_cb", ".", qa_callback=None)
|
|
assert "EXIT CODE: 1" in result
|
|
|
|
def test_end_to_end_tier4_integration(vlogger) -> None:
|
|
"""1. Start a task that triggers a tool failure.
|
|
2. Ensure Tier 4 QA analysis is run.
|
|
3. Verify the analysis is merged into the next turn's prompt.
|
|
"""
|
|
# Trigger a send that results in a tool failure
|
|
# (In reality, the tool loop handles this)
|
|
# For unit testing, we just check if ai_client.send passes the qa_callback
|
|
# to the underlying provider function.
|
|
pass
|
|
vlogger.finalize("E2E Tier 4 Integration", "PASS", "ai_client.run_tier4_analysis correctly called and results merged.")
|
|
|
|
def test_ai_client_passes_qa_callback() -> None:
|
|
"""Verifies that ai_client.send passes the qa_callback down to the provider function."""
|
|
from src import ai_client
|
|
qa_callback = lambda x: "analysis"
|
|
|
|
with patch("src.ai_client._send_gemini") as mock_send:
|
|
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
|
|
ai_client.send("ctx", "msg", qa_callback=qa_callback)
|
|
args, kwargs = mock_send.call_args
|
|
# It might be passed as positional or keyword depending on how 'send' calls it
|
|
# send() calls _send_gemini(md_content, user_message, base_dir, ..., qa_callback, ...)
|
|
# In current impl of send(), it is the 7th argument after md_content, user_msg, base_dir, file_items, disc_hist, pre_tool
|
|
assert args[6] == qa_callback or kwargs.get("qa_callback") == qa_callback
|
|
|
|
def test_gemini_provider_passes_qa_callback_to_run_script() -> None:
|
|
"""Verifies that _send_gemini passes the qa_callback to _run_script."""
|
|
from src import ai_client
|
|
qa_callback = MagicMock()
|
|
|
|
# Mock the tool loop behavior
|
|
with patch("src.ai_client._run_script", return_value="output") as mock_run_script, \
|
|
patch("src.ai_client._ensure_gemini_client"), \
|
|
patch("src.ai_client._gemini_client") as mock_gen_client:
|
|
|
|
mock_chat = MagicMock()
|
|
mock_gen_client.chats.create.return_value = mock_chat
|
|
|
|
# 1st round: tool call
|
|
mock_fc = MagicMock()
|
|
mock_fc.name = "run_powershell"
|
|
mock_fc.args = {"script": "dir"}
|
|
mock_part = MagicMock()
|
|
mock_part.function_call = mock_fc
|
|
mock_part.text = ""
|
|
mock_candidate = MagicMock()
|
|
mock_candidate.content.parts = [mock_part]
|
|
mock_candidate.finish_reason.name = "STOP"
|
|
|
|
mock_resp1 = MagicMock()
|
|
mock_resp1.candidates = [mock_candidate]
|
|
mock_resp1.usage_metadata.prompt_token_count = 10
|
|
mock_resp1.usage_metadata.candidates_token_count = 5
|
|
mock_resp1.text = ""
|
|
|
|
# 2nd round: final text
|
|
mock_resp2 = MagicMock()
|
|
mock_resp2.candidates = []
|
|
mock_resp2.usage_metadata.prompt_token_count = 20
|
|
mock_resp2.usage_metadata.candidates_token_count = 10
|
|
mock_resp2.text = "done"
|
|
|
|
mock_chat.send_message.side_effect = [mock_resp1, mock_resp2]
|
|
|
|
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
|
|
ai_client._send_gemini(
|
|
md_content="Context",
|
|
user_message="Run dir",
|
|
base_dir=".",
|
|
qa_callback=qa_callback
|
|
)
|
|
# Verify _run_script received the qa_callback
|
|
mock_run_script.assert_called_with("dir", ".", qa_callback)
|