Jony
ba920f2b66
Replace the prior release tree with the current plugin, frontend, tests, and documentation. Document retired node IDs and the public Gitea update source.
152 lines
4.6 KiB
Python
152 lines
4.6 KiB
Python
"""统一 Seedance HC 素材创建/查询接口测试。"""
|
|
|
|
import json
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
PACKAGE_DIR = Path(__file__).resolve().parents[1]
|
|
COMFY_ROOT = PACKAGE_DIR.parent.parent
|
|
sys.path.insert(0, str(COMFY_ROOT))
|
|
sys.path.insert(0, str(PACKAGE_DIR.parent))
|
|
|
|
package = types.ModuleType("comfyui_o1key")
|
|
package.__path__ = [str(PACKAGE_DIR)]
|
|
sys.modules["comfyui_o1key"] = package
|
|
for child in ("utils", "clients"):
|
|
module = types.ModuleType(f"comfyui_o1key.{child}")
|
|
module.__path__ = [str(PACKAGE_DIR / child)]
|
|
sys.modules[f"comfyui_o1key.{child}"] = module
|
|
|
|
from comfyui_o1key.clients.seedance_element_client import SeedanceElementClient
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload, status=200):
|
|
self.status = status
|
|
self._text = json.dumps(payload)
|
|
|
|
async def text(self):
|
|
return self._text
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
|
|
class _FakeSession:
|
|
def __init__(self, post_payload=None, get_payload=None):
|
|
self.post_payload = post_payload
|
|
self.get_payload = get_payload
|
|
self.post_call = None
|
|
self.get_call = None
|
|
|
|
def post(self, url, **kwargs):
|
|
self.post_call = (url, kwargs)
|
|
return _FakeResponse(self.post_payload)
|
|
|
|
def get(self, url, **kwargs):
|
|
self.get_call = (url, kwargs)
|
|
return _FakeResponse(self.get_payload)
|
|
|
|
|
|
class SeedanceHCAssetClientTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.client = SeedanceElementClient(
|
|
api_key="test-key",
|
|
base_url="https://cf-api.o1key.com",
|
|
)
|
|
|
|
async def test_create_uses_unified_seedance_asset_endpoint(self):
|
|
session = _FakeSession(post_payload={
|
|
"success": True,
|
|
"data": {"Id": "asset-test", "base_resp": {"status_code": 0}},
|
|
})
|
|
|
|
result = await self.client.create_hc_asset(
|
|
name="参考视频1",
|
|
asset_url="https://upload.o1key.com/uploads/reference.mp4",
|
|
asset_type="Video",
|
|
session=session,
|
|
)
|
|
|
|
url, kwargs = session.post_call
|
|
self.assertEqual(url, "https://cf-api.o1key.com/v1/seedance/assets")
|
|
self.assertEqual(kwargs["json"], {
|
|
"type": "hc",
|
|
"url": "https://upload.o1key.com/uploads/reference.mp4",
|
|
"name": "参考视频1",
|
|
"asset_type": "video",
|
|
})
|
|
self.assertEqual(result["Id"], "asset-test")
|
|
|
|
async def test_query_uses_hc_type_query_parameter(self):
|
|
session = _FakeSession(get_payload={
|
|
"success": True,
|
|
"data": {"Id": "asset/a b", "Status": "Active"},
|
|
})
|
|
|
|
result = await self.client.get_hc_asset("asset/a b", session=session)
|
|
|
|
url, kwargs = session.get_call
|
|
self.assertEqual(
|
|
url,
|
|
"https://cf-api.o1key.com/v1/seedance/assets/asset%2Fa%20b",
|
|
)
|
|
self.assertEqual(kwargs["params"], {"type": "hc"})
|
|
self.assertEqual(result["Status"], "Active")
|
|
|
|
async def test_create_omits_empty_optional_name(self):
|
|
session = _FakeSession(post_payload={
|
|
"success": True,
|
|
"data": {"Id": "asset-unnamed"},
|
|
})
|
|
|
|
await self.client.create_hc_asset(
|
|
name="",
|
|
asset_url="https://upload.o1key.com/uploads/reference.png",
|
|
asset_type="image",
|
|
session=session,
|
|
)
|
|
|
|
_, kwargs = session.post_call
|
|
self.assertNotIn("name", kwargs["json"])
|
|
|
|
async def test_doubao_type_is_used_for_create_and_query(self):
|
|
session = _FakeSession(
|
|
post_payload={
|
|
"success": True,
|
|
"data": {"Id": "asset-doubao"},
|
|
},
|
|
get_payload={
|
|
"success": True,
|
|
"data": {"Id": "asset-doubao", "Status": "Active"},
|
|
},
|
|
)
|
|
|
|
await self.client.create_hc_asset(
|
|
name="豆包真人素材",
|
|
asset_url="https://upload.o1key.com/uploads/reference.png",
|
|
asset_type="image",
|
|
session=session,
|
|
request_type="doubao",
|
|
)
|
|
_, create_kwargs = session.post_call
|
|
self.assertEqual(create_kwargs["json"]["type"], "doubao")
|
|
|
|
await self.client.get_hc_asset(
|
|
"asset-doubao",
|
|
session=session,
|
|
request_type="doubao",
|
|
)
|
|
_, query_kwargs = session.get_call
|
|
self.assertEqual(query_kwargs["params"], {"type": "doubao"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|