fix(tests): pause reopen timer, lock edits to assignee, restore Jira operator on reopen
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

- reopen_red_review starts the timer paused (paused_at set) instead of
  immediately running, since the operator hasn't resumed working yet —
  blue already did this via blue_work_started_at, red needed the same
  behavior via the existing pause/resume mechanism.
- update_test_red/update_test_blue now lock edits to the actual assigned
  operator for leads too, not just techs — a lead could previously edit
  another operator's in-progress test. Mirrored in TeamTabs.tsx.
- push_test_event reassigns the Jira ticket to the original operator
  (not the reopening lead) when a test returns to red_executing or
  blue_evaluating for rework, instead of leaving it on the lead.
This commit is contained in:
kitos
2026-07-13 10:53:10 +02:00
parent a3f86c7b31
commit 0bb34f6834
8 changed files with 279 additions and 21 deletions
+4
View File
@@ -414,10 +414,14 @@ class TestEntity:
Called when the assigned Red Lead sends the work back for rework.
Resets the red-phase timer for a fresh attempt (the first attempt's
worklog was already recorded at submit time, so this loses nothing).
Starts the timer PAUSED — the operator hasn't resumed working yet,
just because the lead reopened it. It only starts counting once they
explicitly resume it, same as coming back from a hold.
"""
self._transition(TestState.red_executing)
self.red_started_at = datetime.utcnow()
self.red_paused_seconds = 0
self.paused_at = self.red_started_at
self._events.append(DomainEvent("red_review_reopened"))
# Define function submit_blue_evidence
+6 -4
View File
@@ -672,11 +672,13 @@ def update_test_red(
Returns:
TestOut: The updated test with refreshed red-team field values.
"""
# Assignee lock: red_tech cannot work a test assigned to someone else
# Assignee lock: only the operator actually assigned to this test may
# edit it — applies to red_lead too, not just red_tech, since a lead
# editing another operator's in-progress test is exactly as wrong as
# a different tech doing it.
_pre_test = crud_get_test_or_raise(db, test_id)
if (
_pre_test.red_tech_assignee is not None
and current_user.role == "red_tech"
and _pre_test.red_tech_assignee != current_user.id
):
raise HTTPException(status_code=403, detail="Test is assigned to another operator")
@@ -738,11 +740,11 @@ def update_test_blue(
Returns:
TestOut: The updated test with refreshed blue-team field values.
"""
# Assignee lock: blue_tech cannot work a test assigned to someone else
# Assignee lock: only the operator actually assigned to this test may
# edit it — applies to blue_lead too, not just blue_tech.
_pre_test = crud_get_test_or_raise(db, test_id)
if (
_pre_test.blue_tech_assignee is not None
and current_user.role == "blue_tech"
and _pre_test.blue_tech_assignee != current_user.id
):
raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+29 -8
View File
@@ -828,9 +828,12 @@ def push_test_event(
link.jira_issue_key, target_status, exc_t,
)
# When the operator starts execution: assign the ticket to that operator.
# When the operator starts execution: assign the ticket to that
# operator. On reopen (rework), *actor* is the lead who reopened
# it, not the operator who should keep working it — the caller
# passes the original red_tech_assignee as *assignee* in that case.
if new_state == "red_executing":
jira_account_id = getattr(actor, "jira_account_id", None)
jira_account_id = getattr(assignee or actor, "jira_account_id", None)
if jira_account_id:
try:
jira.assign_issue(link.jira_issue_key, account_id=jira_account_id)
@@ -859,12 +862,30 @@ def push_test_event(
link.jira_issue_key, jira_account_id, exc_a,
)
# Both are hand-off points to a different team with no single owner
# yet: blue_evaluating is queued for Blue Team (whoever reviewed the
# red side is done with it), and in_review is dual-validation by both
# leads at once. Leaving the previous assignee (red/blue reviewer)
# stuck on the ticket is misleading, so clear it.
if new_state in ("blue_evaluating", "in_review"):
# blue_evaluating is reached two different ways: a fresh hand-off
# from Red (no blue_tech_assignee yet — queued, no owner) or a
# reopen/rework of a test some blue_tech already had (assignee is
# still set) — that operator should get the ticket back, not have
# it clear. in_review is always dual-validation by both leads at
# once, so it never has a single owner.
if new_state == "blue_evaluating":
if test.blue_tech_assignee:
reassignee = db.query(User).filter(User.id == test.blue_tech_assignee).first()
jira_account_id = getattr(reassignee, "jira_account_id", None)
else:
jira_account_id = None
try:
jira.assign_issue(link.jira_issue_key, account_id=jira_account_id)
logger.info(
"%s Jira ticket %s (state=%s)",
"Reassigned" if jira_account_id else "Unassigned", link.jira_issue_key, new_state,
)
except Exception as exc_a:
logger.warning(
"Could not update assignee on %s for state %s: %s",
link.jira_issue_key, new_state, exc_a,
)
elif new_state == "in_review":
try:
jira.assign_issue(link.jira_issue_key, account_id=None)
logger.info("Unassigned Jira ticket %s (state=%s)", link.jira_issue_key, new_state)
@@ -410,7 +410,14 @@ def reopen_red_review(db: Session, test: Test, user: User, notes: str) -> Test:
try:
from app.services.jira_service import push_test_event
push_test_event(db, test, user, "red_executing", extra={"notes": notes.strip()})
original_operator = (
db.query(User).filter(User.id == test.red_tech_assignee).first()
if test.red_tech_assignee else None
)
push_test_event(
db, test, user, "red_executing",
extra={"notes": notes.strip()}, assignee=original_operator,
)
except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
+95 -6
View File
@@ -65,6 +65,8 @@ def _make_test(**overrides):
t.blue_validation_status = None
t.red_validation_notes = None
t.blue_validation_notes = None
t.red_tech_assignee = None
t.blue_tech_assignee = None
for k, v in overrides.items():
setattr(t, k, v)
return t
@@ -149,14 +151,12 @@ def test_push_test_event_maps_every_state_to_jira_status(
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, expected_status)
@pytest.mark.parametrize("new_state", ["blue_evaluating", "in_review"])
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_test_event_clears_assignee_on_handoff(mock_get_client, mock_configured, db, new_state):
def test_push_test_event_clears_assignee_on_in_review(mock_get_client, mock_configured, db):
"""Regression: the previous reviewer's assignment must not linger on the
ticket after a hand-off to a different team / dual-validation stage —
e.g. approving red_review left the red_lead reviewer stuck as assignee
all through blue_evaluating."""
ticket once dual-validation starts — both leads act independently, so
there's no single owner."""
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
@@ -164,11 +164,100 @@ def test_push_test_event_clears_assignee_on_handoff(mock_get_client, mock_config
db.add(link)
db.commit()
jira_service.push_test_event(db, test, MagicMock(username="alice", jira_account_id=None), new_state)
jira_service.push_test_event(db, test, MagicMock(username="alice", jira_account_id=None), "in_review")
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id=None)
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_test_event_clears_assignee_on_fresh_blue_queue_entry(mock_get_client, mock_configured, db):
"""A test reaching blue_evaluating for the first time (red just approved)
has no blue_tech_assignee yet — queued, no owner, must not stay stuck on
the red_lead who approved it."""
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test() # blue_tech_assignee defaults to None
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_test_event(db, test, MagicMock(username="alice", jira_account_id=None), "blue_evaluating")
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id=None)
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_test_event_reassigns_blue_operator_on_reopen(mock_get_client, mock_configured, db):
"""Regression: reopening blue_review (rework) sends the test back to
blue_evaluating, but the SAME operator is still assigned in Aegis
(blue_tech_assignee isn't cleared, only blue_work_started_at is) — Jira
must give them the ticket back, not leave it on the blue_lead who
reopened it or clear it as if nobody owned it."""
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
from app.models.user import User
blue_tech = User(
username="bluetech-reopen", email="bluetech-reopen@test.com",
hashed_password="x", role="blue_tech", jira_account_id="blue-tech-account",
)
db.add(blue_tech)
db.commit()
test = _make_test(blue_tech_assignee=blue_tech.id)
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_test_event(
db, test, MagicMock(username="bluelead", jira_account_id=None), "blue_evaluating",
)
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id="blue-tech-account")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_test_event_red_executing_assigns_actor_by_default(mock_get_client, mock_configured, db):
"""The normal start-execution path: no explicit assignee param, so the
acting operator (actor) is who gets the ticket."""
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_test_event(
db, test, MagicMock(username="redtech", jira_account_id="red-tech-account"), "red_executing",
)
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id="red-tech-account")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_test_event_red_executing_reassigns_original_operator_on_reopen(mock_get_client, mock_configured, db):
"""Regression: reopening red_review sends the test back to
red_executing, but *actor* here is the red_lead who reopened it, not
the operator who should keep working it. The caller passes the
original operator as the explicit assignee, which must win over actor."""
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
original_operator = MagicMock(username="redtech", jira_account_id="red-tech-account")
reopening_lead = MagicMock(username="redlead", jira_account_id="red-lead-account")
jira_service.push_test_event(
db, test, reopening_lead, "red_executing", assignee=original_operator,
)
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id="red-tech-account")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_bt_work_started_sets_in_progress(mock_get_client, mock_configured, db):
@@ -0,0 +1,117 @@
"""Only the operator actually assigned to a test may edit its fields —
not just any red_tech/blue_tech, and not the lead either, unless the lead
is themselves the assignee. Reproduces the bug report: a red_lead could
edit a test assigned to a different operator (jesus-huertas)."""
from app.models.enums import TestState
from app.models.test import Test
from app.models.technique import Technique
def _seed_technique(db) -> Technique:
technique = Technique(
mitre_id="T9998",
name="Assignee Lock Test Technique",
tactic="execution",
platforms=["linux"],
)
db.add(technique)
db.commit()
db.refresh(technique)
return technique
def _seed_test(db, technique, created_by, **overrides) -> Test:
test = Test(technique_id=technique.id, name="Locked test", created_by=created_by, **overrides)
db.add(test)
db.commit()
db.refresh(test)
return test
def test_red_lead_cannot_edit_test_assigned_to_other_operator(
client, db, red_lead_headers, red_lead_user, red_tech_user
):
technique = _seed_technique(db)
test = _seed_test(
db, technique, red_lead_user.id,
state=TestState.red_executing, red_tech_assignee=red_tech_user.id,
)
resp = client.patch(
f"/api/v1/tests/{test.id}/red",
json={"procedure_text": "changed by the wrong lead"},
headers=red_lead_headers,
)
assert resp.status_code == 403
assert "assigned to another operator" in resp.json()["detail"]
def test_different_red_tech_cannot_edit_test_assigned_to_other_operator(
client, db, red_lead_headers, red_lead_user, red_tech_user, red_tech_headers
):
technique = _seed_technique(db)
# Assigned to red_lead_user (leads can be operators too via /assign) so
# a *different* red_tech is the one locked out here.
test = _seed_test(
db, technique, red_lead_user.id,
state=TestState.red_executing, red_tech_assignee=red_lead_user.id,
)
resp = client.patch(
f"/api/v1/tests/{test.id}/red",
json={"procedure_text": "changed by a different tech"},
headers=red_tech_headers,
)
assert resp.status_code == 403
def test_assigned_red_tech_can_edit_own_test(
client, db, red_tech_headers, red_tech_user, red_lead_user
):
technique = _seed_technique(db)
test = _seed_test(
db, technique, red_lead_user.id,
state=TestState.red_executing, red_tech_assignee=red_tech_user.id,
)
resp = client.patch(
f"/api/v1/tests/{test.id}/red",
json={"procedure_text": "changed by the assigned operator"},
headers=red_tech_headers,
)
assert resp.status_code == 200, resp.text
def test_blue_lead_cannot_edit_test_assigned_to_other_operator(
client, db, blue_lead_headers, blue_lead_user, blue_tech_user
):
technique = _seed_technique(db)
test = _seed_test(
db, technique, blue_lead_user.id,
state=TestState.blue_evaluating, blue_tech_assignee=blue_tech_user.id,
)
resp = client.patch(
f"/api/v1/tests/{test.id}/blue",
json={"detection_result": "detected"},
headers=blue_lead_headers,
)
assert resp.status_code == 403
assert "assigned to another operator" in resp.json()["detail"]
def test_unassigned_test_is_editable_by_any_eligible_role(
client, db, red_lead_headers, red_lead_user
):
"""No assignee yet (e.g. still in draft/newly executing) — the lock
only kicks in once an operator has actually been assigned."""
technique = _seed_technique(db)
test = _seed_test(db, technique, red_lead_user.id, state=TestState.red_executing)
resp = client.patch(
f"/api/v1/tests/{test.id}/red",
json={"procedure_text": "no assignee set yet"},
headers=red_lead_headers,
)
assert resp.status_code == 200, resp.text
+9
View File
@@ -246,6 +246,15 @@ def test_reopen_red_review_moves_to_red_executing():
assert any(ev.name == "red_review_reopened" for ev in e.events)
def test_reopen_red_review_starts_timer_paused():
"""The timer must not run until the operator explicitly resumes it —
otherwise Tempo worklogs and operation-time metrics count time the
lead spent reviewing, not time the operator spent working."""
e = _entity("red_review")
e.reopen_red_review()
assert e.paused_at == e.red_started_at
def test_reopen_red_review_wrong_state():
e = _entity("blue_evaluating")
with pytest.raises(InvalidStateTransition):