删除 benchmark/ 下 3438~3441 回归验证脚本
四处修复均已验证收敛,脚本不进 CI、裸跑依赖 manager-api,保留仅为仓库噪声;后续若再动 huoshan_double_stream 时以 pytest 覆盖
This commit is contained in:
1 parent
d8f8a4a083
commit
95ce49ffc9
5 files changed
-767
No files matched your search
@@ -1,75 +0,0 @@
|
||||
"""huoshan_double_stream 4 bug 复现脚本的共享 mock 工具。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
SERVER = os.path.join(REPO, "main", "xiaozhi-server")
|
||||
if SERVER not in sys.path:
|
||||
sys.path.insert(0, SERVER)
|
||||
|
||||
|
||||
class FakeConn:
|
||||
sample_rate = 24000
|
||||
stop_event = asyncio.Event()
|
||||
loop = None
|
||||
sentence_id = "test-session-id"
|
||||
client_abort = False
|
||||
|
||||
def __init__(self):
|
||||
self.stop_event = asyncio.Event()
|
||||
|
||||
|
||||
def build_provider() -> Any:
|
||||
"""绕开 TTSProvider.__init__ 复杂依赖,构造一个可直接调方法的实例。"""
|
||||
from core.providers.tts import huoshan_double_stream as mod
|
||||
|
||||
p = mod.TTSProvider.__new__(mod.TTSProvider)
|
||||
p.ws = MagicMock()
|
||||
p.ws.send = AsyncMock()
|
||||
p.ws.recv = AsyncMock()
|
||||
p.ws.close = AsyncMock()
|
||||
p.conn = FakeConn()
|
||||
p.voice = "test-voice"
|
||||
p.activate_session = False
|
||||
p._monitor_task = None
|
||||
p._pending_prefix = ""
|
||||
p.tts_text = ""
|
||||
p.tts_audio_queue = MagicMock()
|
||||
p.before_stop_play_files = []
|
||||
p.tts_timeout = 1.0
|
||||
p.enable_ws_reuse = False
|
||||
p.resource_type = False
|
||||
p.tts_text_buff = []
|
||||
p.processed_chars = 0
|
||||
p._correct_words_pattern = None
|
||||
p.audio_params = {"speech_rate": 0, "loudness_rate": 0}
|
||||
p.additions = {"aigc_metadata": {}, "cache_config": {}, "post_process": {"pitch": 0}}
|
||||
p.mix_speaker = {}
|
||||
p.opus_encoder = MagicMock()
|
||||
p.opus_encoder.reset_state = MagicMock()
|
||||
p.opus_encoder.encode_pcm_to_opus_stream = MagicMock(return_value=b"")
|
||||
p.handle_opus = MagicMock()
|
||||
p._process_before_stop_play_files = MagicMock()
|
||||
p.text_to_speak = AsyncMock()
|
||||
p._ensure_connection = AsyncMock()
|
||||
p.send_event = AsyncMock()
|
||||
p.send_text = AsyncMock()
|
||||
p.finish_connection = AsyncMock()
|
||||
p.close = AsyncMock()
|
||||
p.print_response = MagicMock()
|
||||
return p
|
||||
|
||||
|
||||
def run(coro):
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
@@ -1,117 +0,0 @@
|
||||
"""PR-issue 3438 会话启动失败可感知回归测试。
|
||||
|
||||
复现:
|
||||
服务端拒绝 StartSession(SessionFailed)时,旧逻辑无任何处理分支:
|
||||
整轮无声、activate_session 卡 True,下一轮靠 close 兜底且无日志
|
||||
|
||||
修复(按评审收敛后的形态):
|
||||
- start_session 发送即返回,不等确认帧(无 future 机制)
|
||||
- monitor matched 分支处理 SessionFailed:error 日志 + activate_session 复位
|
||||
|
||||
Usage:
|
||||
python benchmark/check_huoshan_3438.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import websockets
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def _load_mock():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
_mock = _load_mock()
|
||||
build_provider = _mock.build_provider
|
||||
run = _mock.run
|
||||
|
||||
from core.providers.tts.huoshan_double_stream import ( # noqa: E402
|
||||
EVENT_SessionFailed,
|
||||
EVENT_SessionStarted,
|
||||
)
|
||||
|
||||
FULL_SERVER_RESPONSE = 0b1001
|
||||
MsgTypeFlagWithEvent = 0b100
|
||||
|
||||
|
||||
def _build_session_event_frame(event: int, session_id: str, meta: str = "{}") -> bytes:
|
||||
res = bytearray()
|
||||
res.append((0b0001 << 4) | 1)
|
||||
res.append((FULL_SERVER_RESPONSE << 4) | MsgTypeFlagWithEvent)
|
||||
res.append((0b0001 << 4) | 0)
|
||||
res.append(0)
|
||||
res.extend(event.to_bytes(4, "big", signed=True))
|
||||
enc = session_id.encode("utf-8")
|
||||
res.extend(len(enc).to_bytes(4, "big", signed=True))
|
||||
res.extend(enc)
|
||||
meta_bytes = meta.encode("utf-8")
|
||||
res.extend(len(meta_bytes).to_bytes(4, "big", signed=True))
|
||||
res.extend(meta_bytes)
|
||||
return bytes(res)
|
||||
|
||||
|
||||
def test_start_session_fire_and_forget():
|
||||
"""start_session 发送 StartSession 后立即返回,不等确认帧。"""
|
||||
p = build_provider()
|
||||
p.send_event = AsyncMock()
|
||||
p._ensure_connection = AsyncMock()
|
||||
run(p.start_session("test-session-id"))
|
||||
assert p.send_event.await_count == 1, "应恰好发送一次 StartSession"
|
||||
print("OK: start_session 发送即返回,不阻塞等确认")
|
||||
|
||||
|
||||
def test_monitor_session_failed_resets_flag():
|
||||
"""matched SessionFailed 帧:monitor 记 error 并复位 activate_session。"""
|
||||
p = build_provider()
|
||||
p.activate_session = True
|
||||
p.ws.recv = AsyncMock(
|
||||
side_effect=[
|
||||
_build_session_event_frame(EVENT_SessionFailed, p.conn.sentence_id),
|
||||
websockets.ConnectionClosed(None, None),
|
||||
]
|
||||
)
|
||||
run(p._start_monitor_tts_response())
|
||||
assert p.activate_session is False, (
|
||||
f"SessionFailed 应复位 activate_session,实际 {p.activate_session}"
|
||||
)
|
||||
print("OK: SessionFailed 复位 activate_session,状态不卡 True")
|
||||
|
||||
|
||||
def test_monitor_session_started_keeps_flag():
|
||||
"""matched SessionStarted 帧:不触碰 activate_session(仍为 True)。"""
|
||||
p = build_provider()
|
||||
p.activate_session = True
|
||||
p.ws.recv = AsyncMock(
|
||||
side_effect=[
|
||||
_build_session_event_frame(EVENT_SessionStarted, p.conn.sentence_id),
|
||||
websockets.ConnectionClosed(None, None),
|
||||
]
|
||||
)
|
||||
run(p._start_monitor_tts_response())
|
||||
assert p.activate_session is True, (
|
||||
f"SessionStarted 不应触碰 activate_session,实际 {p.activate_session}"
|
||||
)
|
||||
print("OK: SessionStarted 不影响 activate_session")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
test_start_session_fire_and_forget()
|
||||
test_monitor_session_failed_resets_flag()
|
||||
test_monitor_session_started_keeps_flag()
|
||||
print("\nALL PASS: 会话启动失败可感知(monitor 侧处理,无 future 机制)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,224 +0,0 @@
|
||||
"""PR-issue 3439 finish_session 残留 + opus 尾帧冲刷回归测试。
|
||||
|
||||
复现:
|
||||
1. finish_session 不发送 _pending_prefix 残留:句尾替换词恰好被扣留时整段丢
|
||||
2. monitor 收 SessionFinished 不调 is_end=True 冲刷:opus 流式编码器不足一帧 PCM
|
||||
残量永远不下发,每句固定丢一小段
|
||||
3. to_tts 路径 SessionFinished 不冲刷:非流式生成的音频文件结尾缺一截
|
||||
4. FIRST 分支不重置 opus_encoder:跨轮残留影响下一轮首帧
|
||||
|
||||
修复:
|
||||
- finish_session 开头先 text_to_speak(_pending_prefix)
|
||||
- monitor SessionFinished 调 wav_to_opus_data_audio_raw_stream(b"", is_end=True, ...)
|
||||
- to_tts 路径 SessionFinished 同样 is_end=True 冲刷
|
||||
- FIRST 分支调 reset_stream_state + opus_encoder.reset_state()
|
||||
|
||||
Usage:
|
||||
python benchmark/check_huoshan_3439.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import websockets
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def _load_mock():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
_mock = _load_mock()
|
||||
build_provider = _mock.build_provider
|
||||
run = _mock.run
|
||||
|
||||
# protocol constants
|
||||
FULL_SERVER_RESPONSE = 0b1001
|
||||
MsgTypeFlagWithEvent = 0b100
|
||||
EVENT_SessionFinished = 152
|
||||
|
||||
|
||||
def _build_session_finished_frame(session_id: str, meta: str = '{"e":"ok"}') -> bytes:
|
||||
res = bytearray()
|
||||
res.append((0b0001 << 4) | 1)
|
||||
res.append((FULL_SERVER_RESPONSE << 4) | MsgTypeFlagWithEvent)
|
||||
res.append((0b0001 << 4) | 0)
|
||||
res.append(0)
|
||||
res.extend(EVENT_SessionFinished.to_bytes(4, "big", signed=True))
|
||||
enc = session_id.encode("utf-8")
|
||||
res.extend(len(enc).to_bytes(4, "big", signed=True))
|
||||
res.extend(enc)
|
||||
meta_bytes = meta.encode("utf-8")
|
||||
res.extend(len(meta_bytes).to_bytes(4, "big", signed=True))
|
||||
res.extend(meta_bytes)
|
||||
return bytes(res)
|
||||
|
||||
|
||||
def _async_iter_mock(items):
|
||||
"""AsyncMock whose side_effect 依次返回 items,耗尽后抛 ConnectionClosed。"""
|
||||
m = AsyncMock()
|
||||
it = iter(items)
|
||||
|
||||
async def _next(*a, **k):
|
||||
try:
|
||||
v = next(it)
|
||||
except StopIteration:
|
||||
raise websockets.ConnectionClosed(None, None)
|
||||
if isinstance(v, Exception):
|
||||
raise v
|
||||
return v
|
||||
|
||||
m.side_effect = _next
|
||||
return m
|
||||
|
||||
|
||||
def test_finish_session_sends_pending_prefix():
|
||||
p = build_provider()
|
||||
p._pending_prefix = "末尾残留"
|
||||
p.text_to_speak = AsyncMock()
|
||||
|
||||
run(p.finish_session("sid"))
|
||||
assert p.text_to_speak.await_count == 1, (
|
||||
f"finish_session 应先调 1 次 text_to_speak,实际 {p.text_to_speak.await_count}"
|
||||
)
|
||||
arg = p.text_to_speak.await_args.args[0]
|
||||
assert arg == "末尾残留", f"应传 _pending_prefix,实际 {arg!r}"
|
||||
print("OK: finish_session 开头先发 _pending_prefix 残留")
|
||||
|
||||
|
||||
def test_monitor_session_finished_flushes():
|
||||
p = build_provider()
|
||||
calls = []
|
||||
|
||||
def _spy_encode(pcm, end_of_stream, callback=None):
|
||||
calls.append((pcm, end_of_stream))
|
||||
return b""
|
||||
|
||||
p.opus_encoder.encode_pcm_to_opus_stream = _spy_encode
|
||||
p.ws.recv = _async_iter_mock(
|
||||
[_build_session_finished_frame(p.conn.sentence_id)]
|
||||
)
|
||||
|
||||
run(p._start_monitor_tts_response())
|
||||
flush_calls = [c for c in calls if c[1] is True]
|
||||
assert len(flush_calls) >= 1, (
|
||||
f"monitor 收 SessionFinished 应至少 1 次 is_end=True 冲刷,实际 {calls}"
|
||||
)
|
||||
assert flush_calls[0][0] == b"", f"冲刷调用应传空 pcm,实际 {flush_calls[0][0]!r}"
|
||||
print("OK: monitor SessionFinished 调 is_end=True 冲刷流式编码器")
|
||||
|
||||
|
||||
def test_to_tts_session_finished_flushes():
|
||||
"""to_tts 路径 SessionFinished 同样 is_end=True 冲刷,复用真实 SessionFinished
|
||||
分支逻辑(与 monitor 一致;非流式通过 wav_to_opus_data_audio_raw_stream is_end
|
||||
把编码器残量追加到 audio_data)。"""
|
||||
p = build_provider()
|
||||
calls = []
|
||||
audio_data = []
|
||||
|
||||
def _spy_encode(pcm, end_of_stream, callback=None):
|
||||
calls.append((pcm, end_of_stream))
|
||||
if callback:
|
||||
callback(b"frame")
|
||||
return b""
|
||||
|
||||
p.opus_encoder.encode_pcm_to_opus_stream = _spy_encode
|
||||
|
||||
async def _drive():
|
||||
ws = p.ws
|
||||
ws.recv = _async_iter_mock(
|
||||
[_build_session_finished_frame(p.conn.sentence_id)]
|
||||
)
|
||||
while True:
|
||||
try:
|
||||
msg = await ws.recv()
|
||||
except websockets.ConnectionClosed:
|
||||
break
|
||||
res = p.parser_response(msg)
|
||||
if res.optional.event == 152: # SessionFinished
|
||||
if hasattr(p, "opus_encoder") and p.opus_encoder:
|
||||
p.wav_to_opus_data_audio_raw_stream(
|
||||
b"",
|
||||
is_end=True,
|
||||
callback=lambda opus_frame: audio_data.append(opus_frame),
|
||||
)
|
||||
break
|
||||
|
||||
run(_drive())
|
||||
flush_calls = [c for c in calls if c[1] is True]
|
||||
assert len(flush_calls) == 1, (
|
||||
f"to_tts 路径 SessionFinished 应 1 次 is_end=True 冲刷,实际 {calls}"
|
||||
)
|
||||
assert audio_data == [b"frame"], f"to_tts 收集到的帧应为 [b'frame'],实际 {audio_data}"
|
||||
print("OK: to_tts 路径 SessionFinished 同样 is_end=True 冲刷")
|
||||
|
||||
|
||||
def test_first_resets_opus_encoder():
|
||||
"""FIRST 分支执行后 opus_encoder.reset_state() 被实际调用一次。"""
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from core.providers.tts.dto.dto import ContentType, SentenceType
|
||||
|
||||
p = build_provider()
|
||||
p.opus_encoder.reset_state.reset_mock()
|
||||
p.reset_stream_state = MagicMock() # mock 父类避免依赖
|
||||
p.tts_text_queue = queue.Queue()
|
||||
# start_session 走 mock,run_coroutine_threadsafe 替换为立刻完成
|
||||
|
||||
def _fake_rcts(coro, loop):
|
||||
if asyncio.iscoroutine(coro):
|
||||
coro.close()
|
||||
f = asyncio.Future()
|
||||
f.set_result(None)
|
||||
return f
|
||||
|
||||
msg = SimpleNamespace(
|
||||
sentence_type=SentenceType.FIRST,
|
||||
content_type=ContentType.TEXT,
|
||||
content_detail="x",
|
||||
content_file=None,
|
||||
sentence_id=p.conn.sentence_id,
|
||||
)
|
||||
p.tts_text_queue.put(msg)
|
||||
|
||||
with patch("asyncio.run_coroutine_threadsafe", side_effect=_fake_rcts):
|
||||
t = threading.Thread(target=p.tts_text_priority_thread, daemon=True)
|
||||
t.start()
|
||||
time.sleep(0.3)
|
||||
p.conn.stop_event.set()
|
||||
t.join(timeout=2)
|
||||
|
||||
assert p.opus_encoder.reset_state.call_count == 1, (
|
||||
f"FIRST 分支应调 opus_encoder.reset_state() 1 次,实际 {p.opus_encoder.reset_state.call_count}"
|
||||
)
|
||||
assert p.reset_stream_state.call_count == 1, (
|
||||
f"FIRST 分支应先调 reset_stream_state() 1 次,实际 {p.reset_stream_state.call_count}"
|
||||
)
|
||||
print("OK: FIRST 分支调 reset_stream_state + opus_encoder.reset_state() 行为断言通过")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
test_finish_session_sends_pending_prefix()
|
||||
test_monitor_session_finished_flushes()
|
||||
test_to_tts_session_finished_flushes()
|
||||
test_first_resets_opus_encoder()
|
||||
print("\nALL PASS: 3439 finish_session 残留 + opus 尾帧冲刷(修复点 2)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,163 +0,0 @@
|
||||
"""PR-issue 3440 旧会话事件不影响当前会话 + 取消屏障回归测试。
|
||||
|
||||
复现:
|
||||
旧会话结束事件(SessionCanceled/Failed/Finished)即使 sessionId 不匹配也会
|
||||
reset 当前 activate_session,破坏 start_session 的"上个会话激活时关闭
|
||||
新建"判断
|
||||
|
||||
修复(按评审收敛后的形态):
|
||||
- 旧 session 事件 continue 跳过,不再 reset activate_session
|
||||
- 旧连接误关/偷读由建连前的取消屏障(_cancel_monitor_task 等待任务退出)
|
||||
保证,不依赖 snapshot 判 id
|
||||
|
||||
Usage:
|
||||
python benchmark/check_huoshan_3440.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import websockets
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def _load_mock():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
_mock = _load_mock()
|
||||
build_provider = _mock.build_provider
|
||||
run = _mock.run
|
||||
|
||||
FULL_SERVER_RESPONSE = 0b1001
|
||||
MsgTypeFlagWithEvent = 0b100
|
||||
EVENT_SessionFinished = 152
|
||||
EVENT_SessionCanceled = 151
|
||||
|
||||
|
||||
def _build_session_event_frame(event: int, session_id: str, meta: str = "{}") -> bytes:
|
||||
res = bytearray()
|
||||
res.append((0b0001 << 4) | 1)
|
||||
res.append((FULL_SERVER_RESPONSE << 4) | MsgTypeFlagWithEvent)
|
||||
res.append((0b0001 << 4) | 0)
|
||||
res.append(0)
|
||||
res.extend(event.to_bytes(4, "big", signed=True))
|
||||
enc = session_id.encode("utf-8")
|
||||
res.extend(len(enc).to_bytes(4, "big", signed=True))
|
||||
res.extend(enc)
|
||||
meta_bytes = meta.encode("utf-8")
|
||||
res.extend(len(meta_bytes).to_bytes(4, "big", signed=True))
|
||||
res.extend(meta_bytes)
|
||||
return bytes(res)
|
||||
|
||||
|
||||
def _async_iter_mock(items):
|
||||
m = AsyncMock()
|
||||
it = iter(items)
|
||||
|
||||
async def _next(*a, **k):
|
||||
try:
|
||||
v = next(it)
|
||||
except StopIteration:
|
||||
raise websockets.ConnectionClosed(None, None)
|
||||
if isinstance(v, Exception):
|
||||
raise v
|
||||
return v
|
||||
|
||||
m.side_effect = _next
|
||||
return m
|
||||
|
||||
|
||||
def test_old_session_finished_does_not_reset_activate():
|
||||
"""旧 session 的 SessionFinished 不应 reset 当前 activate_session。"""
|
||||
p = build_provider()
|
||||
p.activate_session = True
|
||||
p.ws.recv = _async_iter_mock(
|
||||
[_build_session_event_frame(EVENT_SessionFinished, "OLD-sid-not-mine")]
|
||||
)
|
||||
run(p._start_monitor_tts_response())
|
||||
assert p.activate_session is True, (
|
||||
f"旧 session 结束事件不应 reset 当前 activate_session,实际 {p.activate_session}"
|
||||
)
|
||||
print("OK: 旧 session SessionFinished 不 reset 当前 activate_session")
|
||||
|
||||
|
||||
def test_old_session_canceled_does_not_reset_activate():
|
||||
p = build_provider()
|
||||
p.activate_session = True
|
||||
p.ws.recv = _async_iter_mock(
|
||||
[_build_session_event_frame(EVENT_SessionCanceled, "OLD-sid-not-mine")]
|
||||
)
|
||||
run(p._start_monitor_tts_response())
|
||||
assert p.activate_session is True
|
||||
print("OK: 旧 session SessionCanceled 不 reset 当前 activate_session")
|
||||
|
||||
|
||||
def test_cancel_barrier_awaits_monitor_exit():
|
||||
"""取消屏障原语:_cancel_monitor_task 返回时旧监听任务必须已完全退出。
|
||||
|
||||
这是"旧监听任务不会读新连接/误关新连接"的根本保证:建新连接前
|
||||
旧任务已被 cancel + await 收尸,其后才允许重赋 self.ws。
|
||||
"""
|
||||
p = build_provider()
|
||||
|
||||
async def _block_recv(*a, **k):
|
||||
await asyncio.Event().wait() # 模拟阻塞在 recv
|
||||
|
||||
p.ws = MagicMock()
|
||||
p.ws.recv = AsyncMock(side_effect=_block_recv)
|
||||
p.ws.close = AsyncMock()
|
||||
|
||||
async def _drive():
|
||||
m = asyncio.create_task(p._start_monitor_tts_response())
|
||||
p._monitor_task = m
|
||||
await asyncio.sleep(0.01)
|
||||
assert not m.done(), "监听任务应仍在运行"
|
||||
await p._cancel_monitor_task()
|
||||
assert m.done() and m.cancelled(), (
|
||||
"_cancel_monitor_task 返回时旧监听任务必须已退出"
|
||||
)
|
||||
assert p._monitor_task is None, "屏障应清空 _monitor_task 引用"
|
||||
|
||||
run(_drive())
|
||||
print("OK: 取消屏障等待旧监听任务完全退出后才返回")
|
||||
|
||||
|
||||
def test_source_has_cancel_barrier_and_skip():
|
||||
"""静态检查:mismatch 纯跳过存在;monitor finally 不再复位 activate_session。"""
|
||||
src = open(
|
||||
os.path.join(REPO, "main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py"),
|
||||
encoding="utf-8",
|
||||
).read()
|
||||
assert "self.conn.sentence_id != res.optional.sessionId" in src, (
|
||||
"mismatch 分支应纯跳过(continue),不复位 activate_session"
|
||||
)
|
||||
monitor_src = src.split("async def _start_monitor_tts_response")[1]
|
||||
finally_src = monitor_src.split("finally:")[1]
|
||||
assert "activate_session" not in finally_src, (
|
||||
"monitor finally 不应触碰 activate_session(避免覆盖新会话状态)"
|
||||
)
|
||||
print("OK: mismatch 纯跳过存在,finally 不触碰 activate_session")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
test_old_session_finished_does_not_reset_activate()
|
||||
test_old_session_canceled_does_not_reset_activate()
|
||||
test_cancel_barrier_awaits_monitor_exit()
|
||||
test_source_has_cancel_barrier_and_skip()
|
||||
print("\nALL PASS: 3440 旧会话事件不影响当前 + 取消屏障保证")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,188 +0,0 @@
|
||||
"""PR-issue 3441 parser_response 四处二进制协议解析错误回归测试。
|
||||
|
||||
复现:
|
||||
1. serialization_method 恒为 0:旧 num=0b00001111 被错用为移位数,>> 15 恒 0
|
||||
2. message_type 判断恒真:旧 `== FULL_SERVER_RESPONSE or AUDIO_ONLY_RESPONSE` 优先级下
|
||||
AUDIO_ONLY_RESPONSE=0b1011 作为真值使条件恒真
|
||||
3. offset 硬编码 4:旧 `res[offset:8]` 与 `header_size * 4` 在 header_size=1 时重合但
|
||||
header_size=2 时全错位
|
||||
4. SessionCanceled 走 else 分支被当作 sessionId+payload 解析:协议层实际为
|
||||
sessionId+meta_json
|
||||
|
||||
Usage:
|
||||
python benchmark/check_huoshan_parser_3441.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
SERVER = os.path.join(REPO, "main", "xiaozhi-server")
|
||||
if SERVER not in sys.path:
|
||||
sys.path.insert(0, SERVER)
|
||||
|
||||
from core.providers.tts.huoshan_double_stream import ( # noqa: E402
|
||||
TTSProvider,
|
||||
FULL_SERVER_RESPONSE,
|
||||
AUDIO_ONLY_RESPONSE,
|
||||
ERROR_INFORMATION,
|
||||
EVENT_SessionStarted,
|
||||
EVENT_SessionFailed,
|
||||
EVENT_SessionFinished,
|
||||
EVENT_SessionCanceled,
|
||||
EVENT_ConnectionStarted,
|
||||
MsgTypeFlagWithEvent,
|
||||
)
|
||||
|
||||
|
||||
def make_header_byte0(protocol_version=0b0001, header_size=1) -> int:
|
||||
return (protocol_version << 4) | header_size
|
||||
|
||||
|
||||
def make_header_byte1(message_type, flags) -> int:
|
||||
return (message_type << 4) | flags
|
||||
|
||||
|
||||
def make_header_byte2(serialization_method, compression_type) -> int:
|
||||
return (serialization_method << 4) | compression_type
|
||||
|
||||
|
||||
def make_str_field(s: str) -> bytes:
|
||||
enc = s.encode("utf-8")
|
||||
return len(enc).to_bytes(4, "big", signed=True) + enc
|
||||
|
||||
|
||||
def build_frame(
|
||||
message_type: int,
|
||||
event: int | None,
|
||||
session_id: str | None,
|
||||
meta_json: str | None = None,
|
||||
payload: bytes | None = None,
|
||||
header_size: int = 1,
|
||||
serialization_method: int = 0b0001,
|
||||
flags: int = MsgTypeFlagWithEvent,
|
||||
) -> bytes:
|
||||
res = bytearray()
|
||||
res.append(make_header_byte0(header_size=header_size))
|
||||
res.append(make_header_byte1(message_type, flags))
|
||||
res.append(make_header_byte2(serialization_method, 0b0000))
|
||||
res.append(0)
|
||||
if header_size > 1:
|
||||
res.extend(b"\x00" * (header_size - 1) * 4)
|
||||
if event is not None:
|
||||
res.extend(event.to_bytes(4, "big", signed=True))
|
||||
if session_id is not None:
|
||||
res.extend(make_str_field(session_id))
|
||||
if meta_json is not None:
|
||||
res.extend(make_str_field(meta_json))
|
||||
if payload is not None:
|
||||
res.extend(len(payload).to_bytes(4, "big", signed=True))
|
||||
res.extend(payload)
|
||||
return bytes(res)
|
||||
|
||||
|
||||
def fresh_provider() -> TTSProvider:
|
||||
import importlib.util
|
||||
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"_huoshan_mock",
|
||||
os.path.join(os.path.dirname(__file__), "_huoshan_mock.py"),
|
||||
)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod.build_provider()
|
||||
|
||||
|
||||
def test_serialization_method_extracted_correctly():
|
||||
p = fresh_provider()
|
||||
frame = build_frame(
|
||||
message_type=FULL_SERVER_RESPONSE,
|
||||
event=EVENT_SessionStarted,
|
||||
session_id="sid",
|
||||
meta_json='{"k":"v"}',
|
||||
serialization_method=0b0010,
|
||||
)
|
||||
res = p.parser_response(frame)
|
||||
assert res.header.serialization_method == 0b0010, (
|
||||
f"serialization_method 应为 0b0010=2,实际 {res.header.serialization_method:#06b}"
|
||||
)
|
||||
assert res.optional.event == EVENT_SessionStarted
|
||||
assert res.optional.sessionId == "sid"
|
||||
assert res.optional.response_meta_json == '{"k":"v"}'
|
||||
print("OK: serialization_method 正确从 res[2]>>4 提取(修复点 1)")
|
||||
|
||||
|
||||
def test_message_type_branch_includes_audio_only():
|
||||
"""修复前 `== FULL or AUDIO_ONLY_RESPONSE` 优先级下恒真。
|
||||
现版必须用 in (...) 限定,否则别的不支持的 message_type 也进同一分支。"""
|
||||
p = fresh_provider()
|
||||
# message_type = 0b0001 (FULL_CLIENT_REQUEST) 不应进 FULL/AUDIO 分支,
|
||||
# 应进 ERROR_INFORMATION 之外的默认(不解析 event)
|
||||
frame = build_frame(
|
||||
message_type=0b0001,
|
||||
event=EVENT_SessionStarted,
|
||||
session_id="sid",
|
||||
)
|
||||
res = p.parser_response(frame)
|
||||
assert res.optional.event == 0, (
|
||||
f"message_type=FULL_CLIENT_REQUEST 不应解析 event,实际 event={res.optional.event}"
|
||||
)
|
||||
# AUDIO_ONLY_RESPONSE 仍能进 FULL/AUDIO 分支解析
|
||||
frame2 = build_frame(
|
||||
message_type=AUDIO_ONLY_RESPONSE,
|
||||
event=EVENT_SessionFinished,
|
||||
session_id="sid2",
|
||||
meta_json='{"e":"ok"}',
|
||||
)
|
||||
res2 = p.parser_response(frame2)
|
||||
assert res2.optional.event == EVENT_SessionFinished
|
||||
assert res2.optional.sessionId == "sid2"
|
||||
print("OK: message_type in (FULL_SERVER_RESPONSE, AUDIO_ONLY_RESPONSE) 正确分支(修复点 2)")
|
||||
|
||||
|
||||
def test_offset_respects_header_size():
|
||||
"""header_size=2 时 optional 起始偏移应为 8,不是 4。"""
|
||||
p = fresh_provider()
|
||||
frame = build_frame(
|
||||
message_type=FULL_SERVER_RESPONSE,
|
||||
event=EVENT_SessionStarted,
|
||||
session_id="s",
|
||||
meta_json="m",
|
||||
header_size=2,
|
||||
)
|
||||
res = p.parser_response(frame)
|
||||
assert res.optional.event == EVENT_SessionStarted
|
||||
assert res.optional.sessionId == "s"
|
||||
assert res.optional.response_meta_json == "m"
|
||||
print("OK: offset = header.header_size * 4 支持扩展头(修复点 3)")
|
||||
|
||||
|
||||
def test_session_canceled_parses_sessionid_and_meta():
|
||||
"""修复前 SessionCanceled 落 else 分支被误读为 sessionId+payload。"""
|
||||
p = fresh_provider()
|
||||
meta = '{"reason":"user cancel"}'
|
||||
frame = build_frame(
|
||||
message_type=FULL_SERVER_RESPONSE,
|
||||
event=EVENT_SessionCanceled,
|
||||
session_id="cancel-sid",
|
||||
meta_json=meta,
|
||||
)
|
||||
res = p.parser_response(frame)
|
||||
assert res.optional.event == EVENT_SessionCanceled
|
||||
assert res.optional.sessionId == "cancel-sid"
|
||||
assert res.optional.response_meta_json == meta
|
||||
print("OK: EVENT_SessionCanceled 按协议解析 sessionId+meta(修复点 4)")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
test_serialization_method_extracted_correctly()
|
||||
test_message_type_branch_includes_audio_only()
|
||||
test_offset_respects_header_size()
|
||||
test_session_canceled_parses_sessionid_and_meta()
|
||||
print("\nALL PASS: parser_response 4 处协议解析错误全修")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user