fix: harden Kling integration verification

Isolate Kling contract tests from the singleton registry so discovery state cannot leak into later selector tests. Align lip-sync face, audio, and timing payloads with the current official API and extend the live smoke coverage.
This commit is contained in:
xucailiang
2026-07-10 20:53:45 +08:00
parent 2b6d717f00
commit b9b9b82b64
11 changed files with 676 additions and 90 deletions

View File

@@ -0,0 +1,15 @@
"""Shared fixtures for contract tests."""
from __future__ import annotations
import pytest
from tools.tool_registry import ToolRegistry
@pytest.fixture()
def isolated_tool_registry(monkeypatch) -> ToolRegistry:
"""Provide a registry singleton replacement scoped to one test."""
test_registry = ToolRegistry()
monkeypatch.setattr("tools.tool_registry.registry", test_registry)
return test_registry

View File

@@ -7,6 +7,8 @@ import json
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
@@ -14,14 +16,12 @@ from tools.avatar.kling_avatar import KlingAvatar
from tools.avatar.kling_lip_sync import KlingLipSync
from tools.avatar.lip_sync import LipSync
from tools.avatar.talking_head import TalkingHead
from tools.tool_registry import registry
def test_registry_discovers_kling_avatar(monkeypatch):
def test_registry_discovers_kling_avatar(monkeypatch, isolated_tool_registry):
monkeypatch.delenv("KLING_API_KEY", raising=False)
registry.clear()
registry.discover("tools")
tool = registry.get("kling_avatar")
isolated_tool_registry.discover("tools")
tool = isolated_tool_registry.get("kling_avatar")
assert tool is not None
assert tool.capability == "avatar"
assert tool.provider == "kling_official"
@@ -123,11 +123,10 @@ def test_avatar_cost_estimate_is_not_zero():
assert tool.dry_run({"image_url": "x", "audio_id": "a"})["cost_estimate_confidence"] == "low"
def test_registry_discovers_kling_lip_sync(monkeypatch):
def test_registry_discovers_kling_lip_sync(monkeypatch, isolated_tool_registry):
monkeypatch.delenv("KLING_API_KEY", raising=False)
registry.clear()
registry.discover("tools")
tool = registry.get("kling_lip_sync")
isolated_tool_registry.discover("tools")
tool = isolated_tool_registry.get("kling_lip_sync")
assert tool is not None
assert tool.capability == "avatar"
assert tool.provider == "kling_official"
@@ -135,8 +134,12 @@ def test_registry_discovers_kling_lip_sync(monkeypatch):
def test_lip_sync_schema_and_local_tool_are_distinct():
tool = KlingLipSync()
props = tool.input_schema["properties"]
assert "kling-official" in tool.agent_skills
assert "avatar-video" in tool.agent_skills
assert "sound_start_time" in props
assert "sound_end_time" in props
assert "sound_insert_time" in props
assert tool.runtime.value == "api"
assert LipSync().provider == "wav2lip"
assert LipSync().runtime.value == "local_gpu"
@@ -165,17 +168,99 @@ def test_advanced_lip_sync_payload_uses_face_and_audio_path(tmp_path):
"session_id": "session-a",
"face_id": "face-a",
"audio_path": str(audio_path),
"sound_start_time": 0,
"sound_end_time": 2500,
"sound_insert_time": 500,
"callback_url": "https://example.com/kling/callback",
}
)
assert request["path"] == "/v1/videos/advanced-lip-sync"
assert request["payload"]["session_id"] == "session-a"
assert request["payload"]["face_choose"] == [{"face_id": "face-a"}]
assert request["payload"]["sound_file"] == base64.b64encode(b"audio").decode("ascii")
assert request["payload"]["face_choose"] == [
{
"face_id": "face-a",
"sound_file": base64.b64encode(b"audio").decode("ascii"),
"sound_start_time": 0,
"sound_end_time": 2500,
"sound_insert_time": 500,
}
]
assert "sound_file" not in request["payload"]
assert request["payload"]["callback_url"] == "https://example.com/kling/callback"
def test_advanced_lip_sync_accepts_official_nested_face_choose():
face_choose = {
"face_id": "face-a",
"audio_id": "audio-a",
"sound_start_time": 0,
"sound_end_time": 4000,
"sound_insert_time": 500,
}
request = KlingLipSync()._build_advanced_request(
{"session_id": "session-a", "face_choose": [face_choose]}
)
assert request["payload"]["face_choose"] == [face_choose]
assert request["audio_source"] == {"type": "audio_id", "value": "audio-a"}
def test_full_lip_sync_preserves_nested_face_timing_defaults():
inputs = {
"session_id": "session-a",
"face_choose": [
{
"face_id": "face-a",
"audio_id": "audio-a",
"sound_start_time": 250,
"sound_end_time": 4250,
"sound_insert_time": 750,
}
],
}
tool = KlingLipSync()
tool._apply_face_timing_defaults(
inputs, {"face_id": "face-a", "start_time": 0, "end_time": 0}
)
request = tool._build_advanced_request(inputs)
assert request["payload"]["face_choose"][0]["sound_start_time"] == 250
assert request["payload"]["face_choose"][0]["sound_end_time"] == 4250
assert request["payload"]["face_choose"][0]["sound_insert_time"] == 750
@pytest.mark.parametrize(
("top_level", "nested", "message"),
[
({"audio_id": "audio-top"}, {"audio_id": "audio-nested"}, "audio input"),
({"sound_end_time": 5000}, {"sound_end_time": 4000}, "sound_end_time"),
],
)
def test_advanced_lip_sync_rejects_conflicting_top_level_and_nested_values(
top_level, nested, message
):
inputs = {
"session_id": "session-a",
"face_choose": [
{
"face_id": "face-a",
"audio_id": "audio-a",
"sound_start_time": 0,
"sound_end_time": 4000,
"sound_insert_time": 0,
**nested,
}
],
**top_level,
}
with pytest.raises(ValueError, match=message):
KlingLipSync()._build_advanced_request(inputs)
def test_identify_face_execute_writes_faces_artifact(monkeypatch, tmp_path):
class FakeClient:
def post(self, path, payload):
@@ -184,7 +269,14 @@ def test_identify_face_execute_writes_faces_artifact(monkeypatch, tmp_path):
"code": 0,
"data": {
"session_id": "session-a",
"faces": [{"face_id": "face-a", "bbox": [0, 0, 100, 100]}],
"face_data": [
{
"face_id": "face-a",
"face_image": "https://example.com/face.png",
"start_time": 0,
"end_time": 5200,
}
],
},
}
@@ -214,9 +306,9 @@ def test_full_lip_sync_requires_confirmation_for_multiple_faces(monkeypatch, tmp
"code": 0,
"data": {
"session_id": "session-a",
"faces": [
{"face_id": "face-small", "bbox": [0, 0, 50, 50]},
{"face_id": "face-large", "bbox": [0, 0, 200, 200]},
"face_data": [
{"face_id": "face-small", "bbox": [0, 0, 50, 50], "start_time": 0, "end_time": 4000},
{"face_id": "face-large", "bbox": [0, 0, 200, 200], "start_time": 500, "end_time": 5500},
],
},
}
@@ -247,9 +339,9 @@ def test_full_lip_sync_auto_selects_largest_face_and_downloads(monkeypatch, tmp_
"code": 0,
"data": {
"session_id": "session-a",
"faces": [
{"face_id": "face-small", "bbox": [0, 0, 50, 50]},
{"face_id": "face-large", "bbox": [0, 0, 200, 200]},
"face_data": [
{"face_id": "face-small", "bbox": [0, 0, 50, 50], "start_time": 0, "end_time": 4000},
{"face_id": "face-large", "bbox": [0, 0, 200, 200], "start_time": 500, "end_time": 5500},
],
},
}
@@ -257,7 +349,15 @@ def test_full_lip_sync_auto_selects_largest_face_and_downloads(monkeypatch, tmp_
def create_classic_task(self, path, payload):
self.path = path
self.payload = payload
assert payload["face_choose"] == [{"face_id": "face-large"}]
assert payload["face_choose"] == [
{
"face_id": "face-large",
"audio_id": "audio-a",
"sound_start_time": 0,
"sound_end_time": 5000,
"sound_insert_time": 500,
}
]
return "lip-task-1"
def poll_classic(self, path, task_id, result_key, timeout_seconds, poll_interval):
@@ -285,7 +385,15 @@ def test_full_lip_sync_auto_selects_largest_face_and_downloads(monkeypatch, tmp_
assert result.success
assert result.data["task_id"] == "lip-task-1"
assert result.data["face_selection"]["selection_method"] == "auto_selected"
assert result.data["face_choose"] == [{"face_id": "face-large"}]
assert result.data["face_choose"] == [
{
"face_id": "face-large",
"audio_id": "audio-a",
"sound_start_time": 0,
"sound_end_time": 5000,
"sound_insert_time": 500,
}
]
assert Path(result.artifacts[0]).read_bytes() == b"video"
faces_artifact = next(Path(path) for path in result.artifacts if Path(path).name == "kling_lip_sync_faces.json")
assert json.loads(faces_artifact.read_text())["selection"]["selection_method"] == "auto_selected"
@@ -300,10 +408,21 @@ def test_auto_select_face_area_avoids_position_inflation():
def test_advanced_lip_sync_execute_downloads_video(monkeypatch, tmp_path):
audio_path = tmp_path / "voice.mp3"
audio_path.write_bytes(b"audio")
class FakeClient:
def create_classic_task(self, path, payload):
assert path == "/v1/videos/advanced-lip-sync"
assert payload["face_choose"] == [{"face_id": "face-a"}]
assert payload["face_choose"] == [
{
"face_id": "face-a",
"sound_file": base64.b64encode(b"audio").decode("ascii"),
"sound_start_time": 0,
"sound_end_time": 4000,
"sound_insert_time": 0,
}
]
return "lip-task-1"
def poll_classic(self, path, task_id, result_key, timeout_seconds, poll_interval):
@@ -322,7 +441,10 @@ def test_advanced_lip_sync_execute_downloads_video(monkeypatch, tmp_path):
"operation": "advanced_lip_sync",
"session_id": "session-a",
"face_id": "face-a",
"audio_id": "audio-a",
"audio_path": str(audio_path),
"sound_start_time": 0,
"sound_end_time": 4000,
"sound_insert_time": 0,
"output_path": str(tmp_path / "lip.mp4"),
}
)
@@ -331,6 +453,15 @@ def test_advanced_lip_sync_execute_downloads_video(monkeypatch, tmp_path):
assert result.data["provider"] == "kling_official"
assert result.data["task_id"] == "lip-task-1"
assert result.data["duration_seconds"] == 4.0
assert result.data["face_choose"] == [
{
"face_id": "face-a",
"sound_start_time": 0,
"sound_end_time": 4000,
"sound_insert_time": 0,
"sound_file_provided": True,
}
]
assert result.cost_usd > 0

View File

@@ -27,7 +27,6 @@ from tools.audio.kling_tts import KlingTTS
from tools.avatar.kling_avatar import KlingAvatar
from tools.avatar.kling_lip_sync import KlingLipSync
from tools.graphics.kling_official_image import KlingOfficialImage
from tools.tool_registry import registry
from tools.video.kling_official_video import KlingOfficialVideo
@@ -194,9 +193,9 @@ def test_turbo_create_and_poll_parse_result_paths(monkeypatch):
def test_schema_snapshot_contains_phase1_contract_fields():
fixture = PROJECT_ROOT / "tests/fixtures/kling_official/schema_snapshot.json"
data = json.loads(fixture.read_text())
assert data["build_id"] == "97344324"
assert "index-B9E4in0e.js" in data["chunk_names"]
assert "document-navigation-nxVgwiS5.js" in data["chunk_names"]
assert data["build_id"] == "97939672"
assert "index-0m3slU3p.js" in data["chunk_names"]
assert "document-navigation-Dk7H_V3n.js" in data["chunk_names"]
assert data["api_base"]["auth_env"] == "KLING_API_KEY"
assert data["task_statuses"]["classic"] == ["submitted", "processing", "succeed", "failed"]
assert data["task_statuses"]["turbo"] == ["submitted", "processing", "succeeded", "failed"]
@@ -211,6 +210,7 @@ def test_schema_snapshot_contains_phase1_contract_fields():
assert data["endpoints"]["video_effects"]["path"] == "/v1/videos/effects"
assert data["result_paths"]["classic_audio_results"] == "data.task_result.audios[]"
assert data["result_paths"]["identify_face_session"] == "data.session_id"
assert data["result_paths"]["identify_face_results"] == "data.face_data[]"
assert data["core_field_enums"]["tts_voice_language"] == ["zh", "en"]
assert data["core_field_enums"]["avatar_mode"] == ["std", "pro"]
@@ -251,7 +251,7 @@ def test_elements_helper_normalizes_and_records_metadata(tmp_path):
raise AssertionError("element_list items without element_id must be rejected")
def test_elements_helper_read_only_endpoints_do_not_enter_registry():
def test_elements_helper_read_only_endpoints_do_not_enter_registry(isolated_tool_registry):
fake = HelperFakeClient()
assert get_custom_element(123, client=fake)["data"]["element_id"] == 123
assert list_custom_elements(client=fake)["data"][0]["element_id"] == 456
@@ -266,10 +266,9 @@ def test_elements_helper_read_only_endpoints_do_not_enter_registry():
assert not hasattr(elements_module, "create_element")
assert not hasattr(elements_module, "delete_element")
registry.clear()
registry.discover("tools")
assert registry.get("kling_elements") is None
assert registry.get("kling_account_usage") is None
isolated_tool_registry.discover("tools")
assert isolated_tool_registry.get("kling_elements") is None
assert isolated_tool_registry.get("kling_account_usage") is None
def test_account_usage_helper_uses_endpoint_cache_and_throttle():
@@ -402,8 +401,7 @@ def test_provider_agent_skills_reference_kling_official():
assert "kling-official" in KlingLipSync().agent_skills
def test_phase3_does_not_register_audio_or_video_effect_tools():
registry.clear()
registry.discover("tools")
assert registry.get("kling_audio") is None
assert registry.get("kling_effects") is None
def test_phase3_does_not_register_audio_or_video_effect_tools(isolated_tool_registry):
isolated_tool_registry.discover("tools")
assert isolated_tool_registry.get("kling_audio") is None
assert isolated_tool_registry.get("kling_effects") is None

View File

@@ -43,6 +43,8 @@ def test_cli_modes_are_explicit_and_non_paid_by_default():
assert script._execution_mode(script._parse_args(["--live-tts"])) == "live_tts"
assert script._execution_mode(script._parse_args(["--live-full"])) == "live_full"
assert script._execution_mode(script._parse_args(["--live"])) == "live_full"
assert script._execution_mode(script._parse_args(["--live-avatar"])) == "live_avatar"
assert script._execution_mode(script._parse_args(["--live-all"])) == "live_all"
def test_video_duration_aligns_to_narration_within_kling_limits():
@@ -52,3 +54,39 @@ def test_video_duration_aligns_to_narration_within_kling_limits():
assert script._aligned_video_duration("10", 6.05) == "10"
assert script._aligned_video_duration("3", None) == "3"
assert script._aligned_video_duration("3", 30.0) == "15"
def test_live_all_combines_core_and_avatar_results(monkeypatch, tmp_path):
script = _load_script()
core = {
"artifacts": {"narration": "narration.mp3", "final": "final.mp4"},
"ffprobe": {"final": {"duration": 6.0}},
"estimated_cost_usd": 0.25,
}
avatar_suite = {
"artifacts": {"avatar": "avatar.mp4", "lip_sync": "lip.mp4"},
"ffprobe": {"avatar": {"duration": 7.0}, "lip_sync": {"duration": 7.0}},
"estimated_cost_usd": 0.75,
}
monkeypatch.setattr(script, "_run_live_full", lambda *args, **kwargs: core)
monkeypatch.setattr(
script, "_run_live_avatar_suite", lambda *args, **kwargs: avatar_suite
)
result = script._run_live_all(
tmp_path,
voice_id="voice-a",
voice_language="en",
voice_speed=1.0,
text="hello",
timeout_seconds=30,
poll_interval=1.0,
include_account_usage=False,
video_duration="3",
)
assert result["core"] is core
assert result["avatar_suite"] is avatar_suite
assert result["artifacts"]["final"] == "final.mp4"
assert result["artifacts"]["lip_sync"] == "lip.mp4"
assert result["estimated_cost_usd"] == 1.0

View File

@@ -9,16 +9,13 @@ from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from tools.graphics.image_selector import ImageSelector
from tools.graphics.kling_official_image import KlingOfficialImage
from tools.tool_registry import registry
def test_registry_discovers_kling_official_image(monkeypatch):
def test_registry_discovers_kling_official_image(monkeypatch, isolated_tool_registry):
monkeypatch.delenv("KLING_API_KEY", raising=False)
registry.clear()
registry.discover("tools")
tool = registry.get("kling_official_image")
isolated_tool_registry.discover("tools")
tool = isolated_tool_registry.get("kling_official_image")
assert tool is not None
assert tool.capability == "image_generation"
assert tool.provider == "kling_official"
@@ -255,12 +252,9 @@ def test_execute_image_omni_series_records_references_callback_and_artifacts(mon
assert Path(result.artifacts[1]).name == "series_2.png"
def test_image_selector_prefers_official_provider(monkeypatch):
def test_image_selector_prefers_official_provider(monkeypatch, isolated_tool_registry):
monkeypatch.setenv("KLING_API_KEY", "test-key")
registry.clear()
registry.register(KlingOfficialImage())
registry.register(ImageSelector())
registry._discovered_packages.add("tools")
isolated_tool_registry.discover("tools")
def fake_execute(self, inputs):
from tools.base_tool import ToolResult
@@ -268,10 +262,11 @@ def test_image_selector_prefers_official_provider(monkeypatch):
return ToolResult(success=True, data={"output_path": "out.png"}, artifacts=["out.png"])
monkeypatch.setattr(KlingOfficialImage, "execute", fake_execute)
result = registry.get("image_selector").execute(
result = isolated_tool_registry.get("image_selector").execute(
{
"prompt": "official image",
"preferred_provider": "kling_official",
"allowed_providers": ["kling_official"],
"api_family": "omni",
"image_reference": "subject",
}

View File

@@ -10,17 +10,14 @@ PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from tools._kling.account import reset_account_usage_cache
from tools.tool_registry import registry
from tools._kling.errors import KlingAPIError
from tools.video.kling_official_video import KlingOfficialVideo
from tools.video.video_selector import VideoSelector
def test_registry_discovers_kling_official_video(monkeypatch):
def test_registry_discovers_kling_official_video(monkeypatch, isolated_tool_registry):
monkeypatch.delenv("KLING_API_KEY", raising=False)
registry.clear()
registry.discover("tools")
tool = registry.get("kling_official_video")
isolated_tool_registry.discover("tools")
tool = isolated_tool_registry.get("kling_official_video")
assert tool is not None
assert tool.capability == "video_generation"
assert tool.provider == "kling_official"
@@ -355,14 +352,13 @@ def test_execute_downloads_all_omni_video_outputs_and_records_metadata(monkeypat
assert Path(result.artifacts[1]).name == "out_2.mp4"
def test_video_selector_prefers_official_provider_without_fal_upload(monkeypatch, tmp_path):
def test_video_selector_prefers_official_provider_without_fal_upload(
monkeypatch, tmp_path, isolated_tool_registry
):
monkeypatch.setenv("KLING_API_KEY", "test-key")
image_path = tmp_path / "ref.png"
image_path.write_bytes(b"fake")
registry.clear()
registry.register(KlingOfficialVideo())
registry.register(VideoSelector())
registry._discovered_packages.add("tools")
isolated_tool_registry.discover("tools")
seen = {}
@@ -377,11 +373,12 @@ def test_video_selector_prefers_official_provider_without_fal_upload(monkeypatch
monkeypatch.setattr(KlingOfficialVideo, "execute", fake_execute)
monkeypatch.setattr("tools.video._shared.upload_image_fal", fail_upload)
result = registry.get("video_selector").execute(
result = isolated_tool_registry.get("video_selector").execute(
{
"prompt": "animate",
"operation": "image_to_video",
"preferred_provider": "kling_official",
"allowed_providers": ["kling_official"],
"reference_image_path": str(image_path),
}
)

View File

@@ -9,15 +9,12 @@ PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from tools.audio.kling_tts import KlingTTS
from tools.audio.tts_selector import TTSSelector
from tools.tool_registry import registry
def test_registry_discovers_kling_tts(monkeypatch):
def test_registry_discovers_kling_tts(monkeypatch, isolated_tool_registry):
monkeypatch.delenv("KLING_API_KEY", raising=False)
registry.clear()
registry.discover("tools")
tool = registry.get("kling_tts")
isolated_tool_registry.discover("tools")
tool = isolated_tool_registry.get("kling_tts")
assert tool is not None
assert tool.capability == "tts"
assert tool.provider == "kling_official"
@@ -153,12 +150,9 @@ def test_execute_accepts_synchronous_create_response(monkeypatch, tmp_path):
assert result.data["audio_duration_seconds"] == 2.5
def test_tts_selector_prefers_kling_official(monkeypatch):
def test_tts_selector_prefers_kling_official(monkeypatch, isolated_tool_registry):
monkeypatch.setenv("KLING_API_KEY", "test-key")
registry.clear()
registry.register(KlingTTS())
registry.register(TTSSelector())
registry._discovered_packages.add("tools")
isolated_tool_registry.discover("tools")
def fake_execute(self, inputs):
from tools.base_tool import ToolResult
@@ -166,11 +160,12 @@ def test_tts_selector_prefers_kling_official(monkeypatch):
return ToolResult(success=True, data={"output_path": "out.mp3"}, artifacts=["out.mp3"])
monkeypatch.setattr(KlingTTS, "execute", fake_execute)
result = registry.get("tts_selector").execute(
result = isolated_tool_registry.get("tts_selector").execute(
{
"text": "official speech",
"voice_id": "voice-a",
"preferred_provider": "kling_official",
"allowed_providers": ["kling_official"],
}
)
assert result.success

View File

@@ -1,5 +1,5 @@
{
"build_id": "97344324",
"build_id": "97939672",
"source_urls": [
"https://kling.ai/document-api/api/get-started/authentication",
"https://kling.ai/document-api/api/get-started/error-codes",
@@ -18,10 +18,10 @@
"https://kling.ai/document-api/api/video/effects"
],
"chunk_names": [
"index-B9E4in0e.js",
"document-navigation-nxVgwiS5.js"
"index-0m3slU3p.js",
"document-navigation-Dk7H_V3n.js"
],
"extracted_at": "2026-07-03T08:12:55Z",
"extracted_at": "2026-07-10T08:34:20Z",
"api_base": {
"default": "https://api-singapore.klingai.com",
"env_override": "KLING_API_BASE_URL",
@@ -145,6 +145,7 @@
"classic_image_results": "data.task_result.images[]",
"classic_audio_results": "data.task_result.audios[]",
"identify_face_session": "data.session_id",
"identify_face_results": "data.face_data[]",
"turbo_results": "data[0].outputs[]"
},
"core_field_enums": {
@@ -220,7 +221,8 @@
]
},
"notes": [
"Official docs were fetched as a SPA on 2026-07-03. Current HTML exposes buildId 97344324.",
"Official docs were fetched as a SPA on 2026-07-10. Current HTML exposes buildId 97939672.",
"Advanced lip sync currently accepts one face_choose item. The item contains face_id, audio_id or sound_file, and sound_start_time, sound_end_time, and sound_insert_time.",
"The current entry/navigation bundles no longer expose the older api-*.js chunk names as literal references, so this fixture records the current entry assets and the Phase 1 core schema facts used by implementation and tests."
]
}