knacky/Metamorph
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
00b7557e300f92630087de1d697e682db1b58f08
Metamorph/backend/app/services/missions.py
Knacky 00b7557e30 feat(m6): missions + snapshot CRUD, membership visibility, status state machine 
Adds the mission layer that materialises template snapshots, plus the SPA
list / 3-step wizard / detail page.

Backend:
- app/services/missions.py — create_mission snapshots scenarios, tests, MITRE
  tags in a 4-query write; list/get apply a non-admin membership filter that
  collapses to 404 (no existence leak); status state machine enforces
  draft → in_progress → completed → archived with archived as a sink; the
  non-admin creator is auto-added as role_hint='red' to retain visibility.
- app/api/missions.py — 8 endpoints (list, get, create, update, add
  scenarios, set members, transition, soft-delete) with strict pydantic
  schemas. The transition endpoint splits the perm gate manually so
  archive requires mission.archive while other targets use mission.update.
- app/api/users.py — new GET /users/roster returning (id, email,
  display_name) only, gated by user.read OR mission.create OR
  mission.update — lets non-admin wizard users see assignable peers
  without exposing the admin /users payload.
- app/api/diag.py — /diag/reset truncates the mission_* tables before the
  template tables because the source_*_template_id FKs are ON DELETE SET
  NULL, which is cheaper to short-circuit by removing the children first.

Frontend:
- lib/missions.ts — typed client, queryKey factory, status accent map.
- pages/MissionsListPage.tsx — list cards with status accent + filters
  (q, client, status).
- pages/MissionsCreatePage.tsx — 3-step wizard (meta → scenarios → members)
  with member roster fed by /users/roster.
- pages/MissionDetailPage.tsx — header + transition buttons (legal next
  states only) + Tests/Members/Synthesis/Export tabs.
- Routes + nav entry (visible to anyone with mission.read or admin).

Tests:
- backend/tests/test_missions.py — 22 pytest covering snapshot fidelity,
  MITRE propagation, membership visibility, transition state machine,
  perm gating, member set replace, append scenarios, soft-delete, partial
  update, inverted-date rejection.
- e2e/tests/m6-missions.spec.ts — 5 Playwright (snapshot freezing, non-admin
  visibility, status transitions + 409, SPA wizard end-to-end, list filter).

Docs:
- CHANGELOG, tasks/testing-m6.md, tasks/lessons.md (snapshot tradeoffs,
  membership=404 pattern, /diag/reset order, auto-creator add).
- README + tasks/todo.md updated.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 15:07:32 +02:00

899 lines
29 KiB
Python
Raw Blame History

