kitos
129 lines
5.4 KiB
Python
129 lines
5.4 KiB
Python
"""Tests for the campaign manager-approval workflow."""
|
|
|
|
import pytest
|
|
|
|
from app.models.campaign import Campaign, 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",
|
|
)
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
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")
|