refactor: extract reusable test-result disposition service

Move test-result disposition logic — the disposition write, the parent
test-definition active/lock coupling, and the Passed-row exclusion — out
of the test_results UI view into a Streamlit-free service under common/
so it can be reused outside the UI. The view now delegates to it;
behavior is unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: luis-dk

testgen/common/test_result_disposition_service.py

@@ -0,0 +1,101 @@
+"""Shared test-result disposition service.
+
+Sets the disposition on test results and keeps the parent test definition's
+active/lock state coupled: a "Muted" (``Disposition.INACTIVE``) disposition
+deactivates the test definition and locks it against auto-regeneration; any
+other value — or clearing the disposition — reactivates and unlocks it. Passed
+test results are never dispositioned. Used by both the Streamlit UI
+(test_results page) and the MCP tools. Must not import Streamlit.
+"""
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from uuid import UUID
+
+from sqlalchemy import func, select, update
+
+from testgen.common.enums import Disposition
+from testgen.common.models import get_current_session
+from testgen.common.models.test_definition import TestDefinition
+from testgen.common.models.test_result import TestResult, TestResultStatus
+
+
+@dataclass(frozen=True)
+class DispositionUpdate:
+    """Outcome of a disposition write.
+
+    ``matched`` is the number of non-Passed results updated; ``passed_skipped`` is
+    the number of Passed results that matched but were left unchanged.
+    """
+
+    matched: int
+    passed_skipped: int
+
+
+def coupled_test_definition_state(disposition: Disposition | None) -> tuple[bool, bool]:
+    """Return the parent test definition's ``(test_active, lock_refresh)`` for a disposition.
+
+    Muted (``INACTIVE``) → ``(False, True)``: deactivate and lock against
+    auto-regeneration. Any other value, or cleared (``None``) → ``(True, False)``.
+    """
+    deactivate = disposition == Disposition.INACTIVE
+    return (not deactivate, deactivate)
+
+
+def coerce_ui_disposition(value: str | None) -> Disposition | None:
+    """Map a UI disposition string to the stored value.
+
+    The UI passes ``Confirmed`` / ``Dismissed`` / ``Inactive``, or ``No Decision`` /
+    empty / ``None`` to clear. Unknown values raise ``ValueError`` (caller's bug).
+    """
+    if value in (None, "", "No Decision"):
+        return None
+    return Disposition(value)
+
+
+def set_test_results_disposition(
+    test_result_ids: Sequence[str | UUID],
+    disposition: Disposition | None,
+) -> DispositionUpdate:
+    """Set ``disposition`` on the given results and couple their parent test definitions.
+
+    Passed results are excluded from the write. ``disposition=None`` clears it (NULL).
+    Returns the matched (non-Passed, updated) and passed-skipped counts.
+    """
+    ids = [UUID(str(rid)) for rid in test_result_ids]
+    if not ids:
+        return DispositionUpdate(matched=0, passed_skipped=0)
+
+    session = get_current_session()
+
+    passed_skipped = session.scalar(
+        select(func.count())
+        .select_from(TestResult)
+        .where(TestResult.id.in_(ids), TestResult.status == TestResultStatus.Passed)
+    ) or 0
+
+    # NULL result_status rows (e.g. training-mode results) are excluded by `!= Passed`,
+    # matching the prior UI behavior — disposition only applies to evaluated results.
+    non_passed = (TestResult.id.in_(ids), TestResult.status != TestResultStatus.Passed)
+
+    tr_stmt = (
+        update(TestResult)
+        .where(*non_passed)
+        .values(disposition=disposition.value if disposition is not None else None)
+    )
+    matched = session.execute(tr_stmt).rowcount
+
+    test_active, lock_refresh = coupled_test_definition_state(disposition)
+    affected_td_ids = select(TestResult.test_definition_id).where(*non_passed)
+    td_stmt = (
+        update(TestDefinition)
+        .where(TestDefinition.id.in_(affected_td_ids))
+        .values(
+            test_active=test_active,
+            lock_refresh=lock_refresh,
+            last_manual_update=datetime.now(UTC).replace(tzinfo=None),
+        )
+    )
+    session.execute(td_stmt)
+
+    return DispositionUpdate(matched=matched, passed_skipped=passed_skipped)

testgen/ui/views/test_results.py

@@ -13,6 +13,10 @@
 from testgen.common.models.test_definition import TestDefinition, TestDefinitionNote, TestDefinitionSummary
 from testgen.common.models.test_suite import TestSuiteMinimal
 from testgen.common.pii_masking import get_pii_columns, mask_profiling_pii
+from testgen.common.test_result_disposition_service import (
+    coerce_ui_disposition,
+    set_test_results_disposition,
+)
 from testgen.ui.components import widgets as testgen
 from testgen.ui.components.widgets.download_dialog import (
     FILE_DATA_TYPE,
@@ -31,7 +35,7 @@
     get_test_issue_source_query,
     get_test_issue_source_query_custom,
 )
