Spaces:
Paused
Paused
File size: 3,853 Bytes
397b4ac 994d3d9 397b4ac 994d3d9 397b4ac 994d3d9 397b4ac 994d3d9 397b4ac 994d3d9 397b4ac 994d3d9 397b4ac | 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 | """Dependency-light tests for the Ask/Q&A server flow."""
import logging
import tempfile
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
logging.disable(logging.CRITICAL)
from qa_flow import build_qa_response, claim_audio_submission, release_audio_submission
def check(label, condition):
if not condition:
raise AssertionError(label)
print(f"[OK] {label}")
def fake_answer_fn(**kwargs):
check("story context forwarded", "Three Little Pigs" in kwargs["story_context"])
check("max_new_tokens forwarded", kwargs["max_new_tokens"] == 12)
if kwargs["question_audio_path"]:
return "The wolf huffed and puffed.", object(), 24000
return "The pig built a house.", None, 24000
def fake_writer(path: Path, waveform, sample_rate: int):
check("audio writer receives sample rate", sample_rate == 24000)
path.write_bytes(b"RIFF fake wav")
def failing_answer_fn(**kwargs):
raise RuntimeError("model unavailable")
def test_empty_question():
result = build_qa_response(
question_text="",
question_audio_path=None,
paragraphs=[],
output_dir=Path(tempfile.gettempdir()),
answer_fn=fake_answer_fn,
)
check("empty question is rejected", result["ok"] is False)
check("empty question has no audio", result["audio_path"] is None)
def test_text_question():
result = build_qa_response(
question_text="What did the pig build?",
question_audio_path=None,
paragraphs=["The Three Little Pigs built houses."],
output_dir=Path(tempfile.gettempdir()),
answer_fn=fake_answer_fn,
max_new_tokens=12,
)
check("text question succeeds", result["ok"] is True)
check("text answer returned", result["answer_text"] == "The pig built a house.")
check("text question does not write audio when model returns none", result["audio_path"] is None)
def test_audio_question_writes_answer():
with tempfile.TemporaryDirectory() as tmp:
result = build_qa_response(
question_text="",
question_audio_path=Path(tmp) / "question.wav",
paragraphs=["The Three Little Pigs met a wolf."],
output_dir=Path(tmp),
answer_fn=fake_answer_fn,
audio_writer=fake_writer,
max_new_tokens=12,
)
check("audio question succeeds", result["ok"] is True)
check("audio question display fallback", result["display_question"] == "(audio question)")
check("audio answer path created", result["audio_path"] is not None)
check("audio answer file exists", Path(result["audio_path"]).exists())
def test_duplicate_audio_claim():
with tempfile.TemporaryDirectory() as tmp:
audio = Path(tmp) / "same-question.wav"
check("first audio claim accepted", claim_audio_submission(audio) is True)
check("duplicate audio claim rejected", claim_audio_submission(audio) is False)
release_audio_submission(audio)
check("released audio claim can retry", claim_audio_submission(audio) is True)
def test_failed_answer_marks_error():
result = build_qa_response(
question_text="",
question_audio_path="question.wav",
paragraphs=["The Three Little Pigs met a wolf."],
output_dir=Path(tempfile.gettempdir()),
answer_fn=failing_answer_fn,
)
check("failed answer still returns display result", result["ok"] is True)
check("failed answer records error", result["error"] == "RuntimeError")
check("failed answer has no audio", result["audio_path"] is None)
if __name__ == "__main__":
test_empty_question()
test_text_question()
test_audio_question_writes_answer()
test_duplicate_audio_claim()
test_failed_answer_marks_error()
print("Q&A flow tests passed")
|