# -*- 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)