Merge branch 'mcp-testresult-disposition' into 'enterprise'

feat(mcp): add test-result disposition tools (TG-1088)

See merge request dkinternal/testgen/dataops-testgen!535

Author: ci bot

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/mcp/server.py

@@ -220,12 +220,14 @@ def build_mcp_server(
         validate_custom_test,
     )
     from testgen.mcp.tools.test_results import (
+        bulk_update_test_results,
         compare_test_runs,
         get_failure_summary,
         get_failure_trend,
         list_test_result_history,
         list_test_results,
         search_test_results,
+        update_test_result,
     )
     from testgen.mcp.tools.test_runs import get_test_run, list_test_runs
 
@@ -266,6 +268,8 @@ def safe_prompt(fn):
     safe_tool(search_test_results)
     safe_tool(get_failure_trend)
     safe_tool(compare_test_runs)
+    safe_tool(update_test_result)
+    safe_tool(bulk_update_test_results)
     safe_tool(get_test_type)
     safe_tool(get_source_data)
     safe_tool(get_source_data_query)

testgen/mcp/tools/common.py

@@ -30,7 +30,7 @@
 from testgen.common.models.scores import ScoreCategory, ScoreDefinition
 from testgen.common.models.table_group import TableGroup
 from testgen.common.models.test_definition import TestDefinition, TestDefinitionNote, TestType
-from testgen.common.models.test_result import TestResultStatus
+from testgen.common.models.test_result import TestResult, TestResultStatus
 from testgen.common.models.test_suite import TestSuite
 from testgen.mcp.exceptions import MCPResourceNotAccessible, MCPUserError
 from testgen.mcp.permissions import get_project_permissions
@@ -353,6 +353,25 @@ def parse_disposition(value: str) -> Disposition:
     return db_value
 
 
+_NO_DECISION = "No Decision"
+
+
+def parse_test_result_disposition(value: str) -> Disposition | None:
+    """Validate a user-facing test-result disposition and return the stored value.
+
+    Accepts ``Confirmed``, ``Dismissed``, ``Muted``, and ``No Decision``. ``Muted``
+    maps to ``Disposition.INACTIVE``; ``No Decision`` clears the disposition (returns
+    ``None`` → NULL).
+    """
+    if value == _NO_DECISION:
+        return None
+    db_value = _DISPOSITION_USER_TO_DB.get(value)
+    if db_value is None:
+        valid = ", ".join([*_DISPOSITION_USER_TO_DB, _NO_DECISION])
+        raise MCPUserError(f"Invalid disposition `{value}`. Valid values: {valid}")
+    return db_value
+
+
 def format_disposition(value: Disposition | str) -> str:
     """Map a stored disposition to its user-facing label (``INACTIVE`` → "Muted")."""
     try:
@@ -623,6 +642,28 @@ def resolve_test_definition(test_definition_id: str) -> TestDefinition:
     return td
 
 