-from testgen.ui.services.database_service import execute_db_query, fetch_df_from_db, fetch_one_from_db
+from testgen.ui.services.database_service import fetch_df_from_db, fetch_one_from_db
 from testgen.ui.services.query_cache import (
     get_table_group_minimal,
     get_test_definition,
@@ -911,43 +915,4 @@ def update_result_disposition(
     test_result_ids: list[str],
     disposition: str,
 ) -> None:
-    execute_db_query(
-        """
-        WITH selects
-            AS (SELECT UNNEST(ARRAY [:test_result_ids]) AS selected_id)
-        UPDATE test_results
-        SET disposition = NULLIF(:disposition, 'No Decision')
-        FROM test_results r
-        INNER JOIN selects s
-            ON (r.id = s.selected_id::UUID)
-        WHERE r.id = test_results.id
-            AND r.result_status != 'Passed';
-        """,
-        {
-            "test_result_ids": test_result_ids,
-            "disposition": disposition,
-        },
-    )
-
-    execute_db_query(
-        """
-        WITH selects
-            AS (SELECT UNNEST(ARRAY [:test_result_ids]) AS selected_id)
-        UPDATE test_definitions
-        SET test_active = :test_active,
-            last_manual_update = CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
-            lock_refresh = :lock_refresh
-        FROM test_definitions d
-        INNER JOIN test_results r
-            ON (d.id = r.test_definition_id)
-        INNER JOIN selects s
-            ON (r.id = s.selected_id::UUID)
-        WHERE d.id = test_definitions.id
-            AND r.result_status != 'Passed';
-        """,
-        {
-            "test_result_ids": test_result_ids,
-            "test_active": "N" if disposition == "Inactive" else "Y",
-            "lock_refresh": "Y" if disposition == "Inactive" else "N",
-        },
-    )
+    set_test_results_disposition(test_result_ids, coerce_ui_disposition(disposition))

tests/unit/common/test_test_result_disposition_service.py

@@ -0,0 +1,83 @@
+from unittest.mock import MagicMock, patch
+from uuid import uuid4
+
+import pytest
+
+from testgen.common.enums import Disposition
+from testgen.common.test_result_disposition_service import (
+    DispositionUpdate,
+    coerce_ui_disposition,
+    coupled_test_definition_state,
+    set_test_results_disposition,
+)
+
+pytestmark = pytest.mark.unit
+
+MODULE = "testgen.common.test_result_disposition_service"
+
+
+class Test_coupled_test_definition_state:
+    def test_muted_deactivates_and_locks(self):
+        # (test_active, lock_refresh)
+        assert coupled_test_definition_state(Disposition.INACTIVE) == (False, True)
+
+    @pytest.mark.parametrize("disposition", [Disposition.CONFIRMED, Disposition.DISMISSED, None])
+    def test_other_values_reactivate_and_unlock(self, disposition):
+        assert coupled_test_definition_state(disposition) == (True, False)
+
+
+class Test_coerce_ui_disposition:
+    @pytest.mark.parametrize(
+        "value,expected",
+        [
+            ("Confirmed", Disposition.CONFIRMED),
+            ("Dismissed", Disposition.DISMISSED),
+            ("Inactive", Disposition.INACTIVE),
+            ("No Decision", None),
+            (None, None),
+            ("", None),
+        ],
+    )
+    def test_maps_ui_string_to_stored_value(self, value, expected):
+        assert coerce_ui_disposition(value) is expected
+
+
+class Test_set_test_results_disposition:
+    def test_empty_ids_is_noop(self):
+        with patch(f"{MODULE}.get_current_session") as get_session:
+            result = set_test_results_disposition([], Disposition.CONFIRMED)
+        assert result == DispositionUpdate(matched=0, passed_skipped=0)
+        get_session.assert_not_called()
+
+    def test_returns_matched_and_passed_counts(self):
+        session = MagicMock()
+        # First call: COUNT of passed rows -> 2. Then the TR update returns rowcount 3.
+        session.scalar.return_value = 2
+        tr_update_result = MagicMock(rowcount=3)
+        session.execute.return_value = tr_update_result
+        with patch(f"{MODULE}.get_current_session", return_value=session):
+            result = set_test_results_disposition([uuid4(), uuid4()], Disposition.DISMISSED)
+        assert result == DispositionUpdate(matched=3, passed_skipped=2)
+        # One TR update + one TD update.
+        assert session.execute.call_count == 2
+
+    def test_muted_couples_td_to_inactive(self):
+        session = MagicMock()
+        session.scalar.return_value = 0
+        session.execute.return_value = MagicMock(rowcount=1)
+        with patch(f"{MODULE}.get_current_session", return_value=session):
+            set_test_results_disposition([uuid4()], Disposition.INACTIVE)
+        td_stmt = session.execute.call_args_list[1].args[0]
+        params = td_stmt.compile().params
+        # YNString stores 'N'/'Y'; bound values are the Python bools before type processing.
+        assert params["test_active"] is False
+        assert params["lock_refresh"] is True
+
+    def test_clear_passes_null_disposition(self):
+        session = MagicMock()
+        session.scalar.return_value = 0
+        session.execute.return_value = MagicMock(rowcount=1)
+        with patch(f"{MODULE}.get_current_session", return_value=session):
+            set_test_results_disposition([uuid4()], None)
+        tr_stmt = session.execute.call_args_list[0].args[0]
+        assert tr_stmt.compile().params["disposition"] is None