feat(campaigns): modification-request router endpoints

2026-07-03 12:34:18 +02:00
parent 95b5ac50c6
commit 6186c246a4
2 changed files with 310 additions and 0 deletions
@@ -102,3 +102,155 @@ def test_manager_reject_without_reason_rejected_by_validation(api, db, red_lead_
        json={"reason": ""},
    )
    assert resp.status_code == 400
+
+
+def _activate_campaign_directly(db, campaign):
+    """Test helper — bypass the approval flow to get straight to 'active' for setup."""
+    campaign.status = "active"
+    db.commit()
+    db.refresh(campaign)
+    return campaign
+
+
+def test_lead_can_create_modification_request(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+
+    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=red_lead_user.id)
+    db.add(new_test)
+    db.commit()
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "add_test", "test_id": str(new_test.id), "justification": "Coverage gap"},
+    )
+    assert resp.status_code == 201
+    assert resp.json()["status"] == "pending"
+
+
+def test_modification_request_without_justification_rejected(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+    ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "remove_test", "test_id": str(ct.test_id), "justification": ""},
+    )
+    assert resp.status_code == 400
+
+
+def test_manager_can_list_pending_modification_requests(api, db, red_lead_user, red_lead_headers, manager_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+    ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
+
+    api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
+    )
+
+    resp = api("get", "/api/v1/campaigns/modification-requests/pending", manager_headers)
+    assert resp.status_code == 200
+    assert len(resp.json()) == 1
+
+
+def test_manager_can_approve_modification_request(api, db, red_lead_user, red_lead_headers, manager_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+    ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
+
+    create_resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
+    )
+    request_id = create_resp.json()["id"]
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/modification-requests/{request_id}/approve",
+        manager_headers,
+    )
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "approved"
+
+    remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all()
+    assert len(remaining) == 0
+
+
+def test_manager_can_reject_modification_request(api, db, red_lead_user, red_lead_headers, manager_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+    ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
+
+    create_resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
+    )
+    request_id = create_resp.json()["id"]
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/modification-requests/{request_id}/reject",
+        manager_headers,
+        json={"review_notes": "Still needed"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "rejected"
+
+    remaining = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all()
+    assert len(remaining) == 1
+
+
+def test_lead_cannot_approve_modification_request(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+    ct = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).first()
+
+    create_resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/modification-requests",
+        red_lead_headers,
+        json={"action": "remove_test", "test_id": str(ct.test_id), "justification": "Superseded"},
+    )
+    request_id = create_resp.json()["id"]
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/modification-requests/{request_id}/approve",
+        red_lead_headers,
+    )
+    assert resp.status_code == 403
+
+
+def test_direct_add_test_blocked_on_active_campaign_via_router(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    _activate_campaign_directly(db, campaign)
+
+    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=red_lead_user.id)
+    db.add(new_test)
+    db.commit()
+
+    resp = api(
+        "post",
+        f"/api/v1/campaigns/{campaign.id}/tests",
+        red_lead_headers,
+        json={"test_id": str(new_test.id)},
+    )
+    assert resp.status_code == 400