+def resolve_test_result(test_result_id: str) -> TestResult:
+    """Resolve a test result ID to the live ORM model, collapsing missing-or-inaccessible.
+
+    Filters monitor suites and project access via the result's parent test suite.
+    """
+    result_uuid = parse_uuid(test_result_id, "test_result_id")
+    perms = get_project_permissions()
+    query = (
+        select(TestResult)
+        .join(TestSuite, TestResult.test_suite_id == TestSuite.id)
+        .where(
+            TestResult.id == result_uuid,
+            TestSuite.is_monitor.isnot(True),
+            TestSuite.project_code.in_(perms.allowed_codes),
+        )
+    )
+    result = get_current_session().scalars(query).first()
+    if result is None:
+        raise MCPResourceNotAccessible("Test result", test_result_id)
+    return result
+
+
 def resolve_test_note(test_note_id: str) -> TestDefinitionNote:
     """Resolve a test note ID to the live ORM model, collapsing missing-or-inaccessible.
 

testgen/mcp/tools/test_results.py

@@ -1,13 +1,19 @@
 from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
+from sqlalchemy import select
+
 from testgen.common.enums import JobStatus
 from testgen.common.models import get_current_session, with_database_session
 from testgen.common.models.job_execution import JobExecution
 from testgen.common.models.test_definition import TestType
 from testgen.common.models.test_result import BucketInterval, TestResult, TestResultStatus
 from testgen.common.models.test_run import TestRun, TestRunSummary
 from testgen.common.models.test_suite import TestSuite
+from testgen.common.test_result_disposition_service import (
+    DispositionUpdate,
+    set_test_results_disposition,
+)
 from testgen.mcp.exceptions import MCPResourceNotAccessible, MCPUserError
 from testgen.mcp.permissions import get_project_permissions, mcp_permission
 from testgen.mcp.tools.common import (
@@ -18,8 +24,11 @@
     parse_failure_group_by,
     parse_result_status,
     parse_since_arg,
+    parse_test_result_disposition,
     parse_uuid,
     resolve_aggregate_scope,
+    resolve_test_result,
+    resolve_test_suite,
     resolve_test_type,
     validate_limit,
     validate_page,
@@ -134,6 +143,7 @@ def list_test_results(
             doc.heading(2, f"[{status_str}] {test_name} on `{r.column_names}` in `{r.table_name}`")
         else:
             doc.heading(2, f"[{status_str}] {test_name} on `{r.table_name}`")
+        doc.field("Test result", r.id, code=True)
         doc.field("Test definition", r.test_definition_id, code=True)
         if r.column_names:
             doc.field("Column", r.column_names, code=True)
@@ -636,3 +646,116 @@ def _section(title: str, rows: list) -> None:
     _section("Removed Tests", diff.removed_tests)
 
     return doc.render()
+
+
+@with_database_session
+@mcp_permission("disposition")
+def update_test_result(test_result_id: str, disposition: str) -> str:
+    """Set the disposition on a single test result (confirm, dismiss, mute, or clear).
+
+    Args:
+        test_result_id: UUID of the test result, e.g. from ``list_test_results``.
+        disposition: New disposition. One of 'Confirmed', 'Dismissed', 'Muted',
+            'No Decision' (clears it). 'Muted' deactivates the parent test and locks
+            it against auto-regeneration; any other value reactivates and unlocks it.
+    """
+    result = resolve_test_result(test_result_id)
+    db_disposition = parse_test_result_disposition(disposition)
+
+    update: DispositionUpdate = set_test_results_disposition([result.id], db_disposition)
+
+    doc = MdDoc()
+    if update.matched == 0:
+        doc.text(
+            f"Test result {MdDoc.code(test_result_id)} was not dispositioned — disposition does "
+            f"not apply to passed results. No change made."
+        )
+        return doc.render()
+
+    doc.text(f"Updated test result {MdDoc.code(test_result_id)} disposition to **{disposition}**.")
+    return doc.render()
+
+
+@with_database_session
+@mcp_permission("disposition")
+def bulk_update_test_results(
+    test_suite_id: str,
+    disposition: str,
+    job_execution_id: str | None = None,
+    table_name: str | None = None,
+    test_type: str | None = None,
+    status: str | None = None,
+    test_definition_id: str | None = None,
+) -> str:
+    """Set the disposition on every matching test result in a suite (confirm, dismiss, mute, clear).
+
+    Args:
+        test_suite_id: UUID of the test suite, e.g. from ``list_test_suites``.
+        disposition: New disposition. One of 'Confirmed', 'Dismissed', 'Muted',
+            'No Decision' (clears it). 'Muted' deactivates the parent tests and locks
+            them against auto-regeneration; any other value reactivates and unlocks.
+        job_execution_id: UUID of a test run within the suite. Defaults to the suite's
+            latest completed run when omitted.
+        table_name: Optional table-name filter. Case-sensitive.
+        test_type: Optional test type name (e.g. 'Alpha Truncation').
+        status: Optional result-status filter (Passed, Failed, Warning, Error, Log).
+        test_definition_id: Optional single test-definition filter.
+    """
+    suite = resolve_test_suite(test_suite_id)
+    db_disposition = parse_test_result_disposition(disposition)
+
+    if job_execution_id:
+        run = TestRun.get_by_id_or_job(parse_uuid(job_execution_id, "job_execution_id"))
+        if run is None or run.test_suite_id != suite.id:
+            raise MCPResourceNotAccessible("Test run", job_execution_id)
+    else:
+        run = (
+            TestRun.get_by_id_or_job(suite.last_complete_test_run_id)
+            if suite.last_complete_test_run_id
+            else None
+        )
+        if run is None:
+            raise MCPUserError(f"No completed test runs found for test suite `{test_suite_id}`.")
+
+    clauses = [TestResult.test_suite_id == suite.id, TestResult.test_run_id == run.id]
+    if status:
+        clauses.append(TestResult.status == parse_result_status(status))
+    if table_name:
+        clauses.append(TestResult.table_name == table_name)
+    if test_type:
+        clauses.append(TestResult.test_type == resolve_test_type(test_type))
+    if test_definition_id:
+        clauses.append(TestResult.test_definition_id == parse_uuid(test_definition_id, "test_definition_id"))
+
+    result_ids = list(get_current_session().scalars(select(TestResult.id).where(*clauses)).all())
+    update = set_test_results_disposition(result_ids, db_disposition)
+
+    filters = []
+    if table_name:
+        filters.append(f"table_name=`{table_name}`")
+    if test_type:
+        filters.append(f"test_type=`{test_type}`")
+    if status:
+        filters.append(f"status=`{status}`")
+    if test_definition_id:
+        filters.append(f"test_definition_id=`{test_definition_id}`")
+    filter_str = ", ".join(filters) if filters else "no filter"
+
+    doc = MdDoc()
+    if update.matched == 0 and update.passed_skipped == 0:
+        doc.heading(1, "No test results matched")
+        doc.text(
+            f"No test results in suite `{suite.test_suite}` matched the filter ({filter_str}). Nothing changed."
+        )
+        return doc.render()
+
+    doc.heading(1, f"Updated {update.matched} test results in suite `{suite.test_suite}`")
+    doc.field("Disposition", disposition)
+    doc.field("Run", run.job_execution_id, code=True)
+    doc.field("Filter", filter_str)
+    if update.passed_skipped:
+        doc.text(
+            f"Left {update.passed_skipped} passed results unchanged — disposition is not "
+            f"applied to passed results."
+        )
+    return doc.render()