Aegis/backend/tests/test_review_gates_router.py

"""HTTP-level tests for the red_review / blue_review lead-review gates.

Covers POST /tests/{id}/review-red and POST /tests/{id}/review-blue: the
assigned-reviewer-only guard, approve/reopen/gap decisions, and the state
transitions they drive. Uses the ``api`` fixture (see conftest.py) for any
step that switches role, since the TestClient cookie jar otherwise bleeds
one role's session into the next request.
"""

import uuid

import pytest

from app.models.evidence import Evidence
from app.models.enums import TeamSide


def _create_technique(api, auth_headers, mitre_id="T1059.099"):
    resp = api(
        "post", "/api/v1/techniques", auth_headers,
        json={"mitre_id": mitre_id, "name": "Command Line"},
    )
    assert resp.status_code == 201, resp.text
    return resp.json()["id"]


def _create_test(api, auth_headers, technique_id, name="Review gate test"):
    resp = api(
        "post", "/api/v1/tests", auth_headers,
        json={"technique_id": technique_id, "name": name},
    )
    assert resp.status_code == 201, resp.text
    return resp.json()["id"]


def _add_evidence(db, test_id, team: TeamSide):
    ev = Evidence(
        test_id=uuid.UUID(test_id),
        file_name="proof.txt",
        file_path="s3://bucket/proof.txt",
        sha256_hash="a" * 64,
        team=team,
    )
    db.add(ev)
    db.commit()


@pytest.fixture
def technique(api, auth_headers):
    return _create_technique(api, auth_headers)


def _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique_id, name="Review gate test"):
    """Drive a fresh test from draft to red_review via real HTTP calls."""
    test_id = _create_test(api, auth_headers, technique_id, name)

    api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
    _add_evidence(db, test_id, TeamSide.red)
    resp = api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
    assert resp.status_code == 200, resp.text
    assert resp.json()["state"] == "red_review"
    return test_id


def _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique_id, name="Review gate test"):
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique_id, name)

    approve = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve"})
    assert approve.status_code == 200, approve.text
    assert approve.json()["state"] == "blue_evaluating"

    api("post", f"/api/v1/tests/{test_id}/start-blue-work", blue_tech_headers)
    _add_evidence(db, test_id, TeamSide.blue)
    resp = api("post", f"/api/v1/tests/{test_id}/submit-blue", blue_tech_headers)
    assert resp.status_code == 200, resp.text
    assert resp.json()["state"] == "blue_review"
    return test_id


# ---------------------------------------------------------------------------
# review-red
# ---------------------------------------------------------------------------


def test_review_red_forbidden_for_non_assigned_lead(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
    """A red_lead who isn't the assigned reviewer gets 403."""
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
    # red_lead_user IS the only red_lead fixture, so it WILL be auto-assigned
    # as reviewer (single-candidate load balancing). To exercise the 403
    # path we need a second, non-assigned red_lead.
    from app.auth import hash_password
    from app.models.user import User

    other_lead = User(
        username="otherredlead", email="otherredlead@test.com",
        hashed_password=hash_password("x"), role="red_lead", is_active=True,
        must_change_password=False,
    )
    db.add(other_lead)
    db.commit()

    login = client.post("/api/v1/auth/login", data={"username": "otherredlead", "password": "x"})
    assert login.status_code == 200
    other_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

    resp = api("post", f"/api/v1/tests/{test_id}/review-red", other_headers, json={"decision": "approve"})
    assert resp.status_code == 403


def test_review_red_approve_moves_to_blue_evaluating(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve", "notes": "LGTM"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["state"] == "blue_evaluating"
    assert body["red_review_notes"] == "LGTM"
    assert body["blue_started_at"] is not None


def test_review_red_reopen_requires_notes(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen"})
    assert resp.status_code == 400


def test_review_red_reopen_moves_to_red_executing(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen", "notes": "add more detail"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["state"] == "red_executing"
    assert body["red_review_notes"] == "add more detail"


def test_review_red_admin_can_always_act(
    client, db, api, auth_headers, red_tech_headers, red_lead_user, technique
):
    test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-red", auth_headers, json={"decision": "approve"})
    assert resp.status_code == 200, resp.text
    assert resp.json()["state"] == "blue_evaluating"


# ---------------------------------------------------------------------------
# review-blue
# ---------------------------------------------------------------------------


def test_review_blue_forbidden_for_non_assigned_lead(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    from app.auth import hash_password
    from app.models.user import User

    other_lead = User(
        username="otherbluelead", email="otherbluelead@test.com",
        hashed_password=hash_password("x"), role="blue_lead", is_active=True,
        must_change_password=False,
    )
    db.add(other_lead)
    db.commit()

    login = client.post("/api/v1/auth/login", data={"username": "otherbluelead", "password": "x"})
    other_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

    resp = api("post", f"/api/v1/tests/{test_id}/review-blue", other_headers, json={"decision": "approve"})
    assert resp.status_code == 403


def test_review_blue_approve_moves_to_in_review(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "approve"})
    assert resp.status_code == 200, resp.text
    assert resp.json()["state"] == "in_review"


def test_review_blue_reopen_requires_notes(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen"})
    assert resp.status_code == 400


def test_review_blue_reopen_moves_to_blue_evaluating(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen", "notes": "redo detection"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["state"] == "blue_evaluating"
    assert body["blue_work_started_at"] is None


def test_review_blue_gap_requires_system_gaps(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "gap"})
    assert resp.status_code == 400


def test_review_blue_gap_moves_to_in_review_with_system_gaps(
    client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
    red_lead_user, blue_lead_user, technique,
):
    test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)

    resp = api(
        "post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers,
        json={"decision": "gap", "system_gaps": "Missing EDR agent on host X"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["state"] == "in_review"
    assert body["system_gaps"] == "Missing EDR agent on host X"