| import importlib |
| import json |
| import wave |
|
|
| import pytest |
|
|
| from figment.config import FigmentConfig, OMNI_MODEL_ID |
|
|
|
|
| def test_hosted_omni_provider_payload_becomes_unconfirmed_audio_draft() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| nvidia_api_key="test-nvidia-key", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
| provider_payload = { |
| "transcript": "Adult reports severe headache and visual spots after flood cleanup.", |
| "suggested_fields": [ |
| { |
| "field": "symptoms", |
| "draft_value": "severe headache and visual spots", |
| "source_snippet": "severe headache and visual spots", |
| "source_timecode": "00:01-00:05", |
| } |
| ], |
| "missing_or_unclear_fields": ["blood_pressure", "pregnancy_status"], |
| "provisional_red_flag_mentions": ["severe headache", "visual spots"], |
| } |
|
|
| draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) |
|
|
| assert draft["audio_intake_path"] == "omni_native" |
| assert draft["audio_runtime"] == "omni_native" |
| assert draft["audio_model_id"] == OMNI_MODEL_ID |
| assert draft["field_fill_model_id"] is None |
| assert draft["transcript_source"] == "hosted_omni_provider" |
| assert draft["transcript"] == provider_payload["transcript"] |
| assert draft["missing_or_unclear_fields"] == ["blood_pressure", "pregnancy_status"] |
| assert draft["provisional_red_flag_mentions"] == ["severe headache", "visual spots"] |
| assert draft["confirmed_intake_required"] is True |
| assert draft["confirmation_status"] == "unconfirmed" |
| assert draft["raw_audio_stored"] is False |
| assert draft["suggested_fields"] == [ |
| { |
| "field": "symptoms", |
| "draft_value": "severe headache and visual spots", |
| "source_snippet": "severe headache and visual spots", |
| "source_timecode": "00:01-00:05", |
| "status": "audio_draft", |
| "needs_confirmation": True, |
| } |
| ] |
|
|
|
|
| def test_hosted_omni_provider_payload_with_invalid_field_names_falls_back_to_transcript_suggestions() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| nvidia_api_key="test-nvidia-key", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
| provider_payload = { |
| "transcript": ( |
| "Seven-year-old at a shelter clinic after flood cleanup. Child cannot keep fluids down, " |
| "is lethargic, has a very dry mouth, and has no urine since morning." |
| ), |
| "suggested_fields": [ |
| { |
| "field": "chief_concern|symptoms|vitals|patient_age|pregnancy_status|allergies|medications|available_supplies|responder_note", |
| "draft_value": "lethargic, dry mouth, no urine", |
| "source_snippet": "lethargic, dry mouth, no urine", |
| } |
| ], |
| "provisional_red_flag_mentions": ["lethargic", "no urine"], |
| } |
|
|
| draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) |
| fields = {suggestion["field"] for suggestion in draft["suggested_fields"]} |
|
|
| assert "chief_concern|symptoms|vitals|patient_age|pregnancy_status|allergies|medications|available_supplies|responder_note" not in fields |
| assert {"responder_note", "patient_age", "chief_concern", "symptoms"}.issubset(fields) |
| assert next(item for item in draft["suggested_fields"] if item["field"] == "patient_age")["draft_value"] == "7 years" |
|
|
|
|
| def test_hosted_omni_provider_payload_without_transcript_or_valid_suggestions_fails_closed() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| nvidia_api_key="test-nvidia-key", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
| provider_payload = { |
| "suggested_fields": [ |
| { |
| "field": "chief_concern|symptoms", |
| "draft_value": "chest pain", |
| "source_snippet": "chest pain", |
| } |
| ] |
| } |
|
|
| draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) |
|
|
| assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" |
| assert draft["audio_runtime"] == "unprocessed_audio" |
| assert draft["transcript"] == "" |
| assert draft["suggested_fields"] == [] |
| assert draft["confirmed_intake_required"] is False |
| assert draft["confirmation_status"] == "confirmed" |
| assert draft["missing_or_unclear_fields"] == ["transcript_or_valid_provider_suggestions"] |
| assert "valid field suggestions" in draft["processing_status"].lower() |
|
|
|
|
| def test_hosted_omni_provider_payload_drops_non_informative_field_values_and_preserves_red_flag_mentions() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| nvidia_api_key="test-nvidia-key", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
| provider_payload = { |
| "transcript": ( |
| "Seven-year-old at a shelter clinic after flood cleanup. Child cannot keep fluids down, " |
| "is lethargic, has a very dry mouth, and has no urine since morning." |
| ), |
| "suggested_fields": [ |
| { |
| "field": "symptoms", |
| "draft_value": "cannot keep fluids down, lethargic, very dry mouth, no urine since morning", |
| "source_snippet": "cannot keep fluids down, lethargic", |
| }, |
| { |
| "field": "allergies", |
| "draft_value": "Not answerable", |
| "source_snippet": "", |
| }, |
| { |
| "field": "medications", |
| "draft_value": "not provided", |
| "source_snippet": "", |
| }, |
| { |
| "field": "responder_note", |
| "draft_value": "Provisional intake draft for trained responder", |
| "source_snippet": "", |
| }, |
| ], |
| "missing_or_unclear_fields": [], |
| "provisional_red_flag_mentions": [], |
| } |
|
|
| draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) |
|
|
| assert {item["field"] for item in draft["suggested_fields"]} == {"symptoms"} |
| assert set(draft["missing_or_unclear_fields"]) >= {"allergies", "medications"} |
| assert set(draft["provisional_red_flag_mentions"]) >= {"lethargic", "no urine"} |
|
|
|
|
| def test_manual_edit_wins_over_hosted_omni_audio_draft_on_confirmation() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| nvidia_api_key="test-nvidia-key", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
| audio_draft = app.draft_audio_intake( |
| config=config, |
| provider_payload={ |
| "transcript": "Audio says chest pain, but the medic corrects it.", |
| "suggested_fields": [ |
| { |
| "field": "chief_concern", |
| "draft_value": "chest pain", |
| "source_snippet": "chest pain", |
| } |
| ], |
| }, |
| ) |
|
|
| confirmed, _, confirmed_audio = app._confirm_ui_intake( |
| "mobile clinic", |
| "52", |
| "not_applicable", |
| "manual correction: wound concern", |
| "", |
| "", |
| "unknown", |
| "unknown", |
| "basic kit", |
| "Typed note remains source of truth.", |
| audio_draft, |
| ) |
|
|
| assert confirmed["chief_concern"] == "manual correction: wound concern" |
| assert confirmed_audio["suggested_fields"][0]["draft_value"] == "manual correction: wound concern" |
| assert confirmed_audio["suggested_fields"][0]["status"] == "edited" |
| assert confirmed_audio["suggested_fields"][0]["needs_confirmation"] is False |
|
|
|
|
| def test_uploaded_audio_uses_hosted_omni_provider_when_enabled(tmp_path, monkeypatch) -> None: |
| app = importlib.import_module("app") |
| audio_path = tmp_path / "field-note.wav" |
| audio_path.write_bytes(b"fake wav bytes") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| nvidia_api_key="test-nvidia-key", |
| ).validated() |
| seen = {} |
|
|
| class FakeModelClient: |
| def __init__(self, config_arg): |
| seen["config"] = config_arg |
|
|
| def generate_audio_draft(self, audio_file): |
| seen["audio_file"] = audio_file |
| return { |
| "transcript": "Adult reports trouble breathing.", |
| "suggested_fields": [ |
| { |
| "field": "symptoms", |
| "draft_value": "trouble breathing", |
| "source_snippet": "trouble breathing", |
| } |
| ], |
| "missing_or_unclear_fields": ["vitals"], |
| "provisional_red_flag_mentions": ["trouble breathing"], |
| } |
|
|
| monkeypatch.setattr(app, "ModelClient", FakeModelClient) |
|
|
| draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) |
|
|
| assert seen["config"] is config |
| assert seen["audio_file"] == str(audio_path) |
| assert draft["audio_intake_path"] == "omni_native" |
| assert draft["audio_runtime"] == "omni_native" |
| assert draft["audio_model_id"] == OMNI_MODEL_ID |
| assert draft["transcript"] == "Adult reports trouble breathing." |
| assert draft["suggested_fields"][0]["field"] == "symptoms" |
| assert draft["confirmation_status"] == "unconfirmed" |
| assert draft["audio_filename"] == "field-note.wav" |
| assert draft["raw_audio_stored"] is False |
| assert "not written to figment traces" in draft["audio_retention_note"].lower() |
|
|
|
|
| def test_hosted_omni_audio_failure_fails_closed(tmp_path, monkeypatch) -> None: |
| app = importlib.import_module("app") |
| audio_path = tmp_path / "field-note.wav" |
| audio_path.write_bytes(b"fake wav bytes") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| nvidia_api_key="test-nvidia-key", |
| ).validated() |
|
|
| class FailingModelClient: |
| def __init__(self, *_args, **_kwargs): |
| pass |
|
|
| def generate_audio_draft(self, _audio_file): |
| raise app.ModelClientError("endpoint unavailable") |
|
|
| monkeypatch.setattr(app, "ModelClient", FailingModelClient) |
|
|
| draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) |
|
|
| assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" |
| assert draft["audio_runtime"] == "unprocessed_audio" |
| assert draft["suggested_fields"] == [] |
| assert "hosted omni audio draft failed" in draft["processing_status"].lower() |
| assert draft["audio_filename"] == "field-note.wav" |
| assert draft["raw_audio_stored"] is False |
| assert "not written to figment traces" in draft["audio_retention_note"].lower() |
|
|
|
|
| def test_local_parakeet_provider_payload_uses_local_4b_runtime_label_when_gated() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_stack="local_4b_parakeet", |
| model_backend="llama_cpp", |
| enable_audio_intake=True, |
| audio_backend="parakeet_nemo", |
| allow_local_asr=True, |
| ).validated() |
|
|
| draft = app.draft_audio_intake( |
| provider_payload={ |
| "transcript": "Local Parakeet transcript says the patient has trouble breathing.", |
| "suggested_fields": [ |
| { |
| "field": "symptoms", |
| "draft_value": "trouble breathing", |
| "source_snippet": "trouble breathing", |
| } |
| ], |
| }, |
| config=config, |
| ) |
|
|
| assert draft["audio_intake_path"] == "parakeet_rnnt_plus_text_nemotron" |
| assert draft["audio_runtime"] == "local_4b_parakeet" |
| assert draft["audio_model_id"] == "nvidia/parakeet-rnnt-1.1b" |
| assert draft["field_fill_model_id"] == "nvidia/NVIDIA-Nemotron-3-Nano-4B-BF16" |
| assert draft["transcript_source"] == "local_parakeet_asr_provider" |
| assert draft["confirmation_status"] == "unconfirmed" |
|
|
|
|
| def test_typed_transcript_under_parakeet_config_stays_heuristic_not_asr() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_stack="local_4b_parakeet", |
| model_backend="llama_cpp", |
| enable_audio_intake=True, |
| audio_backend="parakeet_nemo", |
| allow_local_asr=True, |
| ).validated() |
|
|
| draft = app.draft_audio_intake( |
| transcript="Typed note says the patient has trouble breathing.", |
| config=config, |
| ) |
|
|
| assert draft["audio_intake_path"] == "typed_transcript_heuristic" |
| assert draft["audio_runtime"] == "typed_transcript_heuristic" |
| assert draft["audio_model_id"] is None |
| assert draft["field_fill_model_id"] is None |
| assert draft["draft_source"] == "typed_transcript_heuristic" |
| assert draft["transcript_source"] == "typed_transcript" |
|
|
|
|
| def test_confirm_intake_does_not_silently_bulk_accept_audio_suggestions() -> None: |
| app = importlib.import_module("app") |
| audio_draft = { |
| "audio_intake_path": "omni_native", |
| "audio_runtime": "omni_native", |
| "transcript": "Adult with chest pain.", |
| "suggested_fields": [ |
| { |
| "field": "chief_concern", |
| "draft_value": "chest pain", |
| "source_snippet": "chest pain", |
| "status": "audio_draft", |
| "needs_confirmation": True, |
| } |
| ], |
| "confirmed_intake_required": True, |
| "confirmation_status": "unconfirmed", |
| "raw_audio_stored": False, |
| } |
|
|
| confirmed, _, confirmed_audio = app._confirm_ui_intake( |
| "mobile clinic", |
| "52", |
| "not_applicable", |
| "", |
| "", |
| "", |
| "unknown", |
| "unknown", |
| "basic kit", |
| "Responder typed note remains source of truth.", |
| audio_draft, |
| ) |
|
|
| assert confirmed["chief_concern"] == "" |
| assert confirmed["responder_note"] == "Responder typed note remains source of truth." |
| assert confirmed_audio["suggested_fields"][0]["status"] == "rejected" |
| assert confirmed_audio["suggested_fields"][0]["needs_confirmation"] is False |
|
|
|
|
| def test_rejected_audio_transcript_is_not_passed_to_navigator_prompt(monkeypatch) -> None: |
| app = importlib.import_module("app") |
| import figment.navigator as navigator |
|
|
| audio_draft = { |
| "audio_intake_path": "omni_native", |
| "audio_runtime": "omni_native", |
| "transcript": "Adult says chest pain from audio only.", |
| "suggested_fields": [ |
| { |
| "field": "chief_concern", |
| "draft_value": "chest pain", |
| "source_snippet": "chest pain", |
| "status": "audio_draft", |
| "needs_confirmation": True, |
| } |
| ], |
| "confirmed_intake_required": True, |
| "confirmation_status": "unconfirmed", |
| "raw_audio_stored": False, |
| } |
| confirmed, _, confirmed_audio = app._confirm_ui_intake( |
| "mobile clinic", |
| "52", |
| "not_applicable", |
| "", |
| "", |
| "", |
| "unknown", |
| "unknown", |
| "basic kit", |
| "Typed note remains the source of truth.", |
| audio_draft, |
| ) |
| captured: dict[str, str] = {} |
|
|
| class CapturePromptModelClient: |
| def __init__(self, *_args, **_kwargs) -> None: |
| pass |
|
|
| def generate_json(self, prompt, context): |
| captured["prompt"] = prompt |
| return navigator.canned_navigator_output( |
| context["intake"], |
| context["rule_results"], |
| context["retrieved_cards"], |
| context["urgency_floor"], |
| ) |
|
|
| monkeypatch.setattr(navigator, "ModelClient", CapturePromptModelClient) |
|
|
| navigator.run_navigation( |
| confirmed, |
| [], |
| audio_draft=confirmed_audio, |
| config=FigmentConfig(model_backend="hosted_omni", nvidia_api_key="unused"), |
| retrieved_cards=[ |
| { |
| "card_id": "SAFETY-BOUNDARIES-v1", |
| "card": {"card_id": "SAFETY-BOUNDARIES-v1", "title": "Safety boundaries"}, |
| } |
| ], |
| ) |
|
|
| assert "Adult says chest pain from audio only." not in captured["prompt"] |
| assert "chest pain" not in captured["prompt"].lower() |
|
|
|
|
| def test_apply_audio_draft_prefills_empty_fields_without_confirming() -> None: |
| app = importlib.import_module("app") |
| audio_draft = { |
| "audio_intake_path": "omni_native", |
| "audio_runtime": "omni_native", |
| "transcript": "Adult with chest pain.", |
| "suggested_fields": [ |
| { |
| "field": "chief_concern", |
| "draft_value": "chest pain", |
| "source_snippet": "chest pain", |
| "status": "audio_draft", |
| "needs_confirmation": True, |
| } |
| ], |
| "confirmed_intake_required": True, |
| "confirmation_status": "unconfirmed", |
| "raw_audio_stored": False, |
| } |
|
|
| *fields, audio_json, audio_state = app._apply_audio_draft_ui( |
| "mobile clinic", |
| "52", |
| "not_applicable", |
| "", |
| "", |
| "", |
| "unknown", |
| "unknown", |
| "basic kit", |
| "", |
| audio_draft, |
| ) |
|
|
| assert fields[3] == "chest pain" |
| assert audio_json["confirmation_status"] == "unconfirmed" |
| assert audio_json["suggested_fields"][0]["status"] == "applied_unreviewed" |
| assert audio_json["suggested_fields"][0]["needs_confirmation"] is True |
| assert audio_state is audio_json |
|
|
| intake = app.collect_intake( |
| fields[0], |
| fields[1], |
| fields[2], |
| fields[3], |
| fields[4], |
| fields[5], |
| fields[6], |
| fields[7], |
| fields[8], |
| fields[9], |
| ) |
| with pytest.raises(ValueError, match="audio-derived intake must be confirmed"): |
| app.confirm_intake(intake, audio_draft=audio_state) |
|
|
|
|
| def test_loading_demo_case_resets_audio_and_downstream_state() -> None: |
| app = importlib.import_module("app") |
|
|
| values = app._load_demo_case_and_reset("Rural clinic: pregnancy danger sign") |
|
|
| assert values[:10] == app._load_demo_case("Rural clinic: pregnancy danger sign") |
| assert values[10] is None |
| assert values[11] == "" |
| assert values[12] is None |
| assert values[13] is None |
| assert values[14] == {"red_flags": [], "protocol_urgency": "routine"} |
| assert "PROTOCOL_URGENCY" in values[15] |
| assert values[16] == [] |
| assert values[17] == "" |
| assert "Selected Protocol Card" in values[18] |
| assert values[19] == {} |
| assert values[20] == "" |
| assert "Protocol Urgency" in values[21] |
| assert values[22] == {} |
| assert values[23] is None |
| assert "Run the navigator" in values[24] |
| assert values[25] == {} |
| assert values[26] is None |
| assert values[27] == {} |
|
|
|
|
| def test_source_and_audio_changes_clear_stale_pipeline_outputs() -> None: |
| app = importlib.import_module("app") |
|
|
| assert app._clear_source_outputs() == [ |
| None, |
| {"red_flags": [], "protocol_urgency": "routine"}, |
| app._risk_summary_html({"red_flags": [], "protocol_urgency": "routine"}), |
| [], |
| "", |
| app._protocol_results_html([]), |
| {}, |
| "", |
| app._navigator_summary_html({}), |
| {}, |
| None, |
| app._trace_audit_html({}), |
| {}, |
| {}, |
| ] |
| assert app._clear_audio_outputs() == [ |
| None, |
| None, |
| {"red_flags": [], "protocol_urgency": "routine"}, |
| app._risk_summary_html({"red_flags": [], "protocol_urgency": "routine"}), |
| [], |
| "", |
| app._protocol_results_html([]), |
| {}, |
| "", |
| app._navigator_summary_html({}), |
| {}, |
| None, |
| app._trace_audit_html({}), |
| {}, |
| None, |
| {}, |
| ] |
|
|
|
|
| def test_uploaded_audio_without_transcript_or_payload_is_not_fabricated_as_omni() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="canned", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| ).validated() |
|
|
| draft = app.draft_audio_intake( |
| transcript="", |
| config=config, |
| audio_file="/tmp/field-note.wav", |
| provider_payload={}, |
| ) |
|
|
| assert draft["audio_file_received"] is True |
| assert draft["audio_filename"] == "field-note.wav" |
| assert draft["transcript"] == "" |
| assert draft["suggested_fields"] == [] |
| assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" |
| assert draft["audio_runtime"] == "unprocessed_audio" |
| assert "needs transcript/model support" in draft["processing_status"].lower() |
|
|
|
|
| def test_canned_audio_backend_is_labeled_as_canned_demo_runtime() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="canned", |
| enable_audio_intake=True, |
| audio_backend="canned", |
| ).validated() |
|
|
| draft = app.draft_audio_intake(transcript="", config=config) |
|
|
| assert draft["audio_runtime"] == "canned" |
| assert draft["audio_intake_path"] == "canned_audio_demo" |
| assert draft["audio_model_id"] is None |
| assert draft["transcript_source"] == "synthetic_demo_manifest" |
| assert draft["transcript"] |
| assert draft["suggested_fields"] |
|
|
|
|
| def test_canned_demo_transcript_matches_committed_manifest_pediatric_age() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="canned", |
| enable_audio_intake=True, |
| audio_backend="canned", |
| ).validated() |
| manifest = json.loads((app.PROJECT_ROOT / "data" / "demo_audio" / "manifest.json").read_text(encoding="utf-8")) |
| pediatric_case = next(item for item in manifest["cases"] if item["slug"] == "pediatric_dehydration") |
|
|
| draft = app.draft_audio_intake(transcript="", config=config, provider_payload=None) |
|
|
| assert draft["transcript"] == pediatric_case["source_script"] |
| assert "Seven year old" in draft["transcript"] |
| assert "Four-year-old" not in draft["transcript"] |
| assert next(item for item in draft["suggested_fields"] if item["field"] == "patient_age")["draft_value"] == "7 years" |
|
|
|
|
| def test_hosted_audio_ui_discloses_endpoint_and_deidentified_limit() -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| nvidia_api_key="test-nvidia-key", |
| ).validated() |
|
|
| subtitle = app._audio_section_subtitle(config).lower() |
| chips = app._audio_runtime_chips_html(config).lower() |
|
|
| assert "sent to the configured hosted endpoint" in subtitle |
| assert "synthetic or de-identified" in subtitle |
| assert "10 mb" in subtitle |
| assert "60 seconds" in subtitle |
| assert "hosted endpoint" in chips |
|
|
|
|
| def test_hosted_audio_oversize_file_fails_closed_before_provider_call(tmp_path, monkeypatch) -> None: |
| app = importlib.import_module("app") |
| import figment.model_client as model_client |
|
|
| audio_path = tmp_path / "too-large.wav" |
| audio_path.write_bytes(b"0123456789") |
| monkeypatch.setattr(model_client, "HOSTED_AUDIO_MAX_BYTES", 8) |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| nvidia_api_key="test-nvidia-key", |
| ).validated() |
|
|
| class FailIfCalledModelClient: |
| def __init__(self, *_args, **_kwargs): |
| pass |
|
|
| def generate_audio_draft(self, _audio_file): |
| raise AssertionError("oversize audio should be rejected before provider call") |
|
|
| monkeypatch.setattr(app, "ModelClient", FailIfCalledModelClient) |
|
|
| draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) |
|
|
| assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" |
| assert "exceeds hosted audio size limit" in draft["processing_status"].lower() |
| assert draft["audio_filename"] == "too-large.wav" |
|
|
|
|
| def test_hosted_audio_long_wav_fails_closed_before_provider_call(tmp_path, monkeypatch) -> None: |
| app = importlib.import_module("app") |
| import figment.model_client as model_client |
|
|
| audio_path = tmp_path / "too-long.wav" |
| with wave.open(str(audio_path), "wb") as wav_file: |
| wav_file.setnchannels(1) |
| wav_file.setsampwidth(2) |
| wav_file.setframerate(8000) |
| wav_file.writeframes(b"\0\0" * 8000) |
| monkeypatch.setattr(model_client, "HOSTED_AUDIO_MAX_SECONDS", 0.01) |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| enable_audio_intake=True, |
| audio_backend="omni_native", |
| nvidia_api_key="test-nvidia-key", |
| ).validated() |
|
|
| class FailIfCalledModelClient: |
| def __init__(self, *_args, **_kwargs): |
| pass |
|
|
| def generate_audio_draft(self, _audio_file): |
| raise AssertionError("long audio should be rejected before provider call") |
|
|
| monkeypatch.setattr(app, "ModelClient", FailIfCalledModelClient) |
|
|
| draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) |
|
|
| assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" |
| assert "exceeds hosted audio duration limit" in draft["processing_status"].lower() |
| assert draft["audio_filename"] == "too-long.wav" |
|
|
|
|
| def test_navigator_ui_uses_supplied_runtime_config(monkeypatch) -> None: |
| app = importlib.import_module("app") |
| config = FigmentConfig( |
| model_backend="hosted_omni", |
| audio_backend="omni_native", |
| enable_audio_intake=True, |
| omni_endpoint_url="http://omni.example.test/v1", |
| ).validated() |
| intake = { |
| "confirmed": True, |
| "setting": "mobile clinic", |
| "chief_concern": "typed chest pain concern", |
| "responder_note": "typed source of truth", |
| } |
| seen = {} |
|
|
| def fake_run_case(intake_arg, config_arg=None, audio_draft=None): |
| seen["config"] = config_arg |
| return { |
| "navigator_output": {"protocol_urgency": "routine"}, |
| "sbar": "handoff", |
| "trace": {"model_route": {"model_backend": config_arg.model_backend}}, |
| } |
|
|
| monkeypatch.setattr(app, "run_case", fake_run_case) |
|
|
| output, sbar, trace, state = app._navigate_ui(intake, None, config=config) |
|
|
| assert seen["config"] is config |
| assert output["protocol_urgency"] == "routine" |
| assert sbar == "handoff" |
| assert trace == state |
|
|