Mike-Solar
e0db5ab2cf
- Core types: Rational time, EntityId, ActionBatch, PngBytes (no Pillow dep) - Config: ~/.quercus/config.toml with 0600 perms, env var override for CI - LLMProvider: Claude (anthropic SDK) and OpenAI-compatible endpoint (covers custom gateways and llama.cpp server mode); graceful degradation without API keys - Tool schemas: 26 curated tools with JSON Schema and read_only flags - AgentLoop: sync dialogue loop with confirmation gate (read-only batches bypass, rejected mutations are reported back to the LLM), PNG frame feedback for multimodal providers - Session log: JSONL recording with frame sha256 hashes, replayable - HostAdapter ABC (signatures only, aligned with OPP/1 method families) - Mock LLM provider (scripted/replay/record) for network-free CI Tests: 69 passed via uv run pytest (no network access)
56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
"""mock_llm：脚本化回复、录制/回放往返、脚本文件序列化。"""
|
|
from quercus_core.providers.base import AssistantTurn, Message, Role, ToolCall
|
|
|
|
from mock_llm import MockProvider, RecordingProvider
|
|
|
|
|
|
def test_mock_provider_scripted():
|
|
mock = MockProvider(
|
|
script=[
|
|
AssistantTurn(
|
|
text="", tool_calls=(ToolCall(id="c1", name="get_frame", arguments={"time": {"num": 1, "den": 1}}),)
|
|
),
|
|
AssistantTurn(text="done", tool_calls=()),
|
|
]
|
|
)
|
|
turn1 = mock.generate([Message(role=Role.USER, text="hi")])
|
|
assert turn1.tool_calls[0].name == "get_frame"
|
|
assert mock.observed[0][0].text == "hi"
|
|
turn2 = mock.generate([Message(role=Role.USER, text="hi")])
|
|
assert turn2.text == "done" and not turn2.tool_calls
|
|
assert mock.exhausted
|
|
|
|
|
|
def test_mock_provider_empty_script_graceful():
|
|
mock = MockProvider()
|
|
turn = mock.generate([Message(role=Role.USER, text="hi")])
|
|
assert turn.text and not turn.tool_calls # 空脚本不抛异常，结束本轮
|
|
|
|
|
|
def test_script_file_roundtrip(tmp_path):
|
|
turns = [
|
|
AssistantTurn(
|
|
text="", tool_calls=(ToolCall(id="c1", name="place_clip", arguments={"x": 1}),)
|
|
),
|
|
AssistantTurn(text="ok", tool_calls=()),
|
|
]
|
|
path = tmp_path / "script.json"
|
|
MockProvider.dump_script(turns, path)
|
|
loaded = MockProvider.load_script(path)
|
|
assert loaded == turns
|
|
|
|
|
|
def test_recording_provider_records_and_replays(tmp_path):
|
|
inner = MockProvider(
|
|
script=[AssistantTurn(text="a", tool_calls=()), AssistantTurn(text="b", tool_calls=())]
|
|
)
|
|
rec = RecordingProvider(inner)
|
|
rec.generate([Message(role=Role.USER, text="1")])
|
|
rec.generate([Message(role=Role.USER, text="2")])
|
|
assert [t.text for t in rec.recorded] == ["a", "b"]
|
|
|
|
path = tmp_path / "recorded.json"
|
|
MockProvider.dump_script(rec.recorded, path)
|
|
replayed = MockProvider.load_script(path)
|
|
assert replayed == rec.recorded
|