File size: 7,789 Bytes
5b4386a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7fb75a2
34b6a6d
 
 
 
 
 
 
 
 
 
 
 
 
 
cbf3389
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7fb75a2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
"""画面まわり。文字起こし本体は動かさず、組み立てと片付けだけを確かめる。"""

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