006c4c2c5f
- Simulation model with full field set (redteam + SOC sides) and cascade delete - Alembic migration 0002 for simulations table - simulation_workflow service: PATCH RBAC field-level + auto-transition pending→in_progress + state machine - mitre service: STIX bundle loader (boot-safe) + ranked search (exact-id > prefix-id > name) - 7 new API endpoints: list/create/get/patch/delete simulations, transition, MITRE autocomplete - serialize_simulation added to serializers.py - Makefile update-mitre target with real curl + optional docker restart - Dockerfile updated to copy backend/data/ into image - MITRE enterprise-attack.json bundle committed (~45 MB) - 67 new tests (total 130 passing), ruff clean, mypy introduces no new errors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
237 lines
7.7 KiB
Python
237 lines
7.7 KiB
Python
"""Simulation CRUD tests: create, list, get, delete, cascade."""
|
|
from __future__ import annotations
|
|
|
|
from flask.testing import FlaskClient
|
|
|
|
from backend.app.extensions import db
|
|
from backend.app.models import User
|
|
from backend.app.models.simulation import Simulation
|
|
from backend.tests.conftest import auth_headers as _h
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_engagement(client: FlaskClient, token: str) -> dict:
|
|
resp = client.post(
|
|
"/api/engagements",
|
|
headers=_h(token),
|
|
json={"name": "Op Alpha", "start_date": "2026-06-01"},
|
|
)
|
|
assert resp.status_code == 201, resp.get_json()
|
|
return resp.get_json()
|
|
|
|
|
|
def _make_sim(client: FlaskClient, token: str, eid: int, **kw) -> dict:
|
|
payload = {"name": "Sim 1", **kw}
|
|
resp = client.post(
|
|
f"/api/engagements/{eid}/simulations", headers=_h(token), json=payload
|
|
)
|
|
assert resp.status_code == 201, resp.get_json()
|
|
return resp.get_json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_simulation_as_redteam(
|
|
client: FlaskClient, redteam_user: User, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
body = _make_sim(client, redteam_token, eng["id"])
|
|
assert body["name"] == "Sim 1"
|
|
assert body["status"] == "pending"
|
|
assert body["engagement_id"] == eng["id"]
|
|
assert body["created_by"] == {"id": redteam_user.id, "username": "redteam1"}
|
|
|
|
|
|
def test_create_simulation_as_admin(
|
|
client: FlaskClient, admin_user: User, admin_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, admin_token)
|
|
body = _make_sim(client, admin_token, eng["id"])
|
|
assert body["created_by"]["username"] == "admin1"
|
|
|
|
|
|
def test_create_simulation_soc_forbidden(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
resp = client.post(
|
|
f"/api/engagements/{eng['id']}/simulations",
|
|
headers=_h(soc_token),
|
|
json={"name": "x"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_create_simulation_unauth(client: FlaskClient, redteam_token: str) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
resp = client.post(
|
|
f"/api/engagements/{eng['id']}/simulations", json={"name": "x"}
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_create_simulation_missing_name(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
resp = client.post(
|
|
f"/api/engagements/{eng['id']}/simulations",
|
|
headers=_h(redteam_token),
|
|
json={"name": ""},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_create_simulation_engagement_not_found(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
resp = client.post(
|
|
"/api/engagements/9999/simulations",
|
|
headers=_h(redteam_token),
|
|
json={"name": "x"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_simulations_empty(client: FlaskClient, redteam_token: str) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
resp = client.get(
|
|
f"/api/engagements/{eng['id']}/simulations", headers=_h(redteam_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == []
|
|
|
|
|
|
def test_list_simulations_ordered_desc(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
_make_sim(client, redteam_token, eng["id"], name="First")
|
|
_make_sim(client, redteam_token, eng["id"], name="Second")
|
|
resp = client.get(
|
|
f"/api/engagements/{eng['id']}/simulations", headers=_h(redteam_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
items = resp.get_json()
|
|
assert len(items) == 2
|
|
# Most recent first
|
|
assert items[0]["name"] == "Second"
|
|
assert items[1]["name"] == "First"
|
|
|
|
|
|
def test_list_simulations_soc_can_read(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
_make_sim(client, redteam_token, eng["id"])
|
|
resp = client.get(
|
|
f"/api/engagements/{eng['id']}/simulations", headers=_h(soc_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
assert len(resp.get_json()) == 1
|
|
|
|
|
|
def test_list_simulations_engagement_not_found(
|
|
client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
resp = client.get(
|
|
"/api/engagements/9999/simulations", headers=_h(redteam_token)
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Get
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_simulation_ok(client: FlaskClient, redteam_token: str) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
resp = client.get(f"/api/simulations/{sim['id']}", headers=_h(redteam_token))
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["id"] == sim["id"]
|
|
|
|
|
|
def test_get_simulation_404(client: FlaskClient, redteam_token: str) -> None:
|
|
resp = client.get("/api/simulations/9999", headers=_h(redteam_token))
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_get_simulation_soc_can_read(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
resp = client.get(f"/api/simulations/{sim['id']}", headers=_h(soc_token))
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_delete_simulation_redteam(client: FlaskClient, redteam_token: str) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
resp = client.delete(
|
|
f"/api/simulations/{sim['id']}", headers=_h(redteam_token)
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
|
|
def test_delete_simulation_admin(
|
|
client: FlaskClient, redteam_token: str, admin_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
resp = client.delete(f"/api/simulations/{sim['id']}", headers=_h(admin_token))
|
|
assert resp.status_code == 204
|
|
|
|
|
|
def test_delete_simulation_soc_forbidden(
|
|
client: FlaskClient, redteam_token: str, soc_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
resp = client.delete(f"/api/simulations/{sim['id']}", headers=_h(soc_token))
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_delete_simulation_404(client: FlaskClient, redteam_token: str) -> None:
|
|
resp = client.delete("/api/simulations/9999", headers=_h(redteam_token))
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cascade delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cascade_delete_engagement_removes_simulations(
|
|
app, client: FlaskClient, redteam_token: str
|
|
) -> None:
|
|
eng = _make_engagement(client, redteam_token)
|
|
sim = _make_sim(client, redteam_token, eng["id"])
|
|
sim_id = sim["id"]
|
|
|
|
resp = client.delete(
|
|
f"/api/engagements/{eng['id']}", headers=_h(redteam_token)
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
with app.app_context():
|
|
assert db.session.get(Simulation, sim_id) is None
|