From 190d165d8bd5675c06ee949f93fd1911f2892c11 Mon Sep 17 00:00:00 2001 From: calesthio Date: Thu, 13 Aug 2026 09:16:22 -0700 Subject: [PATCH] fix: enforce Hunyuan provider contracts --- docs/PROVIDERS.md | 3 +- tests/contracts/test_hunyuan_cloud_video.py | 87 +++++++++++++++------ tools/video/hunyuan_cloud_video.py | 14 +++- 3 files changed, 76 insertions(+), 28 deletions(-) diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index a9b86874..4d285c92 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -355,7 +355,7 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon > simple Bearer-token authentication. No TC3-HMAC-SHA256 signing required. **Tools unlocked:** `hunyuan_cloud_video` -**Env vars:** `TENCENT_TOKENHUB_API_KEY` (required), `TENCENT_TOKENHUB_MODEL` (optional, overrides the default model) +**Env var:** `TENCENT_TOKENHUB_API_KEY` #### Setup @@ -365,7 +365,6 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon 4. Add to `.env`: ```bash TENCENT_TOKENHUB_API_KEY=your-tokenhub-api-key - # TENCENT_TOKENHUB_MODEL=hy-video-1.5 # optional model override ``` #### What It's Best For diff --git a/tests/contracts/test_hunyuan_cloud_video.py b/tests/contracts/test_hunyuan_cloud_video.py index e1915fe1..dc9f6d8c 100644 --- a/tests/contracts/test_hunyuan_cloud_video.py +++ b/tests/contracts/test_hunyuan_cloud_video.py @@ -199,7 +199,10 @@ class TestIdempotencyKeys: def test_includes_all_output_affecting_fields(self): fields = HunyuanCloudVideo().idempotency_key_fields - for field in ("prompt", "operation", "model", "image_url", "resolution"): + for field in ( + "prompt", "operation", "model", "image_url", "image_path", + "resolution", "logo_add", + ): assert field in fields, f"missing idempotency field: {field}" def test_excludes_execution_only_fields(self): @@ -358,6 +361,28 @@ class TestToolSpecific: assert result.success is False assert "image_url" in result.error or "image_path" in result.error + @pytest.mark.parametrize( + ("operation", "model", "expected"), + [ + ("text_to_video", "yt-video-2.0", "hy-video-1.5"), + ("image_to_video", "hy-video-1.5", "yt-video-2.0"), + ], + ) + def test_incompatible_model_operation_fails_before_submit( + self, hunyuan_env, monkeypatch, operation, model, expected + ): + monkeypatch.setattr( + HunyuanCloudVideo, + "_generate", + lambda *args, **kwargs: pytest.fail("must not submit a paid task"), + ) + inputs = {"prompt": "test", "operation": operation, "model": model} + if operation == "image_to_video": + inputs["image_url"] = "https://example.com/frame.png" + result = HunyuanCloudVideo().execute(inputs) + assert not result.success + assert expected in (result.error or "") + def test_i2v_both_url_and_path_fails(self, hunyuan_env, tmp_path): img = tmp_path / "ref.jpg" img.write_bytes(b"fake-jpeg") @@ -587,37 +612,16 @@ class TestExecuteWithMocks: assert result.success, result.error assert output_path.read_bytes() == b"fake-i2v-local-video" - def test_explicit_model_takes_priority(self, hunyuan_env, tmp_path, monkeypatch): - """When model is explicitly set, it should be used for both T2V and I2V.""" - task_id = "model-override-1" - calls = _install_fake_requests( - monkeypatch, - post_responses=[ - FakeResponse({ - "id": task_id, "request_id": "req-sub", "object": "video", - "created_at": 1700000000, "status": "queued", - }), - FakeResponse({ - "request_id": "req-poll", "object": "video", - "status": "completed", "progress": 100, - "data": {"url": "https://example.com/out.mp4"}, - }), - ], - get_responses=[FakeResponse(content=b"data")], - ) - + def test_explicit_incompatible_model_is_rejected(self, hunyuan_env): result = HunyuanCloudVideo().execute({ "prompt": "test", "operation": "image_to_video", "model": "hy-video-1.5", "image_url": "https://example.com/frame.jpg", - "poll_interval_seconds": 0.1, - "output_path": str(tmp_path / "out.mp4"), }) - assert result.success, result.error - assert result.data["model"] == "hy-video-1.5" - assert calls["post"][0]["json"]["model"] == "hy-video-1.5" + assert not result.success + assert "yt-video-2.0" in (result.error or "") def test_polling_retries_until_success(self, hunyuan_env, tmp_path, monkeypatch): """Polling should retry when status is queued/running, then succeed.""" @@ -731,6 +735,39 @@ class TestRegistryDiscovery: assert cloud.runtime == ToolRuntime.API assert local.runtime == ToolRuntime.LOCAL_GPU + def test_video_selector_routes_to_hunyuan_cloud(self, hunyuan_env, monkeypatch): + from tools.base_tool import ToolResult + from tools.video.video_selector import VideoSelector + + tool = HunyuanCloudVideo() + selector = VideoSelector() + monkeypatch.setattr(selector, "_providers", lambda: [tool]) + monkeypatch.setattr( + selector, + "_select_best_tool", + lambda _inputs, _candidates, _context: (tool, None), + ) + monkeypatch.setattr( + tool, + "execute", + lambda inputs: ToolResult( + success=True, + data={"received": inputs}, + artifacts=[inputs["output_path"]], + ), + ) + + result = selector.execute( + { + "prompt": "test", + "preferred_provider": "hunyuan_cloud", + "output_path": "out.mp4", + } + ) + assert result.success + assert result.data["selected_tool"] == "hunyuan_cloud_video" + assert result.data["selected_provider"] == "hunyuan_cloud" + # ------------------------------------------------------------------ # Schema validation diff --git a/tools/video/hunyuan_cloud_video.py b/tools/video/hunyuan_cloud_video.py index 5a55dd22..ff443b88 100644 --- a/tools/video/hunyuan_cloud_video.py +++ b/tools/video/hunyuan_cloud_video.py @@ -53,7 +53,7 @@ class HunyuanCloudVideo(BaseTool): determinism = Determinism.STOCHASTIC runtime = ToolRuntime.API - dependencies = [] + dependencies = ["env:TENCENT_TOKENHUB_API_KEY"] install_instructions = ( "Set TENCENT_TOKENHUB_API_KEY to your Tencent Cloud TokenHub API key.\n" " Get it at https://console.cloud.tencent.com/tokenhub" @@ -171,7 +171,9 @@ class HunyuanCloudVideo(BaseTool): "operation", "model", "image_url", + "image_path", "resolution", + "logo_add", ] side_effects = [ "writes video file to output_path", @@ -251,6 +253,16 @@ class HunyuanCloudVideo(BaseTool): ) operation = inputs.get("operation", "text_to_video") + model = self._resolve_model(inputs) + expected_model = _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V + if model != expected_model: + return ToolResult( + success=False, + error=( + f"Model '{model}' is not compatible with operation '{operation}'. " + f"Use '{expected_model}'." + ), + ) if operation == "image_to_video" and not inputs.get("image_url") and not inputs.get("image_path"): return ToolResult( success=False,