wehub-resource-sync
c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Waiting to run
Pre-commit / run (ubuntu-latest) (push) Waiting to run
Python Unittest Coverage / test (macos-15, 3.11) (push) Waiting to run
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Waiting to run
Python Unittest Coverage / test (windows-latest, 3.11) (push) Waiting to run
Web UI / check (push) Waiting to run
642 lines
20 KiB
Python
642 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
# pylint: disable=protected-access
|
|
"""Unit tests for DeepSeekChatModel with mocked API responses.
|
|
|
|
Tests cover both non-streaming and streaming modes, verifying that:
|
|
- Non-stream mode returns a single ChatResponse with is_last=True.
|
|
- Stream mode yields n delta ChatResponses (is_last=False) followed by
|
|
1 final ChatResponse (is_last=True) with the full accumulated content.
|
|
"""
|
|
from typing import Any
|
|
import unittest
|
|
from unittest import IsolatedAsyncioTestCase
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from utils import AnyString
|
|
|
|
from agentscope.message import TextBlock, ToolCallBlock, ThinkingBlock
|
|
from agentscope.model import DeepSeekChatModel
|
|
from agentscope.credential import DeepSeekCredential
|
|
from agentscope.tool import ToolChoice
|
|
|
|
A = AnyString()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_model(stream: bool = False) -> Any:
|
|
return DeepSeekChatModel(
|
|
credential=DeepSeekCredential(api_key="test"),
|
|
model="deepseek-v4-pro",
|
|
stream=stream,
|
|
context_size=65_536,
|
|
)
|
|
|
|
|
|
def _mock_completion(
|
|
text: Any = None,
|
|
tool_calls: Any = None,
|
|
reasoning: Any = None,
|
|
response_id: str = "deepseek-1",
|
|
) -> MagicMock:
|
|
"""Build a mock non-streaming ChatCompletion response."""
|
|
msg = MagicMock()
|
|
msg.content = text
|
|
msg.reasoning_content = reasoning
|
|
msg.tool_calls = None
|
|
|
|
if tool_calls:
|
|
tc_mocks = []
|
|
for tc in tool_calls:
|
|
m = MagicMock()
|
|
m.id = tc["id"]
|
|
m.function.name = tc["name"]
|
|
m.function.arguments = tc["arguments"]
|
|
tc_mocks.append(m)
|
|
msg.tool_calls = tc_mocks
|
|
|
|
choice = MagicMock()
|
|
choice.message = msg
|
|
|
|
resp = MagicMock()
|
|
resp.id = response_id
|
|
resp.choices = [choice]
|
|
resp.usage.prompt_tokens = 10
|
|
resp.usage.completion_tokens = 5
|
|
resp.usage.prompt_cache_hit_tokens = 0
|
|
return resp
|
|
|
|
|
|
def _make_stream_chunk(
|
|
delta_text: str | None = None,
|
|
delta_reasoning: str | None = None,
|
|
tool_calls: list | None = None,
|
|
response_id: str = "deepseek-1",
|
|
usage: dict | None = None,
|
|
has_choices: bool = True,
|
|
) -> MagicMock:
|
|
"""Build a single mock streaming chunk."""
|
|
chunk = MagicMock()
|
|
chunk.id = response_id
|
|
|
|
if usage:
|
|
chunk.usage = MagicMock()
|
|
chunk.usage.prompt_tokens = usage.get("prompt_tokens", 0)
|
|
chunk.usage.completion_tokens = usage.get("completion_tokens", 0)
|
|
chunk.usage.prompt_cache_hit_tokens = usage.get(
|
|
"prompt_cache_hit_tokens",
|
|
0,
|
|
)
|
|
else:
|
|
chunk.usage = None
|
|
|
|
if has_choices:
|
|
delta = MagicMock()
|
|
delta.content = delta_text
|
|
delta.reasoning_content = delta_reasoning
|
|
delta.tool_calls = tool_calls
|
|
choice = MagicMock()
|
|
choice.delta = delta
|
|
chunk.choices = [choice]
|
|
else:
|
|
chunk.choices = []
|
|
|
|
return chunk
|
|
|
|
|
|
def _make_tool_call_delta(
|
|
index: int,
|
|
tc_id: str | None = None,
|
|
name: str | None = None,
|
|
arguments: str | None = None,
|
|
) -> MagicMock:
|
|
"""Build a tool_call delta item for streaming."""
|
|
tc = MagicMock()
|
|
tc.index = index
|
|
tc.id = tc_id
|
|
tc.function.name = name
|
|
tc.function.arguments = arguments
|
|
return tc
|
|
|
|
|
|
class _MockAsyncStream:
|
|
"""Mock async stream that acts as an async context manager + iterator."""
|
|
|
|
def __init__(self, chunks: list) -> None:
|
|
self._chunks = chunks
|
|
self._index = 0
|
|
|
|
async def __aenter__(self) -> "_MockAsyncStream":
|
|
return self
|
|
|
|
async def __aexit__(self, *args: Any) -> None:
|
|
pass
|
|
|
|
def __aiter__(self) -> "_MockAsyncStream":
|
|
return self
|
|
|
|
async def __anext__(self) -> Any:
|
|
if self._index >= len(self._chunks):
|
|
raise StopAsyncIteration
|
|
chunk = self._chunks[self._index]
|
|
self._index += 1
|
|
return chunk
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Non-streaming tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeepSeekNonStream(IsolatedAsyncioTestCase):
|
|
"""Tests for DeepSeekChatModel in non-streaming mode."""
|
|
|
|
def setUp(self) -> None:
|
|
self.model = _make_model(stream=False)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_text_response(self, mock_client_cls: MagicMock) -> None:
|
|
"""Non-stream text response returns a single ChatResponse."""
|
|
mock_create = AsyncMock(
|
|
return_value=_mock_completion(text="Hello world!"),
|
|
)
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
result = await self.model([])
|
|
|
|
self.assertEqual(
|
|
(result.is_last, result.content),
|
|
(True, [TextBlock.model_construct(id=A, text="Hello world!")]),
|
|
)
|
|
self.assertEqual(result.id, "deepseek-1")
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_tool_call_response(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Non-stream tool call response creates ToolCallBlocks."""
|
|
mock_create = AsyncMock(
|
|
return_value=_mock_completion(
|
|
tool_calls=[
|
|
{
|
|
"id": "call-1",
|
|
"name": "get_weather",
|
|
"arguments": '{"city":"Beijing"}',
|
|
},
|
|
{
|
|
"id": "call-2",
|
|
"name": "get_time",
|
|
"arguments": '{"tz":"UTC"}',
|
|
},
|
|
],
|
|
),
|
|
)
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
result = await self.model([])
|
|
|
|
self.assertEqual(
|
|
(result.is_last, result.content),
|
|
(
|
|
True,
|
|
[
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="get_weather",
|
|
input='{"city":"Beijing"}',
|
|
),
|
|
ToolCallBlock(
|
|
id="call-2",
|
|
name="get_time",
|
|
input='{"tz":"UTC"}',
|
|
),
|
|
],
|
|
),
|
|
)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_thinking_response(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Non-stream response with reasoning creates ThinkingBlock."""
|
|
mock_create = AsyncMock(
|
|
return_value=_mock_completion(
|
|
text="The answer is 42.",
|
|
reasoning="Let me think step by step...",
|
|
),
|
|
)
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
result = await self.model([])
|
|
|
|
self.assertEqual(
|
|
(result.is_last, result.content),
|
|
(
|
|
True,
|
|
[
|
|
ThinkingBlock.model_construct(
|
|
id=A,
|
|
thinking="Let me think step by step...",
|
|
),
|
|
TextBlock.model_construct(id=A, text="The answer is 42."),
|
|
],
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Streaming tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeepSeekStream(IsolatedAsyncioTestCase):
|
|
"""Tests for DeepSeekChatModel in streaming mode."""
|
|
|
|
def setUp(self) -> None:
|
|
self.model = _make_model(stream=True)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_stream_text_response(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Stream text yields n deltas (is_last=False) + 1 final
|
|
(is_last=True) with full content."""
|
|
chunks = [
|
|
_make_stream_chunk(delta_text="Hello"),
|
|
_make_stream_chunk(delta_text=" world"),
|
|
_make_stream_chunk(delta_text="!"),
|
|
_make_stream_chunk(
|
|
has_choices=False,
|
|
usage={"prompt_tokens": 10, "completion_tokens": 3},
|
|
),
|
|
]
|
|
mock_create = AsyncMock(return_value=_MockAsyncStream(chunks))
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
gen = await self.model([])
|
|
responses = [r async for r in gen]
|
|
|
|
self.assertListEqual(
|
|
[(r.is_last, r.content) for r in responses],
|
|
[
|
|
(False, [TextBlock.model_construct(id=A, text="Hello")]),
|
|
(False, [TextBlock.model_construct(id=A, text=" world")]),
|
|
(False, [TextBlock.model_construct(id=A, text="!")]),
|
|
(True, [TextBlock.model_construct(id=A, text="Hello world!")]),
|
|
],
|
|
)
|
|
self.assertEqual(responses[-1].id, "deepseek-1")
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_stream_thinking_and_text(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Stream with thinking + text yields deltas then final with both."""
|
|
chunks = [
|
|
_make_stream_chunk(delta_reasoning="Think"),
|
|
_make_stream_chunk(delta_reasoning="ing..."),
|
|
_make_stream_chunk(delta_text="Answer"),
|
|
_make_stream_chunk(delta_text=" here."),
|
|
_make_stream_chunk(
|
|
has_choices=False,
|
|
usage={"prompt_tokens": 10, "completion_tokens": 8},
|
|
),
|
|
]
|
|
mock_create = AsyncMock(return_value=_MockAsyncStream(chunks))
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
gen = await self.model([])
|
|
responses = [r async for r in gen]
|
|
|
|
self.assertListEqual(
|
|
[(r.is_last, r.content) for r in responses],
|
|
[
|
|
(
|
|
False,
|
|
[ThinkingBlock.model_construct(id=A, thinking="Think")],
|
|
),
|
|
(
|
|
False,
|
|
[ThinkingBlock.model_construct(id=A, thinking="ing...")],
|
|
),
|
|
(False, [TextBlock.model_construct(id=A, text="Answer")]),
|
|
(False, [TextBlock.model_construct(id=A, text=" here.")]),
|
|
(
|
|
True,
|
|
[
|
|
ThinkingBlock.model_construct(
|
|
id=A,
|
|
thinking="Thinking...",
|
|
),
|
|
TextBlock.model_construct(id=A, text="Answer here."),
|
|
],
|
|
),
|
|
],
|
|
)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_stream_tool_calls(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Stream tool calls accumulate across chunks into final response."""
|
|
chunks = [
|
|
_make_stream_chunk(
|
|
delta_text=None,
|
|
tool_calls=[
|
|
_make_tool_call_delta(0, "call-1", "get_weather", '{"ci'),
|
|
],
|
|
),
|
|
_make_stream_chunk(
|
|
delta_text=None,
|
|
tool_calls=[
|
|
_make_tool_call_delta(0, None, None, 'ty":"BJ"}'),
|
|
],
|
|
),
|
|
_make_stream_chunk(
|
|
has_choices=False,
|
|
usage={"prompt_tokens": 10, "completion_tokens": 5},
|
|
),
|
|
]
|
|
mock_create = AsyncMock(return_value=_MockAsyncStream(chunks))
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
gen = await self.model([])
|
|
responses = [r async for r in gen]
|
|
|
|
self.assertListEqual(
|
|
[(r.is_last, r.content) for r in responses],
|
|
[
|
|
(
|
|
False,
|
|
[
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="get_weather",
|
|
input='{"ci',
|
|
),
|
|
],
|
|
),
|
|
(
|
|
False,
|
|
[
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="get_weather",
|
|
input='ty":"BJ"}',
|
|
),
|
|
],
|
|
),
|
|
(
|
|
True,
|
|
[
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="get_weather",
|
|
input='{"city":"BJ"}',
|
|
),
|
|
],
|
|
),
|
|
],
|
|
)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_stream_text_then_tool_call(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Stream with text followed by tool call accumulates both."""
|
|
chunks = [
|
|
_make_stream_chunk(delta_reasoning="Let me check"),
|
|
_make_stream_chunk(delta_text="I'll look it up."),
|
|
_make_stream_chunk(
|
|
delta_text=None,
|
|
tool_calls=[
|
|
_make_tool_call_delta(
|
|
0,
|
|
"call-1",
|
|
"search",
|
|
'{"q":"weather"}',
|
|
),
|
|
],
|
|
),
|
|
_make_stream_chunk(
|
|
has_choices=False,
|
|
usage={"prompt_tokens": 15, "completion_tokens": 10},
|
|
),
|
|
]
|
|
mock_create = AsyncMock(return_value=_MockAsyncStream(chunks))
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
gen = await self.model([])
|
|
responses = [r async for r in gen]
|
|
|
|
self.assertListEqual(
|
|
[(r.is_last, r.content) for r in responses],
|
|
[
|
|
(
|
|
False,
|
|
[
|
|
ThinkingBlock.model_construct(
|
|
id=A,
|
|
thinking="Let me check",
|
|
),
|
|
],
|
|
),
|
|
(
|
|
False,
|
|
[TextBlock.model_construct(id=A, text="I'll look it up.")],
|
|
),
|
|
(
|
|
False,
|
|
[
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="search",
|
|
input='{"q":"weather"}',
|
|
),
|
|
],
|
|
),
|
|
(
|
|
True,
|
|
[
|
|
ThinkingBlock.model_construct(
|
|
id=A,
|
|
thinking="Let me check",
|
|
),
|
|
TextBlock.model_construct(
|
|
id=A,
|
|
text="I'll look it up.",
|
|
),
|
|
ToolCallBlock(
|
|
id="call-1",
|
|
name="search",
|
|
input='{"q":"weather"}',
|
|
),
|
|
],
|
|
),
|
|
],
|
|
)
|
|
|
|
@patch("openai.AsyncClient")
|
|
async def test_stream_usage_in_final(
|
|
self,
|
|
mock_client_cls: MagicMock,
|
|
) -> None:
|
|
"""Usage information is captured and present in final response."""
|
|
chunks = [
|
|
_make_stream_chunk(delta_text="Hi"),
|
|
_make_stream_chunk(
|
|
has_choices=False,
|
|
usage={
|
|
"prompt_tokens": 100,
|
|
"completion_tokens": 20,
|
|
"prompt_cache_hit_tokens": 50,
|
|
},
|
|
),
|
|
]
|
|
mock_create = AsyncMock(return_value=_MockAsyncStream(chunks))
|
|
mock_client_cls.return_value.chat.completions.create = mock_create
|
|
|
|
gen = await self.model([])
|
|
responses = [r async for r in gen]
|
|
|
|
self.assertListEqual(
|
|
[(r.is_last, r.content) for r in responses],
|
|
[
|
|
(False, [TextBlock.model_construct(id=A, text="Hi")]),
|
|
(True, [TextBlock.model_construct(id=A, text="Hi")]),
|
|
],
|
|
)
|
|
self.assertEqual(responses[-1].usage.input_tokens, 100)
|
|
self.assertEqual(responses[-1].usage.output_tokens, 20)
|
|
self.assertEqual(responses[-1].usage.cache_input_tokens, 50)
|
|
|
|
|
|
class TestDeepSeekModelParameters(unittest.TestCase):
|
|
"""Tests for DeepSeekChatModel.Parameters."""
|
|
|
|
def test_thinking_enable_stored_on_model(self) -> None:
|
|
"""thinking_enable is accessible through model.parameters."""
|
|
model = DeepSeekChatModel(
|
|
credential=DeepSeekCredential(api_key="test"),
|
|
model="deepseek-reasoner",
|
|
stream=False,
|
|
context_size=65_536,
|
|
parameters=DeepSeekChatModel.Parameters(thinking_enable=True),
|
|
)
|
|
self.assertTrue(model.parameters.thinking_enable)
|
|
|
|
def test_reasoning_effort_stored_on_model(self) -> None:
|
|
"""reasoning_effort is accessible through model.parameters."""
|
|
model = DeepSeekChatModel(
|
|
credential=DeepSeekCredential(api_key="test"),
|
|
model="deepseek-reasoner",
|
|
stream=False,
|
|
context_size=65_536,
|
|
parameters=DeepSeekChatModel.Parameters(reasoning_effort="max"),
|
|
)
|
|
self.assertEqual(model.parameters.reasoning_effort, "max")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared _format_tools fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FT_TOOLS = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather",
|
|
"description": "Get the weather",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"city": {"type": "string"}},
|
|
"required": ["city"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_time",
|
|
"description": "Get the time",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"timezone": {"type": "string"}},
|
|
"required": ["timezone"],
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
class TestDeepSeekFormatTools(unittest.TestCase):
|
|
"""Tests for DeepSeekChatModel._format_tools."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up model instance."""
|
|
self.model = _make_model()
|
|
|
|
def test_auto_mode(self) -> None:
|
|
"""Auto mode returns tools unchanged and string 'auto'."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(
|
|
_FT_TOOLS,
|
|
ToolChoice(mode="auto"),
|
|
)
|
|
self.assertEqual(fmt_tools, _FT_TOOLS)
|
|
self.assertEqual(fmt_choice, "auto")
|
|
|
|
def test_none_mode(self) -> None:
|
|
"""None mode returns tools unchanged and string 'none'."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(
|
|
_FT_TOOLS,
|
|
ToolChoice(mode="none"),
|
|
)
|
|
self.assertEqual(fmt_tools, _FT_TOOLS)
|
|
self.assertEqual(fmt_choice, "none")
|
|
|
|
def test_required_mode(self) -> None:
|
|
"""Required mode returns tools unchanged and string 'required'."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(
|
|
_FT_TOOLS,
|
|
ToolChoice(mode="required"),
|
|
)
|
|
self.assertEqual(fmt_tools, _FT_TOOLS)
|
|
self.assertEqual(fmt_choice, "required")
|
|
|
|
def test_str_mode_force_call(self) -> None:
|
|
"""A specific tool name returns a type=function dict."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(
|
|
_FT_TOOLS,
|
|
ToolChoice(mode="get_weather"),
|
|
)
|
|
self.assertEqual(fmt_tools, _FT_TOOLS)
|
|
self.assertEqual(
|
|
fmt_choice,
|
|
{"type": "function", "function": {"name": "get_weather"}},
|
|
)
|
|
|
|
def test_tools_filtered(self) -> None:
|
|
"""When tool_choice.tools is set, only those tools are included."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(
|
|
_FT_TOOLS,
|
|
ToolChoice(mode="auto", tools=["get_weather"]),
|
|
)
|
|
self.assertEqual(len(fmt_tools), 1)
|
|
self.assertEqual(fmt_tools[0]["function"]["name"], "get_weather")
|
|
self.assertEqual(fmt_choice, "auto")
|
|
|
|
def test_no_tool_choice(self) -> None:
|
|
"""Without tool_choice, returns tools and None."""
|
|
fmt_tools, fmt_choice = self.model._format_tools(_FT_TOOLS, None)
|
|
self.assertEqual(fmt_tools, _FT_TOOLS)
|
|
self.assertIsNone(fmt_choice)
|