feat(backend): enforce campaign scheduling, add reviewer queue filter, tactic order, duplicate-test counts
- 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
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
"""Tests for the campaign start_date scheduling gate on test execution.
|
||||
|
||||
Block 3 fix: `Campaign.start_date` was persisted and displayed but never
|
||||
actually enforced — a test belonging to a campaign scheduled for the future
|
||||
could still be started immediately. `start_execution` now blocks this.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from app.domain.errors import BusinessRuleViolation
|
||||
from app.models.campaign import Campaign, CampaignTest
|
||||
from app.models.enums import TestState
|
||||
from app.models.technique import Technique
|
||||
from app.models.test import Test
|
||||
from app.services.test_workflow_service import start_execution
|
||||
|
||||
|
||||
def _seed_test_in_campaign(db, owner_id, start_date):
|
||||
tech = Technique(mitre_id="T1059.003", name="Windows Command Shell", tactic="execution", platforms=["windows"])
|
||||
db.add(tech)
|
||||
db.flush()
|
||||
|
||||
campaign = Campaign(name="Scheduled Campaign", type="custom", status="active", created_by=owner_id, start_date=start_date)
|
||||
db.add(campaign)
|
||||
db.flush()
|
||||
|
||||
test = Test(technique_id=tech.id, name="Scheduled test", state=TestState.draft, created_by=owner_id)
|
||||
db.add(test)
|
||||
db.flush()
|
||||
|
||||
db.add(CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0))
|
||||
db.commit()
|
||||
db.refresh(test)
|
||||
return test
|
||||
|
||||
|
||||
def test_start_execution_blocked_before_campaign_start_date(db, red_tech_user):
|
||||
future = datetime.utcnow() + timedelta(days=7)
|
||||
test = _seed_test_in_campaign(db, red_tech_user.id, future)
|
||||
|
||||
with pytest.raises(BusinessRuleViolation):
|
||||
start_execution(db, test, red_tech_user)
|
||||
|
||||
|
||||
def test_start_execution_allowed_after_campaign_start_date(db, red_tech_user):
|
||||
past = datetime.utcnow() - timedelta(days=1)
|
||||
test = _seed_test_in_campaign(db, red_tech_user.id, past)
|
||||
|
||||
result = start_execution(db, test, red_tech_user)
|
||||
|
||||
assert result.state == TestState.red_executing
|
||||
|
||||
|
||||
def test_start_execution_allowed_for_standalone_test(db, red_tech_user):
|
||||
"""A test with no campaign at all is never gated by this check."""
|
||||
tech = Technique(mitre_id="T1059.004", name="Unix Shell", tactic="execution", platforms=["linux"])
|
||||
db.add(tech)
|
||||
db.flush()
|
||||
test = Test(technique_id=tech.id, name="Standalone test", state=TestState.draft, created_by=red_tech_user.id)
|
||||
db.add(test)
|
||||
db.commit()
|
||||
db.refresh(test)
|
||||
|
||||
result = start_execution(db, test, red_tech_user)
|
||||
|
||||
assert result.state == TestState.red_executing
|
||||
|
||||
|
||||
def test_start_execution_allowed_when_campaign_has_no_start_date(db, red_tech_user):
|
||||
test = _seed_test_in_campaign(db, red_tech_user.id, None)
|
||||
|
||||
result = start_execution(db, test, red_tech_user)
|
||||
|
||||
assert result.state == TestState.red_executing
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Tests for canonical MITRE kill-chain ordering in coverage-by-tactic.
|
||||
|
||||
Block 3 fix: TacticCoverageChart.tsx (dashboard) rendered whatever order
|
||||
the backend happened to return (alphabetical, and only tactics with
|
||||
seeded techniques), while ExecutiveDashboardPage.tsx re-sorted client-side
|
||||
into the correct kill-chain order. get_coverage_by_tactic now returns the
|
||||
canonical order — and all 14 tactics, zero-filled — so both consumers get
|
||||
consistent results without needing their own reordering logic.
|
||||
"""
|
||||
|
||||
from app.models.technique import Technique
|
||||
from app.services.metrics_query_service import MITRE_TACTIC_ORDER, get_coverage_by_tactic
|
||||
|
||||
|
||||
def test_coverage_by_tactic_returns_canonical_kill_chain_order(db):
|
||||
# Seed a few tactics out of order to prove the response isn't alphabetical.
|
||||
db.add(Technique(mitre_id="T3001", name="A", tactic="impact"))
|
||||
db.add(Technique(mitre_id="T3002", name="B", tactic="execution"))
|
||||
db.add(Technique(mitre_id="T3003", name="C", tactic="reconnaissance"))
|
||||
db.commit()
|
||||
|
||||
rows = get_coverage_by_tactic(db)
|
||||
tactics = [r.tactic for r in rows]
|
||||
|
||||
canonical_present = [t for t in tactics if t in MITRE_TACTIC_ORDER]
|
||||
assert canonical_present == MITRE_TACTIC_ORDER
|
||||
|
||||
|
||||
def test_coverage_by_tactic_zero_fills_tactics_with_no_techniques(db):
|
||||
db.add(Technique(mitre_id="T3004", name="D", tactic="execution"))
|
||||
db.commit()
|
||||
|
||||
rows = get_coverage_by_tactic(db)
|
||||
by_tactic = {r.tactic: r for r in rows}
|
||||
|
||||
assert "persistence" in by_tactic
|
||||
assert by_tactic["persistence"].total == 0
|
||||
assert by_tactic["execution"].total == 1
|
||||
|
||||
|
||||
def test_coverage_by_tactic_unknown_tactic_sorts_last(db):
|
||||
db.add(Technique(mitre_id="T3005", name="E", tactic=None))
|
||||
db.commit()
|
||||
|
||||
rows = get_coverage_by_tactic(db)
|
||||
tactics = [r.tactic for r in rows]
|
||||
|
||||
assert tactics[-1] == "unknown"
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Tests for the `reviewer_id` "my reviews" queue filter on GET /tests.
|
||||
|
||||
Block 3 fix: red_lead/blue_lead had no way to see tests specifically
|
||||
assigned to THEM at the red_review/blue_review lead-review gate — the
|
||||
existing `pending_validation_side` filter only covers the later, unrelated
|
||||
in_review manager-validation stage and ignores assignment entirely.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
from app.models.enums import TestState
|
||||
from app.models.technique import Technique
|
||||
from app.models.test import Test
|
||||
from app.services.test_crud_service import list_tests
|
||||
|
||||
|
||||
def _make_test(db, owner_id, state, name, red_reviewer=None, blue_reviewer=None):
|
||||
tech = Technique(mitre_id=f"T9{uuid.uuid4().hex[:6]}", name="X", tactic="execution", platforms=["windows"])
|
||||
db.add(tech)
|
||||
db.flush()
|
||||
test = Test(
|
||||
technique_id=tech.id,
|
||||
name=name,
|
||||
state=state,
|
||||
created_by=owner_id,
|
||||
red_reviewer_assignee=red_reviewer,
|
||||
blue_reviewer_assignee=blue_reviewer,
|
||||
)
|
||||
db.add(test)
|
||||
db.commit()
|
||||
db.refresh(test)
|
||||
return test
|
||||
|
||||
|
||||
def test_reviewer_id_with_explicit_state_filters_by_matching_assignee(db, red_lead_user, blue_lead_user):
|
||||
mine = _make_test(db, red_lead_user.id, TestState.red_review, "Mine", red_reviewer=red_lead_user.id)
|
||||
_make_test(db, red_lead_user.id, TestState.red_review, "Someone else's", red_reviewer=blue_lead_user.id)
|
||||
|
||||
results = list_tests(db, state="red_review", reviewer_id=red_lead_user.id)
|
||||
|
||||
assert [t.id for t in results] == [mine.id]
|
||||
|
||||
|
||||
def test_reviewer_id_without_state_covers_both_red_and_blue_review(db, red_lead_user, blue_lead_user):
|
||||
red = _make_test(db, red_lead_user.id, TestState.red_review, "Red review mine", red_reviewer=red_lead_user.id)
|
||||
blue = _make_test(db, red_lead_user.id, TestState.blue_review, "Blue review mine", blue_reviewer=red_lead_user.id)
|
||||
_make_test(db, red_lead_user.id, TestState.red_review, "Not mine", red_reviewer=blue_lead_user.id)
|
||||
_make_test(db, red_lead_user.id, TestState.draft, "Unrelated state")
|
||||
|
||||
results = list_tests(db, reviewer_id=red_lead_user.id)
|
||||
|
||||
assert {t.id for t in results} == {red.id, blue.id}
|
||||
|
||||
|
||||
def test_reviewer_id_excludes_unassigned_tests_in_that_state(db, red_lead_user):
|
||||
_make_test(db, red_lead_user.id, TestState.red_review, "Unassigned", red_reviewer=None)
|
||||
|
||||
results = list_tests(db, state="red_review", reviewer_id=red_lead_user.id)
|
||||
|
||||
assert results == []
|
||||
@@ -60,7 +60,11 @@ def _make_user(role: str = "red_tech") -> MagicMock:
|
||||
|
||||
|
||||
def _make_db() -> MagicMock:
|
||||
return 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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Tests for the `existing_test_count` field on TestTemplateSummary.
|
||||
|
||||
Block 3 fix: the Test Catalog page had no way to know a technique already
|
||||
had one or more tests, so operators could unknowingly instantiate a
|
||||
duplicate from a template. list_templates now attaches a per-technique
|
||||
test count so the frontend can show a warning badge.
|
||||
"""
|
||||
|
||||
from app.models.enums import TestState
|
||||
from app.models.technique import Technique
|
||||
from app.models.test import Test
|
||||
from app.models.test_template import TestTemplate
|
||||
from app.services.test_template_service import list_templates
|
||||
|
||||
|
||||
def test_existing_test_count_reflects_tests_for_that_technique(db, red_lead_user):
|
||||
tech = Technique(mitre_id="T1055", name="Process Injection", tactic="defense-evasion", platforms=["windows"])
|
||||
db.add(tech)
|
||||
db.flush()
|
||||
db.add(TestTemplate(mitre_technique_id="T1055", name="Injection template", source="custom"))
|
||||
db.add(Test(technique_id=tech.id, name="Existing test 1", state=TestState.draft, created_by=red_lead_user.id))
|
||||
db.add(Test(technique_id=tech.id, name="Existing test 2", state=TestState.validated, created_by=red_lead_user.id))
|
||||
db.commit()
|
||||
|
||||
templates = list_templates(db, mitre_technique_id="T1055")
|
||||
|
||||
assert len(templates) == 1
|
||||
assert templates[0].existing_test_count == 2
|
||||
|
||||
|
||||
def test_existing_test_count_zero_when_no_tests(db):
|
||||
tech = Technique(mitre_id="T1056", name="Input Capture", tactic="collection", platforms=["windows"])
|
||||
db.add(tech)
|
||||
db.flush()
|
||||
db.add(TestTemplate(mitre_technique_id="T1056", name="Capture template", source="custom"))
|
||||
db.commit()
|
||||
|
||||
templates = list_templates(db, mitre_technique_id="T1056")
|
||||
|
||||
assert len(templates) == 1
|
||||
assert templates[0].existing_test_count == 0
|
||||
@@ -153,7 +153,11 @@ def _make_user(role: str = "red_tech") -> MagicMock:
|
||||
|
||||
|
||||
def _make_db() -> MagicMock:
|
||||
return 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
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
|
||||
Reference in New Issue
Block a user