kitos
4ddd7f9ec3
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
Previously only red_executing (-> "In Progress") ever changed the Jira issue's status; every other Aegis transition (red_review, blue_evaluating, blue_review, in_review, validated, rejected, disputed) only posted a comment, leaving the Jira board stuck on whatever status it started with. Expands push_test_event() to a full TestState -> Jira-status table covering the whole lifecycle (To-Do -> In Progress -> RT Test Review -> Queued Blue Team -> In Progress -> Blue Team Test Review -> Validation -> Done/Rejected/Dispute), matching the Purple Team Jira workflow. Adds two new lifecycle hooks that the generic dispatch can't handle correctly: - push_bt_work_started(): blue_evaluating covers both "queued, unclaimed" and "actively being worked" — needs an explicit push when the blue tech clicks Start Evaluation, or it never leaves "Queued Blue Team". - disputed state push: a conflicting lead vote previously produced zero Jira signal at all. Also fills two real gaps: reopen_red_review and reopen_blue_review never synced anything to Jira before. Their target status differs on purpose — reopen_red_review pushes "In Progress" (Aegis resumes the same operator immediately, no re-claim), while reopen_blue_review pushes "Queued Blue Team" (Aegis clears blue_work_started_at, forcing a fresh "Start Evaluation" click) — verified against the actual field-reset behavior in test_entity.py rather than assumed symmetry between the two teams.
933 lines
36 KiB
Python
933 lines
36 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)
|
|
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"
|
|
|
|
|
|
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_can_resolve_dispute_regardless_of_vote(self, mock_log):
|
|
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
|
|
result = resolve_dispute(db, test, admin, "red")
|
|
assert result.state == TestState.red_executing
|
|
|
|
|
|
# ===========================================================================
|
|
# 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!")
|