diff --git a/backend/app/services/campaign_crud_service.py b/backend/app/services/campaign_crud_service.py index 8ec30d5..384d12a 100644 --- a/backend/app/services/campaign_crud_service.py +++ b/backend/app/services/campaign_crud_service.py @@ -306,6 +306,98 @@ def create_campaign( return serialize_campaign(db, campaign) +def submit_campaign_for_approval( + db: Session, + campaign_id: str, + *, + submitter_id: uuid.UUID, + submitter_role: str, +) -> Campaign: + """Submit a draft campaign into the manager's approval queue. + + Raises EntityNotFoundError, PermissionViolation, BusinessRuleViolation. + Does not commit; caller commits. + """ + campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() + if not campaign: + raise EntityNotFoundError("Campaign", campaign_id) + + if str(campaign.created_by) != str(submitter_id) and submitter_role != "admin": + raise PermissionViolation("Only the creator or admin can submit this campaign for approval") + + if campaign.status != "draft": + raise BusinessRuleViolation("Only draft campaigns can be submitted for approval") + + test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).count() + if test_count == 0: + raise BusinessRuleViolation("Campaign must have at least one test to submit for approval") + + campaign.status = "pending_approval" + db.flush() + return campaign + + +def approve_campaign( + db: Session, + campaign_id: str, + *, + approver_id: uuid.UUID, + start_date: str, +) -> Campaign: + """Approve a pending campaign: fix its start date and activate it. + + Raises EntityNotFoundError, BusinessRuleViolation. + Does not commit; caller commits. + """ + campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() + if not campaign: + raise EntityNotFoundError("Campaign", campaign_id) + + if campaign.status != "pending_approval": + raise BusinessRuleViolation("Only campaigns pending approval can be approved") + + if not start_date: + raise BusinessRuleViolation("start_date is required to approve a campaign") + + campaign.start_date = datetime.fromisoformat(start_date) + campaign.status = "active" + campaign.approved_by = approver_id + campaign.approved_at = datetime.utcnow() + campaign.rejection_reason = None + db.flush() + return campaign + + +def reject_campaign( + db: Session, + campaign_id: str, + *, + rejecter_id: uuid.UUID, + reason: str, +) -> Campaign: + """Reject a pending campaign, returning it to draft with a mandatory reason. + + Raises EntityNotFoundError, BusinessRuleViolation. + Does not commit; caller commits. + """ + campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first() + if not campaign: + raise EntityNotFoundError("Campaign", campaign_id) + + if campaign.status != "pending_approval": + raise BusinessRuleViolation("Only campaigns pending approval can be rejected") + + if not reason or not reason.strip(): + raise BusinessRuleViolation("A rejection reason is required") + + campaign.status = "draft" + campaign.rejection_reason = reason.strip() + campaign.approved_by = None + campaign.approved_at = None + db.flush() + return campaign + + # Define function get_campaign_detail def get_campaign_detail(db: Session, campaign_id: str) -> dict: """Get detailed campaign info including tests and progress. diff --git a/backend/tests/test_campaign_approval.py b/backend/tests/test_campaign_approval.py new file mode 100644 index 0000000..c920fe0 --- /dev/null +++ b/backend/tests/test_campaign_approval.py @@ -0,0 +1,128 @@ +"""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")