"""TestEntity — pure domain object for the test lifecycle state machine. This entity owns ALL state-transition logic and business rules for a security test. It has **no** dependency on FastAPI, SQLAlchemy, or any infrastructure concern. Usage:: entity = TestEntity.from_orm(test_orm_model) entity.start_execution() # draft → red_executing entity.submit_red_evidence() # red_executing → blue_evaluating entity.pause_timer() entity.resume_timer() entity.submit_blue_evidence() # blue_evaluating → in_review entity.validate_red("approved") entity.validate_blue("approved") # triggers dual-validation → validated entity.reopen() # rejected → draft After mutations, the service layer copies ``entity.changes`` back onto the ORM model and persists via Unit of Work. """ # Enable future language features for compatibility from __future__ import annotations # Import enum import enum # Import uuid import uuid # Import dataclass, field from dataclasses from dataclasses import dataclass, field # Import datetime from datetime from datetime import datetime # Import TYPE_CHECKING, Any from typing from typing import TYPE_CHECKING, Any # Import from app.domain.errors from app.domain.errors import ( BusinessRuleViolation, InvalidOperationError, InvalidStateTransition, ) # Check: TYPE_CHECKING if TYPE_CHECKING: # Import Test as TestORM from app.models.test from app.models.test import Test as TestORM # ── Value objects ──────────────────────────────────────────────────── class TestState(str, enum.Enum): """Ordered lifecycle states for a security test.""" # Assign draft = "draft" draft = "draft" # Assign red_executing = "red_executing" red_executing = "red_executing" # Assign blue_evaluating = "blue_evaluating" blue_evaluating = "blue_evaluating" # Assign in_review = "in_review" in_review = "in_review" # Assign validated = "validated" validated = "validated" # Assign rejected = "rejected" rejected = "rejected" disputed = "disputed" # one lead approved, the other rejected # Assign VALID_TRANSITIONS = { VALID_TRANSITIONS: dict[TestState, list[TestState]] = { TestState.draft: [TestState.red_executing], TestState.red_executing: [TestState.blue_evaluating], TestState.blue_evaluating: [TestState.in_review], TestState.in_review: [TestState.validated, TestState.rejected, TestState.disputed], TestState.disputed: [TestState.validated, TestState.rejected], TestState.rejected: [TestState.draft], TestState.validated: [], } # Assign _PAUSABLE_STATES = frozenset({TestState.red_executing, TestState.blue_evaluating}) _PAUSABLE_STATES = frozenset({TestState.red_executing, TestState.blue_evaluating}) # ── Domain events (lightweight records of what happened) ───────────── @dataclass(frozen=True) # Define class DomainEvent class DomainEvent: """Immutable record of a domain-level event emitted by the test entity.""" # name: str name: str # Assign payload = field(default_factory=dict) payload: dict[str, Any] = field(default_factory=dict) # ── Entity ─────────────────────────────────────────────────────────── @dataclass # Define class TestEntity class TestEntity: """Pure domain representation of a security test.""" # id: uuid.UUID id: uuid.UUID # state: TestState state: TestState # Red validation red_validation_status: str | None = None # Assign red_validated_by = None red_validated_by: uuid.UUID | None = None # Assign red_validated_at = None red_validated_at: datetime | None = None # Assign red_validation_notes = None red_validation_notes: str | None = None # Blue validation blue_validation_status: str | None = None # Assign blue_validated_by = None blue_validated_by: uuid.UUID | None = None # Assign blue_validated_at = None blue_validated_at: datetime | None = None # Assign blue_validation_notes = None blue_validation_notes: str | None = None # Phase timing execution_date: datetime | None = None # Assign red_started_at = None red_started_at: datetime | None = None # Assign blue_started_at = None blue_started_at: datetime | None = None # Assign paused_at = None paused_at: datetime | None = None # Assign red_paused_seconds = 0 red_paused_seconds: int = 0 # Assign blue_paused_seconds = 0 blue_paused_seconds: int = 0 # Internal bookkeeping (not persisted as-is) _events: list[DomainEvent] = field(default_factory=list, repr=False) # -- Factory -------------------------------------------------------- @classmethod # Define function from_orm def from_orm(cls, model: TestORM) -> TestEntity: """Build a TestEntity from a SQLAlchemy ``Test`` model instance. Args: model (TestORM): The ORM model whose fields will be copied into the entity. Returns: TestEntity: A fully populated domain entity reflecting the ORM state. """ # Assign raw_state = model.state raw_state = model.state # Assign state = raw_state if isinstance(raw_state, TestState) else TestState(raw_st... state = raw_state if isinstance(raw_state, TestState) else TestState(raw_state) # Return cls( return cls( # Keyword argument: id id=model.id, # Keyword argument: state state=state, # Keyword argument: red_validation_status red_validation_status=model.red_validation_status, # Keyword argument: red_validated_by red_validated_by=model.red_validated_by, # Keyword argument: red_validated_at red_validated_at=model.red_validated_at, # Keyword argument: red_validation_notes red_validation_notes=model.red_validation_notes, # Keyword argument: blue_validation_status blue_validation_status=model.blue_validation_status, # Keyword argument: blue_validated_by blue_validated_by=model.blue_validated_by, # Keyword argument: blue_validated_at blue_validated_at=model.blue_validated_at, # Keyword argument: blue_validation_notes blue_validation_notes=model.blue_validation_notes, # Keyword argument: execution_date execution_date=model.execution_date, # Keyword argument: red_started_at red_started_at=model.red_started_at, # Keyword argument: blue_started_at blue_started_at=model.blue_started_at, # Keyword argument: paused_at paused_at=model.paused_at, # Keyword argument: red_paused_seconds red_paused_seconds=model.red_paused_seconds or 0, # Keyword argument: blue_paused_seconds blue_paused_seconds=model.blue_paused_seconds or 0, ) # Define function apply_to def apply_to(self, model: TestORM) -> None: """Copy the entity's mutable fields back onto the ORM model. Args: model (TestORM): The ORM model to update in-place. Returns: None """ # Assign model.state = self.state model.state = self.state # Assign model.red_validation_status = self.red_validation_status model.red_validation_status = self.red_validation_status # Assign model.red_validated_by = self.red_validated_by model.red_validated_by = self.red_validated_by # Assign model.red_validated_at = self.red_validated_at model.red_validated_at = self.red_validated_at # Assign model.red_validation_notes = self.red_validation_notes model.red_validation_notes = self.red_validation_notes # Assign model.blue_validation_status = self.blue_validation_status model.blue_validation_status = self.blue_validation_status # Assign model.blue_validated_by = self.blue_validated_by model.blue_validated_by = self.blue_validated_by # Assign model.blue_validated_at = self.blue_validated_at model.blue_validated_at = self.blue_validated_at # Assign model.blue_validation_notes = self.blue_validation_notes model.blue_validation_notes = self.blue_validation_notes # Assign model.execution_date = self.execution_date model.execution_date = self.execution_date # Assign model.red_started_at = self.red_started_at model.red_started_at = self.red_started_at # Assign model.blue_started_at = self.blue_started_at model.blue_started_at = self.blue_started_at # Assign model.paused_at = self.paused_at model.paused_at = self.paused_at # Assign model.red_paused_seconds = self.red_paused_seconds model.red_paused_seconds = self.red_paused_seconds # Assign model.blue_paused_seconds = self.blue_paused_seconds model.blue_paused_seconds = self.blue_paused_seconds # -- Query helpers -------------------------------------------------- @property # Define function events def events(self) -> list[DomainEvent]: """Return a snapshot of all domain events raised on this entity. Returns: list[DomainEvent]: Ordered list of events emitted since the entity was constructed or last cleared. """ # Return list(self._events) return list(self._events) # Define function can_transition def can_transition(self, target: TestState) -> bool: """Check whether a transition from the current state to *target* is valid. Args: target (TestState): The desired next state. Returns: bool: True if the transition is allowed, False otherwise. """ # Return target in VALID_TRANSITIONS.get(self.state, []) return target in VALID_TRANSITIONS.get(self.state, []) # Apply the @property decorator @property # Define function is_terminal def is_terminal(self) -> bool: """Return True if the test has reached its final (validated) state. Returns: bool: True when state is ``validated``, False for all other states. """ # Return self.state == TestState.validated return self.state == TestState.validated # -- Core transition ------------------------------------------------ def transition_to(self, target: TestState | str) -> str: """Validate and apply a state transition. Accepts either a :class:`TestState` member or its string value (so callers using ``models.enums.TestState`` work transparently). Returns the *previous* state value as a plain string. Raises :class:`InvalidStateTransition` when the move is illegal. Args: target (TestState | str): The desired next state, as an enum member or its string equivalent. Returns: str: The previous state value before the transition. """ # Assign value = target.value if hasattr(target, "value") else str(target) value = target.value if hasattr(target, "value") else str(target) # Assign resolved = target if isinstance(target, TestState) else TestState(value) resolved = target if isinstance(target, TestState) else TestState(value) # Return self._transition(resolved) return self._transition(resolved) # Define function _transition def _transition(self, target: TestState) -> str: """Validate and apply a state transition, returning the previous state value. Args: target (TestState): The desired next state enum member. Returns: str: The previous state value before the transition was applied. """ # Check: not self.can_transition(target) if not self.can_transition(target): # Assign valid = [s.value for s in VALID_TRANSITIONS.get(self.state, [])] valid = [s.value for s in VALID_TRANSITIONS.get(self.state, [])] # Raise InvalidStateTransition raise InvalidStateTransition( # Keyword argument: current_state current_state=self.state.value, # Keyword argument: target_state target_state=target.value, # Keyword argument: valid_transitions valid_transitions=valid, ) # Assign previous = self.state.value previous = self.state.value # Assign self.state = target self.state = target # Call self._events.append() self._events.append(DomainEvent( # Literal argument value "state_changed", {"previous": previous, "new": target.value}, )) # Return previous return previous # -- Lifecycle commands -------------------------------------------- def start_execution(self) -> None: """Transition the test from ``draft`` to ``red_executing``. Returns: None """ # Call self._transition() self._transition(TestState.red_executing) # Assign now = datetime.utcnow() now = datetime.utcnow() # Assign self.execution_date = now self.execution_date = now # Assign self.red_started_at = now self.red_started_at = now # Call self._events.append() self._events.append(DomainEvent("execution_started")) # Define function submit_red_evidence def submit_red_evidence(self) -> int: """Transition the test from ``red_executing`` to ``blue_evaluating``. Auto-resumes if paused. Returns paused seconds accumulated during this phase (for worklog calculation). Returns: int: Total seconds the red phase was paused. """ # Assign paused_extra = self._auto_resume() paused_extra = self._auto_resume() # Call self._transition() self._transition(TestState.blue_evaluating) # Assign total_paused = self.red_paused_seconds + paused_extra total_paused = self.red_paused_seconds + paused_extra # Assign self.blue_started_at = datetime.utcnow() self.blue_started_at = datetime.utcnow() # Assign self.blue_paused_seconds = 0 self.blue_paused_seconds = 0 # Call self._events.append() self._events.append(DomainEvent( # Literal argument value "red_evidence_submitted", {"red_paused_seconds": total_paused}, )) # Return total_paused return total_paused # Define function submit_blue_evidence def submit_blue_evidence(self) -> int: """Transition the test from ``blue_evaluating`` to ``in_review``. Auto-resumes if paused. Returns paused seconds accumulated during this phase (for worklog calculation). Returns: int: Total seconds the blue phase was paused. """ # Assign paused_extra = self._auto_resume() paused_extra = self._auto_resume() # Call self._transition() self._transition(TestState.in_review) # Assign total_paused = self.blue_paused_seconds + paused_extra total_paused = self.blue_paused_seconds + paused_extra # Call self._events.append() self._events.append(DomainEvent( # Literal argument value "blue_evidence_submitted", {"blue_paused_seconds": total_paused}, )) # Return total_paused return total_paused # Define function pause_timer def pause_timer(self) -> None: """Pause the active phase timer. Returns: None """ # Check: self.state not in _PAUSABLE_STATES if self.state not in _PAUSABLE_STATES: # Raise BusinessRuleViolation raise BusinessRuleViolation( f"Cannot pause timer in '{self.state.value}' state" ) # Check: self.paused_at is not None if self.paused_at is not None: # Raise BusinessRuleViolation raise BusinessRuleViolation("Timer is already paused") # Assign self.paused_at = datetime.utcnow() self.paused_at = datetime.utcnow() # Call self._events.append() self._events.append(DomainEvent("timer_paused")) # Define function resume_timer def resume_timer(self) -> int: """Resume a paused timer. Returns: int: Number of seconds the timer was paused for. """ # Check: self.paused_at is None if self.paused_at is None: # Raise BusinessRuleViolation raise BusinessRuleViolation("Timer is not paused") # Assign now = datetime.utcnow() now = datetime.utcnow() # Assign paused_seconds = max(int((now - self.paused_at).total_seconds()), 0) paused_seconds = max(int((now - self.paused_at).total_seconds()), 0) # Check: self.state == TestState.red_executing if self.state == TestState.red_executing: # Assign self.red_paused_seconds = paused_seconds self.red_paused_seconds += paused_seconds # Alternative: self.state == TestState.blue_evaluating elif self.state == TestState.blue_evaluating: # Assign self.blue_paused_seconds = paused_seconds self.blue_paused_seconds += paused_seconds # Assign self.paused_at = None self.paused_at = None # Call self._events.append() self._events.append(DomainEvent("timer_resumed", {"paused_seconds": paused_seconds})) # Return paused_seconds return paused_seconds # Define function validate_red def validate_red(self, status: str, *, by: uuid.UUID, notes: str | None = None) -> None: """Record Red Lead's validation decision. Args: status (str): Validation outcome; must be ``"approved"`` or ``"rejected"``. by (uuid.UUID): UUID of the Red Lead recording the decision. notes (str | None): Optional free-text notes about the decision. Returns: None """ # Call self._assert_in_review() self._assert_in_review("red") # Call self._assert_valid_vote() self._assert_valid_vote(status) # Assign now = datetime.utcnow() now = datetime.utcnow() # Assign self.red_validation_status = status self.red_validation_status = status # Assign self.red_validated_by = by self.red_validated_by = by # Assign self.red_validated_at = now self.red_validated_at = now # Assign self.red_validation_notes = notes self.red_validation_notes = notes # Call self._events.append() self._events.append(DomainEvent("red_validated", {"status": status})) # Call self._check_dual_validation() self._check_dual_validation() # Define function validate_blue def validate_blue(self, status: str, *, by: uuid.UUID, notes: str | None = None) -> None: """Record Blue Lead's validation decision. Args: status (str): Validation outcome; must be ``"approved"`` or ``"rejected"``. by (uuid.UUID): UUID of the Blue Lead recording the decision. notes (str | None): Optional free-text notes about the decision. Returns: None """ # Call self._assert_in_review() self._assert_in_review("blue") # Call self._assert_valid_vote() self._assert_valid_vote(status) # Assign now = datetime.utcnow() now = datetime.utcnow() # Assign self.blue_validation_status = status self.blue_validation_status = status # Assign self.blue_validated_by = by self.blue_validated_by = by # Assign self.blue_validated_at = now self.blue_validated_at = now # Assign self.blue_validation_notes = notes self.blue_validation_notes = notes # Call self._events.append() self._events.append(DomainEvent("blue_validated", {"status": status})) # Call self._check_dual_validation() self._check_dual_validation() # Define function reopen def reopen(self) -> None: """Transition the test from ``rejected`` back to ``draft``, clearing all validation and timing fields. Returns: None """ # Call self._transition() self._transition(TestState.draft) # Assign self.red_validation_status = None self.red_validation_status = None # Assign self.red_validated_by = None self.red_validated_by = None # Assign self.red_validated_at = None self.red_validated_at = None # Assign self.red_validation_notes = None self.red_validation_notes = None # Assign self.blue_validation_status = None self.blue_validation_status = None # Assign self.blue_validated_by = None self.blue_validated_by = None # Assign self.blue_validated_at = None self.blue_validated_at = None # Assign self.blue_validation_notes = None self.blue_validation_notes = None # Assign self.red_started_at = None self.red_started_at = None # Assign self.blue_started_at = None self.blue_started_at = None # Assign self.paused_at = None self.paused_at = None # Assign self.red_paused_seconds = 0 self.red_paused_seconds = 0 # Assign self.blue_paused_seconds = 0 self.blue_paused_seconds = 0 # Call self._events.append() self._events.append(DomainEvent("test_reopened")) # -- Private ------------------------------------------------------- def _auto_resume(self) -> int: """Accumulate pause time and clear the paused timestamp if currently paused. Returns: int: Extra seconds that were accumulated from the current pause, or 0 if the timer was not paused. """ # Check: self.paused_at is None if self.paused_at is None: # Return 0 return 0 # Assign now = datetime.utcnow() now = datetime.utcnow() # Assign extra = max(int((now - self.paused_at).total_seconds()), 0) extra = max(int((now - self.paused_at).total_seconds()), 0) # Assign self.paused_at = None self.paused_at = None # Return extra return extra # Define function check_dual_validation def check_dual_validation(self) -> None: """Evaluate both leads' votes and advance state if appropriate. Rules (v2 — consensus required): - Both **approved** -> ``validated`` - Both **rejected** -> ``rejected`` - One approved + one rejected -> ``disputed`` (conflict, needs discussion) - Otherwise (one or both still pending) -> no change Called automatically by :meth:`validate_red` and :meth:`validate_blue`. """ # Call self._check_dual_validation() self._check_dual_validation() # Define function _assert_in_review def _assert_in_review(self, side: str) -> None: if self.state not in (TestState.in_review, TestState.disputed): raise InvalidOperationError( f"Cannot validate {side} side while test is in " f"'{self.state.value}' state (must be in_review or disputed)" ) # Apply the @staticmethod decorator @staticmethod # Define function _assert_valid_vote def _assert_valid_vote(status: str) -> None: """Raise InvalidOperationError if *status* is not a valid vote value. Args: status (str): The vote value to validate; must be ``"approved"`` or ``"rejected"``. Returns: None """ # Check: status not in ("approved", "rejected") if status not in ("approved", "rejected"): # Raise InvalidOperationError raise InvalidOperationError( # Literal argument value "validation_status must be 'approved' or 'rejected'" ) # Define function _check_dual_validation def _check_dual_validation(self) -> None: """Advance the test state once both leads have voted.""" r, b = self.red_validation_status, self.blue_validation_status if r == "approved" and b == "approved": self.state = TestState.validated # Call self._events.append() self._events.append(DomainEvent("dual_validation_approved")) elif r == "rejected" or b == "rejected": # Any rejection is a veto — one lead can reject without waiting for the other self.state = TestState.rejected self._events.append(DomainEvent("dual_validation_rejected"))