445 lines
18 KiB
Python
445 lines
18 KiB
Python
import json
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from scripts import run_mvp_meeting
|
|
from src.meeting_lab.audio import PreparedAudio
|
|
from src.meeting_lab.audio.preparation import (
|
|
DEFAULT_NORMALIZATION_FILTER,
|
|
DEFAULT_NORMALIZATION_METHOD,
|
|
)
|
|
from src.meeting_lab.orchestration import mvp as mvp_api
|
|
from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
|
|
from src.meeting_lab.diarization.backend import DiarizationResult
|
|
from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult
|
|
|
|
|
|
VALID_CONTEXT = """schema_version: "1"
|
|
meeting:
|
|
meeting_id: "mvp-test"
|
|
title: "MVP Test"
|
|
language: "de"
|
|
participants: []
|
|
mentioned_people: []
|
|
organization:
|
|
departments: []
|
|
known_entities: {}
|
|
"""
|
|
|
|
|
|
def protocol_result(model: str = "chosen:model") -> DirectProtocolResult:
|
|
text = "# Protokoll\n\nUnverändert. \n"
|
|
return DirectProtocolResult(
|
|
protocol_text=text,
|
|
exact_prompt="exact prompt\n",
|
|
model_metadata={"model": model},
|
|
runtime_metadata={"model": model, "request_count": 1, "client_wall_time_seconds": 0.5},
|
|
raw_response={"response": text, "done": True},
|
|
transcript_input="selected transcript\n",
|
|
)
|
|
|
|
|
|
def fake_transcribe(
|
|
audio_path: Path,
|
|
model_path: Path,
|
|
output_dir: Path,
|
|
language: str,
|
|
*,
|
|
executable: str,
|
|
threads: str | int,
|
|
) -> TranscriptionResult:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
raw = output_dir / "whisper_raw.json"
|
|
transcript = output_dir / "transcript.json"
|
|
text = output_dir / "transcript.txt"
|
|
metadata = output_dir / "runtime_metadata.json"
|
|
raw.write_text('{"transcription": []}\n', encoding="utf-8")
|
|
transcript.write_text(
|
|
json.dumps(
|
|
{
|
|
"text": "Besprechungstext.",
|
|
"segments": [
|
|
{"id": 0, "start": 0.0, "end": 1.0, "text": "Besprechungstext."}
|
|
],
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
text.write_text("Besprechungstext.\n", encoding="utf-8")
|
|
metadata.write_text('{"runtime_seconds": 1.25}\n', encoding="utf-8")
|
|
return TranscriptionResult(output_dir, raw, transcript, text, metadata, 1.25)
|
|
|
|
|
|
def fake_diarize(audio_path, output_dir, device_mode, **kwargs):
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
metadata = {
|
|
"backend": "pyannote.audio",
|
|
"model": "pyannote/speaker-diarization-community-1",
|
|
"requested_device_mode": device_mode,
|
|
"actual_device": "cuda",
|
|
"device_name": "Fake GPU",
|
|
"runtime_seconds": 2.5,
|
|
"speaker_count": 1,
|
|
"credentials_persisted": False,
|
|
}
|
|
paths = {
|
|
"metadata": output_dir / "metadata.json",
|
|
"ordinary": output_dir / "diarization.rttm",
|
|
"exclusive": output_dir / "exclusive_diarization.rttm",
|
|
"turns": output_dir / "turns.json",
|
|
"exclusive_turns": output_dir / "exclusive_turns.json",
|
|
}
|
|
paths["metadata"].write_text(json.dumps(metadata), encoding="utf-8")
|
|
paths["ordinary"].write_text("", encoding="utf-8")
|
|
paths["exclusive"].write_text("", encoding="utf-8")
|
|
paths["turns"].write_text("[]", encoding="utf-8")
|
|
paths["exclusive_turns"].write_text(
|
|
json.dumps([{"start": 0, "end": 10, "speaker_id": "SPEAKER_00"}]),
|
|
encoding="utf-8",
|
|
)
|
|
return DiarizationResult(
|
|
output_dir,
|
|
paths["metadata"],
|
|
paths["ordinary"],
|
|
paths["exclusive"],
|
|
paths["turns"],
|
|
paths["exclusive_turns"],
|
|
metadata,
|
|
)
|
|
|
|
|
|
def fake_prepare(source, destination, **kwargs):
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(source, destination)
|
|
normalization_enabled = kwargs.get("normalization_enabled", True)
|
|
return PreparedAudio(
|
|
source,
|
|
source.suffix.removeprefix("."),
|
|
destination,
|
|
"ffmpeg",
|
|
kwargs.get("ffmpeg_executable", "ffmpeg"),
|
|
normalization_enabled,
|
|
DEFAULT_NORMALIZATION_METHOD if normalization_enabled else None,
|
|
DEFAULT_NORMALIZATION_FILTER if normalization_enabled else None,
|
|
)
|
|
|
|
|
|
class MvpOrchestratorTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
patcher = patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare)
|
|
self.prepare_audio = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def create_inputs(self, root: Path) -> tuple[Path, Path, Path]:
|
|
audio = root / "team meeting.wav"
|
|
whisper_model = root / "ggml-model.bin"
|
|
context = root / "source-context.yaml"
|
|
audio.write_bytes(b"audio")
|
|
whisper_model.write_bytes(b"model")
|
|
context.write_text(VALID_CONTEXT, encoding="utf-8")
|
|
return audio, whisper_model, context
|
|
|
|
def args(self, root: Path, extra: list[str] | None = None):
|
|
audio, whisper_model, context = self.create_inputs(root)
|
|
values = [
|
|
str(audio),
|
|
"--whisper-model", str(whisper_model),
|
|
"--context", str(context),
|
|
"--output-root", str(root / "runs"),
|
|
"--language", "de",
|
|
"--model", "chosen:model",
|
|
"--ollama-endpoint", "http://ollama.test:11434",
|
|
]
|
|
if extra:
|
|
values.extend(extra)
|
|
return run_mvp_meeting.parse_args(values)
|
|
|
|
def test_successful_full_orchestration_and_artifact_layout(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
) as protocol,
|
|
):
|
|
code, run_dir, protocol_path = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertIsNotNone(run_dir)
|
|
self.assertEqual(protocol_path, run_dir / "protocol.md")
|
|
expected = {
|
|
"run_metadata.json",
|
|
"audio/input_manifest.json",
|
|
"audio/prepared.wav",
|
|
"audio/preparation_metadata.json",
|
|
"transcript/whisper_raw.json",
|
|
"transcript/transcript.json",
|
|
"transcript/transcript.txt",
|
|
"transcript/runtime_metadata.json",
|
|
"context/meeting_context.yaml",
|
|
"protocol/exact_prompt.txt",
|
|
"protocol/transcript_input.txt",
|
|
"protocol/raw_response.json",
|
|
"protocol/runtime_metadata.json",
|
|
"protocol.md",
|
|
}
|
|
self.assertTrue(all((run_dir / item).is_file() for item in expected))
|
|
metadata = json.loads((run_dir / "run_metadata.json").read_text())
|
|
self.assertEqual(metadata["status"], "completed")
|
|
self.assertEqual(metadata["model"], "chosen:model")
|
|
self.assertIsNone(metadata["failure"])
|
|
self.assertEqual(whisper.call_count, 1)
|
|
self.assertEqual(
|
|
whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
|
|
)
|
|
self.assertEqual(protocol.call_count, 1)
|
|
self.assertTrue(
|
|
self.prepare_audio.call_args.kwargs["normalization_enabled"]
|
|
)
|
|
preparation = json.loads(
|
|
(run_dir / "audio" / "preparation_metadata.json").read_text()
|
|
)
|
|
self.assertTrue(preparation["normalization_enabled"])
|
|
|
|
def test_cli_can_disable_audio_normalization_without_bypassing_preparation(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root, ["--no-audio-normalization"])
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
),
|
|
):
|
|
code, run_dir, _ = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.prepare_audio.assert_called_once()
|
|
self.assertFalse(
|
|
self.prepare_audio.call_args.kwargs["normalization_enabled"]
|
|
)
|
|
self.assertEqual(
|
|
whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
|
|
)
|
|
preparation = json.loads(
|
|
(run_dir / "audio" / "preparation_metadata.json").read_text()
|
|
)
|
|
self.assertFalse(preparation["normalization_enabled"])
|
|
self.assertIsNone(preparation["normalization_method"])
|
|
self.assertIsNone(preparation["normalization_filter"])
|
|
|
|
def test_context_model_endpoint_and_whisper_options_are_forwarded(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(
|
|
root,
|
|
[
|
|
"--whisper-executable",
|
|
"/tools/whisper-cli",
|
|
"--ffmpeg-executable",
|
|
"/tools/ffmpeg",
|
|
"--threads",
|
|
"4",
|
|
],
|
|
)
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
) as protocol,
|
|
):
|
|
code, run_dir, _ = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(whisper.call_args.args[3], "de")
|
|
self.assertEqual(whisper.call_args.kwargs["executable"], "/tools/whisper-cli")
|
|
self.assertEqual(whisper.call_args.kwargs["threads"], "4")
|
|
self.assertEqual(
|
|
self.prepare_audio.call_args.kwargs["ffmpeg_executable"],
|
|
"/tools/ffmpeg",
|
|
)
|
|
self.assertEqual(protocol.call_args.args[1], run_dir / "context/meeting_context.yaml")
|
|
self.assertEqual(protocol.call_args.kwargs["model"], "chosen:model")
|
|
self.assertEqual(
|
|
protocol.call_args.kwargs["endpoint"], "http://ollama.test:11434"
|
|
)
|
|
|
|
def test_diarization_is_off_by_default_and_preserves_protocol_input(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
|
|
patch.object(mvp_api, "diarize_audio") as diarization,
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
) as protocol,
|
|
):
|
|
code, run_dir, _ = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
diarization.assert_not_called()
|
|
self.assertEqual(protocol.call_args.args[0], run_dir / "transcript/transcript.json")
|
|
self.assertFalse((run_dir / "diarization").exists())
|
|
|
|
def test_diarization_cli_propagates_and_uses_derived_protocol_input(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(
|
|
root,
|
|
[
|
|
"--diarization", "gpu",
|
|
"--diarization-runtime", "container",
|
|
"--diarization-container-image", "rocm/test",
|
|
"--diarization-container-arg=--device=/dev/kfd",
|
|
],
|
|
)
|
|
with (
|
|
patch.object(
|
|
mvp_api, "transcribe_audio", side_effect=fake_transcribe
|
|
),
|
|
patch.object(
|
|
mvp_api, "diarize_audio", side_effect=fake_diarize
|
|
) as diarization,
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
) as protocol,
|
|
):
|
|
code, run_dir, _ = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(diarization.call_args.args[2], "gpu")
|
|
self.assertEqual(
|
|
diarization.call_args.args[0], run_dir / "audio" / "prepared.wav"
|
|
)
|
|
self.assertEqual(diarization.call_args.kwargs["runtime"], "container")
|
|
self.assertEqual(
|
|
diarization.call_args.kwargs["container_args"], ("--device=/dev/kfd",)
|
|
)
|
|
derived = run_dir / "diarization/transcript_diarized.json"
|
|
self.assertEqual(protocol.call_args.args[0], derived)
|
|
self.assertIn("SPEAKER_00", derived.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
json.loads((run_dir / "transcript/transcript.json").read_text())["text"],
|
|
"Besprechungstext.",
|
|
)
|
|
run_metadata = json.loads((run_dir / "run_metadata.json").read_text())
|
|
self.assertTrue(run_metadata["diarization"]["enabled"])
|
|
self.assertNotIn("HF_TOKEN", json.dumps(run_metadata))
|
|
|
|
def test_whisper_failure_is_recorded_and_protocol_is_not_called(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
with (
|
|
patch.object(
|
|
mvp_api,
|
|
"transcribe_audio",
|
|
side_effect=TranscriptionError("whisper stopped"),
|
|
),
|
|
patch.object(mvp_api, "generate_direct_protocol") as protocol,
|
|
):
|
|
code, run_dir, protocol_path = run_mvp_meeting.run(args)
|
|
|
|
metadata = json.loads((run_dir / "run_metadata.json").read_text())
|
|
self.assertEqual(code, 2)
|
|
self.assertIsNone(protocol_path)
|
|
self.assertEqual(metadata["status"], "failed")
|
|
self.assertEqual(metadata["failure"]["stage"], "whisper")
|
|
self.assertIn("whisper stopped", metadata["failure"]["message"])
|
|
protocol.assert_not_called()
|
|
self.assertTrue((run_dir / "audio/input_manifest.json").is_file())
|
|
|
|
def test_protocol_failure_preserves_transcript_and_failure_metadata(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
side_effect=ValueError("generation stopped"),
|
|
),
|
|
):
|
|
code, run_dir, protocol_path = run_mvp_meeting.run(args)
|
|
|
|
metadata = json.loads((run_dir / "run_metadata.json").read_text())
|
|
self.assertEqual(code, 2)
|
|
self.assertIsNone(protocol_path)
|
|
self.assertEqual(metadata["failure"]["stage"], "protocol")
|
|
self.assertTrue((run_dir / "transcript/transcript.json").is_file())
|
|
self.assertFalse((run_dir / "protocol.md").exists())
|
|
|
|
def test_unique_run_directories_do_not_overwrite(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
fixed = datetime(2026, 8, 21, 9, 15, 30)
|
|
first = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
|
|
marker = first / "keep.txt"
|
|
marker.write_text("keep", encoding="utf-8")
|
|
second = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
|
|
self.assertEqual(first.name, "team_meeting_20260821_091530")
|
|
self.assertEqual(second.name, "team_meeting_20260821_091530_01")
|
|
self.assertEqual(marker.read_text(), "keep")
|
|
|
|
def test_semantic_pipeline_functions_are_never_invoked(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
with (
|
|
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
|
|
patch.object(
|
|
mvp_api,
|
|
"generate_direct_protocol",
|
|
return_value=protocol_result(),
|
|
),
|
|
patch("src.meeting_lab.chunking.chunk_transcript.build_chunks") as chunking,
|
|
patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction,
|
|
patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation,
|
|
):
|
|
code, _, _ = run_mvp_meeting.run(args)
|
|
|
|
self.assertEqual(code, 0)
|
|
chunking.assert_not_called()
|
|
extraction.assert_not_called()
|
|
consolidation.assert_not_called()
|
|
|
|
def test_main_returns_nonzero_for_whisper_failure(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
args = self.args(root)
|
|
argv = [
|
|
str(args.audio_file),
|
|
"--whisper-model", str(args.whisper_model),
|
|
"--context", str(args.context),
|
|
"--output-root", str(args.output_root),
|
|
]
|
|
with patch.object(
|
|
mvp_api,
|
|
"transcribe_audio",
|
|
side_effect=TranscriptionError("failed"),
|
|
):
|
|
self.assertEqual(run_mvp_meeting.main(argv), 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|