"""Mission CRUD + snapshot service.
A mission is a *materialised* run of one or more scenario templates: when the
mission is created (or scenarios are appended later), the service copies the
template rows into `mission_scenarios` / `mission_tests` / `mission_test_mitre_tags`
verbatim. Editing the source templates afterwards does not touch the mission —
that's the snapshot contract from spec §11.
Visibility rule (spec §4, last bullet): a non-admin user can only see a mission
they are a member of. The decorator layer enforces *which type of action* is
allowed (perm codes); this service enforces *which mission* is visible.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from datetime import date, datetime, timezone
from typing import Any, Iterable
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, selectinload
from app.db.session import session_scope
from app.db.types import (
MISSION_ROLE_HINTS,
MISSION_STATUSES,
)
from app.models.auth import User
from app.models.mission import (
Mission,
MissionMember,
MissionScenario,
MissionTest,
MissionTestMitreTag,
)
from app.models.mitre import MitreSubtechnique, MitreTactic, MitreTechnique
from app.models.template import (
ScenarioTemplate,
TestTemplate,
TestTemplateMitreTag,
)
_UNSET: Any = object()
# Status transition graph. A target status that's not in the source's set is
# rejected as InvalidTransition. `archived` is a one-way sink (un-archiving
# would require an explicit restore endpoint, out of M6 scope).
_VALID_TRANSITIONS: dict[str, frozenset[str]] = {
"draft": frozenset({"in_progress", "archived"}),
"in_progress": frozenset({"completed", "archived"}),
"completed": frozenset({"archived"}),
"archived": frozenset(),
}
# --------------------------------------------------------------------------- #
# Exceptions
# --------------------------------------------------------------------------- #
class MissionNotFound(Exception):
"""Mission missing, soft-deleted, or not visible to the viewer."""
class UnknownScenarioTemplate(Exception):
pass
class UnknownUser(Exception):
pass
class InvalidTransition(Exception):
pass
class InvalidMemberPayload(Exception):
pass
# --------------------------------------------------------------------------- #
# Views (detached dataclasses — safe to return after session_scope exits)
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class MemberAssignment:
"""Inbound member spec. The service resolves the user and validates the hint."""
user_id: uuid.UUID
role_hint: str
@dataclass(frozen=True)
class MissionMemberView:
user_id: uuid.UUID
user_email: str
user_display_name: str | None
role_hint: str
@dataclass(frozen=True)
class MissionMitreTagView:
kind: str
external_id: str
name: str
url: str | None
@dataclass(frozen=True)
class MissionTestView:
id: uuid.UUID
position: int
snapshot_name: str
snapshot_description: str | None
snapshot_objective: str | None
snapshot_procedure_md: str | None
snapshot_prerequisites_md: str | None
snapshot_expected_red_md: str | None
snapshot_expected_blue_md: str | None
snapshot_opsec_level: str
snapshot_tags: list[str]
snapshot_expected_iocs: list[str]
state: str
executed_at: datetime | None
executed_at_overridden: bool
mitre_tags: list[MissionMitreTagView]
source_test_template_id: uuid.UUID | None
@dataclass(frozen=True)
class MissionScenarioView:
id: uuid.UUID
position: int
snapshot_name: str
snapshot_description: str | None
tests: list[MissionTestView]
source_scenario_template_id: uuid.UUID | None
@dataclass(frozen=True)
class MissionListItemView:
id: uuid.UUID
name: str
client_target: str | None
date_start: date | None
date_end: date | None
status: str
description_md: str | None
visibility_mode: str
scenarios_count: int
tests_count: int
members_count: int
deleted_at: datetime | None
created_at: datetime
updated_at: datetime
@dataclass(frozen=True)
class MissionView:
id: uuid.UUID
name: str
client_target: str | None
date_start: date | None
date_end: date | None
status: str
description_md: str | None
visibility_mode: str
scenarios_count: int
tests_count: int
members_count: int
deleted_at: datetime | None
created_at: datetime
updated_at: datetime
scenarios: list[MissionScenarioView]
members: list[MissionMemberView]
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def _opt_str(value: str | None) -> str | None:
if value is None:
return None
v = value.strip()
return v or None
def _normalize_name(value: str) -> str:
name = (value or "").strip()
if not name:
raise ValueError("name is required")
if len(name) > 255:
raise ValueError("name must be ≤ 255 characters")
return name
def _validate_dates(date_start: date | None, date_end: date | None) -> None:
if date_start and date_end and date_end < date_start:
raise ValueError("date_end must be on or after date_start")
def _validate_status(value: str) -> str:
if value not in MISSION_STATUSES:
raise ValueError(f"status must be one of {MISSION_STATUSES}")
return value
def _validate_role_hint(value: str) -> str:
if value not in MISSION_ROLE_HINTS:
raise InvalidMemberPayload(f"role_hint must be one of {MISSION_ROLE_HINTS}")
return value
def _is_member(s: Session, mission_id: uuid.UUID, viewer_id: uuid.UUID) -> bool:
return (
s.scalar(
select(func.count())
.select_from(MissionMember)
.where(
MissionMember.mission_id == mission_id,
MissionMember.user_id == viewer_id,
)
)
or 0
) > 0
def _membership_filter(viewer_id: uuid.UUID):
"""SQL predicate restricting to missions where viewer_id is a member."""
return Mission.id.in_(
select(MissionMember.mission_id).where(MissionMember.user_id == viewer_id)
)
def _load_users_map(s: Session, ids: Iterable[uuid.UUID]) -> dict[uuid.UUID, User]:
ids_list = [i for i in ids]
if not ids_list:
return {}
rows = s.scalars(
select(User).where(User.id.in_(ids_list), User.deleted_at.is_(None))
).all()
return {u.id: u for u in rows}
# --------------------------------------------------------------------------- #
# MITRE denormalisation
# --------------------------------------------------------------------------- #
def _collect_mitre_ids(
tag_rows: Iterable[TestTemplateMitreTag],
) -> tuple[set[uuid.UUID], set[uuid.UUID], set[uuid.UUID]]:
tactic_ids: set[uuid.UUID] = set()
technique_ids: set[uuid.UUID] = set()
sub_ids: set[uuid.UUID] = set()
for tag in tag_rows:
if tag.mitre_kind == "tactic" and tag.tactic_id is not None:
tactic_ids.add(tag.tactic_id)
elif tag.mitre_kind == "technique" and tag.technique_id is not None:
technique_ids.add(tag.technique_id)
elif tag.mitre_kind == "subtechnique" and tag.subtechnique_id is not None:
sub_ids.add(tag.subtechnique_id)
return tactic_ids, technique_ids, sub_ids
def _resolve_mitre_lookup(
s: Session,
tactic_ids: set[uuid.UUID],
technique_ids: set[uuid.UUID],
sub_ids: set[uuid.UUID],
) -> tuple[
dict[uuid.UUID, MitreTactic],
dict[uuid.UUID, MitreTechnique],
dict[uuid.UUID, MitreSubtechnique],
]:
"""Batch-load all MITRE rows referenced by a snapshot in 3 queries."""
tactic_map: dict[uuid.UUID, MitreTactic] = {}
technique_map: dict[uuid.UUID, MitreTechnique] = {}
sub_map: dict[uuid.UUID, MitreSubtechnique] = {}
if tactic_ids:
tactic_map = {
r.id: r
for r in s.scalars(
select(MitreTactic).where(MitreTactic.id.in_(tactic_ids))
).all()
}
if technique_ids:
technique_map = {
r.id: r
for r in s.scalars(
select(MitreTechnique).where(MitreTechnique.id.in_(technique_ids))
).all()
}
if sub_ids:
sub_map = {
r.id: r
for r in s.scalars(
select(MitreSubtechnique).where(MitreSubtechnique.id.in_(sub_ids))
).all()
}
return tactic_map, technique_map, sub_map
def _snapshot_tag(
tag: TestTemplateMitreTag,
tactic_map: dict[uuid.UUID, MitreTactic],
technique_map: dict[uuid.UUID, MitreTechnique],
sub_map: dict[uuid.UUID, MitreSubtechnique],
) -> MissionTestMitreTag | None:
"""Convert a template's polymorphic MITRE tag into a frozen mission tag.
Returns None if the referenced MITRE row vanished between read and snapshot
(paranoid: should not happen inside one tx).
"""
if tag.mitre_kind == "tactic" and tag.tactic_id in tactic_map:
r = tactic_map[tag.tactic_id]
return MissionTestMitreTag(
mitre_kind="tactic",
mitre_external_id=r.external_id,
mitre_name=r.name,
mitre_url=r.url,
)
if tag.mitre_kind == "technique" and tag.technique_id in technique_map:
r = technique_map[tag.technique_id]
return MissionTestMitreTag(
mitre_kind="technique",
mitre_external_id=r.external_id,
mitre_name=r.name,
mitre_url=r.url,
)
if tag.mitre_kind == "subtechnique" and tag.subtechnique_id in sub_map:
r = sub_map[tag.subtechnique_id]
return MissionTestMitreTag(
mitre_kind="subtechnique",
mitre_external_id=r.external_id,
mitre_name=r.name,
mitre_url=r.url,
)
return None
# --------------------------------------------------------------------------- #
# Snapshot
# --------------------------------------------------------------------------- #
def _load_scenario_templates_for_snapshot(
s: Session, scenario_ids: list[uuid.UUID]
) -> dict[uuid.UUID, ScenarioTemplate]:
"""Load scenarios in eager-load mode and reject unknowns/soft-deleted upfront."""
if not scenario_ids:
return {}
rows = s.scalars(
select(ScenarioTemplate)
.options(selectinload(ScenarioTemplate.tests))
.where(ScenarioTemplate.id.in_(scenario_ids))
).all()
by_id = {sc.id: sc for sc in rows}
missing = set(scenario_ids) - by_id.keys()
if missing:
raise UnknownScenarioTemplate(
f"unknown scenario_template ids: {sorted(str(m) for m in missing)}"
)
deleted = [sc.id for sc in rows if sc.deleted_at is not None]
if deleted:
raise UnknownScenarioTemplate(
f"cannot snapshot soft-deleted scenario_template ids: "
f"{sorted(str(d) for d in deleted)}"
)
return by_id
def _snapshot_scenarios(
s: Session,
mission_id: uuid.UUID,
scenario_ids: list[uuid.UUID],
start_position: int,
) -> None:
"""Append `scenario_ids` as new MissionScenario+MissionTest rows under the mission.
Position counter continues from `start_position`. Each scenario_template's
`tests` order is preserved 1:1. MITRE tags on the source templates are
copied as denormalised `MissionTestMitreTag` rows (frozen external_id/name/url).
"""
if not scenario_ids:
return
sc_by_id = _load_scenario_templates_for_snapshot(s, scenario_ids)
# Collect the underlying test_template ids in stable order.
ordered_test_ids: list[uuid.UUID] = []
for sid in scenario_ids:
sc = sc_by_id[sid]
for link in sc.tests:
ordered_test_ids.append(link.test_template_id)
test_template_map: dict[uuid.UUID, TestTemplate] = {}
if ordered_test_ids:
test_template_rows = s.scalars(
select(TestTemplate)
.options(selectinload(TestTemplate.mitre_tags))
.where(TestTemplate.id.in_(set(ordered_test_ids)))
).all()
test_template_map = {t.id: t for t in test_template_rows}
# A test_template may be soft-deleted between the scenario authoring and
# the mission creation. We do not refuse the snapshot (the user expects
# the scenario's planned tests to appear); we just freeze the last
# known content, which is what a snapshot is for.
missing_t = set(ordered_test_ids) - test_template_map.keys()
if missing_t:
raise UnknownScenarioTemplate(
f"scenario references missing test_template ids: "
f"{sorted(str(m) for m in missing_t)}"
)
# Pre-load all MITRE rows referenced by any tag across all involved templates.
all_tag_rows: list[TestTemplateMitreTag] = []
for t in test_template_map.values():
all_tag_rows.extend(t.mitre_tags)
tactic_map, technique_map, sub_map = _resolve_mitre_lookup(
s, *_collect_mitre_ids(all_tag_rows)
)
pos = start_position
for sid in scenario_ids:
sc = sc_by_id[sid]
ms = MissionScenario(
mission_id=mission_id,
source_scenario_template_id=sc.id,
snapshot_name=sc.name,
snapshot_description=sc.description,
position=pos,
)
s.add(ms)
s.flush() # populate ms.id for the child tests
test_pos = 0
for link in sc.tests:
tt = test_template_map[link.test_template_id]
mt = MissionTest(
scenario_id=ms.id,
source_test_template_id=tt.id,
position=test_pos,
snapshot_name=tt.name,
snapshot_description=tt.description,
snapshot_objective=tt.objective,
snapshot_procedure_md=tt.procedure_md,
snapshot_prerequisites_md=tt.prerequisites_md,
snapshot_expected_red_md=tt.expected_result_red_md,
snapshot_expected_blue_md=tt.expected_detection_blue_md,
snapshot_opsec_level=tt.opsec_level,
snapshot_tags=list(tt.tags or []),
snapshot_expected_iocs=list(tt.expected_iocs or []),
state="pending",
)
s.add(mt)
s.flush()
for src_tag in tt.mitre_tags:
snap = _snapshot_tag(src_tag, tactic_map, technique_map, sub_map)
if snap is not None:
snap.mission_test_id = mt.id
s.add(snap)
test_pos += 1
pos += 1
s.flush()
# --------------------------------------------------------------------------- #
# View assembly
# --------------------------------------------------------------------------- #
def _member_views(s: Session, mission: Mission) -> list[MissionMemberView]:
if not mission.members:
return []
users = _load_users_map(s, [m.user_id for m in mission.members])
out: list[MissionMemberView] = []
for m in mission.members:
u = users.get(m.user_id)
out.append(
MissionMemberView(
user_id=m.user_id,
user_email=u.email if u else "<deleted>",
user_display_name=(u.display_name if u else None),
role_hint=m.role_hint,
)
)
out.sort(key=lambda mv: (mv.role_hint, mv.user_email))
return out
def _scenario_views(scenarios: list[MissionScenario]) -> list[MissionScenarioView]:
views: list[MissionScenarioView] = []
for sc in sorted(scenarios, key=lambda s_: s_.position):
test_views: list[MissionTestView] = []
for t in sorted(sc.tests, key=lambda t_: t_.position):
tag_views = [
MissionMitreTagView(
kind=tag.mitre_kind,
external_id=tag.mitre_external_id,
name=tag.mitre_name,
url=tag.mitre_url,
)
for tag in sorted(
t.mitre_tags, key=lambda tg: (tg.mitre_kind, tg.mitre_external_id)
)
]
test_views.append(
MissionTestView(
id=t.id,
position=t.position,
snapshot_name=t.snapshot_name,
snapshot_description=t.snapshot_description,
snapshot_objective=t.snapshot_objective,
snapshot_procedure_md=t.snapshot_procedure_md,
snapshot_prerequisites_md=t.snapshot_prerequisites_md,
snapshot_expected_red_md=t.snapshot_expected_red_md,
snapshot_expected_blue_md=t.snapshot_expected_blue_md,
snapshot_opsec_level=t.snapshot_opsec_level,
snapshot_tags=list(t.snapshot_tags or []),
snapshot_expected_iocs=list(t.snapshot_expected_iocs or []),
state=t.state,
executed_at=t.executed_at,
executed_at_overridden=t.executed_at_overridden,
mitre_tags=tag_views,
source_test_template_id=t.source_test_template_id,
)
)
views.append(
MissionScenarioView(
id=sc.id,
position=sc.position,
snapshot_name=sc.snapshot_name,
snapshot_description=sc.snapshot_description,
tests=test_views,
source_scenario_template_id=sc.source_scenario_template_id,
)
)
return views
def _to_detail_view(s: Session, m: Mission) -> MissionView:
scenarios = [sc for sc in m.scenarios if sc.deleted_at is None]
members = _member_views(s, m)
scenario_views = _scenario_views(scenarios)
tests_count = sum(len(sc.tests) for sc in scenario_views)
return MissionView(
id=m.id,
name=m.name,
client_target=m.client_target,
date_start=m.date_start,
date_end=m.date_end,
status=m.status,
description_md=m.description_md,
visibility_mode=m.visibility_mode,
scenarios_count=len(scenario_views),
tests_count=tests_count,
members_count=len(members),
deleted_at=m.deleted_at,
created_at=m.created_at,
updated_at=m.updated_at,
scenarios=scenario_views,
members=members,
)
def _to_list_item(m: Mission) -> MissionListItemView:
# Cheap counts via the loaded relationships (selectinloaded by the caller).
live_scenarios = [sc for sc in m.scenarios if sc.deleted_at is None]
tests_count = sum(len(sc.tests) for sc in live_scenarios)
return MissionListItemView(
id=m.id,
name=m.name,
client_target=m.client_target,
date_start=m.date_start,
date_end=m.date_end,
status=m.status,
description_md=m.description_md,
visibility_mode=m.visibility_mode,
scenarios_count=len(live_scenarios),
tests_count=tests_count,
members_count=len(m.members),
deleted_at=m.deleted_at,
created_at=m.created_at,
updated_at=m.updated_at,
)
# --------------------------------------------------------------------------- #
# Public API — list / get
# --------------------------------------------------------------------------- #
def list_missions(
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
q: str | None = None,
status: str | None = None,
client: str | None = None,
include_deleted: bool = False,
limit: int = 100,
offset: int = 0,
) -> tuple[list[MissionListItemView], int]:
with session_scope() as s:
stmt = (
select(Mission)
.options(
selectinload(Mission.scenarios).selectinload(MissionScenario.tests),
selectinload(Mission.members),
)
.order_by(Mission.created_at.desc(), Mission.id.desc())
)
count_stmt = select(func.count()).select_from(Mission)
if not include_deleted:
stmt = stmt.where(Mission.deleted_at.is_(None))
count_stmt = count_stmt.where(Mission.deleted_at.is_(None))
if not viewer_is_admin:
stmt = stmt.where(_membership_filter(viewer_id))
count_stmt = count_stmt.where(_membership_filter(viewer_id))
if status:
_validate_status(status)
stmt = stmt.where(Mission.status == status)
count_stmt = count_stmt.where(Mission.status == status)
if client:
like = f"%{client.lower()}%"
cond = func.lower(Mission.client_target).like(like)
stmt = stmt.where(cond)
count_stmt = count_stmt.where(cond)
if q:
like = f"%{q.lower()}%"
cond = or_(
func.lower(Mission.name).like(like),
func.lower(Mission.description_md).like(like),
)
stmt = stmt.where(cond)
count_stmt = count_stmt.where(cond)
total = s.scalar(count_stmt) or 0
rows = s.scalars(
stmt.limit(max(1, min(limit, 500))).offset(max(0, offset))
).all()
return [_to_list_item(m) for m in rows], int(total)
def get_mission(
mission_id: uuid.UUID,
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
include_deleted: bool = False,
) -> MissionView:
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None:
raise MissionNotFound()
if m.deleted_at is not None and not include_deleted:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
return _to_detail_view(s, m)
# --------------------------------------------------------------------------- #
# Public API — write
# --------------------------------------------------------------------------- #
def _validate_members(s: Session, members: list[MemberAssignment]) -> None:
"""Reject duplicates, bad role hints, unknown/soft-deleted users."""
seen: set[uuid.UUID] = set()
for m in members:
if m.user_id in seen:
raise InvalidMemberPayload(f"duplicate user_id: {m.user_id}")
seen.add(m.user_id)
_validate_role_hint(m.role_hint)
if not members:
return
user_map = _load_users_map(s, seen)
missing = seen - user_map.keys()
if missing:
raise UnknownUser(f"unknown or deleted user_ids: {sorted(str(u) for u in missing)}")
def create_mission(
*,
name: str,
creator_id: uuid.UUID,
creator_is_admin: bool,
client_target: str | None = None,
date_start: date | None = None,
date_end: date | None = None,
description_md: str | None = None,
scenario_template_ids: list[uuid.UUID] | None = None,
members: list[MemberAssignment] | None = None,
) -> MissionView:
"""Create a mission and snapshot the requested scenarios + their tests.
Side effect: if `creator_is_admin` is False and the creator is not in
`members`, they are added with `role_hint='red'`. This prevents the
non-admin creator from immediately losing visibility on the mission they
just created (membership-based visibility, see spec §4).
"""
name_norm = _normalize_name(name)
_validate_dates(date_start, date_end)
scenarios = list(scenario_template_ids or [])
members_list = list(members or [])
with session_scope() as s:
_validate_members(s, members_list)
# Auto-add the non-admin creator as a member so they retain visibility.
if not creator_is_admin and not any(m.user_id == creator_id for m in members_list):
members_list = [
MemberAssignment(user_id=creator_id, role_hint="red"),
*members_list,
]
# Defensive re-validation in case the creator id was bogus.
_validate_members(s, members_list)
mission = Mission(
name=name_norm,
client_target=_opt_str(client_target),
date_start=date_start,
date_end=date_end,
description_md=_opt_str(description_md),
status="draft",
visibility_mode="whitebox",
)
s.add(mission)
s.flush()
for m in members_list:
s.add(
MissionMember(
mission_id=mission.id,
user_id=m.user_id,
role_hint=m.role_hint,
)
)
if scenarios:
_snapshot_scenarios(s, mission.id, scenarios, start_position=0)
s.flush()
s.refresh(mission)
return _to_detail_view(s, mission)
def update_mission_metadata(
mission_id: uuid.UUID,
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
name: str | None = None,
client_target: Any = _UNSET,
date_start: Any = _UNSET,
date_end: Any = _UNSET,
description_md: Any = _UNSET,
) -> MissionView:
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None or m.deleted_at is not None:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
if name is not None:
m.name = _normalize_name(name)
if client_target is not _UNSET:
m.client_target = _opt_str(client_target)
if date_start is not _UNSET:
m.date_start = date_start
if date_end is not _UNSET:
m.date_end = date_end
# Validate the combined date pair regardless of which side was passed.
_validate_dates(m.date_start, m.date_end)
if description_md is not _UNSET:
m.description_md = _opt_str(description_md)
s.flush()
s.refresh(m)
return _to_detail_view(s, m)
def add_scenarios_to_mission(
mission_id: uuid.UUID,
scenario_template_ids: list[uuid.UUID],
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
) -> MissionView:
"""Append more snapshot scenarios to an existing mission.
They land at `current_max_position + 1` and onwards. Empty list is a no-op
and just returns the current view.
"""
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None or m.deleted_at is not None:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
if scenario_template_ids:
max_pos = s.scalar(
select(func.coalesce(func.max(MissionScenario.position), -1)).where(
MissionScenario.mission_id == mission_id
)
)
_snapshot_scenarios(
s,
mission_id,
list(scenario_template_ids),
start_position=int(max_pos) + 1,
)
s.flush()
s.refresh(m)
return _to_detail_view(s, m)
def set_mission_members(
mission_id: uuid.UUID,
members: list[MemberAssignment],
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
) -> MissionView:
"""Replace the entire member set. Wipe + insert, like the scenario reorder."""
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None or m.deleted_at is not None:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
_validate_members(s, members)
for link in list(m.members):
s.delete(link)
s.flush()
for assignment in members:
s.add(
MissionMember(
mission_id=m.id,
user_id=assignment.user_id,
role_hint=assignment.role_hint,
)
)
s.flush()
s.refresh(m)
return _to_detail_view(s, m)
def transition_mission_status(
mission_id: uuid.UUID,
target_status: str,
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
) -> MissionView:
"""Move the mission's status one step along the lifecycle graph."""
_validate_status(target_status)
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None or m.deleted_at is not None:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
if target_status == m.status:
# No-op transitions are valid: a client retry should not 409.
s.refresh(m)
return _to_detail_view(s, m)
allowed = _VALID_TRANSITIONS.get(m.status, frozenset())
if target_status not in allowed:
raise InvalidTransition(
f"cannot transition from {m.status!r} to {target_status!r}"
)
m.status = target_status
s.flush()
s.refresh(m)
return _to_detail_view(s, m)
def soft_delete_mission(
mission_id: uuid.UUID,
*,
viewer_id: uuid.UUID,
viewer_is_admin: bool,
) -> None:
with session_scope() as s:
m = s.get(Mission, mission_id)
if m is None or m.deleted_at is not None:
raise MissionNotFound()
if not viewer_is_admin and not _is_member(s, mission_id, viewer_id):
raise MissionNotFound()
m.deleted_at = datetime.now(tz=timezone.utc)
Reference in New Issue View Git Blame Copy Permalink