229 lines
12 KiB
Python
229 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""t-261004160531: видео→текст в voice-пайплайне (регресс t-261004152714).
|
||
|
||
Покрытие:
|
||
1. unit: extract_audio_track — реальный ffmpeg на живых фикстурах
|
||
(mp4 с дорожкой → ogg/opus байты; mp4 без дорожки и мусор → None);
|
||
2. unit: роутинг kind=video ≤60с → сразу faster-whisper turbo
|
||
(webservice-ярус не зовётся, движок = large-v3-turbo);
|
||
3. unit: длинное видео (>60с) → обычная лестница (webservice base первым);
|
||
4. unit: контейнер без дорожки → MessageTranscribed(audio_track_failed);
|
||
5. e2e на фикстуре: MessageReceived(kind=video) → download (мок) →
|
||
ffmpeg-извлечение → ASR (мок) → транскрипт в проекции inbox;
|
||
идемпотентность повтора (второй вызов — из read model, без хождений).
|
||
|
||
Запуск: ./venv/bin/python tests/test_voice_video.py | pytest tests/test_voice_video.py
|
||
ASR-слои мокнуты (нет сети/моделей), ffmpeg — настоящий (он на хосте).
|
||
"""
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
# ── изоляция: своя БД в /tmp, прод-стор не трогаем ──
|
||
_DST = "/tmp/simbiot-test-voice-video.db"
|
||
for p in (_DST, _DST + "-wal", _DST + "-shm"):
|
||
if os.path.exists(p):
|
||
os.unlink(p)
|
||
os.environ["SIMBIOT_DB"] = _DST
|
||
sys.path.insert(0, "/root/simbiot-harness")
|
||
|
||
from harness import es, tg_inbox
|
||
import harness.voice as voice
|
||
|
||
fails = []
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
print(("PASS" if cond else "FAIL"), name, ("| " + extra) if not cond else "", flush=True)
|
||
if not cond:
|
||
fails.append(name)
|
||
raise AssertionError(f"{name} {extra}")
|
||
|
||
|
||
# ── фикстуры: генерируем ffmpeg'ом при запуске (детерминировано, без бинарников в репо) ──
|
||
_TMP = "/tmp/simbiot-test-voice-video"
|
||
os.makedirs(_TMP, exist_ok=True)
|
||
WITH_AUDIO = os.path.join(_TMP, "with_audio.mp4") # testsrc + sine 440Hz
|
||
NO_AUDIO = os.path.join(_TMP, "no_audio.mp4") # только видео
|
||
|
||
|
||
def _ff(args):
|
||
r = subprocess.run(["ffmpeg", "-y", "-v", "error"] + args,
|
||
capture_output=True, timeout=120)
|
||
if r.returncode != 0:
|
||
raise RuntimeError((r.stderr or b"").decode()[:300])
|
||
|
||
|
||
_ff(["-f", "lavfi", "-i", "testsrc=duration=2:size=128x96:rate=10",
|
||
"-f", "lavfi", "-i", "sine=frequency=440:duration=2",
|
||
"-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac",
|
||
"-shortest", WITH_AUDIO])
|
||
_ff(["-f", "lavfi", "-i", "testsrc=duration=2:size=128x96:rate=10",
|
||
"-c:v", "libx264", "-preset", "ultrafast", "-an", NO_AUDIO])
|
||
|
||
with open(WITH_AUDIO, "rb") as f:
|
||
VIDEO_WITH_AUDIO = f.read()
|
||
with open(NO_AUDIO, "rb") as f:
|
||
VIDEO_NO_AUDIO = f.read()
|
||
|
||
es.init_db()
|
||
|
||
|
||
# ────────────────────────── 1. extract_audio_track ──────────────────────────
|
||
|
||
def test_extract_audio_track_from_container():
|
||
"""mp4 с дорожкой → ogg/opus байты (заголовок OggS + OpusHead, звук есть)."""
|
||
track = voice.extract_audio_track(VIDEO_WITH_AUDIO)
|
||
check("контейнер с дорожкой → байты", bool(track), "None/пусто")
|
||
assert track[:4] == b"OggS", f"не ogg: {track[:8]!r}"
|
||
assert b"OpusHead" in track[:64], "не opus-кодек"
|
||
check("ogg/opus формат дорожки", True)
|
||
|
||
|
||
def test_extract_audio_track_no_track():
|
||
"""mp4 без аудио → None (ffmpeg ничего не вынул)."""
|
||
r = voice.extract_audio_track(VIDEO_NO_AUDIO)
|
||
check("контейнер без дорожки → None", r is None, f"получили {type(r)}")
|
||
r = voice.extract_audio_track(b"not a video at all")
|
||
check("мусор на входе → None (без исключения)", r is None, f"получили {type(r)}")
|
||
|
||
|
||
# ─────────────────────── 2-4. роутинг kind=video ────────────────────────────
|
||
|
||
def _video_msg(mid, duration):
|
||
"""MessageReceived c video-вложением в Telegram-формате."""
|
||
return {"update_id": int(mid.split("-")[1]), "message": {
|
||
"message_id": int(mid.split("-")[1]) % 100000, "date": 1780000000,
|
||
"chat": {"id": 42}, "from": {"id": 1, "username": "alex", "first_name": "Alex"},
|
||
"video": {"file_id": f"fid-{mid}", "file_unique_id": f"fuid-{mid}",
|
||
"duration": duration}}}
|
||
|
||
|
||
class _Rec:
|
||
"""Записываем вызовы ASR-слоёв и подменяем их ответами."""
|
||
|
||
def __init__(self, fw_text=None, ws_text=None):
|
||
self.calls = {"fw": [], "ws": [], "dl": []}
|
||
self._orig = {}
|
||
self._patch("_download_voice",
|
||
lambda fid, fuid="": (self.calls["dl"].append(fid) or VIDEO_WITH_AUDIO, None))
|
||
self._patch("_asr_faster_whisper",
|
||
lambda audio, fn: (self.calls["fw"].append(audio) or fw_text))
|
||
self._patch("_asr_webservice",
|
||
lambda audio, fn: (self.calls["ws"].append(audio) or ws_text))
|
||
|
||
def _patch(self, name, fn):
|
||
self._orig[name] = getattr(voice, name)
|
||
setattr(voice, name, fn)
|
||
|
||
def restore(self):
|
||
for k, v in self._orig.items():
|
||
setattr(voice, k, v)
|
||
self._orig = {}
|
||
|
||
|
||
def test_short_video_routes_to_turbo():
|
||
"""kind=video ≤60с: сразу faster-whisper turbo, webservice не зовётся."""
|
||
tg_inbox.receive_message(_video_msg("m-910001", 5))
|
||
rec = _Rec(fw_text="привет из видео")
|
||
# детерминированный маркер: сам ogg-заголовок между запусками ffmpeg
|
||
# не байт-стабилен — проверяем передачу ровно того, что вернул extract
|
||
marker = b"OggS-EXTRACTED-TRACK"
|
||
orig_extract = voice.extract_audio_track
|
||
voice.extract_audio_track = lambda c: (rec.calls.setdefault("ex", []).append(1) or marker)
|
||
try:
|
||
text, engine = voice.transcribe_message(
|
||
"m-910001", "fid-m-910001", "fuid-m-910001", duration=5, kind="video")
|
||
finally:
|
||
voice.extract_audio_track = orig_extract
|
||
rec.restore()
|
||
check("транскрипт получен", text == "привет из видео", f"text={text!r}")
|
||
check("движок = large-v3-turbo", engine == "faster-whisper-large-v3-turbo",
|
||
f"engine={engine!r}")
|
||
check("turbo-ярус вызван ровно 1 раз", len(rec.calls["fw"]) == 1,
|
||
f"{len(rec.calls['fw'])}")
|
||
check("webservice-ярус пропущен", len(rec.calls["ws"]) == 0,
|
||
f"{len(rec.calls['ws'])}")
|
||
check("в ASR ушла ИЗВЛЕЧЁННАЯ дорожка (не контейнер)",
|
||
rec.calls["fw"] and rec.calls["fw"][0] == marker, "байты != extract()")
|
||
check("extract вызван для видео", len(rec.calls.get("ex", [])) == 1)
|
||
m = es.run_projections()["inbox"]["m-910001"]
|
||
check("проекция: transcript в read model", m["transcript"] == "привет из видео")
|
||
check("проекция: text = транскрипт", m["text"] == "привет из видео")
|
||
check("проекция: без transcribe_reason", not m.get("transcribe_reason"))
|
||
|
||
|
||
def test_long_video_uses_tier_ladder():
|
||
"""kind=video >60с: сначала webservice base (обычная лестница)."""
|
||
tg_inbox.receive_message(_video_msg("m-910002", 120))
|
||
rec = _Rec(ws_text="длинное видео base")
|
||
try:
|
||
text, engine = voice.transcribe_message(
|
||
"m-910002", "fid-m-910002", "fuid-m-910002", duration=120, kind="video")
|
||
finally:
|
||
rec.restore()
|
||
check("транскрипт из base-яруса", text == "длинное видео base", f"text={text!r}")
|
||
check("движок = whisper-asr-base", engine == "whisper-asr-base", f"engine={engine!r}")
|
||
check("webservice вызван", len(rec.calls["ws"]) == 1, f"{len(rec.calls['ws'])}")
|
||
check("turbo не понадобился", len(rec.calls["fw"]) == 0, f"{len(rec.calls['fw'])}")
|
||
|
||
|
||
def test_video_without_track_marks_failure():
|
||
"""Дорожки нет → MessageTranscribed(reason=audio_track_failed), движок None."""
|
||
tg_inbox.receive_message(_video_msg("m-910003", 7))
|
||
rec = _Rec(fw_text="не должен вызываться")
|
||
rec._patch("_download_voice",
|
||
lambda fid, fuid="": (self_dl := rec.calls["dl"].append(fid) or VIDEO_NO_AUDIO, None))
|
||
try:
|
||
text, engine = voice.transcribe_message(
|
||
"m-910003", "fid-m-910003", "fuid-m-910003", duration=7, kind="video")
|
||
finally:
|
||
rec.restore()
|
||
check("результат — отказ (None)", text is None, f"text={text!r}")
|
||
m = es.run_projections()["inbox"]["m-910003"]
|
||
check("reason=audio_track_failed", m.get("transcribe_reason") == "audio_track_failed",
|
||
f"reason={m.get('transcribe_reason')!r}")
|
||
check("ASR-слои не звались", not rec.calls["fw"] and not rec.calls["ws"])
|
||
|
||
|
||
# ──────────────────────────── 5. e2e на фикстуре ────────────────────────────
|
||
|
||
def test_e2e_video_message_to_transcript():
|
||
"""Полный путь: приём видео-сообщения → транскрипция → идемпотентность."""
|
||
mid = tg_inbox.receive_message(_video_msg("m-910004", 12))
|
||
rec = _Rec(fw_text="текст из видео e2e")
|
||
try:
|
||
text, engine = voice.transcribe_message(
|
||
mid, "fid-m-910004", "fuid-m-910004", duration=12, kind="video")
|
||
finally:
|
||
rec.restore()
|
||
check("e2e: транскрипт вернулся", text == "текст из видео e2e", f"text={text!r}")
|
||
m = es.run_projections()["inbox"][mid]
|
||
check("e2e: kind=video в проекции", m["kind"] == "video", f"kind={m['kind']!r}")
|
||
check("e2e: движок записан", m["engine"] == "faster-whisper-large-v3-turbo")
|
||
check("e2e: минуты увидат текст", m["text"] == "текст из видео e2e", f"text={m['text']!r}")
|
||
# идемпотентность: повтор берёт транскрипт из read model — без download/ASR
|
||
rec2 = _Rec(fw_text="другой текст (не должен использоваться)")
|
||
try:
|
||
text2, engine2 = voice.transcribe_message(
|
||
mid, "fid-m-910004", "fuid-m-910004", duration=12, kind="video")
|
||
finally:
|
||
rec2.restore()
|
||
check("повтор: транскрипт из кэша read model", text2 == "текст из видео e2e",
|
||
f"text={text2!r}")
|
||
check("повтор: без повторного download", not rec2.calls["dl"], "качал заново")
|
||
check("повтор: без повторного ASR", not rec2.calls["fw"] and not rec2.calls["ws"])
|
||
|
||
|
||
# ────────────────────────────── plain-run режим ──────────────────────────────
|
||
|
||
if __name__ == "__main__":
|
||
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||
for t in tests:
|
||
try:
|
||
t()
|
||
except AssertionError:
|
||
pass
|
||
print(f"\n{'='*60}")
|
||
print(f"{'ALL OK' if not fails else 'FAILED: ' + ', '.join(fails)} ({len(tests)} тестов)")
|
||
sys.exit(1 if fails else 0)
|