"""Seedance 视频适配器的测试：领域请求 ↔ 厂商请求体的翻译，以及状态映射。 夹具里的响应体逐字取自官方文档的「返回格式」示例，避免测试通过而真实字段名对不上。 """ from __future__ import annotations import httpx import pytest from app.domain.generation import ( GenerationRequest, MediaKind, Retry, TaskState, UserFacing, ) from app.providers.volcengine.client import ArkClient from app.providers.volcengine.models import SeedanceSubmission from app.providers.volcengine.video import PROVIDER_CODE, SeedanceVideoProvider FAKE_API_KEY = "dummy-api-key-for-tests" #: 官方查询文档「返回格式」一节的原样示例。 OFFICIAL_SUCCESS_PAYLOAD = { "id": "cgt-2026****-****", "model": "doubao-seedance-2-0-260128", "status": "succeeded", "content": { "video_url": "https://ark-content-generation-cn-beijing.tos-cn-beijing.volces.com/xxx" }, "usage": {"completion_tokens": 108900, "total_tokens": 108900}, "created_at": 1779348818, "updated_at": 1779348874, "seed": 78674, "resolution": "720p", "ratio": "16:9", "duration": 5, "framespersecond": 24, "service_tier": "default", "execution_expires_after": 172800, "generate_audio": True, "draft": False, "priority": 0, } def provider_with(handler, **kwargs) -> SeedanceVideoProvider: client = ArkClient( api_key=FAKE_API_KEY, base_url="https://ark.example.invalid/api/v3", client=httpx.Client(transport=httpx.MockTransport(handler)), ) return SeedanceVideoProvider(client, **kwargs) def video_request(**kwargs) -> GenerationRequest: defaults = { "job_id": "job-1", "media_kind": MediaKind.VIDEO, "model": "doubao-seedance-x", "prompt": "一只猫", } return GenerationRequest(**{**defaults, **kwargs}) class TestIdentity: def test_provider_code_matches_the_catalog_column(self): assert ( provider_with(lambda r: httpx.Response(200, json={})).provider_code == PROVIDER_CODE ) def test_declares_itself_as_a_video_provider(self): assert ( provider_with(lambda r: httpx.Response(200, json={})).media_kind is MediaKind.VIDEO ) class TestSubmitTranslation: def test_text_to_video_sends_only_text(self): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: import json captured.update(json.loads(request.content)) return httpx.Response(200, json={"id": "cgt-1"}) provider_with(handler).submit(video_request()) assert captured["content"] == [{"type": "text", "text": "一只猫"}] def test_input_url_becomes_first_frame_with_explicit_role(self): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: import json captured.update(json.loads(request.content)) return httpx.Response(200, json={"id": "cgt-1"}) provider_with(handler).submit(video_request(input_url="https://x/a.png")) image = captured["content"][1] assert image["type"] == "image_url" assert image["image_url"]["url"] == "https://x/a.png" # role 省略时厂商按 first_frame 处理，但显式写出以免依赖默认行为。 assert image["role"] == "first_frame" def test_model_from_request_wins_over_the_configured_default(self): """请求里的模型来自任务行冻结的快照，必须优先。 否则「任务记录里写的模型」与「实际调用的模型」会不一致，事后无法对账。 """ captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: import json captured.update(json.loads(request.content)) return httpx.Response(200, json={"id": "cgt-1"}) provider_with(handler, default_model="fallback-model").submit( video_request(model="snapshot-model") ) assert captured["model"] == "snapshot-model" def test_configured_default_is_only_a_fallback(self): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: import json captured.update(json.loads(request.content)) return httpx.Response(200, json={"id": "cgt-1"}) provider_with(handler, default_model="fallback-model").submit( video_request(model="") ) assert captured["model"] == "fallback-model" def test_no_model_anywhere_fails_before_calling_the_vendor(self): with pytest.raises(ValueError, match="未指定模型"): provider_with(lambda r: httpx.Response(200, json={})).submit( video_request(model="") ) def test_task_expiry_comes_from_config_not_the_vendor_default(self): # 厂商默认 48 小时，对交互式应用过长——被放弃的任务会在厂商侧挂两天。 captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: import json captured.update(json.loads(request.content)) return httpx.Response(200, json={"id": "cgt-1"}) provider_with(handler, task_expiry_sec=7200).submit(video_request()) assert captured["execution_expires_after"] == 7200 def test_rejects_a_misrouted_image_request(self): with pytest.raises(ValueError, match="只处理视频生成"): provider_with(lambda r: httpx.Response(200, json={})).submit( video_request(media_kind=MediaKind.IMAGE) ) def test_returns_vendor_task_id(self): def handler(request): return httpx.Response(200, json={"id": "cgt-123"}) assert provider_with(handler).submit(video_request()) == "cgt-123" def test_success_without_task_id_is_an_error(self): # 2xx 但没有 ID：我们再也查不到这个任务，而它可能已经在计费。 def handler(request): return httpx.Response(200, json={"unexpected": True}) with pytest.raises(ValueError, match="缺少任务 ID"): provider_with(handler).submit(video_request()) class TestSubmissionBody: """`SeedanceSubmission` 自身的行为，不经过 HTTP。""" def test_empty_content_is_rejected_locally(self): # 空内容必然被厂商拒，但那要花一次往返（还可能计入 RPM）。 with pytest.raises(ValueError, match="至少要有一个"): SeedanceSubmission(model="m").to_body() def test_last_frame_requested_by_default_for_a_free_cover_image(self): assert ( SeedanceSubmission(model="m", prompt="x").to_body()["return_last_frame"] is True ) def test_optional_fields_are_omitted_when_unset(self): body = SeedanceSubmission(model="m", prompt="x").to_body() assert "safety_identifier" not in body assert "callback_url" not in body class TestFetchStateMapping: def test_succeeded(self): result = provider_with( lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) ).fetch("cgt-1") assert result.state is TaskState.SUCCEEDED assert result.progress == 100 assert result.asset_url.endswith("/xxx") assert result.ratio == "16:9" assert result.resolution == "720p" assert result.billing_tokens == 108900 assert result.failure is None def test_asset_url_carries_its_expiry_so_callers_cannot_treat_it_as_permanent(self): """厂商 URL 只有 24 小时有效期，存成 external_url 会变死链。 把时效随 URL 一起给出，上层就没有理由把它当永久地址。 """ result = provider_with( lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) ).fetch("cgt-1") assert result.asset_url_expires_in_sec == 24 * 3600 def test_queued_maps_to_pending(self): def handler(request): return httpx.Response( 200, json={"id": "x", "status": "queued", "content": {}} ) assert provider_with(handler).fetch("x").state is TaskState.PENDING def test_running_has_zero_progress_because_the_vendor_gives_none(self): # 不是缺陷：客户端契约本就不承诺平滑进度（ADR-023 §10）。 def handler(request): return httpx.Response( 200, json={"id": "x", "status": "running", "content": {}} ) result = provider_with(handler).fetch("x") assert result.state is TaskState.RUNNING assert result.progress == 0 def test_cancelled_is_terminal_without_a_failure(self): def handler(request): return httpx.Response( 200, json={"id": "x", "status": "cancelled", "content": {}} ) result = provider_with(handler).fetch("x") assert result.state is TaskState.CANCELLED assert result.state.is_terminal assert result.failure is None def test_expired_maps_to_failed_and_stays_retryable(self): """`expired` 只见于创建接口的回调说明，查询接口文档没列。 只按查询文档实现会让超时任务掉进未知分支。它是超时而非内容问题，故可重试； 且厂商 expired 时不一定给 error.code，故单独分类而非走通用路径。 """ def handler(request): return httpx.Response( 200, json={"id": "x", "status": "expired", "content": {}} ) result = provider_with(handler).fetch("x") assert result.state is TaskState.FAILED assert result.failure.retry is Retry.BACKOFF assert result.failure.provider_code == "TaskExpired" def test_unknown_status_keeps_the_task_alive(self): # 厂商新增状态时把在途任务判死更糟，最坏是等到我们自己的超时。 def handler(request): return httpx.Response( 200, json={"id": "x", "status": "brand_new", "content": {}} ) assert provider_with(handler).fetch("x").state is TaskState.RUNNING def test_succeeded_without_a_url_is_treated_as_a_retryable_failure(self): """厂商说成功却没给 URL——那不是成功。 放一个没有产物的「成功」给上层，只会把空指针推到更远的地方。 """ def handler(request): return httpx.Response( 200, json={"id": "x", "status": "succeeded", "content": {}} ) result = provider_with(handler).fetch("x") assert result.state is TaskState.FAILED assert result.failure.retry is Retry.BACKOFF assert result.failure.provider_code == "MissingVideoUrl" class TestFetchFailureClassification: def test_content_rejection_is_not_retried(self): def handler(request): return httpx.Response( 200, json={ "id": "x", "status": "failed", "content": {}, "error": { "code": "OutputVideoSensitiveContentDetected", "message": "output may contain sensitive information", }, }, ) result = provider_with(handler).fetch("x") assert result.failure.retry is Retry.NEVER assert result.failure.user_facing is UserFacing.CONTENT_REJECTED # 原始码与 message 都要带到上层——排查时它们是唯一线索。 assert result.failure.provider_code == "OutputVideoSensitiveContentDetected" assert "sensitive" in result.failure.message class TestDurationPrecision: def test_frames_over_fps_is_exact(self): # 官方例子：133 帧 / 24fps = 5.54 秒。 def handler(request): return httpx.Response( 200, json={ "id": "x", "status": "succeeded", "frames": 133, "framespersecond": 24, "content": {"video_url": "https://v"}, }, ) assert provider_with(handler).fetch("x").duration_ms == 5542 def test_falls_back_to_the_vendors_floored_seconds(self): # duration 与 frames 只返回其一。只有 duration 时它是「帧数/24 向下取整」 # 的约数，最多可能短 1 秒——可用，但不可当精确值。 def handler(request): return httpx.Response( 200, json={ "id": "x", "status": "succeeded", "duration": 5, "content": {"video_url": "https://v"}, }, ) assert provider_with(handler).fetch("x").duration_ms == 5000 def test_none_rather_than_a_made_up_zero(self): def handler(request): return httpx.Response( 200, json={"id": "x", "status": "running", "content": {}} ) assert provider_with(handler).fetch("x").duration_ms is None class TestCoverImage: def test_last_frame_becomes_the_cover_url(self): """尾帧图可免费充当视频封面——iteration-4/03 R13 把缩略图列为视频成本项， 实际上厂商白给。""" def handler(request): return httpx.Response( 200, json={ "id": "x", "status": "succeeded", "content": { "video_url": "https://v", "last_frame_url": "https://f.png", }, }, ) assert provider_with(handler).fetch("x").cover_url == "https://f.png" class TestPixelDimensionsAreLeftBlankOnPurpose: def test_no_invented_dimensions(self): """火山只返回 ratio 与 resolution，不返回像素值。 换算需要一张随模型族变化的对照表；在 V7 目录种子定下模型族之前， 留空比编一个可能错的尺寸好。 """ result = provider_with( lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) ).fetch("x") assert result.width_px is None assert result.height_px is None # 但原始信息要透出，供上层或后续实现换算。 assert (result.ratio, result.resolution) == ("16:9", "720p")