73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
"""The admin exchange log keeps one day of request and response text."""
|
|
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from deerflow.models.model_exchange_log import (
|
|
activate_exchange,
|
|
deactivate_exchange,
|
|
query_model_exchanges,
|
|
remember_provider_chunk,
|
|
remember_request,
|
|
use_exchange_store,
|
|
)
|
|
|
|
|
|
def test_stream_request_and_split_response_are_stored(tmp_path: Path):
|
|
use_exchange_store(tmp_path / "model-exchanges.jsonl")
|
|
capture, token = activate_exchange()
|
|
try:
|
|
remember_request(
|
|
{
|
|
"model": "qwen-local",
|
|
"extra_body": {"enable_thinking": True},
|
|
"tools": [{"type": "function", "function": {"name": "write_file"}}],
|
|
"messages": [
|
|
{"role": "system", "content": "你是助手"},
|
|
{"role": "user", "content": "帮我生成一个1000字的md小作文"},
|
|
],
|
|
}
|
|
)
|
|
remember_provider_chunk({"choices": [{"delta": {"reasoning": "先想标题"}}]})
|
|
remember_provider_chunk({"choices": [{"delta": {"content": "正文开始"}}]})
|
|
finally:
|
|
deactivate_exchange(capture, token)
|
|
|
|
records = query_model_exchanges()
|
|
assert len(records) == 1
|
|
record = records[0]
|
|
assert record["model"] == "qwen-local"
|
|
assert record["enable_thinking"] == "true"
|
|
assert record["request_preview"] == "帮我生成一个1000字的md小作文"
|
|
assert "[system]\n你是助手" in record["request_text"]
|
|
assert "[user]\n帮我生成一个1000字的md小作文" in record["request_text"]
|
|
assert "tools: write_file" in record["request_text"]
|
|
request_body = json.loads(record["request_body"])
|
|
assert request_body["model"] == "qwen-local"
|
|
assert request_body["enable_thinking"] is True
|
|
assert "extra_body" not in request_body
|
|
assert request_body["messages"][1]["content"] == "帮我生成一个1000字的md小作文"
|
|
assert request_body["tools"][0]["function"]["name"] == "write_file"
|
|
parts = record["response_raw"].split("\n\n")
|
|
assert json.loads(parts[0]) == {"choices": [{"delta": {"reasoning": "先想标题"}}]}
|
|
assert json.loads(parts[1]) == {"choices": [{"delta": {"content": "正文开始"}}]}
|
|
assert record["response_text"] == record["response_raw"]
|
|
assert record["reasoning_field"] == "reasoning"
|
|
assert record["response_content"] == "正文开始"
|
|
assert record["response_reasoning"] == "先想标题"
|
|
use_exchange_store(None)
|
|
|
|
|
|
def test_records_older_than_one_day_are_dropped(tmp_path: Path):
|
|
path = tmp_path / "model-exchanges.jsonl"
|
|
use_exchange_store(path)
|
|
old = (datetime.now(UTC) - timedelta(days=2)).isoformat(timespec="seconds")
|
|
path.write_text(
|
|
'{"id":"old","at":"' + old + '","model":"old","user_id":"","enable_thinking":"unset",'
|
|
'"request_preview":"旧","response_preview":"","request_text":"旧请求","response_text":"旧返回"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert query_model_exchanges() == []
|
|
use_exchange_store(None)
|