245 lines
8.2 KiB
Python
245 lines
8.2 KiB
Python
import importlib.util
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ADAPTER_PATH = Path(__file__).with_name("glm-kimi-adapter.py")
|
|
SPEC = importlib.util.spec_from_file_location("glm_kimi_adapter", ADAPTER_PATH)
|
|
ADAPTER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(ADAPTER)
|
|
|
|
|
|
def response(message, finish_reason="tool_calls", tokens=10):
|
|
return {
|
|
"model": "test/model",
|
|
"usage": {
|
|
"prompt_tokens": tokens,
|
|
"completion_tokens": 1,
|
|
"total_tokens": tokens + 1,
|
|
"prompt_tokens_details": {"cached_tokens": 0},
|
|
"completion_tokens_details": {"reasoning_tokens": 0},
|
|
},
|
|
"choices": [{"finish_reason": finish_reason, "message": message}],
|
|
}
|
|
|
|
|
|
def subagent_result(agent_id):
|
|
return {
|
|
"result": f"Ergebnis {agent_id}",
|
|
"usage": {
|
|
"prompt_tokens": 2,
|
|
"completion_tokens": 1,
|
|
"total_tokens": 3,
|
|
"cached_tokens": 0,
|
|
"reasoning_tokens": 0,
|
|
},
|
|
"turns": 1,
|
|
"tool_calls": 0,
|
|
"errors": [],
|
|
"status": "completed",
|
|
}
|
|
|
|
|
|
class AdapterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.provider = {
|
|
"base_url": "https://example.invalid/v1",
|
|
"__id": "test",
|
|
}
|
|
self.liveness = ADAPTER.LivenessMonitor(interval_seconds=0)
|
|
|
|
def test_timeout_zero_is_forwarded_as_no_requests_timeout(self):
|
|
captured = {}
|
|
|
|
class FakeResponse:
|
|
status_code = 200
|
|
text = ""
|
|
|
|
@staticmethod
|
|
def json():
|
|
return {"ok": True}
|
|
|
|
def fake_post(*args, **kwargs):
|
|
captured["timeout"] = kwargs["timeout"]
|
|
return FakeResponse()
|
|
|
|
with patch.object(ADAPTER.requests, "post", side_effect=fake_post):
|
|
result = ADAPTER.post_chat_completion(
|
|
"https://example.invalid", {}, {}, 0, self.liveness,
|
|
"test", "Testaufruf",
|
|
)
|
|
|
|
self.assertEqual({"ok": True}, result)
|
|
self.assertIsNone(captured["timeout"])
|
|
|
|
def test_requested_tensorx_models_have_explicit_effort_mapping(self):
|
|
self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["qwen"])
|
|
self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["z-ai"])
|
|
|
|
def test_custom_mode_exposes_spawn_subagent(self):
|
|
captured = {}
|
|
|
|
def fake_post(url, headers, body, timeout, liveness,
|
|
activity_id, activity_label):
|
|
captured["tools"] = body["tools"]
|
|
return response(
|
|
{"role": "assistant", "content": "Fertig."},
|
|
finish_reason="stop",
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir, \
|
|
patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post):
|
|
result = ADAPTER.run_agent_loop(
|
|
provider=self.provider,
|
|
model="qwen/qwen3.8-flash-next",
|
|
system_prompt="System",
|
|
user_prompt="Aufgabe",
|
|
api_key="key",
|
|
effort="high",
|
|
root=temp_dir,
|
|
output_dir=temp_dir,
|
|
max_turns=0,
|
|
temperature=1.0,
|
|
timeout=0,
|
|
liveness=self.liveness,
|
|
mode="custom",
|
|
)
|
|
|
|
tool_names = {
|
|
tool["function"]["name"] for tool in captured["tools"]
|
|
}
|
|
self.assertIn("spawn_subagent", tool_names)
|
|
self.assertFalse(result["is_error"])
|
|
|
|
def test_liveness_monitor_reports_active_main_and_subagent_operations(self):
|
|
lines = []
|
|
monitor = ADAPTER.LivenessMonitor(interval_seconds=0.02)
|
|
with patch.object(ADAPTER, "log_stderr", side_effect=lines.append):
|
|
monitor.set("main", "Hauptagent wartet auf Subagenten")
|
|
monitor.set("subagent-1", "Subagent 1 wartet auf API-Antwort")
|
|
monitor.start()
|
|
time.sleep(0.06)
|
|
monitor.stop()
|
|
|
|
output = "\n".join(lines)
|
|
self.assertIn("LIFESIGN: Prozess lebt", output)
|
|
self.assertIn("Hauptagent wartet auf Subagenten", output)
|
|
self.assertIn("Subagent 1 wartet auf API-Antwort", output)
|
|
self.assertIn("keinen serverseitigen Fortschritt", output)
|
|
|
|
def test_subagents_from_one_turn_run_in_parallel_without_count_limit(self):
|
|
first_message = {
|
|
"role": "assistant",
|
|
"content": "Ich delegiere.",
|
|
"tool_calls": [
|
|
{
|
|
"id": f"spawn-{index}",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "spawn_subagent",
|
|
"arguments": (
|
|
'{"description":"Aufgabe %d",'
|
|
'"subagent_type":"explore"}' % index
|
|
),
|
|
},
|
|
}
|
|
for index in range(12)
|
|
],
|
|
}
|
|
responses = [
|
|
response(first_message),
|
|
response(
|
|
{"role": "assistant", "content": "Fertig."},
|
|
finish_reason="stop",
|
|
),
|
|
]
|
|
starts = []
|
|
|
|
def fake_post(*args, **kwargs):
|
|
return responses.pop(0)
|
|
|
|
def fake_subagent(*args, **kwargs):
|
|
agent_id = args[10]
|
|
starts.append(time.monotonic())
|
|
time.sleep(0.15)
|
|
return subagent_result(agent_id)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
started = time.monotonic()
|
|
with patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post), \
|
|
patch.object(ADAPTER, "run_subagent", side_effect=fake_subagent):
|
|
result = ADAPTER.run_agent_loop(
|
|
provider=self.provider,
|
|
model="test/model",
|
|
system_prompt="System",
|
|
user_prompt="Aufgabe",
|
|
api_key="key",
|
|
effort="high",
|
|
root=temp_dir,
|
|
output_dir=temp_dir,
|
|
max_turns=0,
|
|
temperature=1.0,
|
|
timeout=0,
|
|
liveness=self.liveness,
|
|
mode="builtin",
|
|
subagent_max_turns=0,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
|
|
self.assertEqual(12, result["subagent_stats"]["spawned"])
|
|
self.assertEqual(12, result["subagent_stats"]["completed"])
|
|
self.assertEqual(0, result["subagent_stats"]["failed"])
|
|
self.assertLess(max(starts) - min(starts), 0.12)
|
|
self.assertLess(elapsed, 0.6)
|
|
|
|
def test_api_error_is_not_masked_by_earlier_content(self):
|
|
first_message = {
|
|
"role": "assistant",
|
|
"content": "Zwischenstand",
|
|
"tool_calls": [{
|
|
"id": "list-1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "list_directory",
|
|
"arguments": '{"path":""}',
|
|
},
|
|
}],
|
|
}
|
|
calls = [response(first_message), RuntimeError("Transportfehler")]
|
|
|
|
def fake_post(*args, **kwargs):
|
|
item = calls.pop(0)
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir, \
|
|
patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post):
|
|
result = ADAPTER.run_agent_loop(
|
|
provider=self.provider,
|
|
model="test/model",
|
|
system_prompt="System",
|
|
user_prompt="Aufgabe",
|
|
api_key="key",
|
|
effort="high",
|
|
root=temp_dir,
|
|
output_dir=temp_dir,
|
|
max_turns=0,
|
|
temperature=1.0,
|
|
timeout=0,
|
|
liveness=self.liveness,
|
|
mode="solo",
|
|
)
|
|
|
|
self.assertEqual("Zwischenstand", result["result"])
|
|
self.assertTrue(result["is_error"])
|
|
self.assertEqual("error", result["subtype"])
|
|
self.assertIn("Transportfehler", result["errors"][0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|