feat(campaigns): gate test edits to draft, add modification request workflow

backend/tests/test_campaign_approval.py

@@ -168,3 +168,157 @@ class TestRejectCampaign:
     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()
+
+        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
+
+    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_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"