feat(backend): sprint 5 — SimulationTemplate CRUD + instantiation
- SimulationTemplate model + migration 0005 (CREATE TABLE + name index) - 5 CRUD endpoints under /api/templates (admin|redteam only, SOC 403) - POST /api/engagements/<eid>/simulations extended with optional template_id - serialize_template() reusing _enrich_techniques/_enrich_tactics helpers - IntegrityError → 409 for duplicate name on both POST and PATCH - 28 new tests (CRUD, RBAC, dedup, instantiation, migration round-trip) - 221 tests pass; ruff clean; mypy clean Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Knacky
239
backend/tests/test_simulation_templates_crud.py
Normal file
239
backend/tests/test_simulation_templates_crud.py
Normal file
@@ -0,0 +1,239 @@
|
||||
"""SimulationTemplate CRUD: list, create, get, patch, delete + RBAC + dedup."""
|
||||
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_template import SimulationTemplate
|
||||
from backend.tests.conftest import auth_headers as _h # noqa: E402
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_template(client: FlaskClient, token: str, **kw) -> dict:
|
||||
payload = {"name": "Template Alpha", **kw}
|
||||
resp = client.post("/api/templates", headers=_h(token), json=payload)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# List
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_list_templates_empty(client: FlaskClient, admin_token: str) -> None:
|
||||
resp = client.get("/api/templates", headers=_h(admin_token))
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json() == []
|
||||
|
||||
|
||||
def test_list_templates_soc_forbidden(client: FlaskClient, soc_token: str) -> None:
|
||||
resp = client.get("/api/templates", headers=_h(soc_token))
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_list_templates_unauthenticated(client: FlaskClient) -> None:
|
||||
resp = client.get("/api/templates")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Create
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_create_template_as_admin(
|
||||
client: FlaskClient, admin_user: User, admin_token: str
|
||||
) -> None:
|
||||
body = _make_template(
|
||||
client,
|
||||
admin_token,
|
||||
description="desc",
|
||||
commands="cmd",
|
||||
prerequisites="prereq",
|
||||
)
|
||||
assert body["name"] == "Template Alpha"
|
||||
assert body["description"] == "desc"
|
||||
assert body["commands"] == "cmd"
|
||||
assert body["prerequisites"] == "prereq"
|
||||
assert body["techniques"] == []
|
||||
assert body["tactics"] == []
|
||||
assert body["created_by"] == {"id": admin_user.id, "username": "admin1"}
|
||||
assert body["id"] is not None
|
||||
|
||||
|
||||
def test_create_template_as_redteam(
|
||||
client: FlaskClient, redteam_user: User, redteam_token: str
|
||||
) -> None:
|
||||
body = _make_template(client, redteam_token)
|
||||
assert body["created_by"]["username"] == "redteam1"
|
||||
|
||||
|
||||
def test_create_template_soc_forbidden(client: FlaskClient, soc_token: str) -> None:
|
||||
resp = client.post(
|
||||
"/api/templates", headers=_h(soc_token), json={"name": "T"}
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_create_template_missing_name(client: FlaskClient, admin_token: str) -> None:
|
||||
resp = client.post("/api/templates", headers=_h(admin_token), json={})
|
||||
assert resp.status_code == 400
|
||||
assert "name" in resp.get_json()["error"]
|
||||
|
||||
|
||||
def test_create_template_duplicate_name_409(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
_make_template(client, admin_token)
|
||||
resp = client.post(
|
||||
"/api/templates", headers=_h(admin_token), json={"name": "Template Alpha"}
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
assert "already exists" in resp.get_json()["error"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Get single
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_template(client: FlaskClient, admin_token: str) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.get(f"/api/templates/{created['id']}", headers=_h(admin_token))
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json()["id"] == created["id"]
|
||||
|
||||
|
||||
def test_get_template_not_found(client: FlaskClient, admin_token: str) -> None:
|
||||
resp = client.get("/api/templates/9999", headers=_h(admin_token))
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_get_template_soc_forbidden(
|
||||
client: FlaskClient, admin_token: str, soc_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.get(f"/api/templates/{created['id']}", headers=_h(soc_token))
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Patch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_patch_template_name(client: FlaskClient, admin_token: str) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.patch(
|
||||
f"/api/templates/{created['id']}",
|
||||
headers=_h(admin_token),
|
||||
json={"name": "Renamed"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json()["name"] == "Renamed"
|
||||
assert resp.get_json()["updated_at"] is not None
|
||||
|
||||
|
||||
def test_patch_template_empty_name_rejected(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.patch(
|
||||
f"/api/templates/{created['id']}",
|
||||
headers=_h(admin_token),
|
||||
json={"name": ""},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_patch_template_unknown_field_rejected(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.patch(
|
||||
f"/api/templates/{created['id']}",
|
||||
headers=_h(admin_token),
|
||||
json={"bogus_field": "x"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "unknown fields" in resp.get_json()["error"]
|
||||
|
||||
|
||||
def test_patch_template_duplicate_name_409(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
_make_template(client, admin_token, name="T1")
|
||||
t2 = _make_template(client, admin_token, name="T2")
|
||||
resp = client.patch(
|
||||
f"/api/templates/{t2['id']}",
|
||||
headers=_h(admin_token),
|
||||
json={"name": "T1"},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
def test_patch_template_soc_forbidden(
|
||||
client: FlaskClient, admin_token: str, soc_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.patch(
|
||||
f"/api/templates/{created['id']}",
|
||||
headers=_h(soc_token),
|
||||
json={"name": "X"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_patch_template_not_found(client: FlaskClient, admin_token: str) -> None:
|
||||
resp = client.patch(
|
||||
"/api/templates/9999", headers=_h(admin_token), json={"name": "X"}
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Delete
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_delete_template(
|
||||
client: FlaskClient, app, admin_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.delete(f"/api/templates/{created['id']}", headers=_h(admin_token))
|
||||
assert resp.status_code == 204
|
||||
with app.app_context():
|
||||
assert db.session.get(SimulationTemplate, created["id"]) is None
|
||||
|
||||
|
||||
def test_delete_template_not_found(client: FlaskClient, admin_token: str) -> None:
|
||||
resp = client.delete("/api/templates/9999", headers=_h(admin_token))
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_delete_template_soc_forbidden(
|
||||
client: FlaskClient, admin_token: str, soc_token: str
|
||||
) -> None:
|
||||
created = _make_template(client, admin_token)
|
||||
resp = client.delete(f"/api/templates/{created['id']}", headers=_h(soc_token))
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# List returns ordered by name
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_list_templates_ordered_by_name(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
for name in ("Zebra", "Alpha", "Midpoint"):
|
||||
_make_template(client, admin_token, name=name)
|
||||
body = client.get("/api/templates", headers=_h(admin_token)).get_json()
|
||||
names = [t["name"] for t in body]
|
||||
assert names == sorted(names)
|
||||
195
backend/tests/test_simulations_from_template.py
Normal file
195
backend/tests/test_simulations_from_template.py
Normal file
@@ -0,0 +1,195 @@
|
||||
"""Tests for creating simulations from a template (POST /api/engagements/<eid>/simulations)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from alembic.operations import Operations
|
||||
from alembic.runtime.migration import MigrationContext
|
||||
from flask.testing import FlaskClient
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
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 Bravo", "start_date": "2026-06-01"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()
|
||||
|
||||
|
||||
def _make_template(client: FlaskClient, token: str, **kw) -> dict:
|
||||
payload = {"name": "Base Template", **kw}
|
||||
resp = client.post("/api/templates", headers=_h(token), json=payload)
|
||||
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 From Template", **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()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Instantiation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_create_simulation_from_template_copies_fields(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
tmpl = _make_template(
|
||||
client,
|
||||
admin_token,
|
||||
description="template desc",
|
||||
commands="template cmd",
|
||||
prerequisites="template prereq",
|
||||
)
|
||||
sim = _make_sim(client, admin_token, eng["id"], template_id=tmpl["id"])
|
||||
|
||||
assert sim["description"] == "template desc"
|
||||
assert sim["commands"] == "template cmd"
|
||||
assert sim["prerequisites"] == "template prereq"
|
||||
assert sim["techniques"] == []
|
||||
assert sim["tactics"] == []
|
||||
assert sim["status"] == "pending"
|
||||
|
||||
|
||||
def test_create_simulation_name_overrides_template(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
tmpl = _make_template(client, admin_token)
|
||||
sim = _make_sim(
|
||||
client, admin_token, eng["id"], name="Custom Name", template_id=tmpl["id"]
|
||||
)
|
||||
assert sim["name"] == "Custom Name"
|
||||
|
||||
|
||||
def test_create_simulation_template_not_found(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
resp = client.post(
|
||||
f"/api/engagements/{eng['id']}/simulations",
|
||||
headers=_h(admin_token),
|
||||
json={"name": "S", "template_id": 9999},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
assert "Template not found" in resp.get_json()["error"]
|
||||
|
||||
|
||||
def test_create_simulation_without_template_unaffected(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
sim = _make_sim(client, admin_token, eng["id"])
|
||||
assert sim["description"] is None
|
||||
assert sim["commands"] is None
|
||||
assert sim["prerequisites"] is None
|
||||
|
||||
|
||||
def test_create_simulation_from_template_status_is_pending(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
tmpl = _make_template(client, admin_token)
|
||||
sim = _make_sim(client, admin_token, eng["id"], template_id=tmpl["id"])
|
||||
assert sim["status"] == "pending"
|
||||
|
||||
|
||||
def test_delete_template_does_not_cascade_to_simulations(
|
||||
client: FlaskClient, admin_token: str
|
||||
) -> None:
|
||||
eng = _make_engagement(client, admin_token)
|
||||
tmpl = _make_template(client, admin_token)
|
||||
sim = _make_sim(client, admin_token, eng["id"], template_id=tmpl["id"])
|
||||
sid = sim["id"]
|
||||
|
||||
# Delete the template.
|
||||
del_resp = client.delete(
|
||||
f"/api/templates/{tmpl['id']}", headers=_h(admin_token)
|
||||
)
|
||||
assert del_resp.status_code == 204
|
||||
|
||||
# Simulation must still be retrievable.
|
||||
get_resp = client.get(f"/api/simulations/{sid}", headers=_h(admin_token))
|
||||
assert get_resp.status_code == 200
|
||||
assert get_resp.get_json()["id"] == sid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Migration round-trip
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_migration_0005_round_trip() -> None:
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
migration_file = (
|
||||
Path(__file__).parent.parent
|
||||
/ "migrations"
|
||||
/ "versions"
|
||||
/ "0005_simulation_templates.py"
|
||||
)
|
||||
|
||||
import importlib.util
|
||||
|
||||
spec = importlib.util.spec_from_file_location("m0005", migration_file)
|
||||
assert spec is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module) # type: ignore[union-attr]
|
||||
|
||||
with engine.begin() as conn:
|
||||
ctx = MigrationContext.configure(conn)
|
||||
import alembic.op as op_module
|
||||
|
||||
op_module._proxy = Operations(ctx) # type: ignore[attr-defined]
|
||||
|
||||
# Create users table (FK dependency).
|
||||
conn.execute(
|
||||
text(
|
||||
"CREATE TABLE users ("
|
||||
"id INTEGER PRIMARY KEY, "
|
||||
"username TEXT NOT NULL, "
|
||||
"password_hash TEXT NOT NULL, "
|
||||
"role TEXT NOT NULL DEFAULT 'redteam', "
|
||||
"created_at DATETIME"
|
||||
")"
|
||||
)
|
||||
)
|
||||
|
||||
module.upgrade()
|
||||
|
||||
tables_after = conn.execute(
|
||||
text("SELECT name FROM sqlite_master WHERE type='table'")
|
||||
).fetchall()
|
||||
table_names = {r[0] for r in tables_after}
|
||||
assert "simulation_templates" in table_names
|
||||
|
||||
cols = conn.execute(
|
||||
text("PRAGMA table_info(simulation_templates)")
|
||||
).fetchall()
|
||||
col_names = {c[1] for c in cols}
|
||||
for expected in ("id", "name", "techniques", "tactic_ids", "created_by_id"):
|
||||
assert expected in col_names, f"missing column: {expected}"
|
||||
|
||||
module.downgrade()
|
||||
|
||||
tables_after_down = conn.execute(
|
||||
text("SELECT name FROM sqlite_master WHERE type='table'")
|
||||
).fetchall()
|
||||
table_names_down = {r[0] for r in tables_after_down}
|
||||
assert "simulation_templates" not in table_names_down
|
||||
Reference in New Issue
Block a user