"""Campaign domain entity with lifecycle validation. Pure domain logic — no framework imports. """ from __future__ import annotations import enum import uuid from dataclasses import dataclass, field from typing import Any from app.domain.errors import BusinessRuleViolation, InvalidStateTransition class CampaignStatus(str, enum.Enum): draft = "draft" active = "active" completed = "completed" archived = "archived" class CampaignType(str, enum.Enum): custom = "custom" apt_emulation = "apt_emulation" kill_chain = "kill_chain" compliance = "compliance" VALID_TRANSITIONS: dict[CampaignStatus, list[CampaignStatus]] = { CampaignStatus.draft: [CampaignStatus.active], CampaignStatus.active: [CampaignStatus.completed], CampaignStatus.completed: [CampaignStatus.archived], CampaignStatus.archived: [], } @dataclass class CampaignEntity: name: str type: CampaignType = CampaignType.custom status: CampaignStatus = CampaignStatus.draft id: uuid.UUID | None = None description: str | None = None threat_actor_id: uuid.UUID | None = None created_by: uuid.UUID | None = None target_platform: str | None = None tags: list[str] = field(default_factory=list) test_count: int = 0 def can_transition_to(self, target: CampaignStatus) -> bool: return target in VALID_TRANSITIONS.get(self.status, []) def activate(self) -> None: if not self.can_transition_to(CampaignStatus.active): raise InvalidStateTransition( self.status.value, CampaignStatus.active.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) if self.test_count == 0: raise BusinessRuleViolation( "Campaign must have at least one test to activate" ) self.status = CampaignStatus.active def complete(self) -> None: if not self.can_transition_to(CampaignStatus.completed): raise InvalidStateTransition( self.status.value, CampaignStatus.completed.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) self.status = CampaignStatus.completed def archive(self) -> None: if not self.can_transition_to(CampaignStatus.archived): raise InvalidStateTransition( self.status.value, CampaignStatus.archived.value, [s.value for s in VALID_TRANSITIONS[self.status]], ) self.status = CampaignStatus.archived def ensure_modifiable(self) -> None: if self.status not in (CampaignStatus.draft, CampaignStatus.active): raise BusinessRuleViolation( f"Cannot modify campaign in '{self.status.value}' state" ) @classmethod def from_orm(cls, orm: Any) -> CampaignEntity: """Build a CampaignEntity from a SQLAlchemy Campaign model.""" test_count = len(getattr(orm, "campaign_tests", None) or []) return cls( id=orm.id, name=orm.name, type=CampaignType(orm.type) if orm.type else CampaignType.custom, status=CampaignStatus(orm.status) if orm.status else CampaignStatus.draft, description=orm.description, threat_actor_id=orm.threat_actor_id, created_by=orm.created_by, target_platform=orm.target_platform, tags=orm.tags or [], test_count=test_count, )