kitos
8fe38dc661
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
- Reopening a test for rework (red or blue) now archives the round that's ending into a new test_round_history table before resetting the live fields — procedure/results/detection data is preserved instead of overwritten, and Phase Timeline renders every past round alongside the current one, so Blue Team keeps visibility into earlier Red attempts. - Each archived round also posts a permanent Jira comment (push_round_archived) since Jira's custom fields only ever show the latest value once the next round resubmits — the comment thread is what preserves full history there. - push_pause_event no longer transitions the Jira issue to "On Hold" — a timer pause is not a real Hold, and flipping Jira status for it was misleading. Only the explicit Hold action (push_hold_event) does that now.
1025 lines
41 KiB
Python
1025 lines
41 KiB
Python
"""T-125: Tests del flujo de trabajo Red/Blue.
|
|
|
|
Comprehensive tests covering the full test lifecycle:
|
|
draft -> red_executing -> blue_evaluating -> in_review -> validated/rejected
|
|
|
|
Uses mock objects to test the workflow service and router logic
|
|
without requiring a running database.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import uuid
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from types import ModuleType
|
|
from datetime import datetime
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stub heavy dependencies before importing app modules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if backend_dir not in sys.path:
|
|
sys.path.insert(0, backend_dir)
|
|
|
|
if "pydantic_settings" not in sys.modules:
|
|
_ps = ModuleType("pydantic_settings")
|
|
class _BaseSettings:
|
|
def __init__(self, **kwargs): pass
|
|
def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs)
|
|
_ps.BaseSettings = _BaseSettings
|
|
sys.modules["pydantic_settings"] = _ps
|
|
|
|
if "app.config" not in sys.modules:
|
|
_cfg = ModuleType("app.config")
|
|
class _FakeSettings:
|
|
DATABASE_URL = "sqlite:///:memory:"
|
|
SECRET_KEY = "test"
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 60
|
|
REDIS_URL = "redis://localhost:6379/0"
|
|
MINIO_ENDPOINT = "localhost:9000"
|
|
MINIO_ACCESS_KEY = "test"
|
|
MINIO_SECRET_KEY = "test"
|
|
MINIO_BUCKET = "test"
|
|
MINIO_SECURE = False
|
|
MAX_RETEST_COUNT = 3
|
|
REPORT_TEMPLATES_DIR = "app/templates/reports"
|
|
REPORT_OUTPUT_DIR = "/tmp/aegis_reports"
|
|
COMPANY_NAME = "Test Org"
|
|
COMPANY_LOGO_PATH = "app/templates/reports/assets/logo.png"
|
|
JIRA_ENABLED = False
|
|
JIRA_URL = ""
|
|
JIRA_USERNAME = ""
|
|
JIRA_API_TOKEN = ""
|
|
JIRA_IS_CLOUD = True
|
|
JIRA_DEFAULT_PROJECT = ""
|
|
JIRA_ISSUE_TYPE_TEST = "Task"
|
|
JIRA_ISSUE_TYPE_CAMPAIGN = "Epic"
|
|
TEMPO_ENABLED = False
|
|
TEMPO_API_TOKEN = ""
|
|
TEMPO_DEFAULT_WORK_TYPE = "Red Team"
|
|
NVD_API_KEY = ""
|
|
STALE_THRESHOLD_DAYS = 365
|
|
CORS_ORIGINS = "http://localhost:3000"
|
|
SCORING_WEIGHT_TESTS = 40
|
|
SCORING_WEIGHT_DETECTION_RULES = 20
|
|
SCORING_WEIGHT_D3FEND = 15
|
|
SCORING_WEIGHT_FRESHNESS = 15
|
|
SCORING_WEIGHT_PLATFORM_DIVERSITY = 10
|
|
_cfg.settings = _FakeSettings()
|
|
sys.modules["app.config"] = _cfg
|
|
|
|
if "app.database" not in sys.modules:
|
|
_db = ModuleType("app.database")
|
|
_db.Base = type("Base", (), {"metadata": MagicMock()})
|
|
_db.get_db = MagicMock()
|
|
sys.modules["app.database"] = _db
|
|
|
|
for _mod in [
|
|
"taxii2client", "taxii2client.v20",
|
|
"jose", "boto3", "botocore", "botocore.exceptions",
|
|
"apscheduler", "apscheduler.schedulers",
|
|
"apscheduler.schedulers.background",
|
|
"apscheduler.triggers", "apscheduler.triggers.cron",
|
|
]:
|
|
if _mod not in sys.modules:
|
|
m = ModuleType(_mod)
|
|
if _mod == "taxii2client.v20": m.Server = MagicMock
|
|
elif _mod == "jose": m.JWTError = Exception; m.jwt = MagicMock()
|
|
elif _mod == "boto3": m.client = MagicMock()
|
|
elif _mod == "botocore.exceptions": m.ClientError = Exception
|
|
elif _mod == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock
|
|
elif _mod == "apscheduler.triggers.cron": m.CronTrigger = MagicMock
|
|
sys.modules[_mod] = m
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Imports
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from fastapi import HTTPException
|
|
from app.domain.exceptions import InvalidOperationError, InvalidTransitionError
|
|
from app.models.enums import TestState, TestResult
|
|
from app.models.test import Test
|
|
from app.services.test_workflow_service import (
|
|
VALID_TRANSITIONS,
|
|
can_transition,
|
|
transition_state,
|
|
start_execution,
|
|
submit_red_evidence,
|
|
approve_red_review,
|
|
submit_blue_evidence,
|
|
approve_blue_review,
|
|
validate_as_red_lead,
|
|
validate_as_blue_lead,
|
|
check_dual_validation,
|
|
reopen_test,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_test(state: TestState = TestState.draft, **kwargs) -> MagicMock:
|
|
t = MagicMock()
|
|
t.id = uuid.uuid4()
|
|
t.name = "Test Security Check"
|
|
t.technique_id = uuid.uuid4()
|
|
t.state = state
|
|
t.red_validation_status = kwargs.get("red_validation_status", None)
|
|
t.blue_validation_status = kwargs.get("blue_validation_status", None)
|
|
t.red_validated_by = kwargs.get("red_validated_by", None)
|
|
t.red_validated_at = kwargs.get("red_validated_at", None)
|
|
t.red_validation_notes = kwargs.get("red_validation_notes", None)
|
|
t.blue_validated_by = kwargs.get("blue_validated_by", None)
|
|
t.blue_validated_at = kwargs.get("blue_validated_at", None)
|
|
t.blue_validation_notes = kwargs.get("blue_validation_notes", None)
|
|
t.execution_date = kwargs.get("execution_date", None)
|
|
t.red_started_at = kwargs.get("red_started_at", None)
|
|
t.blue_started_at = kwargs.get("blue_started_at", None)
|
|
t.paused_at = kwargs.get("paused_at", None)
|
|
t.red_paused_seconds = kwargs.get("red_paused_seconds", 0)
|
|
t.blue_paused_seconds = kwargs.get("blue_paused_seconds", 0)
|
|
t.red_round_number = kwargs.get("red_round_number", 1)
|
|
t.blue_round_number = kwargs.get("blue_round_number", 1)
|
|
t.procedure_text = kwargs.get("procedure_text", None)
|
|
t.tool_used = kwargs.get("tool_used", None)
|
|
t.attack_success = kwargs.get("attack_success", None)
|
|
t.red_summary = kwargs.get("red_summary", None)
|
|
t.execution_start_time = kwargs.get("execution_start_time", None)
|
|
t.execution_end_time = kwargs.get("execution_end_time", None)
|
|
t.detection_result = kwargs.get("detection_result", None)
|
|
t.containment_result = kwargs.get("containment_result", None)
|
|
t.detection_time = kwargs.get("detection_time", None)
|
|
t.containment_time = kwargs.get("containment_time", None)
|
|
t.blue_summary = kwargs.get("blue_summary", None)
|
|
t.red_tech_assignee = kwargs.get("red_tech_assignee", None)
|
|
t.blue_tech_assignee = kwargs.get("blue_tech_assignee", None)
|
|
return t
|
|
|
|
|
|
def _make_user(role: str = "red_tech") -> MagicMock:
|
|
user = MagicMock()
|
|
user.id = uuid.uuid4()
|
|
user.role = role
|
|
return user
|
|
|
|
|
|
def _make_db() -> MagicMock:
|
|
db = MagicMock()
|
|
# start_execution's campaign-schedule lookup (.query(Campaign).join(...).filter(...).first())
|
|
# defaults to "no campaign found" so unrelated tests aren't gated by a MagicMock date.
|
|
db.query.return_value.join.return_value.filter.return_value.first.return_value = None
|
|
return db
|
|
|
|
|
|
# ===========================================================================
|
|
# 1. test_full_happy_path
|
|
# draft -> red_executing -> blue_evaluating -> in_review -> validated
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.select_reviewer")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_full_happy_path(mock_log, mock_select_reviewer):
|
|
"""draft -> red_executing -> red_review -> blue_evaluating -> blue_review -> in_review -> validated"""
|
|
test = _make_test(TestState.draft)
|
|
red_tech = _make_user("red_tech")
|
|
blue_tech = _make_user("blue_tech")
|
|
red_lead = _make_user("red_lead")
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
mock_select_reviewer.side_effect = [red_lead, blue_lead]
|
|
|
|
# Step 1: draft -> red_executing
|
|
result = start_execution(db, test, red_tech)
|
|
assert result.state == TestState.red_executing
|
|
assert result.execution_date is not None
|
|
|
|
# Step 2: red_executing -> red_review -> (Red Lead approves) -> blue_evaluating
|
|
result = submit_red_evidence(db, result, red_tech)
|
|
assert result.state == TestState.red_review
|
|
result = approve_red_review(db, result, red_lead)
|
|
assert result.state == TestState.blue_evaluating
|
|
|
|
# Step 3: blue_evaluating -> blue_review -> (Blue Lead approves) -> in_review
|
|
result = submit_blue_evidence(db, result, blue_tech)
|
|
assert result.state == TestState.blue_review
|
|
result = approve_blue_review(db, result, blue_lead)
|
|
assert result.state == TestState.in_review
|
|
|
|
# Step 4: Red Lead approves
|
|
result = validate_as_red_lead(db, result, red_lead, "approved", "Attack well documented")
|
|
assert result.red_validation_status == "approved"
|
|
assert result.red_validated_by == red_lead.id
|
|
assert result.red_validated_at is not None
|
|
assert result.red_validation_notes == "Attack well documented"
|
|
# Still in_review (waiting for blue lead)
|
|
assert result.state == TestState.in_review
|
|
|
|
# Step 5: Blue Lead approves -> validated
|
|
result = validate_as_blue_lead(db, result, blue_lead, "approved", "Detection confirmed")
|
|
assert result.blue_validation_status == "approved"
|
|
assert result.state == TestState.validated
|
|
|
|
# Verify audit logs were generated at each step
|
|
assert mock_log.call_count >= 5
|
|
|
|
|
|
# ===========================================================================
|
|
# 2. test_rejection_and_reopen
|
|
# in_review -> rejected -> draft -> red_executing -> ...
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.select_reviewer")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_rejection_and_reopen(mock_log, mock_select_reviewer):
|
|
"""in_review -> rejected -> draft -> red_executing -> ..."""
|
|
test = _make_test(TestState.draft)
|
|
red_tech = _make_user("red_tech")
|
|
blue_tech = _make_user("blue_tech")
|
|
red_lead = _make_user("red_lead")
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
mock_select_reviewer.side_effect = [red_lead, blue_lead]
|
|
|
|
# Advance to in_review
|
|
start_execution(db, test, red_tech)
|
|
submit_red_evidence(db, test, red_tech)
|
|
approve_red_review(db, test, red_lead)
|
|
submit_blue_evidence(db, test, blue_tech)
|
|
approve_blue_review(db, test, blue_lead)
|
|
assert test.state == TestState.in_review
|
|
|
|
# Red Lead rejects -> rejected
|
|
validate_as_red_lead(db, test, red_lead, "rejected", "Need more evidence")
|
|
assert test.state == TestState.rejected
|
|
|
|
# Reopen -> draft
|
|
reopen_test(db, test, red_lead)
|
|
assert test.state == TestState.draft
|
|
|
|
# Restart the cycle
|
|
start_execution(db, test, red_tech)
|
|
assert test.state == TestState.red_executing
|
|
|
|
|
|
# ===========================================================================
|
|
# 3. test_invalid_transitions
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_invalid_transitions(mock_log):
|
|
"""Verify that invalid state transitions raise InvalidTransitionError."""
|
|
db = _make_db()
|
|
user = _make_user("admin")
|
|
|
|
# draft -> validated (should fail)
|
|
test = _make_test(TestState.draft)
|
|
try:
|
|
transition_state(db, test, TestState.validated, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# draft -> blue_evaluating (should fail)
|
|
test = _make_test(TestState.draft)
|
|
try:
|
|
transition_state(db, test, TestState.blue_evaluating, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# red_executing -> in_review (should fail, must go through blue_evaluating)
|
|
test = _make_test(TestState.red_executing)
|
|
try:
|
|
transition_state(db, test, TestState.in_review, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# validated -> anything (terminal state)
|
|
test = _make_test(TestState.validated)
|
|
try:
|
|
transition_state(db, test, TestState.draft, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# rejected -> red_executing (must go through draft first)
|
|
test = _make_test(TestState.rejected)
|
|
try:
|
|
transition_state(db, test, TestState.red_executing, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
|
|
# ===========================================================================
|
|
# 4. test_red_tech_cannot_access_blue_phase
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_red_tech_cannot_access_blue_phase(mock_log):
|
|
"""Red tech cannot submit blue evidence (wrong transition from wrong state)."""
|
|
db = _make_db()
|
|
red_tech = _make_user("red_tech")
|
|
|
|
# A test in red_executing cannot jump to in_review
|
|
test = _make_test(TestState.red_executing)
|
|
try:
|
|
submit_blue_evidence(db, test, red_tech)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# Red tech cannot validate (test must be in blue_evaluating for submit_blue)
|
|
test2 = _make_test(TestState.draft)
|
|
try:
|
|
submit_blue_evidence(db, test2, red_tech)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
|
|
# ===========================================================================
|
|
# 5. test_blue_tech_cannot_access_red_phase
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_blue_tech_cannot_access_red_phase(mock_log):
|
|
"""Blue tech cannot start execution or submit red evidence."""
|
|
db = _make_db()
|
|
blue_tech = _make_user("blue_tech")
|
|
|
|
# Blue tech cannot start execution (test must be in draft -> red_executing)
|
|
# The workflow service doesn't check role, but the router does.
|
|
# At service level, blue_evaluating -> blue_evaluating is invalid transition:
|
|
test = _make_test(TestState.blue_evaluating)
|
|
try:
|
|
start_execution(db, test, blue_tech)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
# Blue tech cannot submit red evidence on a draft test
|
|
test2 = _make_test(TestState.draft)
|
|
try:
|
|
submit_red_evidence(db, test2, blue_tech)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
|
|
# ===========================================================================
|
|
# 6. test_dual_validation_both_approve
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_both_approve(mock_log):
|
|
"""Both managers approve -> test becomes validated."""
|
|
test = _make_test(TestState.in_review)
|
|
red_lead = _make_user("red_lead")
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
# Red Lead approves first
|
|
validate_as_red_lead(db, test, red_lead, "approved", "LGTM")
|
|
assert test.red_validation_status == "approved"
|
|
# Not yet validated — waiting for blue
|
|
assert test.state == TestState.in_review
|
|
|
|
# Blue Lead approves
|
|
validate_as_blue_lead(db, test, blue_lead, "approved", "Detection verified")
|
|
assert test.blue_validation_status == "approved"
|
|
assert test.state == TestState.validated
|
|
|
|
|
|
# ===========================================================================
|
|
# 7. test_dual_validation_one_rejects
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_one_rejects(mock_log):
|
|
"""One manager rejects -> test becomes rejected immediately."""
|
|
test = _make_test(TestState.in_review)
|
|
red_lead = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
validate_as_red_lead(db, test, red_lead, "rejected", "Insufficient evidence")
|
|
assert test.red_validation_status == "rejected"
|
|
assert test.state == TestState.rejected
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_blue_rejects_first(mock_log):
|
|
"""Blue Lead rejects first -> test becomes rejected immediately."""
|
|
test = _make_test(TestState.in_review)
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
validate_as_blue_lead(db, test, blue_lead, "rejected", "Detection not adequate")
|
|
assert test.blue_validation_status == "rejected"
|
|
assert test.state == TestState.rejected
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_red_approves_blue_rejects(mock_log):
|
|
"""Red approves, then blue rejects -> genuine conflict -> disputed."""
|
|
test = _make_test(TestState.in_review)
|
|
red_lead = _make_user("red_lead")
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
validate_as_red_lead(db, test, red_lead, "approved", "Good attack")
|
|
assert test.state == TestState.in_review # waiting for blue
|
|
|
|
validate_as_blue_lead(db, test, blue_lead, "rejected", "Bad detection")
|
|
assert test.state == TestState.disputed
|
|
|
|
|
|
# ===========================================================================
|
|
# 8. test_evidence_team_separation
|
|
# ===========================================================================
|
|
|
|
|
|
def test_evidence_team_separation():
|
|
"""Verify evidence router logic separates red and blue evidence correctly."""
|
|
from app.domain.errors import BusinessRuleViolation, PermissionViolation
|
|
from app.models.enums import TeamSide
|
|
from app.services.evidence_service import validate_upload_permission
|
|
|
|
# Red tech can upload red evidence in draft
|
|
test = _make_test(TestState.draft)
|
|
red_user = _make_user("red_tech")
|
|
red_user.role = "red_tech"
|
|
validate_upload_permission(test, TeamSide.red, red_user.role) # should not raise
|
|
|
|
# Red tech can upload red evidence in red_executing
|
|
test.state = TestState.red_executing
|
|
validate_upload_permission(test, TeamSide.red, red_user.role) # should not raise
|
|
|
|
# Red tech CANNOT upload red evidence in blue_evaluating (state violation -> 400)
|
|
test.state = TestState.blue_evaluating
|
|
try:
|
|
validate_upload_permission(test, TeamSide.red, red_user.role)
|
|
assert False, "Should have raised BusinessRuleViolation"
|
|
except BusinessRuleViolation:
|
|
pass
|
|
|
|
# Red tech CANNOT upload blue evidence (role violation -> 403)
|
|
test.state = TestState.blue_evaluating
|
|
try:
|
|
validate_upload_permission(test, TeamSide.blue, red_user.role)
|
|
assert False, "Should have raised PermissionViolation"
|
|
except PermissionViolation:
|
|
pass
|
|
|
|
# Blue tech can upload blue evidence in blue_evaluating
|
|
test.state = TestState.blue_evaluating
|
|
blue_user = _make_user("blue_tech")
|
|
blue_user.role = "blue_tech"
|
|
validate_upload_permission(test, TeamSide.blue, blue_user.role) # should not raise
|
|
|
|
# Blue tech CANNOT upload blue evidence in draft (state violation -> 400)
|
|
test.state = TestState.draft
|
|
try:
|
|
validate_upload_permission(test, TeamSide.blue, blue_user.role)
|
|
assert False, "Should have raised BusinessRuleViolation"
|
|
except BusinessRuleViolation:
|
|
pass
|
|
|
|
# Blue tech CANNOT upload red evidence (role violation -> 403)
|
|
test.state = TestState.draft
|
|
try:
|
|
validate_upload_permission(test, TeamSide.red, blue_user.role)
|
|
assert False, "Should have raised PermissionViolation"
|
|
except PermissionViolation:
|
|
pass
|
|
|
|
|
|
# ===========================================================================
|
|
# 9. test_red_edit_allowed_in_draft_and_red_executing
|
|
# ===========================================================================
|
|
|
|
|
|
def test_red_edit_allowed_in_draft_and_red_executing():
|
|
"""Verify the red update checks that state is draft or red_executing."""
|
|
from app.services.test_crud_service import update_test_red
|
|
import inspect
|
|
source = inspect.getsource(update_test_red)
|
|
|
|
# The service must guard against states other than draft/red_executing
|
|
assert "draft" in source, "Red update must allow draft state"
|
|
assert "red_executing" in source, "Red update must allow red_executing state"
|
|
assert "BusinessRuleViolation" in source, "Must raise domain exception for invalid state (mapped to 400)"
|
|
|
|
|
|
# ===========================================================================
|
|
# 10. test_reopen_clears_validation_fields
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_clears_validation_fields(mock_log):
|
|
"""Reopen clears all red/blue validation status, notes, timestamps."""
|
|
test = _make_test(
|
|
TestState.rejected,
|
|
red_validation_status="rejected",
|
|
red_validated_by=uuid.uuid4(),
|
|
red_validated_at=datetime.utcnow(),
|
|
red_validation_notes="Bad attack",
|
|
blue_validation_status="approved",
|
|
blue_validated_by=uuid.uuid4(),
|
|
blue_validated_at=datetime.utcnow(),
|
|
blue_validation_notes="Good detection",
|
|
)
|
|
user = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
result = reopen_test(db, test, user)
|
|
|
|
assert result.state == TestState.draft
|
|
assert result.red_validation_status is None
|
|
assert result.red_validated_by is None
|
|
assert result.red_validated_at is None
|
|
assert result.red_validation_notes is None
|
|
assert result.blue_validation_status is None
|
|
assert result.blue_validated_by is None
|
|
assert result.blue_validated_at is None
|
|
assert result.blue_validation_notes is None
|
|
db.flush.assert_called()
|
|
|
|
|
|
# ===========================================================================
|
|
# 11. test_cannot_validate_outside_in_review
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_cannot_validate_outside_in_review(mock_log):
|
|
"""Managers cannot validate a test that is not in in_review state."""
|
|
db = _make_db()
|
|
red_lead = _make_user("red_lead")
|
|
blue_lead = _make_user("blue_lead")
|
|
|
|
for state in [TestState.draft, TestState.red_executing, TestState.blue_evaluating, TestState.validated, TestState.rejected]:
|
|
test = _make_test(state)
|
|
try:
|
|
validate_as_red_lead(db, test, red_lead, "approved", "OK")
|
|
assert False, f"Red Lead should not validate in {state.value}"
|
|
except InvalidOperationError as exc:
|
|
assert exc.code == "INVALID_OPERATION"
|
|
|
|
test2 = _make_test(state)
|
|
try:
|
|
validate_as_blue_lead(db, test2, blue_lead, "approved", "OK")
|
|
assert False, f"Blue Lead should not validate in {state.value}"
|
|
except InvalidOperationError as exc:
|
|
assert exc.code == "INVALID_OPERATION"
|
|
|
|
|
|
# ===========================================================================
|
|
# 12. test_cannot_reopen_non_rejected_test
|
|
# ===========================================================================
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_cannot_reopen_non_rejected_test(mock_log):
|
|
"""Reopen only works on rejected tests."""
|
|
db = _make_db()
|
|
user = _make_user("red_lead")
|
|
|
|
for state in [TestState.draft, TestState.red_executing, TestState.blue_evaluating, TestState.in_review, TestState.validated]:
|
|
test = _make_test(state)
|
|
try:
|
|
reopen_test(db, test, user)
|
|
assert False, f"Should not reopen from {state.value}"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ===========================================================================
|
|
# 12b. Review-decision functions (approve/reopen/gap)
|
|
# ===========================================================================
|
|
|
|
|
|
class TestReviewDecisions:
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_approve_red_review_moves_to_blue_evaluating(self, mock_log):
|
|
test = _make_test(TestState.red_review)
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import approve_red_review
|
|
result = approve_red_review(db, test, reviewer, notes="looks good")
|
|
|
|
assert result.state == TestState.blue_evaluating
|
|
assert result.red_review_by == reviewer.id
|
|
assert result.red_review_notes == "looks good"
|
|
assert result.blue_started_at is not None
|
|
|
|
def test_reopen_red_review_requires_notes(self):
|
|
test = _make_test(TestState.red_review)
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_red_review
|
|
with pytest.raises(InvalidOperationError):
|
|
reopen_red_review(db, test, reviewer, notes="")
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_red_review_moves_to_red_executing_with_notes(self, mock_log):
|
|
test = _make_test(TestState.red_review, red_tech_assignee=uuid.uuid4())
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_red_review
|
|
result = reopen_red_review(db, test, reviewer, notes="add more detail")
|
|
|
|
assert result.state == TestState.red_executing
|
|
assert result.red_review_notes == "add more detail"
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_approve_blue_review_moves_to_in_review(self, mock_log):
|
|
test = _make_test(TestState.blue_review)
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import approve_blue_review
|
|
result = approve_blue_review(db, test, reviewer)
|
|
|
|
assert result.state == TestState.in_review
|
|
|
|
def test_reopen_blue_review_requires_notes(self):
|
|
test = _make_test(TestState.blue_review)
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_blue_review
|
|
with pytest.raises(InvalidOperationError):
|
|
reopen_blue_review(db, test, reviewer, notes=None)
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_blue_review_moves_to_blue_evaluating(self, mock_log):
|
|
test = _make_test(TestState.blue_review, blue_tech_assignee=uuid.uuid4())
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_blue_review
|
|
result = reopen_blue_review(db, test, reviewer, notes="redo it")
|
|
|
|
assert result.state == TestState.blue_evaluating
|
|
assert result.blue_work_started_at is None
|
|
|
|
def test_flag_blue_review_gap_requires_system_gaps_text(self):
|
|
test = _make_test(TestState.blue_review)
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import flag_blue_review_gap
|
|
with pytest.raises(InvalidOperationError):
|
|
flag_blue_review_gap(db, test, reviewer, system_gaps="")
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_flag_blue_review_gap_moves_to_in_review(self, mock_log):
|
|
test = _make_test(TestState.blue_review)
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import flag_blue_review_gap
|
|
result = flag_blue_review_gap(db, test, reviewer, system_gaps="Missing EDR agent on host X")
|
|
|
|
assert result.state == TestState.in_review
|
|
assert result.system_gaps == "Missing EDR agent on host X"
|
|
|
|
@patch("app.services.jira_service.push_test_event")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_red_review_pushes_in_progress_to_jira(self, mock_log, mock_push):
|
|
"""Sent back for RT rework resumes the same operator immediately —
|
|
must push 'In Progress', not require a fresh queue/re-claim."""
|
|
test = _make_test(TestState.red_review, red_tech_assignee=uuid.uuid4())
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_red_review
|
|
reopen_red_review(db, test, reviewer, notes="add more detail")
|
|
|
|
mock_push.assert_called_once()
|
|
args, kwargs = mock_push.call_args
|
|
assert args[3] == "red_executing"
|
|
|
|
@patch("app.services.jira_service.push_test_event")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_blue_review_pushes_blue_evaluating_to_jira(self, mock_log, mock_push):
|
|
"""Sent back for BT rework clears blue_work_started_at (the operator
|
|
must re-claim via 'Start Evaluation') — must push 'Queued Blue Team',
|
|
the same status a first-time entry gets."""
|
|
test = _make_test(TestState.blue_review, blue_tech_assignee=uuid.uuid4())
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import reopen_blue_review
|
|
reopen_blue_review(db, test, reviewer, notes="redo it")
|
|
|
|
mock_push.assert_called_once()
|
|
args, kwargs = mock_push.call_args
|
|
assert args[3] == "blue_evaluating"
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_red_review_archives_round_and_resets_fields(self, mock_log):
|
|
"""Reopening must not erase what Red already submitted — it archives
|
|
the round (visible via round_history) and only then blanks the live
|
|
fields for a fresh attempt, bumping the round counter."""
|
|
from app.models.test_round_history import TestRoundHistory
|
|
|
|
test = _make_test(
|
|
TestState.red_review,
|
|
procedure_text="ran mimikatz", tool_used="mimikatz",
|
|
attack_success="successful", red_summary="Got a shell.",
|
|
)
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
with patch("app.services.jira_service.push_test_event"), \
|
|
patch("app.services.jira_service.push_round_archived"):
|
|
from app.services.test_workflow_service import reopen_red_review
|
|
reopen_red_review(db, test, reviewer, notes="incomplete evidence")
|
|
|
|
archived = [c.args[0] for c in db.add.call_args_list if isinstance(c.args[0], TestRoundHistory)]
|
|
assert len(archived) == 1
|
|
assert archived[0].team == "red"
|
|
assert archived[0].round_number == 1
|
|
assert archived[0].attack_success == "successful"
|
|
assert archived[0].red_summary == "Got a shell."
|
|
assert archived[0].review_notes == "incomplete evidence"
|
|
|
|
assert test.procedure_text is None
|
|
assert test.tool_used is None
|
|
assert test.attack_success is None
|
|
assert test.red_summary is None
|
|
assert test.red_round_number == 2
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_blue_review_archives_round_and_resets_fields(self, mock_log):
|
|
from app.models.test_round_history import TestRoundHistory
|
|
|
|
test = _make_test(
|
|
TestState.blue_review,
|
|
detection_result="detected", blue_summary="Caught it via EDR.",
|
|
)
|
|
reviewer = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
with patch("app.services.jira_service.push_test_event"), \
|
|
patch("app.services.jira_service.push_round_archived"):
|
|
from app.services.test_workflow_service import reopen_blue_review
|
|
reopen_blue_review(db, test, reviewer, notes="missing containment steps")
|
|
|
|
archived = [c.args[0] for c in db.add.call_args_list if isinstance(c.args[0], TestRoundHistory)]
|
|
assert len(archived) == 1
|
|
assert archived[0].team == "blue"
|
|
assert archived[0].round_number == 1
|
|
assert archived[0].detection_result == "detected"
|
|
assert archived[0].blue_summary == "Caught it via EDR."
|
|
assert archived[0].review_notes == "missing containment steps"
|
|
|
|
assert test.detection_result is None
|
|
assert test.blue_summary is None
|
|
assert test.blue_round_number == 2
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_red_review_pushes_round_archived_comment_to_jira(self, mock_log):
|
|
test = _make_test(TestState.red_review)
|
|
reviewer = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
with patch("app.services.jira_service.push_test_event"), \
|
|
patch("app.services.jira_service.push_round_archived") as mock_archive:
|
|
from app.services.test_workflow_service import reopen_red_review
|
|
reopen_red_review(db, test, reviewer, notes="redo it")
|
|
|
|
mock_archive.assert_called_once()
|
|
|
|
|
|
class TestStartBlueWork:
|
|
@patch("app.services.jira_service.push_bt_work_started")
|
|
@patch("app.services.jira_service.push_bt_started")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_start_blue_work_pushes_in_progress(self, mock_log, mock_push_started, mock_push_work_started):
|
|
test = _make_test(TestState.blue_evaluating)
|
|
test.blue_work_started_at = None
|
|
blue_tech = _make_user("blue_tech")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import start_blue_work
|
|
result = start_blue_work(db, test, blue_tech)
|
|
|
|
assert result.blue_work_started_at is not None
|
|
mock_push_work_started.assert_called_once()
|
|
|
|
|
|
class TestDisputedJiraSync:
|
|
@patch("app.services.jira_service.push_test_event")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_disputed_pushes_dispute_status(self, mock_log, mock_push):
|
|
"""A conflicting vote (one lead approved, the other rejected) must
|
|
surface as a distinct 'Dispute' status in Jira — previously this
|
|
transition posted no Jira signal at all."""
|
|
test = _make_test(
|
|
TestState.in_review,
|
|
red_validation_status="approved",
|
|
)
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import validate_as_blue_lead
|
|
result = validate_as_blue_lead(db, test, blue_lead, "rejected", notes="Not detected")
|
|
|
|
assert result.state == TestState.disputed
|
|
mock_push.assert_called_once()
|
|
args, kwargs = mock_push.call_args
|
|
assert args[3] == "disputed"
|
|
|
|
|
|
# ===========================================================================
|
|
# 12c. resolve_dispute — flip approver's vote to reject, route to a team
|
|
# ===========================================================================
|
|
|
|
|
|
class TestResolveDispute:
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_red_lead_approver_flips_to_red_queue(self, mock_log):
|
|
test = _make_test(
|
|
TestState.disputed,
|
|
red_validation_status="approved",
|
|
blue_validation_status="rejected",
|
|
)
|
|
red_lead = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import resolve_dispute
|
|
result = resolve_dispute(db, test, red_lead, "red", notes="redo the attack")
|
|
|
|
assert result.state == TestState.red_executing
|
|
assert result.red_validation_status is None
|
|
assert result.blue_validation_status is None
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_blue_lead_approver_flips_to_blue_queue(self, mock_log):
|
|
test = _make_test(
|
|
TestState.disputed,
|
|
blue_validation_status="approved",
|
|
red_validation_status="rejected",
|
|
)
|
|
blue_lead = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import resolve_dispute
|
|
result = resolve_dispute(db, test, blue_lead, "blue")
|
|
|
|
assert result.state == TestState.blue_evaluating
|
|
|
|
def test_non_approver_cannot_resolve_dispute(self):
|
|
test = _make_test(
|
|
TestState.disputed,
|
|
red_validation_status="rejected",
|
|
blue_validation_status="approved",
|
|
)
|
|
red_lead = _make_user("red_lead") # red_lead REJECTED, didn't approve
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import resolve_dispute
|
|
with pytest.raises(InvalidOperationError):
|
|
resolve_dispute(db, test, red_lead, "red")
|
|
|
|
def test_invalid_target_team_rejected(self):
|
|
test = _make_test(
|
|
TestState.disputed,
|
|
red_validation_status="approved",
|
|
blue_validation_status="rejected",
|
|
)
|
|
red_lead = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import resolve_dispute
|
|
with pytest.raises(InvalidOperationError):
|
|
resolve_dispute(db, test, red_lead, "purple")
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_admin_cannot_resolve_dispute(self, mock_log):
|
|
"""Admin administers the site — only the lead who approved may flip
|
|
their vote, with no admin override."""
|
|
test = _make_test(
|
|
TestState.disputed,
|
|
red_validation_status="approved",
|
|
blue_validation_status="rejected",
|
|
)
|
|
admin = _make_user("admin")
|
|
db = _make_db()
|
|
|
|
from app.services.test_workflow_service import resolve_dispute
|
|
with pytest.raises(InvalidOperationError):
|
|
resolve_dispute(db, test, admin, "red")
|
|
|
|
|
|
# ===========================================================================
|
|
# 13. select_reviewer — load-balanced reviewer assignment
|
|
# ===========================================================================
|
|
|
|
|
|
class TestReviewerSelection:
|
|
"""Uses the real sqlite `db` fixture from conftest.py (not MagicMock),
|
|
since load-balancing needs real COUNT() queries."""
|
|
|
|
def _make_lead(self, db, username, role="red_lead"):
|
|
from app.models.user import User
|
|
u = User(username=username, role=role, hashed_password="x", is_active=True)
|
|
db.add(u)
|
|
db.flush()
|
|
return u
|
|
|
|
def _make_technique(self, db, mitre_id="T1059"):
|
|
from app.models.technique import Technique
|
|
t = Technique(mitre_id=mitre_id, name="Command Line", tactic="execution")
|
|
db.add(t)
|
|
db.flush()
|
|
return t
|
|
|
|
def test_picks_lead_with_fewest_active_reviews(self, db):
|
|
from app.services.test_workflow_service import select_reviewer
|
|
|
|
lead_a = self._make_lead(db, "reda_selrev")
|
|
lead_b = self._make_lead(db, "redb_selrev")
|
|
tech = self._make_technique(db, "T1059.selrev1")
|
|
|
|
busy_test = Test(
|
|
technique_id=tech.id, name="Busy",
|
|
state=TestState.red_review, red_reviewer_assignee=lead_a.id,
|
|
)
|
|
db.add(busy_test)
|
|
db.commit()
|
|
|
|
chosen = select_reviewer(db, role="red_lead")
|
|
assert chosen.id == lead_b.id
|
|
|
|
def test_excludes_the_submitter_if_they_are_a_lead(self, db):
|
|
from app.services.test_workflow_service import select_reviewer
|
|
from app.domain.exceptions import BusinessRuleViolation
|
|
|
|
lead_a = self._make_lead(db, "reda_excl")
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="No available"):
|
|
select_reviewer(db, role="red_lead", exclude_user_id=lead_a.id)
|
|
|
|
def test_no_candidates_raises_clear_error(self, db):
|
|
from app.services.test_workflow_service import select_reviewer
|
|
from app.domain.exceptions import BusinessRuleViolation
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="No available"):
|
|
select_reviewer(db, role="red_lead")
|
|
|
|
def test_ties_broken_by_username(self, db):
|
|
from app.services.test_workflow_service import select_reviewer
|
|
|
|
self._make_lead(db, "zzz_tie", role="blue_lead")
|
|
self._make_lead(db, "aaa_tie", role="blue_lead")
|
|
db.commit()
|
|
|
|
chosen = select_reviewer(db, role="blue_lead")
|
|
assert chosen.username == "aaa_tie"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("T-125 Validation: Workflow Tests")
|
|
print("=" * 55)
|
|
test_full_happy_path()
|
|
test_rejection_and_reopen()
|
|
test_invalid_transitions()
|
|
test_red_tech_cannot_access_blue_phase()
|
|
test_blue_tech_cannot_access_red_phase()
|
|
test_dual_validation_both_approve()
|
|
test_dual_validation_one_rejects()
|
|
test_dual_validation_blue_rejects_first()
|
|
test_dual_validation_red_approves_blue_rejects()
|
|
test_evidence_team_separation()
|
|
test_red_edit_allowed_in_draft_and_red_executing()
|
|
test_reopen_clears_validation_fields()
|
|
test_cannot_validate_outside_in_review()
|
|
test_cannot_reopen_non_rejected_test()
|
|
print("=" * 55)
|
|
print("ALL T-125 validations PASSED!")
|