mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-09-07 04:10:42 +00:00
265 lines
8.3 KiB
Python
265 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
import runpy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
SMOKE = runpy.run_path(ROOT / "scripts" / "smoke-python-runtime.py")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("prompt_name", "expected"),
|
|
[
|
|
("SNAPSHOT_DIRECT_CHILD_PROMPT", "DIRECT_CHILD_OK"),
|
|
("SNAPSHOT_WORKFLOW_CHILD_PROMPT", "WORKFLOW_CHILD_OK"),
|
|
],
|
|
)
|
|
def test_child_prompt_precedes_runtime_context(prompt_name: str, expected: str) -> None:
|
|
chunks = SMOKE["completion_chunks"]({
|
|
"messages": [
|
|
{"role": "user", "content": SMOKE[prompt_name]},
|
|
{"role": "user", "content": "Current runtime context"},
|
|
],
|
|
})
|
|
|
|
assert any(
|
|
choice.get("delta", {}).get("content") == expected
|
|
for chunk in chunks
|
|
for choice in chunk.get("choices", [])
|
|
)
|
|
|
|
|
|
def test_mcp_smoke_requests_the_discovered_tool() -> None:
|
|
chunks = SMOKE["completion_chunks"]({
|
|
"messages": [{"role": "user", "content": SMOKE["MCP_PROMPT"]}],
|
|
"tools": [{"type": "function", "function": {"name": "mcp__fixture__add"}}],
|
|
})
|
|
|
|
calls = [
|
|
call
|
|
for chunk in chunks
|
|
for choice in chunk.get("choices", [])
|
|
for call in choice.get("delta", {}).get("tool_calls", [])
|
|
]
|
|
assert calls[0]["function"] == {
|
|
"name": "mcp__fixture__add",
|
|
"arguments": '{"a": 19, "b": 23}',
|
|
}
|
|
|
|
|
|
def test_mcp_smoke_accepts_the_external_server_result() -> None:
|
|
chunks = SMOKE["completion_chunks"]({
|
|
"messages": [
|
|
{"role": "user", "content": SMOKE["MCP_PROMPT"]},
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [{
|
|
"id": "mcp-add",
|
|
"type": "function",
|
|
"function": {"name": "mcp__fixture__add", "arguments": '{}'},
|
|
}],
|
|
},
|
|
{"role": "tool", "tool_call_id": "mcp-add", "content": "42"},
|
|
],
|
|
})
|
|
|
|
assert any(
|
|
choice.get("delta", {}).get("content") == SMOKE["MCP_TEXT"]
|
|
for chunk in chunks
|
|
for choice in chunk.get("choices", [])
|
|
)
|
|
|
|
|
|
def test_snapshot_comparison_normalizes_only_session_generation_provenance() -> None:
|
|
normalize = SMOKE["normalize_session_format_comparison"]
|
|
expected = {
|
|
"header": {"type": "session", "version": 0, "otherVersion": 7},
|
|
"accepted": {
|
|
"type": "session-log-deepseek/delivery-accepted",
|
|
"data": {"sessionId": "s", "throughSeq": 4},
|
|
},
|
|
"source": {
|
|
"kind": "session-reference",
|
|
"references": [{"sessionId": "other", "capturedThroughSeq": 8}],
|
|
},
|
|
}
|
|
actual = {
|
|
"header": {"type": "session", "version": 1, "otherVersion": 7},
|
|
"accepted": {
|
|
"type": "session-log-deepseek/delivery-accepted",
|
|
"data": {"sessionId": "s", "sessionFormatVersion": 1, "throughSeq": 4},
|
|
},
|
|
"source": {
|
|
"kind": "session-reference",
|
|
"references": [{
|
|
"sessionId": "other",
|
|
"capturedFormatVersion": 1,
|
|
"capturedThroughSeq": 8,
|
|
}],
|
|
},
|
|
}
|
|
|
|
assert normalize(expected) == normalize(actual)
|
|
assert normalize(expected)["header"]["otherVersion"] == 7
|
|
|
|
|
|
def test_snapshot_value_normalizes_embedded_assistant_stream_timing() -> None:
|
|
normalize = SMOKE["normalize_snapshot_value"]
|
|
event = {
|
|
"type": "assistant/message",
|
|
"seq": 4,
|
|
"time": 100,
|
|
"data": {
|
|
"stream": [
|
|
{"type": "chunk", "time": 101, "chunk": {"type": "finish"}},
|
|
{"type": "text-chunks", "time0": 102, "dt": [1, 2], "texts": ["a", "b", "c"]},
|
|
],
|
|
},
|
|
}
|
|
|
|
normalized = normalize(event, [])
|
|
|
|
assert normalized["time"] == 0
|
|
assert normalized["data"]["stream"] == [
|
|
{"type": "chunk", "time": 0, "chunk": {"type": "finish"}},
|
|
{"type": "text-chunks", "time0": 0, "dt": [0, 0], "texts": ["a", "b", "c"]},
|
|
]
|
|
|
|
|
|
def test_snapshot_comparison_expands_embedded_assistant_streams() -> None:
|
|
normalize = SMOKE["normalize_session_format_comparison"]
|
|
expected = [
|
|
{
|
|
"type": "assistant/chunk",
|
|
"seq": 4,
|
|
"time": 0,
|
|
"data": {"turn": 1, "step": 1, "chunk": {
|
|
"type": "text-delta", "index": 0, "text": "done",
|
|
}},
|
|
},
|
|
{
|
|
"type": "assistant/message",
|
|
"seq": 5,
|
|
"time": 0,
|
|
"data": {"turn": 1, "step": 1, "message": {"role": "assistant"}},
|
|
"sourceEventSeqs": [4],
|
|
"surfaceOp": "append",
|
|
},
|
|
]
|
|
actual = [{
|
|
"type": "assistant/message",
|
|
"seq": 4,
|
|
"time": 0,
|
|
"data": {
|
|
"turn": 1,
|
|
"step": 1,
|
|
"message": {"role": "assistant"},
|
|
"stream": [{
|
|
"type": "text-chunks", "time0": 0, "index": 0, "dt": [], "texts": ["done"],
|
|
}],
|
|
},
|
|
"surfaceOp": "append",
|
|
}]
|
|
|
|
assert normalize(actual, 2) == normalize(expected, 1)
|
|
|
|
tool_result = {
|
|
"type": "tool/result",
|
|
"data": {"turn": 1, "step": 1},
|
|
"sourceEventSeqs": [4],
|
|
}
|
|
assert normalize(tool_result, 1)["sourceEventSeqs"] == [4]
|
|
assert normalize(tool_result, 2)["sourceEventSeqs"] == [4]
|
|
|
|
|
|
def test_snapshot_stream_expands_reasoning_and_tool_call_records() -> None:
|
|
expand = SMOKE["expand_snapshot_stream_member"]
|
|
|
|
assert expand({
|
|
"type": "reasoning-chunks", "time0": 0, "index": 1,
|
|
"dt": [], "texts": ["think"],
|
|
}) == [{"type": "reasoning-delta", "index": 1, "text": "think"}]
|
|
assert expand({
|
|
"type": "tool-call-chunks", "time0": 0, "index": 2,
|
|
"id": "call-1", "name": "read", "dt": [1], "args": ["{", "}"],
|
|
}) == [
|
|
{"type": "tool-call-delta", "index": 2, "id": "call-1", "name": "read", "argumentsDelta": "{"},
|
|
{"type": "tool-call-delta", "index": 2, "id": "call-1", "name": "read", "argumentsDelta": "}"},
|
|
]
|
|
|
|
|
|
def test_snapshot_file_builder_order_is_checked_outside_update_mode(tmp_path: Path) -> None:
|
|
compare = SMOKE["compare_snapshot_files"]
|
|
|
|
with pytest.raises(AssertionError, match="snapshot builder produced"):
|
|
compare({}, False, tmp_path, ("result.json",))
|
|
|
|
|
|
def test_snapshot_comparison_expands_sdk_wrapped_attempts() -> None:
|
|
normalize = SMOKE["normalize_session_format_comparison"]
|
|
actual = [{
|
|
"method": "session.event",
|
|
"payload": {
|
|
"sessionId": "s",
|
|
"event": {
|
|
"type": "assistant/attempt",
|
|
"seq": 7,
|
|
"time": 0,
|
|
"data": {
|
|
"turn": 1,
|
|
"step": 1,
|
|
"stream": [{"type": "chunk", "time": 0, "chunk": {"type": "finish"}}],
|
|
},
|
|
},
|
|
},
|
|
}]
|
|
|
|
assert normalize(actual) == [{
|
|
"method": "session.event",
|
|
"payload": {
|
|
"sessionId": "s",
|
|
"event": {
|
|
"type": "assistant/chunk",
|
|
"data": {"turn": 1, "step": 1, "chunk": {"type": "finish"}},
|
|
},
|
|
},
|
|
}]
|
|
|
|
|
|
def test_snapshot_generation_names_select_highest_role_without_double_counting(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
render = SMOKE["snapshot_session_filename"]
|
|
select = SMOKE["selected_snapshot_session_files"]
|
|
assert render(0, 0) == "session.jsonl"
|
|
assert render(0, 2) == "session.v2.jsonl"
|
|
assert render(3, 0) == "session.3.jsonl"
|
|
assert render(3, 2) == "session.3.v2.jsonl"
|
|
|
|
(tmp_path / "session.jsonl").write_text(
|
|
'{"type":"session","version":0}\n', encoding="utf-8",
|
|
)
|
|
(tmp_path / "session.v1.jsonl").write_text(
|
|
'{"type":"session","version":1}\n', encoding="utf-8",
|
|
)
|
|
(tmp_path / "session.1.jsonl").write_text(
|
|
'{"type":"session","version":0}\n', encoding="utf-8",
|
|
)
|
|
|
|
assert {index: path.name for index, path in select(tmp_path).items()} == {
|
|
0: "session.v1.jsonl",
|
|
1: "session.1.jsonl",
|
|
}
|
|
|
|
|
|
def test_snapshot_generation_filename_must_match_header(tmp_path: Path) -> None:
|
|
(tmp_path / "session.v1.jsonl").write_text(
|
|
'{"type":"session","version":0}\n', encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(AssertionError, match="filename declares Session format v1"):
|
|
SMOKE["selected_snapshot_session_files"](tmp_path)
|