feat(mcp): add test-result disposition tools

Add update_test_result and bulk_update_test_results to disposition test
results (confirm, dismiss, mute, or clear), gated on the disposition
permission and reusing the shared disposition service. Adds the
parse_test_result_disposition and resolve_test_result helpers, surfaces
test_result_id in list_test_results, and registers both tools.

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

Author: luis-dk

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()

tests/unit/mcp/test_tools_common.py

@@ -772,3 +772,65 @@ def test_parse_sql_flavor_invalid_lists_display_values():
     msg = str(exc.value)
     for member in SqlFlavorLabel:
         assert member.value in msg
+
+
+# --- parse_test_result_disposition ---
+
+
+@pytest.mark.parametrize(
+    "user_label,expected",
+    [
+        ("Confirmed", Disposition.CONFIRMED),
+        ("Dismissed", Disposition.DISMISSED),
+        ("Muted", Disposition.INACTIVE),
+        ("No Decision", None),
+    ],
+)
+def test_parse_test_result_disposition_user_labels(user_label, expected):
+    from testgen.mcp.tools.common import parse_test_result_disposition
+
+    assert parse_test_result_disposition(user_label) is expected
+
+
+def test_parse_test_result_disposition_rejects_unknown_and_lists_accepted():
+    from testgen.mcp.tools.common import parse_test_result_disposition
+
+    with pytest.raises(MCPUserError) as exc:
+        parse_test_result_disposition("Inactive")  # DB value, not user-facing
+    msg = str(exc.value)
+    for label in ("Confirmed", "Dismissed", "Muted", "No Decision"):
+        assert label in msg
+
+
+# --- resolve_test_result ---
+
+
+@patch("testgen.mcp.tools.common.get_project_permissions")
+@patch("testgen.mcp.tools.common.get_current_session")
+def test_resolve_test_result_happy_path(mock_session, mock_perms, db_session_mock):
+    from testgen.mcp.tools.common import resolve_test_result
+
+    result = MagicMock()
+    mock_session.return_value.scalars.return_value.first.return_value = result
+    mock_perms.return_value = _mock_perms(allowed_projects=("demo",))
+
+    assert resolve_test_result(str(uuid4())) is result
+
+
+@patch("testgen.mcp.tools.common.get_project_permissions")
+@patch("testgen.mcp.tools.common.get_current_session")
+def test_resolve_test_result_missing_or_inaccessible(mock_session, mock_perms, db_session_mock):
+    from testgen.mcp.tools.common import resolve_test_result
+
+    mock_session.return_value.scalars.return_value.first.return_value = None
+    mock_perms.return_value = _mock_perms(allowed_projects=("demo",))
+
+    with pytest.raises(MCPResourceNotAccessible, match=r"Test result .* not found or not accessible"):
+        resolve_test_result(str(uuid4()))
+
+
+def test_resolve_test_result_invalid_uuid():
+    from testgen.mcp.tools.common import resolve_test_result
+
+    with pytest.raises(MCPUserError, match="Invalid test_result_id"):
+        resolve_test_result("not-a-uuid")