"""Tests for the campaign manager-approval workflow.""" import uuid 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", ) 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")