"""Тесты чистой функции чанкинга транскрипта `chunk_transcript` (ТЗ §1.3). Синтетические транскрипты (строки `[Имя MM:SS] текст`) без сети/БД/токенизатора — подсчёт токенов в тестах подменяется фейковым callable (TDD). """ from core.summarization.chunking import chunk_transcript def _word_count(text: str) -> int: """Фейковый подсчёт токенов для тестов: число слов (детерминированно и без сторонних зависимостей — реальный `QwenTokenCounter` тестируется отдельно).""" return len(text.split()) def _line(speaker: str, offset_s: float, text: str) -> str: """Собрать строку-фразу транскрипта в формате `[Имя MM:SS] текст`.""" minutes, seconds = divmod(int(offset_s), 60) return f"[{speaker} {minutes:02d}:{seconds:02d}] {text}" def test_empty_transcript_returns_empty_list() -> None: """Пустой транскрипт → пустой список чанков.""" assert chunk_transcript("", _word_count) == [] def test_blank_transcript_returns_empty_list() -> None: """Транскрипт из одних пустых строк — тоже пустой вход.""" assert chunk_transcript("\n\n \n", _word_count) == [] def test_short_meeting_fits_in_single_chunk() -> None: """Короткая встреча (мало времени и токенов) → один чанк со всеми фразами.""" lines = [ _line("Аня", 0, "привет всем"), _line("Боря", 3, "привет как дела"), _line("Аня", 8, "всё хорошо начинаем"), ] transcript = "\n".join(lines) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20 ) assert chunks == [transcript] def test_long_meeting_splits_into_several_chunks_respecting_token_limit() -> None: """Длинная встреча: несколько чанков, ни один не превышает max_chunk_tokens, все исходные фразы сохранены без потерь, порядок не нарушен.""" lines = [] for i in range(40): speaker = "Аня" if i % 2 == 0 else "Боря" text = " ".join(["слово"] * 20) # по 20 токенов на фразу lines.append(_line(speaker, i * 5, text)) # фразы каждые 5 секунд transcript = "\n".join(lines) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=100, target_chunk_minutes=20 ) assert len(chunks) > 1 for chunk in chunks: assert _word_count(chunk) <= 100 # все исходные фразы присутствуют ровно один раз, порядок сохранён reconstructed = [line for chunk in chunks for line in chunk.splitlines()] assert reconstructed == lines def test_chunk_closes_on_target_minutes_even_with_tokens_to_spare() -> None: """Чанк закрывается по достижении target_chunk_minutes, даже если лимит токенов ещё далеко не исчерпан; граница — строго на границе фразы.""" lines = [_line("Аня" if i % 2 == 0 else "Боря", i * 5, "слово") for i in range(20)] transcript = "\n".join(lines) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=100_000, target_chunk_minutes=1 ) assert len(chunks) > 1 # каждая строка каждого чанка — целая исходная фраза, ничего не обрезано all_lines_in_chunks = [line for chunk in chunks for line in chunk.splitlines()] assert all_lines_in_chunks == lines def test_hh_mm_ss_label_offset_triggers_boundary_across_hour_gap() -> None: """Метка `[Имя ЧЧ:MM:SS]` парсится корректно: часовой разрыв между двумя фразами закрывает чанк по времени при малом target_chunk_minutes.""" transcript = "\n".join( [ "[Аня 00:00] начало встречи", "[Аня 1:00:05] через час — подведём итоги", ] ) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=1 ) assert len(chunks) == 2 assert chunks[0] == "[Аня 00:00] начало встречи" assert chunks[1] == "[Аня 1:00:05] через час — подведём итоги" def test_mm_ss_label_under_hour_kept_in_one_chunk_when_within_target() -> None: """Метка `[Имя MM:SS]` (до часа): фразы укладываются в target_chunk_minutes и попадают в один чанк.""" transcript = "\n".join( [ "[Аня 00:00] первая фраза", "[Боря 00:30] вторая фраза", ] ) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20 ) assert chunks == [transcript] def test_monologue_exceeding_token_limit_splits_by_sentences_with_repeated_label() -> None: """Монолог одной фразой (40 минут) больше max_chunk_tokens: делится по предложениям, каждая часть — отдельный чанк с повторённой меткой спикера, ни одна часть не превышает лимит, предложения не разорваны.""" sentences = [f"Пункт номер {i} с некоторыми словами тут." for i in range(30)] monologue_text = " ".join(sentences) transcript = _line("Иван", 0, monologue_text) chunks = chunk_transcript(transcript, _word_count, max_chunk_tokens=20, target_chunk_minutes=20) assert len(chunks) > 1 for chunk in chunks: assert chunk.startswith("[Иван 00:00]") assert _word_count(chunk) <= 20 # ни одно предложение не потеряно и не разорвано пополам recombined_text = " ".join(chunk.removeprefix("[Иван 00:00] ") for chunk in chunks) for sentence in sentences: assert sentence in recombined_text def test_monologue_alone_still_produces_single_chunk_when_within_limit() -> None: """Одна длинная фраза, которая тем не менее укладывается в лимит токенов, не должна делиться по предложениям — она атомарна.""" transcript = _line("Иван", 0, "Это не очень длинный монолог. Всего два предложения.") chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20 ) assert chunks == [transcript] def test_normal_phrases_are_never_split_across_or_within_chunks() -> None: """В обычном случае (без аварийного монолога) каждая исходная строка-фраза целиком попадает ровно в один чанк — ни разрывов внутри строки, ни дублирования между чанками.""" lines = [ _line("Аня", 0, "первая фраза без разрывов"), _line("Боря", 2, "вторая фраза целиком"), _line("Аня", 500, "третья фраза после большой паузы"), _line("Боря", 505, "четвёртая фраза"), ] transcript = "\n".join(lines) chunks = chunk_transcript( transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=5 ) all_lines_in_chunks = [line for chunk in chunks for line in chunk.splitlines()] assert all_lines_in_chunks == lines # каждая фраза встречается ровно в одном чанке assert len(all_lines_in_chunks) == len(lines)