"""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. The round is archived (visible via round_history) and only the per-round verdict fields (attack_success, execution start/end) are blanked for a fresh answer — free-text fields (procedure, tool, summary) stay in place for the operator to edit, not retype from scratch.""" 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 == "ran mimikatz" assert test.tool_used == "mimikatz" assert test.red_summary == "Got a shell." assert test.attack_success is None assert test.execution_start_time is None assert test.execution_end_time 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.containment_result is None assert test.blue_summary == "Caught it via EDR." 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!")