Spaces:
Running on Zero
Running on Zero
| """画面まわり。文字起こし本体は動かさず、組み立てと片付けだけを確かめる。""" | |
| 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 "<script>" 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 | |