kitos
12a5484003
Approving a remove_test modification request deletes the underlying Test row, which was cascading through test_id's ON DELETE CASCADE and wiping out the request row itself before it could be read back. Changed to ON DELETE SET NULL so the audit record (justification, reviewer, decision) survives. Adds regression coverage plus double-approve/reject idempotency tests.
368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""Tests for the campaign manager-approval workflow."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.models.campaign import Campaign, CampaignModificationRequest, CampaignTest
|
|
from app.models.technique import Technique
|
|
from app.models.test import Test
|
|
from app.models.enums import TestState
|
|
from app.domain.errors import BusinessRuleViolation, EntityNotFoundError, PermissionViolation
|
|
from app.services.campaign_crud_service import (
|
|
create_campaign,
|
|
submit_campaign_for_approval,
|
|
approve_campaign,
|
|
reject_campaign,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def draft_campaign_with_test(db, admin_user):
|
|
"""A draft campaign with exactly one test, created by admin_user."""
|
|
tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"])
|
|
db.add(tech)
|
|
db.flush()
|
|
|
|
campaign = Campaign(name="Approval Test Campaign", type="custom", status="draft", created_by=admin_user.id)
|
|
db.add(campaign)
|
|
db.flush()
|
|
|
|
test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=admin_user.id)
|
|
db.add(test)
|
|
db.flush()
|
|
|
|
ct = CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0)
|
|
db.add(ct)
|
|
db.commit()
|
|
db.refresh(campaign)
|
|
return campaign
|
|
|
|
|
|
class TestSubmitCampaignForApproval:
|
|
def test_submit_moves_draft_to_pending_approval(self, db, draft_campaign_with_test, admin_user):
|
|
campaign = submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id),
|
|
submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
assert campaign.status == "pending_approval"
|
|
|
|
def test_submit_without_tests_raises(self, db, admin_user):
|
|
campaign = Campaign(name="Empty", type="custom", status="draft", created_by=admin_user.id)
|
|
db.add(campaign)
|
|
db.commit()
|
|
with pytest.raises(BusinessRuleViolation, match="at least one test"):
|
|
submit_campaign_for_approval(
|
|
db, str(campaign.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
|
|
def test_submit_by_non_creator_non_admin_raises(self, db, draft_campaign_with_test, red_lead_user):
|
|
with pytest.raises(PermissionViolation):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id),
|
|
submitter_id=red_lead_user.id, submitter_role="red_lead",
|
|
)
|
|
|
|
def test_submit_non_draft_raises(self, db, draft_campaign_with_test, admin_user):
|
|
draft_campaign_with_test.status = "active"
|
|
db.commit()
|
|
with pytest.raises(BusinessRuleViolation, match="Only draft campaigns"):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id),
|
|
submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
|
|
def test_submit_nonexistent_campaign_raises(self, db, admin_user):
|
|
with pytest.raises(EntityNotFoundError):
|
|
submit_campaign_for_approval(
|
|
db, str(uuid.uuid4()), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
|
|
|
|
class TestApproveCampaign:
|
|
def test_approve_sets_start_date_and_activates(self, db, draft_campaign_with_test, admin_user):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
db.commit()
|
|
campaign = approve_campaign(
|
|
db, str(draft_campaign_with_test.id),
|
|
approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
|
|
)
|
|
assert campaign.status == "active"
|
|
assert campaign.start_date is not None
|
|
assert campaign.approved_by == admin_user.id
|
|
assert campaign.approved_at is not None
|
|
|
|
def test_approve_without_start_date_raises(self, db, draft_campaign_with_test, admin_user):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
db.commit()
|
|
with pytest.raises(BusinessRuleViolation, match="start_date is required"):
|
|
approve_campaign(db, str(draft_campaign_with_test.id), approver_id=admin_user.id, start_date="")
|
|
|
|
def test_approve_non_pending_raises(self, db, draft_campaign_with_test, admin_user):
|
|
with pytest.raises(BusinessRuleViolation, match="pending approval"):
|
|
approve_campaign(
|
|
db, str(draft_campaign_with_test.id),
|
|
approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
|
|
)
|
|
|
|
def test_approve_nonexistent_campaign_raises(self, db, admin_user):
|
|
with pytest.raises(EntityNotFoundError):
|
|
approve_campaign(
|
|
db, str(uuid.uuid4()), approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
|
|
)
|
|
|
|
|
|
class TestRejectCampaign:
|
|
def test_reject_returns_to_draft_with_reason(self, db, draft_campaign_with_test, admin_user):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
db.commit()
|
|
campaign = reject_campaign(
|
|
db, str(draft_campaign_with_test.id),
|
|
rejecter_id=admin_user.id, reason="Missing scope details",
|
|
)
|
|
assert campaign.status == "draft"
|
|
assert campaign.rejection_reason == "Missing scope details"
|
|
|
|
def test_reject_without_reason_raises(self, db, draft_campaign_with_test, admin_user):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
db.commit()
|
|
with pytest.raises(BusinessRuleViolation, match="reason is required"):
|
|
reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="")
|
|
|
|
def test_reject_non_pending_raises(self, db, draft_campaign_with_test, admin_user):
|
|
with pytest.raises(BusinessRuleViolation, match="pending approval"):
|
|
reject_campaign(db, str(draft_campaign_with_test.id), rejecter_id=admin_user.id, reason="No")
|
|
|
|
def test_reject_clears_prior_approval_fields(self, db, draft_campaign_with_test, admin_user):
|
|
submit_campaign_for_approval(
|
|
db, str(draft_campaign_with_test.id), submitter_id=admin_user.id, submitter_role="admin",
|
|
)
|
|
db.commit()
|
|
approve_campaign(
|
|
db, str(draft_campaign_with_test.id),
|
|
approver_id=admin_user.id, start_date="2026-08-01T00:00:00",
|
|
)
|
|
db.commit()
|
|
# Simulate the campaign being sent back into the approval queue
|
|
# (e.g. after a modification request) so reject_campaign has
|
|
# approved_by/approved_at to clear.
|
|
draft_campaign_with_test.status = "pending_approval"
|
|
db.commit()
|
|
|
|
campaign = reject_campaign(
|
|
db, str(draft_campaign_with_test.id),
|
|
rejecter_id=admin_user.id, reason="Needs rework",
|
|
)
|
|
assert campaign.status == "draft"
|
|
assert campaign.approved_by is None
|
|
assert campaign.approved_at is None
|
|
|
|
def test_reject_nonexistent_campaign_raises(self, db, admin_user):
|
|
with pytest.raises(EntityNotFoundError):
|
|
reject_campaign(db, str(uuid.uuid4()), rejecter_id=admin_user.id, reason="No")
|
|
|
|
|
|
from app.services.campaign_crud_service import (
|
|
add_test_to_campaign,
|
|
remove_test_from_campaign,
|
|
create_modification_request,
|
|
approve_modification_request,
|
|
reject_modification_request,
|
|
list_modification_requests,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def active_campaign_with_test(db, draft_campaign_with_test, admin_user):
|
|
"""An active campaign (bypasses the approval flow directly for test setup)."""
|
|
draft_campaign_with_test.status = "active"
|
|
db.commit()
|
|
db.refresh(draft_campaign_with_test)
|
|
return draft_campaign_with_test
|
|
|
|
|
|
class TestDirectTestEditingGate:
|
|
def test_add_test_blocked_on_active_campaign(self, db, active_campaign_with_test, admin_user):
|
|
tech = Technique(mitre_id="T1078", name="Valid Accounts", tactic="persistence", platforms=["windows"])
|
|
db.add(tech)
|
|
db.flush()
|
|
new_test = Test(technique_id=tech.id, name="New test", state=TestState.draft, created_by=admin_user.id)
|
|
db.add(new_test)
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="draft"):
|
|
add_test_to_campaign(db, str(active_campaign_with_test.id), test_id=str(new_test.id))
|
|
|
|
def test_remove_test_blocked_on_active_campaign(self, db, active_campaign_with_test):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
with pytest.raises(BusinessRuleViolation, match="draft"):
|
|
remove_test_from_campaign(db, str(active_campaign_with_test.id), str(ct.id))
|
|
|
|
|
|
class TestModificationRequests:
|
|
def test_create_add_test_request(self, db, active_campaign_with_test, admin_user):
|
|
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
|
|
db.add(tech)
|
|
db.flush()
|
|
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id)
|
|
db.add(new_test)
|
|
db.commit()
|
|
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="add_test",
|
|
test_id=str(new_test.id), justification="Coverage gap found in review",
|
|
)
|
|
assert request.status == "pending"
|
|
|
|
# Underlying campaign is untouched until approval
|
|
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
|
|
assert len(cts) == 1
|
|
|
|
def test_create_request_without_justification_raises(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
with pytest.raises(BusinessRuleViolation, match="justification is required"):
|
|
create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="",
|
|
)
|
|
|
|
def test_create_request_on_non_active_campaign_raises(self, db, draft_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == draft_campaign_with_test.id).first()
|
|
with pytest.raises(BusinessRuleViolation, match="active campaigns"):
|
|
create_modification_request(
|
|
db, str(draft_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Not needed",
|
|
)
|
|
|
|
def test_approve_add_test_request_applies_change(self, db, active_campaign_with_test, admin_user):
|
|
tech = Technique(mitre_id="T1547", name="Boot Autostart", tactic="persistence", platforms=["windows"])
|
|
db.add(tech)
|
|
db.flush()
|
|
new_test = Test(technique_id=tech.id, name="Autostart test", state=TestState.draft, created_by=admin_user.id)
|
|
db.add(new_test)
|
|
db.commit()
|
|
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="add_test",
|
|
test_id=str(new_test.id), justification="Coverage gap found in review",
|
|
)
|
|
db.commit()
|
|
|
|
approved = approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
|
|
assert approved.status == "approved"
|
|
|
|
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
|
|
assert len(cts) == 2
|
|
assert any(ct.test_id == new_test.id for ct in cts)
|
|
|
|
def test_approve_remove_test_request_applies_change(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
request_id = request.id
|
|
|
|
approve_modification_request(db, str(request_id), reviewer_id=admin_user.id)
|
|
|
|
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
|
|
assert len(cts) == 0
|
|
|
|
# The request row must survive the cascade-delete of the underlying Test
|
|
# (ondelete="CASCADE" on CampaignModificationRequest.test_id) — this locks
|
|
# in the fix where the request's status is persisted BEFORE the test is
|
|
# removed, not after.
|
|
reloaded = (
|
|
db.query(CampaignModificationRequest)
|
|
.filter(CampaignModificationRequest.id == request_id)
|
|
.first()
|
|
)
|
|
assert reloaded is not None
|
|
assert reloaded.status == "approved"
|
|
assert reloaded.reviewed_by == admin_user.id
|
|
assert reloaded.reviewed_at is not None
|
|
|
|
def test_approve_already_decided_request_raises(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="Only pending"):
|
|
approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
|
|
|
|
def test_reject_request_leaves_campaign_untouched(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
|
|
rejected = reject_modification_request(
|
|
db, str(request.id), reviewer_id=admin_user.id, review_notes="Still needed for coverage",
|
|
)
|
|
assert rejected.status == "rejected"
|
|
assert rejected.review_notes == "Still needed for coverage"
|
|
|
|
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
|
|
assert len(cts) == 1
|
|
|
|
def test_reject_without_notes_raises(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
with pytest.raises(BusinessRuleViolation, match="Review notes are required"):
|
|
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="")
|
|
|
|
def test_reject_already_decided_request_raises(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
request = create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="Not needed")
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="Only pending"):
|
|
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="Again")
|
|
|
|
def test_list_modification_requests_filters_by_status(self, db, active_campaign_with_test, admin_user):
|
|
ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).first()
|
|
create_modification_request(
|
|
db, str(active_campaign_with_test.id),
|
|
requester_id=admin_user.id, action="remove_test",
|
|
test_id=str(ct.test_id), justification="Test superseded",
|
|
)
|
|
db.commit()
|
|
|
|
pending = list_modification_requests(db, status="pending")
|
|
assert len(pending) == 1
|
|
assert pending[0]["action"] == "remove_test"
|