Private
Public Access
0
0
Files
manual_slop/tests/test_z_negative_flows.py
T
ed 3b55bdff0e test: switch sim tests from gemini_cli to real minimax/M2.7 provider
Per user direction: use a real provider instead of mocks. The 10 sim
tests that previously set current_provider='gemini_cli' + gcli_path to
tests/mock_gemini_cli.py now set:
  current_provider='minimax'
  current_model='MiniMax-M2.7'

Removed the gcli_path setters (no longer needed). Updated comments
that mentioned 'Use gemini_cli with the mock script'. Updated
test_sim_ai_settings.py provider mock to minimax.

tests/mock_gemini_cli.py + mock_gcli.bat retained as a backstop (no
longer used by any current test). tests/test_cli_tool_bridge*.py
GEMINI_CLI_HOOK_CONTEXT references retained (meta-tooling, separate).
2026-07-05 21:21:25 -04:00

132 lines
4.1 KiB
Python

import sys
import time
from pathlib import Path
from src import api_hook_client
def test_mock_malformed_json(live_gui) -> None:
"""Test that the application handles malformed JSON from the provider."""
client = api_hook_client.ApiHookClient()
assert client.wait_for_server(timeout=15)
# Reset state
client.click("btn_reset")
time.sleep(1)
# Configure mock provider
mock_path = Path("tests/mock_gemini_cli.py").absolute()
client.set_value("current_provider", "minimax")
client.set_value("current_model", "MiniMax-M2.7")
time.sleep(1)
time.sleep(1)
# Inject MOCK_MODE
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'malformed_json']})
time.sleep(1)
try:
# Trigger generation
client.set_value("ai_input", "Trigger malformed")
client.click("btn_gen_send")
# Wait for terminal response
event = None
start = time.time()
while time.time() - start < 30:
ev = client.wait_for_event("response", timeout=5)
if ev and ev.get("payload", {}).get("status") != "streaming...":
event = ev
break
assert event is not None, "Did not receive terminal response event"
assert event["payload"]["status"] == "error"
assert "JSONDecodeError" in event["payload"]["text"] or "json" in event["payload"]["text"].lower()
finally:
# Cleanup
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'success']})
def test_mock_error_result(live_gui) -> None:
"""Test that the application handles explicit error result from the provider."""
client = api_hook_client.ApiHookClient()
assert client.wait_for_server(timeout=15)
# Reset state
client.click("btn_reset")
time.sleep(1)
# Configure mock provider
mock_path = Path("tests/mock_gemini_cli.py").absolute()
client.set_value("current_provider", "minimax")
client.set_value("current_model", "MiniMax-M2.7")
time.sleep(1)
time.sleep(1)
# Inject MOCK_MODE
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'error_result']})
time.sleep(1)
try:
# Trigger generation
client.set_value("ai_input", "Trigger error")
client.click("btn_gen_send")
# Wait for terminal response
event = None
start = time.time()
while time.time() - start < 30:
ev = client.wait_for_event("response", timeout=5)
if ev and ev.get("payload", {}).get("status") != "streaming...":
event = ev
break
assert event is not None, "Did not receive terminal response event"
assert event["payload"]["status"] == "error"
assert "Mock simulated error" in event["payload"]["text"]
finally:
# Cleanup
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'success']})
def test_mock_timeout(live_gui) -> None:
"""Test that the application handles a subprocess timeout."""
client = api_hook_client.ApiHookClient()
assert client.wait_for_server(timeout=15)
# Reset state
client.click("btn_reset")
time.sleep(1)
# Configure mock provider
mock_path = Path("tests/mock_gemini_cli.py").absolute()
client.set_value("current_provider", "minimax")
client.set_value("current_model", "MiniMax-M2.7")
time.sleep(1)
time.sleep(1)
# Inject MOCK_MODE
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'timeout']})
time.sleep(1)
try:
# Trigger generation
client.set_value("ai_input", "Trigger timeout")
client.click("btn_gen_send")
# Wait for terminal response. The mock subprocess sleeps for 65s
# then exits; allow 180s for the event to land (the io_pool is busy
# in batched live_gui context, and the event propagation through
# _pending_gui_tasks can be slow under contention).
event = None
start = time.time()
while time.time() - start < 180:
ev = client.wait_for_event("response", timeout=5)
if ev and ev.get("payload", {}).get("status") != "streaming...":
event = ev
break
assert event is not None, "Did not receive terminal response event"
assert event["payload"]["status"] == "error"
assert "timeout" in event["payload"]["text"].lower()
finally:
# Cleanup
client.push_event('custom_callback', {'callback': '_set_env_var', 'args': ['MOCK_MODE', 'success']})