kitos
fd47db7bea
serialize_modification_request crashed toward a string 'None' instead of JSON null for requests whose test_id was nulled by the SET NULL cascade. Also corrects two comments that still described the debunked ordering theory instead of the actual SET NULL fix, and fixes a test that asserted post-cascade state without committing first (the ORM only picks up out-of-band DB-side SET NULL on already-loaded objects after expire_on_commit forces a re-read, matching real router behavior).
380 lines
17 KiB
Python
380 lines
17 KiB
Python
"""Tests for the campaign manager-approval workflow."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.models.campaign import Campaign, CampaignModificationRequest, 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")
|
|
|
|
|
|
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,
|
|
serialize_modification_request,
|
|
)
|
|
|
|
|
|
@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()
|
|
request_id = request.id
|
|
|
|
approve_modification_request(db, str(request_id), reviewer_id=admin_user.id)
|
|
db.commit()
|
|
|
|
cts = db.query(CampaignTest).filter(CampaignTest.campaign_id == active_campaign_with_test.id).all()
|
|
assert len(cts) == 0
|
|
|
|
# The request row must survive deletion of the underlying Test — this
|
|
# locks in test_id's ondelete="SET NULL" (not CASCADE), which is what
|
|
# keeps this audit record intact after the test it referenced is gone.
|
|
# (Commit above first: ON DELETE SET NULL is applied DB-side by the
|
|
# cascade, and the ORM only picks that up on already-loaded objects
|
|
# after expire_on_commit forces a fresh read — exactly what happens
|
|
# for real in the router, which commits before serializing.)
|
|
reloaded = (
|
|
db.query(CampaignModificationRequest)
|
|
.filter(CampaignModificationRequest.id == request_id)
|
|
.first()
|
|
)
|
|
assert reloaded is not None
|
|
assert reloaded.status == "approved"
|
|
assert reloaded.reviewed_by == admin_user.id
|
|
assert reloaded.reviewed_at is not None
|
|
|
|
# Serialization must not crash and must emit a real null, not the
|
|
# string "None", for a request whose test_id has been nulled out.
|
|
assert reloaded.test_id is None
|
|
serialized = serialize_modification_request(db, reloaded)
|
|
assert serialized["test_id"] is None
|
|
assert serialized["test_name"] is None
|
|
|
|
def test_approve_already_decided_request_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()
|
|
approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="Only pending"):
|
|
approve_modification_request(db, str(request.id), reviewer_id=admin_user.id)
|
|
|
|
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_reject_already_decided_request_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()
|
|
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="Not needed")
|
|
db.commit()
|
|
|
|
with pytest.raises(BusinessRuleViolation, match="Only pending"):
|
|
reject_modification_request(db, str(request.id), reviewer_id=admin_user.id, review_notes="Again")
|
|
|
|
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"
|