feat(review): review queue for leads + manual reviewer reassignment
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

Leads get the same two-queue view operators already have: 'Available
to Review' (pending reviews currently assigned to a peer lead) and
'My Assigned Reviews' (assigned to me), replacing the old single
'My Reviews' toggle. Since the load-balanced auto-assignment always
picks a reviewer immediately, 'available' means peer-assigned reviews
a lead could pick up if the assignee can't get to them, not unclaimed
ones (there aren't any).

POST /tests/{id}/assign now also accepts red_reviewer_assignee /
blue_reviewer_assignee, validated against the matching lead role and
synced to Jira the same way operator assignment already is. The
AssigneeControl UI gained a 'reviewer' kind (leads-only picker) shown
on the test detail header while a test sits in red_review/blue_review
— this also fixes the reviewer assignment being invisible in Aegis
even though it was already being pushed to Jira correctly.
This commit is contained in:
kitos
2026-07-10 16:19:14 +02:00
parent be2a3fdced
commit 0fc2427f71
8 changed files with 250 additions and 72 deletions
+32 -11
View File
@@ -1280,44 +1280,65 @@ def assign_test_operators(
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role_strict("manager", "red_lead", "blue_lead")),
):
"""Assign red_tech and/or blue_tech operators to a test. Leads/managers only — not admin, who administers the site rather than coordinating operators."""
"""Assign red/blue tech operators and/or reviewers to a test. Leads/managers only — not admin, who administers the site rather than coordinating people."""
test = crud_get_test_or_raise(db, test_id)
newly_assigned: User | None = None
newly_assigned: list[User] = []
if payload.red_tech_assignee is not None:
u = db.query(User).filter(User.id == payload.red_tech_assignee).first()
if not u or u.role not in ("red_tech", "red_lead"):
raise HTTPException(status_code=400, detail="Invalid red tech assignee")
test.red_tech_assignee = payload.red_tech_assignee
newly_assigned = u
newly_assigned.append(u)
if payload.blue_tech_assignee is not None:
u = db.query(User).filter(User.id == payload.blue_tech_assignee).first()
if not u or u.role not in ("blue_tech", "blue_lead"):
raise HTTPException(status_code=400, detail="Invalid blue tech assignee")
test.blue_tech_assignee = payload.blue_tech_assignee
newly_assigned = u
newly_assigned.append(u)
if payload.red_reviewer_assignee is not None:
u = db.query(User).filter(User.id == payload.red_reviewer_assignee).first()
if not u or u.role != "red_lead":
raise HTTPException(status_code=400, detail="Invalid red reviewer — must be a red_lead")
test.red_reviewer_assignee = payload.red_reviewer_assignee
newly_assigned.append(u)
if payload.blue_reviewer_assignee is not None:
u = db.query(User).filter(User.id == payload.blue_reviewer_assignee).first()
if not u or u.role != "blue_lead":
raise HTTPException(status_code=400, detail="Invalid blue reviewer — must be a blue_lead")
test.blue_reviewer_assignee = payload.blue_reviewer_assignee
newly_assigned.append(u)
# Handle intentional null (clearing) — model_fields_set tracks which keys were sent
if "red_tech_assignee" in payload.model_fields_set and payload.red_tech_assignee is None:
test.red_tech_assignee = None
if "blue_tech_assignee" in payload.model_fields_set and payload.blue_tech_assignee is None:
test.blue_tech_assignee = None
if "red_reviewer_assignee" in payload.model_fields_set and payload.red_reviewer_assignee is None:
test.red_reviewer_assignee = None
if "blue_reviewer_assignee" in payload.model_fields_set and payload.blue_reviewer_assignee is None:
test.blue_reviewer_assignee = None
log_action(db, current_user.id, "assign_test", str(test_id), {
"red_tech_assignee": str(payload.red_tech_assignee) if payload.red_tech_assignee else None,
"blue_tech_assignee": str(payload.blue_tech_assignee) if payload.blue_tech_assignee else None,
"red_reviewer_assignee": str(payload.red_reviewer_assignee) if payload.red_reviewer_assignee else None,
"blue_reviewer_assignee": str(payload.blue_reviewer_assignee) if payload.blue_reviewer_assignee else None,
})
db.commit()
db.refresh(test)
if newly_assigned is not None:
try:
from app.services.jira_service import push_assignee_update
push_assignee_update(db, test, newly_assigned)
db.commit()
except Exception: # nosec B110
pass # jira_service already logs warnings internally
if newly_assigned:
from app.services.jira_service import push_assignee_update
for assignee in newly_assigned:
try:
push_assignee_update(db, test, assignee)
db.commit()
except Exception: # nosec B110
pass # jira_service already logs warnings internally
return test
+3 -1
View File
@@ -171,10 +171,12 @@ class TestRemediationUpdate(BaseModel):
class TestAssign(BaseModel):
"""Payload for assigning operators to a test."""
"""Payload for assigning operators or reviewers to a test."""
red_tech_assignee: uuid.UUID | None = None
blue_tech_assignee: uuid.UUID | None = None
red_reviewer_assignee: uuid.UUID | None = None
blue_reviewer_assignee: uuid.UUID | None = None
class TestHold(BaseModel):
+74
View File
@@ -148,3 +148,77 @@ def test_assigning_operator_syncs_to_jira(client, db, red_lead_headers, red_lead
mock_push.assert_called_once()
assert captured["test_id"] == test.id
assert captured["assignee_id"] == red_tech_user.id
def _make_second_red_lead(db):
from app.auth import hash_password
from app.models.user import User
u = User(
username="redlead2", email="redlead2@test.com",
hashed_password=hash_password("x"), role="red_lead", is_active=True,
must_change_password=False,
)
db.add(u)
db.commit()
db.refresh(u)
return u
def test_lead_can_reassign_red_reviewer_to_peer(client, db, red_lead_headers, red_lead_user):
"""A lead can hand a review off to a peer lead — e.g. if the currently
assigned reviewer can't get to it."""
technique = _seed_technique(db)
test = _seed_test(db, technique, red_lead_user.id)
peer = _make_second_red_lead(db)
test.red_reviewer_assignee = red_lead_user.id
db.commit()
resp = client.post(
f"/api/v1/tests/{test.id}/assign",
json={"red_reviewer_assignee": str(peer.id)},
headers=red_lead_headers,
)
assert resp.status_code == 200, resp.text
assert resp.json()["red_reviewer_assignee"] == str(peer.id)
db.refresh(test)
assert test.red_reviewer_assignee == peer.id
def test_reviewer_reassign_rejects_non_lead(client, db, red_lead_headers, red_lead_user, red_tech_user):
"""Only a red_lead can be the red reviewer — a red_tech is not eligible."""
technique = _seed_technique(db)
test = _seed_test(db, technique, red_lead_user.id)
resp = client.post(
f"/api/v1/tests/{test.id}/assign",
json={"red_reviewer_assignee": str(red_tech_user.id)},
headers=red_lead_headers,
)
assert resp.status_code == 400
def test_reviewer_reassign_rejects_wrong_side(client, db, red_lead_headers, red_lead_user, blue_lead_user):
"""A blue_lead can't be set as the red reviewer."""
technique = _seed_technique(db)
test = _seed_test(db, technique, red_lead_user.id)
resp = client.post(
f"/api/v1/tests/{test.id}/assign",
json={"red_reviewer_assignee": str(blue_lead_user.id)},
headers=red_lead_headers,
)
assert resp.status_code == 400
def test_manager_can_reassign_blue_reviewer(client, db, manager_headers, manager_user, blue_lead_user):
technique = _seed_technique(db)
test = _seed_test(db, technique, manager_user.id)
resp = client.post(
f"/api/v1/tests/{test.id}/assign",
json={"blue_reviewer_assignee": str(blue_lead_user.id)},
headers=manager_headers,
)
assert resp.status_code == 200, resp.text
assert resp.json()["blue_reviewer_assignee"] == str(blue_lead_user.id)