Spaces:
Paused
Paused
File size: 8,382 Bytes
4b03eed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 | # -*- coding: utf-8 -*-
# pylint: disable=protected-access,unused-argument
"""Unit tests for DashScopeEmbeddingModel."""
from dataclasses import asdict
from typing import Any
from unittest import IsolatedAsyncioTestCase
from unittest.mock import AsyncMock, MagicMock, patch
from utils import AnyValue
from agentscope.credential import DashScopeCredential
from agentscope.embedding import (
DashScopeEmbeddingModel,
EmbeddingResponse,
EmbeddingUsage,
)
from agentscope.message import DataBlock, Base64Source, URLSource
A = AnyValue()
def _text_resp(
embeddings: list[list[float]],
total_tokens: int = 10,
status_code: int = 200,
) -> MagicMock:
"""Build a mock DashScope text embedding response."""
resp = MagicMock()
resp.status_code = status_code
resp.output = {"embeddings": [{"embedding": e} for e in embeddings]}
resp.usage = {"total_tokens": total_tokens}
return resp
def _cred() -> DashScopeCredential:
"""Create a test credential."""
return DashScopeCredential(api_key="k")
def _img() -> DataBlock:
"""Create a test image DataBlock."""
return DataBlock(
source=Base64Source(data="aWltYWdl", media_type="image/png"),
)
def _vid() -> DataBlock:
"""Create a test video DataBlock."""
return DataBlock(
source=URLSource(url="https://x.com/v.mp4", media_type="video/mp4"),
)
def _mock_resp(embeddings: list[list[float]]) -> EmbeddingResponse:
"""Create a mock EmbeddingResponse."""
return EmbeddingResponse(
embeddings=embeddings,
usage=EmbeddingUsage(tokens=len(embeddings), time=0.01),
)
class DashScopeListModelsTest(IsolatedAsyncioTestCase):
"""Test list_models for DashScope."""
async def test_list_models(self) -> None:
"""Should list 7 models (text + multimodal)."""
cards = DashScopeEmbeddingModel.list_models()
names = sorted(c.name for c in cards)
self.assertEqual(len(cards), 7)
self.assertIn("text-embedding-v4", names)
self.assertIn("qwen3-vl-embedding", names)
self.assertIn("multimodal-embedding-v1", names)
async def test_hidden_dimensions(self) -> None:
"""multimodal-embedding-v1 declares a fixed dimension (no enum)."""
cards = DashScopeEmbeddingModel.list_models()
v1 = next(c for c in cards if c.name == "multimodal-embedding-v1")
self.assertDictEqual(
v1.model_dump(),
{
"type": "embedding_model",
"name": "multimodal-embedding-v1",
"label": "Multimodal Embedding v1",
"status": "active",
"input_types": [
"text/plain",
"image/jpeg",
"image/png",
"image/bmp",
],
"output_types": ["application/x-embedding"],
"dimensions": 1024,
"supported_dimensions": None,
"context_size": 512,
"parameter_schema": {
"type": "object",
"properties": {},
"required": [],
},
"parameter_overrides": {},
},
)
async def test_visible_dimensions(self) -> None:
"""qwen3-vl-embedding exposes dimension choices on the card."""
cards = DashScopeEmbeddingModel.list_models()
qwen = next(c for c in cards if c.name == "qwen3-vl-embedding")
self.assertEqual(qwen.dimensions, 2560)
self.assertEqual(
qwen.supported_dimensions,
[2560, 2048, 1536, 1024, 768, 512, 256],
)
class DashScopeTextCallTest(IsolatedAsyncioTestCase):
"""Test DashScope text embedding API calls."""
@patch("dashscope.embeddings.TextEmbedding.call")
async def test_text_call(self, mock_api: Any) -> None:
"""Text mode returns correct embeddings."""
mock_api.return_value = _text_resp([[0.1, 0.2], [0.3, 0.4]], 12)
model = DashScopeEmbeddingModel(
credential=_cred(),
model="text-embedding-v4",
dimensions=2,
)
result = await model(["hello", "world"])
self.assertDictEqual(
asdict(result),
{
"embeddings": [[0.1, 0.2], [0.3, 0.4]],
"id": A,
"created_at": A,
"type": "embedding",
"usage": {"tokens": 12, "time": A, "type": "embedding"},
"source": "api",
},
)
@patch("dashscope.embeddings.TextEmbedding.call")
async def test_text_rejects_datablock(self, mock_api: Any) -> None:
"""Text mode rejects DataBlock inputs."""
model = DashScopeEmbeddingModel(
credential=_cred(),
model="text-embedding-v4",
dimensions=1024,
)
with self.assertRaises(ValueError):
await model([_img()])
@patch("dashscope.embeddings.TextEmbedding.call")
async def test_text_api_error_raises(self, mock_api: Any) -> None:
"""Non-200 status code raises RuntimeError after retries."""
mock_api.return_value = _text_resp([], status_code=400)
model = DashScopeEmbeddingModel(
credential=_cred(),
model="text-embedding-v4",
dimensions=1024,
retry_delay=0.0,
)
with self.assertRaises(RuntimeError):
await model(["hello"])
class DashScopeMultimodalCallTest(IsolatedAsyncioTestCase):
"""Test DashScope multimodal embedding via mocked _call_multimodal."""
async def test_multimodal_text_and_image(self) -> None:
"""Multimodal call with text + image."""
model = DashScopeEmbeddingModel(
credential=_cred(),
model="qwen3-vl-embedding",
dimensions=2,
)
model._call_multimodal = AsyncMock(
return_value=_mock_resp([[0.1, 0.2], [0.3, 0.4]]),
)
result = await model(["describe this", _img()])
self.assertDictEqual(
asdict(result),
{
"embeddings": [[0.1, 0.2], [0.3, 0.4]],
"id": A,
"created_at": A,
"type": "embedding",
"usage": {"tokens": 2, "time": 0.01, "type": "embedding"},
"source": "api",
},
)
async def test_multimodal_batching_by_image_limit(self) -> None:
"""8 images with max_images=5 produces 2 batches (5+3)."""
model = DashScopeEmbeddingModel(
credential=_cred(),
model="qwen3-vl-embedding",
dimensions=1,
)
call_count = 0
async def _mock(inputs: list, **_kw: Any) -> EmbeddingResponse:
nonlocal call_count
call_count += 1
return _mock_resp([[0.1]] * len(inputs))
model._call_multimodal = _mock # type: ignore[assignment]
result = await model([_img() for _ in range(8)])
self.assertEqual(result["embeddings"], [[0.1]] * 8)
self.assertEqual(call_count, 2)
async def test_multimodal_batching_by_video_limit(self) -> None:
"""3 videos with max_videos=1 produces 3 batches."""
model = DashScopeEmbeddingModel(
credential=_cred(),
model="qwen3-vl-embedding",
dimensions=1,
)
call_count = 0
async def _mock(inputs: list, **_kw: Any) -> EmbeddingResponse:
nonlocal call_count
call_count += 1
return _mock_resp([[0.1]] * len(inputs))
model._call_multimodal = _mock # type: ignore[assignment]
result = await model([_vid(), _vid(), _vid()])
self.assertEqual(result["embeddings"], [[0.1]] * 3)
self.assertEqual(call_count, 3)
async def test_video_base64_rejected(self) -> None:
"""Video with Base64Source raises ValueError."""
bad = DataBlock(source=Base64Source(data="x", media_type="video/mp4"))
with self.assertRaises(ValueError):
DashScopeEmbeddingModel._format_data_block(bad)
|