kitos/Aegis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tests): add red_review/blue_review states to test state machine

Browse Source
This commit is contained in:
kitos
2026-07-06 10:38:15 +02:00
parent af65681179
commit 58d1da93ce
3 changed files with 197 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/domain/enums.py
+4
View File
@@ -35,8 +35,12 @@ class TestState(str, enum.Enum):
draft = "draft"
# Assign red_executing = "red_executing"
red_executing = "red_executing"
# Red Lead reviews the operator's work before it queues for Blue Team
red_review = "red_review"
# Assign blue_evaluating = "blue_evaluating"
blue_evaluating = "blue_evaluating"
# Blue Lead reviews the operator's work before cross-validation
blue_review = "blue_review"
# Assign in_review = "in_review"
in_review = "in_review"
# Assign validated = "validated"
backend/app/domain/test_entity.py
+71 -11
View File
@@ -60,8 +60,12 @@ class TestState(str, enum.Enum):
draft = "draft"
# Assign red_executing = "red_executing"
red_executing = "red_executing"
# Red Lead reviews the operator's work before it queues for Blue Team
red_review = "red_review"
# Assign blue_evaluating = "blue_evaluating"
blue_evaluating = "blue_evaluating"
# Blue Lead reviews the operator's work before cross-validation
blue_review = "blue_review"
# Assign in_review = "in_review"
in_review = "in_review"
# Assign validated = "validated"
@@ -74,8 +78,10 @@ class TestState(str, enum.Enum):
# 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.red_executing: [TestState.red_review],
TestState.red_review: [TestState.blue_evaluating, TestState.red_executing],
TestState.blue_evaluating: [TestState.blue_review],
TestState.blue_review: [TestState.in_review, TestState.blue_evaluating],
TestState.in_review: [TestState.validated, TestState.rejected, TestState.disputed],
TestState.disputed: [TestState.validated, TestState.rejected],
TestState.rejected: [TestState.draft],
@@ -363,10 +369,12 @@ class TestEntity:
# Define function submit_red_evidence
def submit_red_evidence(self) -> int:
"""Transition the test from ``red_executing`` to ``blue_evaluating``.
"""Transition the test from ``red_executing`` to ``red_review``.
Auto-resumes if paused. Returns paused seconds accumulated
during this phase (for worklog calculation).
during this phase (for worklog calculation). The Blue Team queue
timer does not start yet — that happens in ``approve_red_review``,
once the Red Lead actually releases the test to Blue Team.
Returns:
int: Total seconds the red phase was paused.
@@ -374,13 +382,9 @@ class TestEntity:
# Assign paused_extra = self._auto_resume()
paused_extra = self._auto_resume()
# Call self._transition()
self._transition(TestState.blue_evaluating)
self._transition(TestState.red_review)
# 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
@@ -390,9 +394,32 @@ class TestEntity:
# Return total_paused
return total_paused
def approve_red_review(self) -> None:
"""Transition the test from ``red_review`` to ``blue_evaluating``.
Called when the assigned Red Lead approves the operator's work.
Starts the Blue Team queue timer.
"""
self._transition(TestState.blue_evaluating)
self.blue_started_at = datetime.utcnow()
self.blue_paused_seconds = 0
self._events.append(DomainEvent("red_review_approved"))
def reopen_red_review(self) -> None:
"""Transition the test from ``red_review`` back to ``red_executing``.
Called when the assigned Red Lead sends the work back for rework.
Resets the red-phase timer for a fresh attempt (the first attempt's
worklog was already recorded at submit time, so this loses nothing).
"""
self._transition(TestState.red_executing)
self.red_started_at = datetime.utcnow()
self.red_paused_seconds = 0
self._events.append(DomainEvent("red_review_reopened"))
# Define function submit_blue_evidence
def submit_blue_evidence(self) -> int:
"""Transition the test from ``blue_evaluating`` to ``in_review``.
"""Transition the test from ``blue_evaluating`` to ``blue_review``.
Auto-resumes if paused. Returns paused seconds accumulated
during this phase (for worklog calculation).
@@ -403,7 +430,7 @@ class TestEntity:
# Assign paused_extra = self._auto_resume()
paused_extra = self._auto_resume()
# Call self._transition()
self._transition(TestState.in_review)
self._transition(TestState.blue_review)
# Assign total_paused = self.blue_paused_seconds + paused_extra
total_paused = self.blue_paused_seconds + paused_extra
# Call self._events.append()
@@ -415,6 +442,39 @@ class TestEntity:
# Return total_paused
return total_paused
def approve_blue_review(self) -> None:
"""Transition the test from ``blue_review`` to ``in_review``.
Called when the assigned Blue Lead approves the operator's work.
"""
self._transition(TestState.in_review)
self._events.append(DomainEvent("blue_review_approved"))
def reopen_blue_review(self) -> None:
"""Transition the test from ``blue_review`` back to ``blue_evaluating``.
Called when the assigned Blue Lead sends the work back for rework.
Resets the blue-phase timer for a fresh attempt. Note:
``blue_work_started_at`` (the pickup timestamp) lives only on the
ORM model, not on this entity — the service layer resets it there.
"""
self._transition(TestState.blue_evaluating)
self.blue_started_at = datetime.utcnow()
self.blue_paused_seconds = 0
self._events.append(DomainEvent("blue_review_reopened"))
def flag_blue_review_gap(self) -> None:
"""Transition the test from ``blue_review`` to ``in_review``.
Called when the assigned Blue Lead determines the shortfall is a
missing capability, not operator error — the test still proceeds
to cross-validation since a retry can't fix a capability gap. The
gap description itself is stored on the ORM model
(``Test.system_gaps``), not on this entity.
"""
self._transition(TestState.in_review)
self._events.append(DomainEvent("blue_review_gap_flagged"))
# Define function pause_timer
def pause_timer(self) -> None:
"""Pause the active phase timer.