Spaces:
Runtime error
Runtime error
| """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") | |