Aegis/backend/tests/test_assign_operators.py

"""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