kitos
0fc2427f71
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.
225 lines
7.5 KiB
Python
225 lines
7.5 KiB
Python
"""Tests for POST /tests/{id}/assign — lead/manager manual operator assignment."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.models.test import Test
|
|
from app.models.technique import Technique
|
|
|
|
|
|
def _seed_technique(db, tactic="execution") -> Technique:
|
|
technique = Technique(
|
|
mitre_id="T9999",
|
|
name="Test Technique",
|
|
tactic=tactic,
|
|
platforms=["linux"],
|
|
)
|
|
db.add(technique)
|
|
db.commit()
|
|
db.refresh(technique)
|
|
return technique
|
|
|
|
|
|
def _seed_test(db, technique, created_by) -> Test:
|
|
test = Test(technique_id=technique.id, name="Assignable test", created_by=created_by)
|
|
db.add(test)
|
|
db.commit()
|
|
db.refresh(test)
|
|
return test
|
|
|
|
|
|
def test_red_lead_can_assign_red_tech(client, db, red_lead_headers, red_lead_user, red_tech_user):
|
|
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_tech_assignee": str(red_tech_user.id)},
|
|
headers=red_lead_headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["red_tech_assignee"] == str(red_tech_user.id)
|
|
|
|
db.refresh(test)
|
|
assert test.red_tech_assignee == red_tech_user.id
|
|
|
|
|
|
def test_assign_rejects_wrong_role_for_side(client, db, red_lead_headers, red_lead_user, blue_tech_user):
|
|
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_tech_assignee": str(blue_tech_user.id)},
|
|
headers=red_lead_headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_admin_cannot_be_assigned_as_operator(client, db, red_lead_headers, red_lead_user, admin_user):
|
|
"""Admin administers the site and can't be assigned as an operator either."""
|
|
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_tech_assignee": str(admin_user.id)},
|
|
headers=red_lead_headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_red_tech_cannot_assign(client, db, red_tech_headers, red_tech_user):
|
|
technique = _seed_technique(db)
|
|
test = _seed_test(db, technique, red_tech_user.id)
|
|
|
|
resp = client.post(
|
|
f"/api/v1/tests/{test.id}/assign",
|
|
json={"red_tech_assignee": str(red_tech_user.id)},
|
|
headers=red_tech_headers,
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_cannot_assign(client, db, auth_headers, admin_user, red_tech_user):
|
|
"""Admin administers the site — coordinating operators is a lead/manager call."""
|
|
technique = _seed_technique(db)
|
|
test = _seed_test(db, technique, admin_user.id)
|
|
|
|
resp = client.post(
|
|
f"/api/v1/tests/{test.id}/assign",
|
|
json={"red_tech_assignee": str(red_tech_user.id)},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_manager_can_assign(client, db, manager_headers, manager_user, red_tech_user):
|
|
technique = _seed_technique(db)
|
|
test = _seed_test(db, technique, manager_user.id)
|
|
|
|
resp = client.post(
|
|
f"/api/v1/tests/{test.id}/assign",
|
|
json={"red_tech_assignee": str(red_tech_user.id)},
|
|
headers=manager_headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["red_tech_assignee"] == str(red_tech_user.id)
|
|
|
|
|
|
def test_clearing_assignee_does_not_touch_jira(client, db, red_lead_headers, red_lead_user, red_tech_user):
|
|
"""Sending null explicitly clears the assignee without a Jira sync call."""
|
|
technique = _seed_technique(db)
|
|
test = _seed_test(db, technique, red_lead_user.id)
|
|
test.red_tech_assignee = red_tech_user.id
|
|
db.commit()
|
|
|
|
with patch("app.services.jira_service.push_assignee_update") as mock_push:
|
|
resp = client.post(
|
|
f"/api/v1/tests/{test.id}/assign",
|
|
json={"red_tech_assignee": None},
|
|
headers=red_lead_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["red_tech_assignee"] is None
|
|
mock_push.assert_not_called()
|
|
|
|
|
|
def test_assigning_operator_syncs_to_jira(client, db, red_lead_headers, red_lead_user, red_tech_user):
|
|
technique = _seed_technique(db)
|
|
test = _seed_test(db, technique, red_lead_user.id)
|
|
|
|
# Capture IDs *during* the call, while the endpoint's request-scoped
|
|
# session is still open — reading them afterward hits a
|
|
# DetachedInstanceError once that session closes.
|
|
captured = {}
|
|
|
|
def _capture(_db, test_arg, assignee_arg):
|
|
captured["test_id"] = test_arg.id
|
|
captured["assignee_id"] = assignee_arg.id
|
|
|
|
with patch("app.services.jira_service.push_assignee_update", side_effect=_capture) as mock_push:
|
|
resp = client.post(
|
|
f"/api/v1/tests/{test.id}/assign",
|
|
json={"red_tech_assignee": str(red_tech_user.id)},
|
|
headers=red_lead_headers,
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
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)
|