Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 47 additions & 1 deletion tests/unit/asr_client/endpointing/test_audio_evidence.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import hashlib
import json
import time
import wave
from pathlib import Path

Expand All @@ -12,6 +13,51 @@
)


def _complete_record_count(index_path: Path) -> int:
# 只认已经 flush 完整的整行 JSON:写线程是先 open 再 write 再 flush,
# 光看文件存在会读到空文件或半行。
try:
lines = index_path.read_text("utf-8").splitlines()
except OSError:
return 0
complete = 0
for line in lines:
try:
json.loads(line)
except ValueError:
return complete
complete += 1
return complete


def _wait_for_written_run_dir(
target: Path,
expected_records: int = 1,
timeout_s: float = 10.0,
) -> list[Path]:
# 等落盘:close() 只给写线程一个很短的确认窗口,不是落盘保证。正常路径下写线程
# 几毫秒就写完了,但 Windows CI 上磁盘会抖到超过那个窗口,close() 一返回就
# iterdir 会撞上还没建出来的目录、或是刚 open 出来的空 index。
deadline = time.monotonic() + timeout_s
while True:
try:
run_dirs = list(target.iterdir())
except FileNotFoundError:
run_dirs = []
if run_dirs and all(
_complete_record_count(d / "index.jsonl") >= expected_records
for d in run_dirs
):
return run_dirs
if time.monotonic() >= deadline:
raise AssertionError(
f"等待 {target} 落盘超时({timeout_s}s):"
f"run_dirs={[d.name for d in run_dirs]},"
f"records={[_complete_record_count(d / 'index.jsonl') for d in run_dirs]}"
)
time.sleep(0.01)


async def test_audio_evidence_is_off_without_explicit_opt_in(tmp_path: Path) -> None:
target = tmp_path / "data" / "smart_turn" / "audio-evidence"
recorder = create_smart_turn_audio_evidence_recorder(
Expand Down Expand Up @@ -54,7 +100,7 @@ async def test_audio_evidence_writes_local_wav_and_index_under_data(
)
await recorder.close()

run_dirs = list(target.iterdir())
run_dirs = _wait_for_written_run_dir(target)
assert recorder.enabled is True
assert len(run_dirs) == 1
run_dir = run_dirs[0]
Expand Down
8 changes: 7 additions & 1 deletion tests/unit/test_window_pin_static_contracts.py
Original file line number Diff line number Diff line change
Expand Up @@ -252,8 +252,14 @@ def test_credentials_tabs_are_wired_to_the_single_tab_panel():
template = read_text("templates/cookies_login.html")

tab_buttons = re.findall(r'<button class="tab-btn[^>]*>', template)
assert len(tab_buttons) == 10
# 凭证源会持续增删,锁死具体个数只会让每个新增源都红一次;真正要守的是每个按钮的
# ARIA 接线。和 switchTab 调用数交叉比对,正则一旦失配就会立刻不等,避免测出 0 个假绿。
switch_calls = template.count("switchTab('")
assert switch_calls > 0
assert len(tab_buttons) == switch_calls
Comment thread
coderabbitai[bot] marked this conversation as resolved.
for button in tab_buttons:
# 逐个按钮各带一次调用,才能保证上面那个总数相等不是靠"这里漏一次、别处多一次"凑出来的。
assert button.count("switchTab('") == 1, button
assert 'role="tab"' in button, button
assert 'aria-controls="main-panel"' in button, button
assert re.search(
Expand Down
Loading