feat(assign): let leads manually assign operators + sync Jira; send platform as Jira label
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

Adds GET /users/operators (leads+admin) and wires a lead-only assign
control into the test detail header, calling the existing but
previously unreachable POST /tests/{id}/assign endpoint. Manual
assignment now also pushes the Jira ticket assignee immediately
instead of waiting for the operator to start execution.

Also adds test.platform as a kebab-case label on Jira ticket creation.
This commit is contained in:
kitos
2026-07-10 08:44:14 +02:00
parent 119db6f91d
commit 2afb886cd9
12 changed files with 487 additions and 4 deletions
+111
View File
@@ -0,0 +1,111 @@
"""Tests for POST /tests/{id}/assign — lead/admin 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_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
+83 -1
View File
@@ -383,6 +383,44 @@ def test_auto_create_test_issue_maps_data_classification_to_real_jira_field(
assert fields["customfield_11814"] == {"value": expected_jira_value}
@pytest.mark.parametrize(
"platform,expected_label",
[
("Windows", "windows"),
("linux", "linux"),
("macOS", "macos"),
("Azure AD", "azure-ad"),
(None, None),
],
)
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_jira_project_key", return_value="SEC")
@patch("app.services.jira_service.get_jira_parent_ticket_standalone", return_value=None)
@patch("app.services.jira_service.get_admin_jira_client")
def test_auto_create_test_issue_adds_platform_label(
mock_get_client, mock_parent, mock_project_key, mock_configured, db,
platform, expected_label,
):
mock_jira = MagicMock()
mock_jira.issue_create.return_value = {"key": "SEC-1", "id": "1"}
mock_get_client.return_value = mock_jira
from app.models.user import User
test = _make_test_for_issue(platform=platform)
technique = _make_technique()
actor = User(username="gerardo-ruiz", email="gerardo.ruiz@kaseya.com", hashed_password="x")
db.add(actor)
db.commit()
jira_service.auto_create_test_issue(db, test, actor, technique=technique)
labels = mock_jira.issue_create.call_args.kwargs["fields"]["labels"]
assert ["aegis", "security-test", "T1003-006"] == labels[:3]
if expected_label:
assert expected_label in labels
else:
assert len(labels) == 3
def _make_user(**overrides):
from app.models.user import User
u = MagicMock(spec=User)
@@ -428,4 +466,48 @@ def test_lookup_user_jira_account_id_no_match(mock_get_client, mock_configured,
updated = jira_service.lookup_user_jira_account_id(db, user)
assert updated is False
assert user.jira_account_id is 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_assignee_update_assigns_jira_issue(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
assignee = _make_user(jira_account_id="account-42")
jira_service.push_assignee_update(db, test, assignee)
mock_jira.assign_issue.assert_called_once_with(link.jira_issue_key, account_id="account-42")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_assignee_update_noop_without_jira_account_id(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
assignee = _make_user(jira_account_id=None)
jira_service.push_assignee_update(db, test, assignee)
mock_jira.assign_issue.assert_not_called()
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_assignee_update_noop_without_link(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
assignee = _make_user(jira_account_id="account-42")
jira_service.push_assignee_update(db, test, assignee)
mock_jira.assign_issue.assert_not_called()
+33
View File
@@ -0,0 +1,33 @@
"""Tests for GET /users/operators — lead/admin-reachable operator picker list."""
def test_red_lead_can_list_operators(client, red_lead_headers, red_tech_user, blue_tech_user, red_lead_user):
resp = client.get("/api/v1/users/operators", headers=red_lead_headers)
assert resp.status_code == 200, resp.text
usernames = {u["username"] for u in resp.json()}
assert "redtech" in usernames
assert "bluetech" in usernames
assert "redlead" in usernames
def test_operators_list_excludes_viewers(client, red_lead_headers, db):
from app.auth import hash_password
from app.models.user import User
viewer = User(
username="viewer_ops", email="viewer_ops@test.com",
hashed_password=hash_password("x"), role="viewer", is_active=True,
must_change_password=False,
)
db.add(viewer)
db.commit()
resp = client.get("/api/v1/users/operators", headers=red_lead_headers)
assert resp.status_code == 200
usernames = {u["username"] for u in resp.json()}
assert "viewer_ops" not in usernames
def test_red_tech_cannot_list_operators(client, red_tech_headers):
resp = client.get("/api/v1/users/operators", headers=red_tech_headers)
assert resp.status_code == 403