eigyouonsei / tests /test_app.py
ryota
GPUが混んでいても止まらないようにし、出力をExcelだけにする
7fb75a2
Raw
History Blame Contribute Delete
7.79 kB
"""画面まわり。文字起こし本体は動かさず、組み立てと片付けだけを確かめる。"""
import sys
import time
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import app as web # noqa: E402
import pipeline as pl # noqa: E402
SEGMENTS = [
{"start": 0.0, "end": 8.0, "text": "本日はありがとうございます"},
{"start": 9.0, "end": 14.0, "text": "よろしくお願いします"},
]
TURNS = [(0.0, 8.5, "SPEAKER_00"), (8.5, 14.5, "SPEAKER_01")]
def build(playback: Path | None = None) -> pl.FileResult:
entry = pl.FileResult(
original_name="A社_初回訪問.wav", duration=95.0,
recorded_at="2026-08-22 14:00:00", status="transcribed", playback=playback,
)
entry.utterances = pl.merge(SEGMENTS, TURNS)
entry.speakers = sorted({u.speaker for u in entry.utterances})
return entry
def test_発話の時刻に再生ボタンが付く(tmp_path):
audio = tmp_path / "000.mp3"
audio.write_bytes(b"x")
html = web.render([build(audio)])
assert '<audio id="audio-1"' in html
assert 'data-audio="audio-1"' in html
assert 'data-t="9.00"' in html # 2つ目の発話の開始位置
assert "00:00:09" in html
def test_音声が無い録音には再生機を出さない():
html = web.render([build()])
assert "<audio" not in html
assert "SPEAKER_00" in html # 本文は出る
def test_本文のHTMLを素通しさせない():
"""録音の中身が画面を壊さないように。"""
entry = build()
entry.utterances[0].text = '<script>alert("x")</script>'
html = web.render([entry])
assert "<script>alert" not in html
assert "&lt;script&gt;" in html
def test_除外した録音は理由が出る():
entry = pl.FileResult(original_name="短い.wav", duration=12.0,
status="skipped", reason="60秒未満")
html = web.render([entry])
assert "除外" in html and "60秒未満" in html
def test_集計が出る():
skipped = pl.FileResult(original_name="短い.wav", duration=12.0,
status="skipped", reason="60秒未満")
text = web.summarize([build(), skipped])
assert "書き起こし 1" in text
assert "長さ不足で除外 1" in text
def test_古い作業場は片付けられる(tmp_path, monkeypatch):
"""商談の音声を必要以上に残さないため、時間で消す。"""
monkeypatch.setattr(web, "WORK_ROOT", tmp_path)
old = tmp_path / "job-old"
old.mkdir()
stale = time.time() - web.JOB_TTL_SECONDS - 60
import os
os.utime(old, (stale, stale))
fresh = tmp_path / "job-fresh"
fresh.mkdir()
web.sweep_old_jobs()
assert not old.exists()
assert fresh.exists()
def test_ファイル未選択なら止める():
with pytest.raises(Exception) as err:
web.process([], 60, "large-v3-turbo", 2, False, True, "", "")
assert "選んで" in str(err.value)
def test_トークン無しで話者分離はできない(monkeypatch, tmp_path):
monkeypatch.delenv("HF_TOKEN", raising=False)
with pytest.raises(Exception) as err:
web.process([str(tmp_path / "a.wav")], 60, "large-v3-turbo", 2, True, True, "", "")
assert "トークン" in str(err.value)
def test_GPUの飾りは環境で切り替わる():
"""自分のPCでは素通し。ZeroGPU のときだけ GPU を割り当てる。"""
def plain():
return "ok"
if not web.ON_ZERO_GPU:
assert web.on_gpu(60)(plain) is plain
def test_待ち受け先は環境で変わる(monkeypatch):
"""Spaces で 127.0.0.1 のままだと外から届かず起動に失敗する。"""
monkeypatch.delenv("HOST", raising=False)
monkeypatch.setattr(web, "ON_SPACES", False)
assert web.listen_on() == "127.0.0.1" # 自分のPCでは外に開かない
monkeypatch.setattr(web, "ON_SPACES", True)
assert web.listen_on() == "0.0.0.0" # Spaces では外から届くように
monkeypatch.setenv("HOST", "1.2.3.4")
assert web.listen_on() == "1.2.3.4" # 指定があればそれに従う
def test_GPU側の失敗理由が消えない():
"""ZeroGPU は例外をうまく送り返せず、中身の無い名前だけになる。
成否と本文を組にして持ち帰り、呼び出し側で組み立て直すこと。"""
def boom():
raise RuntimeError("libcudnn が見つかりません")
ok, message = web.carry_errors(boom)()
assert ok is False
assert "libcudnn" in message
with pytest.raises(RuntimeError) as err:
web.unwrap((False, "RuntimeError: libcudnn が見つかりません"))
assert "libcudnn" in str(err.value)
assert web.unwrap((True, [1, 2, 3])) == [1, 2, 3]
def test_時刻が押せることを説明する(tmp_path):
"""押せると分からなければ、機能が無いのと同じなので。"""
audio = tmp_path / "000.mp3"
audio.write_bytes(b"x")
html = web.render([build(audio)])
assert "時刻" in html and "録音が流れます" in html
# 音声が無いときは案内も出さない
assert "録音が流れます" not in web.render([build()])
def test_出力はExcelだけ(tmp_path):
"""使わない形式まで作ると、そのぶん待たせることになる。"""
outputs = pl.build_outputs([build()], tmp_path / "出力", "test", xlsx_only=True)
assert outputs["xlsx"].exists()
assert set(outputs) == {"xlsx", "per_file"}
assert list(tmp_path.glob("出力/*")) == [outputs["xlsx"]]
def test_混雑と本当の失敗を見分ける():
assert web.is_busy("No GPU was available after 60s Retry later")
assert web.is_busy("GPU quota exceeded")
assert not web.is_busy("RuntimeError: libcudnn.so.9 が見つかりません")
def test_GPUが混んでいたらCPUで続ける(monkeypatch, tmp_path):
"""1本の商談を丸ごと落とすより、遅くても終わらせる。"""
monkeypatch.setattr(web, "GPU_RETRIES", 1)
monkeypatch.setattr(web.time, "sleep", lambda _: None)
monkeypatch.setattr(web.pl, "probe_duration", lambda _: 120.0)
tries = {"gpu": 0}
def busy(wav, *args):
tries["gpu"] += 1
return False, "No GPU was available after 60s Retry later"
table = {seconds: busy for seconds in web.GPU_TIERS}
notes = []
result = web.run_on_gpu(table, tmp_path / "a.wav", (), 0.15,
lambda: ["CPUの結果"], notes.append)
assert result == ["CPUの結果"]
assert tries["gpu"] == 2 # 1回目 + 再挑戦1回
assert notes and "CPU" in notes[0] # 画面にも断りを出す
def test_混雑でない失敗はそのまま伝える(monkeypatch, tmp_path):
monkeypatch.setattr(web.pl, "probe_duration", lambda _: 60.0)
def broken(wav, *args):
return False, "RuntimeError: libcudnn.so.9 が見つかりません"
table = {seconds: broken for seconds in web.GPU_TIERS}
with pytest.raises(RuntimeError) as err:
web.run_on_gpu(table, tmp_path / "a.wav", (), 0.15,
lambda: ["使われないはず"], lambda _: None)
assert "libcudnn" in str(err.value)
def test_確保時間は録音の長さで決まる(monkeypatch, tmp_path):
"""長く要求するほど空きが見つかりにくいので、足りる範囲で短く頼む。"""
monkeypatch.setattr(web.pl, "probe_duration", lambda _: 60.0)
short = web.gpu_tier(tmp_path / "a.wav", 0.15)
monkeypatch.setattr(web.pl, "probe_duration", lambda _: 3600.0)
long = web.gpu_tier(tmp_path / "a.wav", 0.15)
assert short < long
assert short in web.GPU_TIERS and long in web.GPU_TIERS