From 7473d0d1bb4b0b36d95217a84499e03f02444d78 Mon Sep 17 00:00:00 2001 From: Yiyabo Date: Tue, 30 Jun 2026 14:35:34 +0800 Subject: [PATCH] dashscope: add contract tests (95 tests, no API calls needed) --- tests/contracts/test_dashscope_tools.py | 449 ++++++++++++++++++++++++ 1 file changed, 449 insertions(+) create mode 100644 tests/contracts/test_dashscope_tools.py diff --git a/tests/contracts/test_dashscope_tools.py b/tests/contracts/test_dashscope_tools.py new file mode 100644 index 00000000..9df984d7 --- /dev/null +++ b/tests/contracts/test_dashscope_tools.py @@ -0,0 +1,449 @@ +"""Contract tests for DashScope (Alibaba Cloud Bailian) provider tools. + +These tests verify that the tools satisfy the BaseTool contract without +requiring a real DashScope API key or making any API calls. They check +class attributes, schemas, status reporting, cost estimates, and the +Layer 3 skill file existence. + +Run: pytest tests/contracts/test_dashscope_tools.py -v +""" + +from pathlib import Path + +import pytest + +from tools.base_tool import ( + BaseTool, + ExecutionMode, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) +from tools.graphics.dashscope_image import DashscopeImage +from tools.audio.dashscope_tts import DashscopeTTS +from tools.analysis.dashscope_asr import DashscopeAsr + +TOOLS = [DashscopeImage, DashscopeTTS, DashscopeAsr] +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent + +EXPECTED_TIER = { + DashscopeImage: ToolTier.GENERATE, + DashscopeTTS: ToolTier.VOICE, + DashscopeAsr: ToolTier.ANALYZE, +} +EXPECTED_CAPABILITY = { + DashscopeImage: "image_generation", + DashscopeTTS: "tts", + DashscopeAsr: "analysis", +} +EXPECTED_EXECUTION_MODE = { + DashscopeImage: ExecutionMode.SYNC, + DashscopeTTS: ExecutionMode.SYNC, + DashscopeAsr: ExecutionMode.ASYNC, +} + + +# ------------------------------------------------------------------ +# Contract compliance (parametrized over all 3 tools) +# ------------------------------------------------------------------ + +@pytest.mark.parametrize("cls", TOOLS, ids=lambda c: c.name) +class TestContract: + + def test_inherits_base_tool(self, cls): + assert issubclass(cls, BaseTool) + + def test_has_required_identity(self, cls): + tool = cls() + assert tool.name + assert tool.version + assert tool.provider == "dashscope" + assert tool.capability == EXPECTED_CAPABILITY[cls] + assert tool.tier == EXPECTED_TIER[cls] + assert tool.stability == ToolStability.EXPERIMENTAL + assert tool.runtime == ToolRuntime.API + + def test_has_input_schema(self, cls): + tool = cls() + schema = tool.input_schema + assert schema.get("type") == "object" + props = schema.get("properties", {}) + required = schema.get("required", []) + # Each tool has at least one required field + assert len(required) >= 1 + for field in required: + assert field in props + + def test_has_capabilities(self, cls): + tool = cls() + assert len(tool.capabilities) > 0 + + def test_has_agent_skills(self, cls): + tool = cls() + assert tool.agent_skills + assert "dashscope" in tool.agent_skills + + def test_dashscope_layer3_skill_exists(self, cls): + skill_path = ( + PROJECT_ROOT / ".agents" / "skills" / "dashscope" / "SKILL.md" + ) + assert skill_path.exists(), f"Missing Layer 3 skill: {skill_path}" + content = skill_path.read_text(encoding="utf-8") + assert "DASHSCOPE_API_KEY" in content + + def test_has_fallbacks(self, cls): + tool = cls() + assert tool.fallback or tool.fallback_tools + + def test_has_install_instructions(self, cls): + tool = cls() + assert tool.install_instructions + assert "DASHSCOPE_API_KEY" in tool.install_instructions + + def test_get_info_returns_dict(self, cls): + tool = cls() + info = tool.get_info() + assert isinstance(info, dict) + assert info["name"] == tool.name + assert info["provider"] == "dashscope" + assert info["runtime"] == "api" + assert info["agent_skills"] == ["dashscope"] + + def test_execution_mode(self, cls): + tool = cls() + assert tool.execution_mode == EXPECTED_EXECUTION_MODE[cls] + + def test_status_unavailable_without_key(self, cls, monkeypatch): + monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False) + tool = cls() + assert tool.get_status() == ToolStatus.UNAVAILABLE + + def test_status_available_with_key(self, cls, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = cls() + assert tool.get_status() == ToolStatus.AVAILABLE + + def test_idempotency_key_fields(self, cls): + tool = cls() + assert len(tool.idempotency_key_fields) > 0 + + def test_has_resource_profile(self, cls): + tool = cls() + assert tool.resource_profile.network_required is True + assert tool.resource_profile.vram_mb == 0 + + def test_has_retry_policy(self, cls): + tool = cls() + assert tool.retry_policy.max_retries >= 0 + + def test_has_side_effects(self, cls): + tool = cls() + assert len(tool.side_effects) > 0 + # Must mention it calls the API + assert any("API" in s for s in tool.side_effects) + + def test_has_user_visible_verification(self, cls): + tool = cls() + assert len(tool.user_visible_verification) > 0 + + def test_lazy_imports_requests(self, cls): + """Tool module must not import requests at top level (registry + discovery must stay fast).""" + import importlib + import sys + # Remove requests from cache to simulate fresh import + mod_name = cls.__module__ + if "requests" in sys.modules: + del sys.modules["requests"] + # Re-import the tool module — should not pull in requests + # (requests is imported inside execute(), not at module level) + importlib.reload(sys.modules[mod_name]) + # The tool module itself should not have imported requests + # (it's inside execute, so module-level reload shouldn't trigger it) + # This is a smoke test — the real proof is that registry.discover() + # works without requests installed, but requests IS installed here. + + def test_estimate_cost_returns_float(self, cls): + tool = cls() + # Use tool-specific minimal inputs + if cls is DashscopeImage: + cost = tool.estimate_cost({"prompt": "test", "n": 1}) + elif cls is DashscopeTTS: + cost = tool.estimate_cost({"text": "test"}) + else: + cost = tool.estimate_cost({"audio_url": "https://x.com/a.mp3"}) + assert isinstance(cost, float) + assert cost >= 0.0 + + def test_dry_run_returns_dict(self, cls): + tool = cls() + if cls is DashscopeImage: + result = tool.dry_run({"prompt": "test"}) + elif cls is DashscopeTTS: + result = tool.dry_run({"text": "test"}) + else: + result = tool.dry_run({"audio_url": "https://x.com/a.mp3"}) + assert isinstance(result, dict) + assert "tool" in result + assert result["tool"] == tool.name + + +# ------------------------------------------------------------------ +# Image-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeImageSpecific: + + def test_default_model_is_qwen_image_2_pro(self): + tool = DashscopeImage() + assert tool.input_schema["properties"]["model"]["default"] == "qwen-image-2.0-pro" + + def test_default_size_uses_asterisk_format(self): + """CRITICAL: DashScope uses W*H (asterisk), not WxH.""" + tool = DashscopeImage() + size_default = tool.input_schema["properties"]["size"]["default"] + assert "*" in size_default + assert "x" not in size_default.lower() + + def test_cost_positive_for_image(self): + tool = DashscopeImage() + assert tool.estimate_cost({"prompt": "test", "n": 1}) > 0.0 + + def test_cost_scales_with_n(self): + tool = DashscopeImage() + cost1 = tool.estimate_cost({"prompt": "test", "n": 1}) + cost3 = tool.estimate_cost({"prompt": "test", "n": 3}) + assert cost3 > cost1 + + def test_build_payload_uses_asterisk_size(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "test"}) + assert "*" in payload["parameters"]["size"] + + def test_build_payload_includes_messages_structure(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "a cat"}) + assert "input" in payload + assert "messages" in payload["input"] + assert payload["input"]["messages"][0]["content"][0]["text"] == "a cat" + + def test_build_payload_optional_negative_prompt(self): + tool = DashscopeImage() + payload = tool._build_payload({ + "prompt": "test", + "negative_prompt": "blurry", + }) + assert payload["parameters"]["negative_prompt"] == "blurry" + + def test_build_payload_omits_negative_prompt_when_absent(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "test"}) + assert "negative_prompt" not in payload["parameters"] + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeImage._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# TTS-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeTtsSpecific: + + def test_default_model_is_qwen3_tts_flash(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["model"]["default"] == "qwen3-tts-flash" + + def test_default_voice_is_cherry(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["voice"]["default"] == "Cherry" + + def test_default_language_is_auto(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["language_type"]["default"] == "Auto" + + def test_cost_scales_with_text_length(self): + tool = DashscopeTTS() + cost_short = tool.estimate_cost({"text": "hi"}) + cost_long = tool.estimate_cost({"text": "hi " * 100}) + assert cost_long > cost_short + + def test_build_payload_includes_input_text_voice(self): + tool = DashscopeTTS() + payload = tool._build_payload({"text": "hello", "voice": "Ethan"}) + assert payload["input"]["text"] == "hello" + assert payload["input"]["voice"] == "Ethan" + + def test_build_payload_adds_instructions_for_instruct_model(self): + tool = DashscopeTTS() + payload = tool._build_payload({ + "text": "hello", + "instructions": "speak softly", + }) + assert payload["input"]["instructions"] == "speak softly" + assert payload["input"]["optimize_instructions"] is True + + def test_fallback_includes_piper(self): + """Piper is the free offline fallback — must be in fallback list.""" + tool = DashscopeTTS() + assert "piper_tts" in tool.fallback_tools + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeTTS._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# ASR-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeAsrSpecific: + + def test_default_model_is_filetrans(self): + """CRITICAL: must use qwen3-asr-flash-filetrans, NOT qwen3-asr-flash. + The sync version does not support word-level timestamps.""" + tool = DashscopeAsr() + assert tool.input_schema["properties"]["model"]["default"] == "qwen3-asr-flash-filetrans" + + def test_execution_mode_is_async(self): + tool = DashscopeAsr() + assert tool.execution_mode == ExecutionMode.ASYNC + + def test_default_enable_words_is_true(self): + """Word-level timestamps must be enabled by default.""" + tool = DashscopeAsr() + assert tool.input_schema["properties"]["enable_words"]["default"] is True + + def test_default_language_hints_includes_zh_en(self): + tool = DashscopeAsr() + hints = tool.input_schema["properties"]["language_hints"]["default"] + assert "zh" in hints + assert "en" in hints + + def test_rejects_local_file_path(self, monkeypatch): + """audio_url must be a public URL — local paths are rejected.""" + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": "/local/path/audio.mp3"}) + assert result.success is False + assert "publicly accessible URL" in result.error + + def test_rejects_relative_path(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": "audio.mp3"}) + assert result.success is False + assert "publicly accessible URL" in result.error + + def test_rejects_empty_url(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": ""}) + assert result.success is False + assert "required" in result.error.lower() + + def test_rejects_no_key(self, monkeypatch): + monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False) + tool = DashscopeAsr() + result = tool.execute({"audio_url": "https://example.com/audio.mp3"}) + assert result.success is False + assert "DASHSCOPE_API_KEY" in result.error + + def test_build_payload_enables_words(self): + tool = DashscopeAsr() + payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"}) + assert payload["parameters"]["enable_words"] is True + + def test_build_payload_includes_file_url(self): + """qwen3-asr-flash-filetrans uses file_url (singular string), + NOT file_urls (plural array) like paraformer-v2.""" + tool = DashscopeAsr() + payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"}) + assert payload["input"]["file_url"] == "https://x.com/a.mp3" + + def test_extract_words_normalizes_ms_to_seconds(self): + """Word timestamps from DashScope are in milliseconds; the tool + must normalize to seconds for downstream subtitle building.""" + fake_transcription = { + "transcripts": [ + { + "sentences": [ + { + "words": [ + {"text": "hello", "begin_time": 1000, "end_time": 1500}, + {"text": "world", "begin_time": 1500, "end_time": 2000}, + ] + } + ] + } + ] + } + words = DashscopeAsr._extract_words(fake_transcription) + assert len(words) == 2 + assert words[0]["text"] == "hello" + assert words[0]["begin_time_seconds"] == 1.0 + assert words[0]["end_time_seconds"] == 1.5 + assert words[1]["begin_time_seconds"] == 1.5 + assert words[1]["end_time_seconds"] == 2.0 + + def test_extract_words_handles_empty_transcription(self): + words = DashscopeAsr._extract_words({}) + assert words == [] + + def test_is_public_url_accepts_https(self): + assert DashscopeAsr._is_public_url("https://example.com/audio.mp3") is True + + def test_is_public_url_rejects_local(self): + assert DashscopeAsr._is_public_url("/local/path/audio.mp3") is False + assert DashscopeAsr._is_public_url("audio.mp3") is False + assert DashscopeAsr._is_public_url("ftp://example.com/audio.mp3") is False + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeAsr._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# Registry discovery +# ------------------------------------------------------------------ + +class TestDashscopeRegistryDiscovery: + + def test_all_three_tools_discoverable(self): + from tools.tool_registry import ToolRegistry + registry = ToolRegistry() + registry.discover() + dashscope_tools = [ + t for t in registry._tools.values() + if t.provider == "dashscope" + ] + names = {t.name for t in dashscope_tools} + assert names == {"dashscope_image", "dashscope_tts", "dashscope_asr"} + + def test_image_selector_finds_dashscope(self): + """image_selector should auto-discover dashscope_image by capability.""" + from tools.graphics.image_selector import ImageSelector + selector = ImageSelector() + # Selector discovers providers by capability="image_generation" + # dashscope_image has that capability, so it should be routable + assert DashscopeImage().capability == "image_generation" + + def test_tts_selector_finds_dashscope(self): + """tts_selector should auto-discover dashscope_tts by capability.""" + from tools.audio.tts_selector import TTSSelector + selector = TTSSelector() + assert DashscopeTTS().capability == "tts"