chore(conductor): Archive gemini_cli_headless_20260224 track and update tests

This commit is contained in:
2026-02-25 18:39:36 -05:00
parent 1c78febd16
commit 94e41d20ff
15 changed files with 173 additions and 71 deletions

View File

@@ -4,15 +4,10 @@ import requests
import pytest
from api_hook_client import ApiHookClient
def test_api_ask_synchronous_flow(live_gui):
def test_api_ask_client_method(live_gui):
"""
Tests the full synchronous lifecycle of the /api/ask endpoint:
1. A client makes a blocking request.
2. An event is emitted with a unique request_id.
3. A separate agent responds to that request_id.
4. The original blocking request completes with the provided data.
Tests the request_confirmation method in ApiHookClient.
"""
# The live_gui fixture starts the Manual Slop application with hooks on 8999.
client = ApiHookClient("http://127.0.0.1:8999")
# Drain existing events
@@ -22,14 +17,11 @@ def test_api_ask_synchronous_flow(live_gui):
def make_blocking_request():
try:
# This POST will block until we call /api/ask/respond
# Note: /api/ask returns {'status': 'ok', 'response': ...}
resp = requests.post(
"http://127.0.0.1:8999/api/ask",
json={"prompt": "Should we proceed with the refactor?"},
timeout=10
# This call should block until we respond
results["response"] = client.request_confirmation(
tool_name="powershell",
args={"command": "echo hello"}
)
results["response"] = resp.json()
except Exception as e:
results["error"] = str(e)
@@ -37,7 +29,7 @@ def test_api_ask_synchronous_flow(live_gui):
t = threading.Thread(target=make_blocking_request)
t.start()
# Poll for the 'ask_received' event to find the generated request_id
# Poll for the 'ask_received' event
request_id = None
start_time = time.time()
while time.time() - start_time < 5:
@@ -52,8 +44,8 @@ def test_api_ask_synchronous_flow(live_gui):
assert request_id is not None, "Timed out waiting for 'ask_received' event"
# Respond to the task via the respond endpoint
expected_response = {"approved": True, "message": "Proceeding as requested."}
# Respond
expected_response = {"approved": True}
resp = requests.post(
"http://127.0.0.1:8999/api/ask/respond",
json={
@@ -63,11 +55,7 @@ def test_api_ask_synchronous_flow(live_gui):
)
assert resp.status_code == 200
# Join the thread and verify the original request received the correct data
t.join(timeout=5)
assert not t.is_alive(), "Background thread failed to unblock"
assert results["error"] is None, f"Request failed: {results['error']}"
# The /api/ask endpoint returns {'status': 'ok', 'response': expected_response}
assert results["response"]["status"] == "ok"
assert results["response"]["response"] == expected_response
assert not t.is_alive()
assert results["error"] is None
assert results["response"] == expected_response