96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
from datetime import date
|
|
|
|
from mka.application.meeting_service import ParticipantInput
|
|
from mka.application.progress_timing import TimingSnapshot
|
|
from mka.application.run_inputs import RunInputState
|
|
from mka.ui import streamlit_app
|
|
|
|
|
|
def test_alias_input_accepts_lines_and_commas() -> None:
|
|
assert streamlit_app._parse_aliases("Sikirgut\nSekugrid HS, Secugrid H S") == (
|
|
"Sikirgut",
|
|
"Sekugrid HS",
|
|
"Secugrid H S",
|
|
)
|
|
|
|
|
|
def test_regenerated_protocol_widget_value_is_deferred_until_next_run(
|
|
monkeypatch,
|
|
) -> None:
|
|
state = {"edited_protocol": "old protocol"}
|
|
monkeypatch.setattr(streamlit_app.st, "session_state", state)
|
|
|
|
streamlit_app._queue_edited_protocol("regenerated protocol")
|
|
|
|
assert state["edited_protocol"] == "old protocol"
|
|
assert state["pending_edited_protocol"] == "regenerated protocol"
|
|
|
|
streamlit_app._apply_pending_edited_protocol()
|
|
|
|
assert state["edited_protocol"] == "regenerated protocol"
|
|
assert "pending_edited_protocol" not in state
|
|
|
|
|
|
def test_imported_inputs_are_applied_via_pending_state_before_widgets(
|
|
monkeypatch,
|
|
) -> None:
|
|
existing_upload = object()
|
|
state = {"source_media_0": existing_upload, "source_media_widget_generation": 0}
|
|
monkeypatch.setattr(streamlit_app.st, "session_state", state)
|
|
imported = RunInputState(
|
|
title="Imported meeting",
|
|
description="Imported context",
|
|
language="en",
|
|
meeting_date=date(2026, 8, 25),
|
|
participants=(ParticipantInput("martin", "Martin"),),
|
|
audio_normalization=False,
|
|
diarization_enabled=True,
|
|
source_file_name="meeting.wav",
|
|
)
|
|
|
|
streamlit_app._queue_run_inputs(imported)
|
|
|
|
assert "meeting_title" not in state
|
|
assert state["pending_run_inputs"] == imported
|
|
|
|
streamlit_app._apply_pending_run_inputs()
|
|
|
|
assert state["meeting_title"] == "Imported meeting"
|
|
assert state["meeting_description"] == "Imported context"
|
|
assert state["meeting_language"] == "en"
|
|
assert state["meeting_has_date"] is True
|
|
assert state["participants"][0]["participant_id"] == "martin"
|
|
assert state["audio_normalization"] is False
|
|
assert state["diarization_enabled"] is True
|
|
assert state["imported_source_file_name"] == "meeting.wav"
|
|
assert state["source_media_widget_generation"] == 1
|
|
assert "source_media_1" not in state
|
|
assert "pending_run_inputs" not in state
|
|
assert state["source_media_0"] is existing_upload
|
|
|
|
|
|
def test_duration_formatting_covers_seconds_minutes_and_hours() -> None:
|
|
assert streamlit_app._format_duration(8.9) == "00:08"
|
|
assert streamlit_app._format_duration(7 * 60 + 18) == "07:18"
|
|
assert streamlit_app._format_duration(3600 + 3 * 60 + 42) == "1:03:42"
|
|
|
|
|
|
def test_completed_regeneration_timing_survives_frontend_rerun(monkeypatch) -> None:
|
|
state = {}
|
|
monkeypatch.setattr(streamlit_app.st, "session_state", state)
|
|
timing = TimingSnapshot(
|
|
stage_durations={"protocol_generation": 4.5},
|
|
active_stage=None,
|
|
total_duration=4.5,
|
|
running=False,
|
|
)
|
|
states = {stage: "skipped" for stage in streamlit_app.STAGES}
|
|
states["protocol_generation"] = "completed"
|
|
|
|
streamlit_app._remember_regeneration_timing(states, "Protocol regenerated", timing)
|
|
|
|
saved_states, saved_message, saved_timing = state["regeneration_progress"]
|
|
assert saved_states["protocol_generation"] == "completed"
|
|
assert saved_message == "Protocol regenerated"
|
|
assert saved_timing == timing
|