kitos
58c0143304
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
Admin administers the site — it no longer has any override on
validate-red/blue, review-red/blue, resolve-dispute, request-discussion,
reopen, hold, resume, assign-operators, or classification-update.
Introduces require_any_role_strict(), a role dependency without the
global admin bypass, so these specific endpoints truly exclude admin
instead of only removing it from the (redundant) role tuple.
Managers gain the ability to assign red_tech/blue_tech operators
(POST /tests/{id}/assign, GET /users/operators) alongside leads, since
that's coordination, not resolving a ticket.
Also enlarges and repositions the operator-assignment controls next
to the Start Execution button, and fixes a literal '\u2014' rendering
as text instead of an em dash in the test detail technique line.
138 lines
4.6 KiB
Python
138 lines
4.6 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_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
|