Improve speaker mapping UX

tests/test_speaker_mapping_ui.py

@@ -0,0 +1,132 @@
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from streamlit.testing.v1 import AppTest
+
+from mka.ui.streamlit_app import _mapping_counts, _speaker_options
+
+
+@pytest.mark.parametrize("label", ["SPEAKER_00", "SPEAKER_01", "SPEAKER_02"])
+def test_unmapped_options_include_all_participants(label):
+    assert _speaker_options(("a", "b"), {}, label) == [None, "a", "b"]
+
+
+def test_options_reserve_other_assignments_and_preserve_current():
+    mappings = {"SPEAKER_00": "a"}
+    assert _speaker_options(("a", "b"), mappings, "SPEAKER_00") == [None, "a", "b"]
+    for label in ("SPEAKER_01", "SPEAKER_02"):
+        assert _speaker_options(("a", "b"), mappings, label) == [None, "b"]
+    mappings["SPEAKER_00"] = "b"
+    assert _speaker_options(("a", "b"), mappings, "SPEAKER_01") == [None, "a"]
+    mappings["SPEAKER_00"] = None
+    assert _speaker_options(("a", "b"), mappings, "SPEAKER_01") == [None, "a", "b"]
+
+
+def test_existing_duplicate_or_missing_participant_is_not_removed():
+    assert _speaker_options(("b",), {"s0": "a", "s1": "a"}, "s0") == [None, "b", "a"]
+
+
+@pytest.mark.parametrize(
+    ("mappings", "expected"),
+    [
+        ({}, (2, 0, 2)),
+        ({"s0": "a", "other": "b"}, (2, 1, 1)),
+        ({"s0": "a", "s1": "b"}, (2, 2, 0)),
+        ({"s0": None}, (2, 0, 2)),
+    ],
+)
+def test_counts_only_include_detected_speakers(mappings, expected):
+    assert _mapping_counts(("s0", "s1"), mappings) == expected
+
+
+def mapping_app():
+    from mka.application.meeting_service import SpeakerMappingReview, SpeakerReview
+    from mka.ui.streamlit_app import _render_speaker_mapping
+
+    _render_speaker_mapping(
+        SpeakerMappingReview(
+            tuple(SpeakerReview(f"SPEAKER_0{i}", ()) for i in range(3)),
+            (("a", "Participant A"), ("b", "Participant B"), ("c", "Participant C")),
+            {"SPEAKER_00": "a"},
+        ),
+        "test",
+    )
+
+
+def test_widget_reruns_filter_release_and_update_status():
+    app = AppTest.from_function(mapping_app).run()
+    assert not app.exception
+    assert app.selectbox[0].value == "a"
+    assert "Participant A" not in app.selectbox[1].options
+    assert len(app.warning) == 1
+    assert "3 speakers detected · 1 assigned · 2 unassigned" in app.markdown[0].value
+    assert "Unassigned" in app.selectbox[1].label
+
+    app.selectbox[2].select("c").run()
+    assert app.selectbox[0].value == "a"
+    assert "Participant C" not in app.selectbox[0].options
+    app.selectbox[1].select("b").run()
+    assert not app.warning
+    assert "3 assigned · 0 unassigned" in app.markdown[0].value
+    app.selectbox[0].select(None).run()
+    assert app.warning
+    assert "Participant A" in app.selectbox[1].options
+    assert app.selectbox[1].value == "b"
+    app.selectbox[1].select("a").run()
+    assert "Participant B" in app.selectbox[0].options
+    assert "Participant A" not in app.selectbox[0].options
+    assert app.selectbox[0].value is None
+    assert not app.exception
+
+
+def test_anonymous_regeneration_remains_allowed(monkeypatch):
+    from mka.application.meeting_service import SpeakerMappingReview, SpeakerReview
+    from mka.ui import streamlit_app as ui
+
+    outcome = SimpleNamespace(
+        succeeded=True,
+        run_dir=Path("/tmp/mapping-test"),
+        speaker_attribution_available=True,
+        original_protocol="Anonymous protocol",
+    )
+    review = SpeakerMappingReview((SpeakerReview("SPEAKER_00", ()),), (("a", "A"),), {})
+    calls = []
+
+    class Service:
+        def __init__(self, *args):
+            pass
+
+        def load_speaker_mapping_review(self, run_dir):
+            return review
+
+        def regenerate_protocol(self, run_dir, selections, **kwargs):
+            calls.append(selections)
+            return outcome
+
+    monkeypatch.setattr(ui, "MeetingProcessingService", Service)
+    monkeypatch.setattr(ui.AppSettings, "from_environment", lambda: None)
+    monkeypatch.setattr(ui, "MeetingLabGateway", lambda: None)
+    monkeypatch.setattr(ui, "_remember_regeneration_timing", lambda *args: None, raising=False)
+    monkeypatch.setattr(
+        ui,
+        "_run_with_live_progress",
+        lambda service, action, states, message: (action(None), states, message, None),
+        raising=False,
+    )
+
+    def result_app(outcome):
+        import streamlit as st
+
+        from mka.ui.streamlit_app import _render_result
+
+        st.session_state.outcome = outcome
+        _render_result()
+
+    app = AppTest.from_function(result_app, args=(outcome,)).run()
+    regenerate = app.button[1]
+    assert not regenerate.disabled
+    assert app.warning
+    regenerate.click().run()
+    assert calls == [{}]
+    assert not app.exception