89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""Exercise the real executor and Streamlit reruns without model calls."""
|
|
|
|
from pathlib import Path
|
|
from threading import current_thread
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import streamlit
|
|
from streamlit.testing.v1 import AppTest
|
|
|
|
from mka.application.meeting_service import AppProgressEvent, SpeakerMappingReview, SpeakerReview
|
|
from mka.application.progress_timing import ProcessingTimer
|
|
from mka.ui import streamlit_app as ui
|
|
|
|
|
|
class GuardedStreamlit:
|
|
"""Fail if a worker touches the UI session proxy, including callback reads."""
|
|
|
|
def __getattr__(self, name):
|
|
if name == "session_state" and current_thread().name.startswith("ThreadPoolExecutor"):
|
|
raise AssertionError("Worker accessed Streamlit session state")
|
|
return getattr(streamlit, name)
|
|
|
|
|
|
@pytest.mark.parametrize("fails", [False, True])
|
|
@pytest.mark.parametrize("mapped", [False, True])
|
|
def test_regeneration_captures_ui_values_before_real_worker_and_retains_timing(
|
|
monkeypatch, fails, mapped
|
|
):
|
|
outcome = SimpleNamespace(
|
|
succeeded=True,
|
|
run_dir=Path("/tmp/alpha-mapping-test"),
|
|
speaker_attribution_available=True,
|
|
original_protocol="Existing protocol",
|
|
)
|
|
review = SpeakerMappingReview((SpeakerReview("SPEAKER_00", ()),), (("a", "A"),), {})
|
|
calls = []
|
|
|
|
class Service:
|
|
def __init__(self, *args):
|
|
self.timer = ProcessingTimer()
|
|
|
|
def load_speaker_mapping_review(self, run_dir):
|
|
return review
|
|
|
|
def regenerate_protocol(self, run_dir, selections, *, progress_sink, performance_profile):
|
|
assert current_thread().name.startswith("ThreadPoolExecutor")
|
|
calls.append((run_dir, selections, performance_profile))
|
|
self.timer.start_run()
|
|
self.timer.start_stage("protocol_generation")
|
|
progress_sink(AppProgressEvent("protocol_generation", "started", 0.0))
|
|
if fails:
|
|
raise RuntimeError("model unavailable")
|
|
self.timer.finish_stage("protocol_generation")
|
|
self.timer.finish_run()
|
|
progress_sink(AppProgressEvent("protocol_generation", "completed", 0.0))
|
|
return outcome
|
|
|
|
monkeypatch.setattr(ui, "st", GuardedStreamlit())
|
|
monkeypatch.setattr(ui, "MeetingProcessingService", Service)
|
|
monkeypatch.setattr(ui.AppSettings, "from_environment", lambda: None)
|
|
monkeypatch.setattr(ui, "MeetingLabGateway", lambda: None)
|
|
|
|
def result_app(outcome):
|
|
import streamlit as st
|
|
|
|
from mka.ui.streamlit_app import _render_result
|
|
|
|
st.session_state.outcome = outcome
|
|
st.session_state.performance_profile = "fast"
|
|
_render_result()
|
|
|
|
app = AppTest.from_function(result_app, args=(outcome,)).run()
|
|
if mapped:
|
|
app.selectbox[0].select("a").run()
|
|
button = next(button for button in app.button if "Regenerate" in button.label)
|
|
assert not button.disabled
|
|
button.click().run()
|
|
assert not app.exception
|
|
assert calls == [(outcome.run_dir, {"SPEAKER_00": "a"} if mapped else {}, "fast")]
|
|
states, message, timing = app.session_state["processing_progress"]
|
|
assert states["protocol_generation"] == ("failed" if fails else "completed")
|
|
assert not timing.running
|
|
assert timing.total_duration >= timing.stage_durations["protocol_generation"] >= 0
|
|
app.run()
|
|
assert not app.exception
|
|
assert app.session_state["processing_progress"][2] == timing
|
|
assert any("total runtime" in item.value for item in app.info)
|