feat(tests): let managers escalate and directly resolve disputed tests
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled

Request Discussion only ever nudged the peer lead, and the one-time
manager notification when a dispute starts had no real follow-up
action attached to it — a manager could be told a dispute existed but
had no way to actually do anything about it. Adds:

- POST /tests/{id}/escalate-to-manager — either lead can explicitly
  ask managers to step in, distinct from the passive initial notice.
- POST /tests/{id}/manager-resolve-dispute — a manager decides the
  final outcome (validated/rejected) directly, bypassing both leads'
  votes entirely, reusing the same dual-validation event dispatch so
  Jira/notifications behave like any other resolution.

Both leads are notified of the manager's final call, win or lose.
This commit is contained in:
kitos
2026-07-15 13:03:01 +02:00
parent 32f4fd25bd
commit 3a01facd46
9 changed files with 444 additions and 1 deletions
+74
View File
@@ -65,6 +65,7 @@ from app.schemas.test import (
TestClassificationUpdate,
TestCreate,
TestHold,
TestManagerResolveDispute,
TestOut,
TestRedReview,
TestRedUpdate,
@@ -152,6 +153,7 @@ from app.services.test_workflow_service import (
validate_as_red_lead as wf_validate_red,
validate_as_blue_lead as wf_validate_blue,
resolve_dispute as wf_resolve_dispute,
resolve_dispute_by_manager as wf_resolve_dispute_by_manager,
reopen_test as wf_reopen,
handle_remediation_completed as wf_handle_remediation,
get_retest_chain as wf_get_retest_chain,
@@ -1192,6 +1194,78 @@ def resolve_dispute(
return test
# ---------------------------------------------------------------------------
# POST /tests/{id}/escalate-to-manager — disputed: notify managers to step in
# ---------------------------------------------------------------------------
@router.post("/{test_id}/escalate-to-manager")
def escalate_to_manager(
test_id: uuid.UUID,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
):
"""Either lead on a disputed test can escalate it for a manager to decide.
Unlike the automatic manager notification sent when a test first becomes
disputed, this is an explicit, repeatable nudge — a lead pulls this when
peer discussion (Request Discussion) isn't going anywhere.
"""
from app.services.notification_service import notify_role_with_email
test = crud_get_test_or_raise(db, test_id)
if test.state.value != "disputed":
from app.domain.errors import BusinessRuleViolation
raise BusinessRuleViolation("Test is not in disputed state")
try:
notify_role_with_email(
db,
role="manager",
type="validation_disputed",
title="Dispute escalated — needs your decision",
message=(
f'{current_user.username} escalated test "{test.name}" — the leads could not '
f'resolve their disagreement. Please review and decide the final outcome.'
),
entity_type="test",
entity_id=test.id,
)
except Exception as e:
import logging
logging.getLogger(__name__).warning("Failed to notify managers of escalation: %s", e)
log_action(
db, user_id=current_user.id, action="escalate_dispute_to_manager",
entity_type="test", entity_id=test.id, details={"test_name": test.name},
)
db.commit()
return {"status": "escalated", "message": "Managers have been notified"}
# ---------------------------------------------------------------------------
# POST /tests/{id}/manager-resolve-dispute — manager makes the final call
# ---------------------------------------------------------------------------
@router.post("/{test_id}/manager-resolve-dispute", response_model=TestOut)
def manager_resolve_dispute(
test_id: uuid.UUID,
payload: TestManagerResolveDispute,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role_strict("manager")),
) -> TestOut:
"""A manager decides the final outcome of a disputed test directly."""
test = crud_get_test_with_technique(db, test_id)
with UnitOfWork(db) as uow:
test = wf_resolve_dispute_by_manager(db, test, current_user, payload.outcome, notes=payload.notes)
uow.commit()
db.refresh(test)
return test
# ---------------------------------------------------------------------------
# POST /tests/{id}/reopen — rejected → draft
# ---------------------------------------------------------------------------