07979bfb09
Replace CreateRecTask polling with recording-file flash API, add TENCENT_APP_ID, remove server-side pydub slicing, and log ASR recognition text at INFO in development. Co-authored-by: Cursor <cursoragent@cursor.com>
107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.adapters.asr.tencent_asr import TencentASRProvider
|
|
from app.core import chapter_pipeline_lock
|
|
from app.features.story import post_commit
|
|
|
|
|
|
def test_chapter_pipeline_lock_delegates_to_token_lock(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
handle = object()
|
|
recorded: dict[str, object] = {}
|
|
|
|
def fake_acquire(key: str, *, ttl_seconds: int) -> object:
|
|
recorded["key"] = key
|
|
recorded["ttl_seconds"] = ttl_seconds
|
|
return handle
|
|
|
|
def fake_release(arg: object) -> None:
|
|
recorded["released"] = arg
|
|
|
|
monkeypatch.setattr(chapter_pipeline_lock, "acquire_redis_lock", fake_acquire)
|
|
monkeypatch.setattr(chapter_pipeline_lock, "release_redis_lock", fake_release)
|
|
|
|
acquired = chapter_pipeline_lock.acquire_chapter_pipeline_lock(
|
|
"user-1", "childhood", ttl_seconds=120
|
|
)
|
|
chapter_pipeline_lock.release_chapter_pipeline_lock(acquired)
|
|
|
|
assert acquired is handle
|
|
assert recorded == {
|
|
"key": "lock:chapter:user-1:childhood",
|
|
"ttl_seconds": 120,
|
|
"released": handle,
|
|
}
|
|
|
|
|
|
def test_post_commit_reuses_singleton_redis_client(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client = object()
|
|
|
|
def fake_get_sync_redis(*, decode_responses: bool):
|
|
assert decode_responses is True
|
|
return client
|
|
|
|
monkeypatch.setattr(post_commit, "get_sync_redis", fake_get_sync_redis)
|
|
|
|
first = post_commit._get_redis()
|
|
second = post_commit._get_redis()
|
|
|
|
assert first is second
|
|
assert first is client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tencent_asr_flash_transcribe(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeAsyncClient:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *args):
|
|
return None
|
|
|
|
async def post(self, url, *, headers=None, content=None, timeout=None):
|
|
captured["url"] = url
|
|
captured["headers"] = headers
|
|
captured["content"] = content
|
|
captured["timeout"] = timeout
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"code": 0,
|
|
"request_id": "req-1",
|
|
"flash_result": [{"channel_id": 0, "text": " 你好，世界 "}],
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
|
|
|
|
provider = TencentASRProvider(
|
|
"sid",
|
|
"skey",
|
|
"1259220000",
|
|
engine_type="16k_zh_large",
|
|
)
|
|
text = await provider.transcribe(b"fake-audio", format="m4a")
|
|
|
|
assert text == "你好，世界"
|
|
assert captured["content"] == b"fake-audio"
|
|
assert captured["timeout"] == 60.0
|
|
url = str(captured["url"])
|
|
assert "engine_type=16k_zh_large" in url
|
|
assert "voice_format=m4a" in url
|
|
assert "/asr/flash/v1/1259220000?" in url
|
|
assert "secretid=sid" in url
|
|
headers = captured["headers"]
|
|
assert headers is not None
|
|
assert headers["Authorization"]
|
|
assert headers["Content-Type"] == "application/octet-stream"
|
|
assert headers["Content-Length"] == str(len(b"fake-audio"))
|