kitos
1c8fa436ab
- start_execution now blocks a test from starting before its campaign's start_date, closing a gap where the field was persisted but never enforced - GET /tests gains a reviewer_id "my reviews" filter for the red_review/ blue_review lead-gate stage, distinct from pending_validation_side (which only ever covered the later in_review stage and ignored assignment entirely) - get_coverage_by_tactic now returns all 14 MITRE tactics in canonical kill-chain order instead of an alphabetical, partial list — the regular Dashboard and Executive Dashboard both consume this endpoint and previously disagreed on order/completeness - test-template listing now includes existing_test_count per technique so the catalog can warn before creating a likely-duplicate test
280 lines
9.5 KiB
Python
280 lines
9.5 KiB
Python
"""Validation tests for T-106: Test Workflow Service.
|
|
|
|
Mocked DB sessions exercise the workflow service without PostgreSQL.
|
|
Uses the same import stack as the rest of the suite (``tests.conftest``) so
|
|
this module must **not** replace entries in ``sys.modules`` — doing so broke
|
|
JWT verification in API tests by splitting :mod:`app.auth` (loaded early via
|
|
conftest) from :mod:`app.routers.auth` / :mod:`app.dependencies.auth`.
|
|
"""
|
|
|
|
import uuid
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.domain.exceptions import InvalidTransitionError
|
|
from app.models.enums import TestState
|
|
from app.services.test_workflow_service import (
|
|
approve_red_review,
|
|
can_transition,
|
|
reopen_test,
|
|
start_execution,
|
|
submit_blue_evidence,
|
|
submit_red_evidence,
|
|
transition_state,
|
|
validate_as_blue_lead,
|
|
validate_as_red_lead,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_test(state: TestState = TestState.draft, **kwargs) -> MagicMock:
|
|
t = MagicMock()
|
|
t.id = uuid.uuid4()
|
|
t.name = "Mock Test"
|
|
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 = None
|
|
t.red_validated_at = None
|
|
t.red_validation_notes = None
|
|
t.blue_validated_by = None
|
|
t.blue_validated_at = None
|
|
t.blue_validation_notes = None
|
|
t.execution_date = None
|
|
t.red_started_at = None
|
|
t.blue_started_at = None
|
|
t.paused_at = None
|
|
t.red_paused_seconds = 0
|
|
t.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. draft -> red_executing works
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_draft_to_red_executing(mock_log):
|
|
test = _make_test(TestState.draft)
|
|
user = _make_user("red_tech")
|
|
db = _make_db()
|
|
|
|
result = start_execution(db, test, user)
|
|
|
|
assert result.state == TestState.red_executing
|
|
assert result.execution_date is not None
|
|
db.flush.assert_called()
|
|
mock_log.assert_called()
|
|
print(" [PASS] Transition draft -> red_executing works")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. draft -> validated fails (not allowed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_draft_to_validated_fails(mock_log):
|
|
test = _make_test(TestState.draft)
|
|
user = _make_user("admin")
|
|
db = _make_db()
|
|
|
|
try:
|
|
transition_state(db, test, TestState.validated, user)
|
|
assert False, "Should have raised InvalidTransitionError"
|
|
except InvalidTransitionError as exc:
|
|
assert exc.code == "INVALID_TRANSITION"
|
|
assert exc.current_state == "draft"
|
|
assert exc.target_state == "validated"
|
|
print(" [PASS] Transition draft -> validated correctly fails")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. red_executing -> blue_evaluating works
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.select_reviewer")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_red_executing_to_blue_evaluating(mock_log, mock_select_reviewer):
|
|
test = _make_test(TestState.red_executing)
|
|
user = _make_user("red_tech")
|
|
db = _make_db()
|
|
mock_select_reviewer.return_value = _make_user("red_lead")
|
|
|
|
result = submit_red_evidence(db, test, user)
|
|
|
|
assert result.state == TestState.red_review
|
|
db.flush.assert_called()
|
|
mock_log.assert_called()
|
|
print(" [PASS] Transition red_executing -> red_review works")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. check_dual_validation -> validated when both approved
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_both_approved(mock_log):
|
|
test = _make_test(TestState.in_review)
|
|
user_red = _make_user("red_lead")
|
|
user_blue = _make_user("blue_lead")
|
|
db = _make_db()
|
|
|
|
validate_as_red_lead(db, test, user_red, "approved", "LGTM")
|
|
validate_as_blue_lead(db, test, user_blue, "approved", "Detection OK")
|
|
|
|
assert test.state == TestState.validated
|
|
print(" [PASS] check_dual_validation -> validated when both approved")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. check_dual_validation -> rejected when one rejects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_dual_validation_one_rejected(mock_log):
|
|
test = _make_test(TestState.in_review)
|
|
user_red = _make_user("red_lead")
|
|
db = _make_db()
|
|
|
|
validate_as_red_lead(db, test, user_red, "rejected", "Insufficient evidence")
|
|
|
|
assert test.state == TestState.rejected
|
|
print(" [PASS] check_dual_validation -> rejected when one rejects")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. reopen_test clears validation fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_reopen_clears_validation(mock_log):
|
|
test = _make_test(
|
|
TestState.rejected,
|
|
red_validation_status="rejected",
|
|
blue_validation_status="approved",
|
|
)
|
|
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.blue_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_validated_by is None
|
|
assert result.blue_validated_at is None
|
|
assert result.blue_validation_notes is None
|
|
db.flush.assert_called()
|
|
print(" [PASS] reopen_test clears validation fields and moves to draft")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7. Every transition generates an audit log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@patch("app.services.test_workflow_service.select_reviewer")
|
|
@patch("app.services.test_workflow_service.log_action")
|
|
def test_transitions_generate_audit_logs(mock_log, mock_select_reviewer):
|
|
test = _make_test(TestState.draft)
|
|
user = _make_user("red_tech")
|
|
db = _make_db()
|
|
mock_select_reviewer.return_value = _make_user("red_lead")
|
|
|
|
start_execution(db, test, user)
|
|
assert mock_log.call_count >= 1
|
|
c1 = mock_log.call_count
|
|
|
|
submit_red_evidence(db, test, user)
|
|
assert mock_log.call_count > c1
|
|
c2 = mock_log.call_count
|
|
|
|
approve_red_review(db, test, _make_user("red_lead"))
|
|
assert mock_log.call_count > c2
|
|
c3 = mock_log.call_count
|
|
|
|
submit_blue_evidence(db, test, user)
|
|
assert mock_log.call_count > c3
|
|
|
|
print(" [PASS] Each transition generates an audit log")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 8. can_transition correctness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_can_transition_map():
|
|
test = _make_test(TestState.draft)
|
|
|
|
assert can_transition(test, TestState.red_executing) is True
|
|
assert can_transition(test, TestState.validated) is False
|
|
assert can_transition(test, TestState.blue_evaluating) is False
|
|
|
|
test.state = TestState.red_executing
|
|
assert can_transition(test, TestState.blue_evaluating) is True
|
|
assert can_transition(test, TestState.draft) is False
|
|
|
|
test.state = TestState.blue_evaluating
|
|
assert can_transition(test, TestState.in_review) is True
|
|
|
|
test.state = TestState.in_review
|
|
assert can_transition(test, TestState.validated) is True
|
|
assert can_transition(test, TestState.rejected) is True
|
|
assert can_transition(test, TestState.draft) is False
|
|
|
|
test.state = TestState.rejected
|
|
assert can_transition(test, TestState.draft) is True
|
|
|
|
test.state = TestState.validated
|
|
assert can_transition(test, TestState.draft) is False
|
|
assert can_transition(test, TestState.rejected) is False
|
|
|
|
print(" [PASS] can_transition map is correct")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
print("T-106 Validation: Test Workflow Service")
|
|
print("=" * 50)
|
|
test_draft_to_red_executing()
|
|
test_draft_to_validated_fails()
|
|
test_red_executing_to_blue_evaluating()
|
|
test_dual_validation_both_approved()
|
|
test_dual_validation_one_rejected()
|
|
test_reopen_clears_validation()
|
|
test_transitions_generate_audit_logs()
|
|
test_can_transition_map()
|
|
print("=" * 50)
|
|
print("ALL T-106 validations PASSED!")
|