kitos
119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""Router-level tests for the campaign manager-approval workflow."""
|
|
|
|
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
|
|
|
|
|
|
def _post(client, url, headers, **kwargs):
|
|
"""POST while forcing auth via the Authorization header.
|
|
|
|
The login endpoint also sets an HttpOnly ``aegis_token`` cookie, and the
|
|
shared ``TestClient`` cookie jar persists across requests within a test.
|
|
``get_current_user`` prefers the cookie over the ``Authorization``
|
|
header, so once a second role's headers fixture logs in (setting its own
|
|
cookie), a request made with an *earlier* role's headers would silently
|
|
authenticate as the later role unless the leftover cookie is cleared.
|
|
"""
|
|
client.cookies.clear()
|
|
return client.post(url, headers=headers, **kwargs)
|
|
|
|
|
|
def _make_draft_campaign(db, owner_id):
|
|
tech = Technique(mitre_id="T1059", name="Command Line", tactic="execution", platforms=["windows"])
|
|
db.add(tech)
|
|
db.flush()
|
|
campaign = Campaign(name="Router Test Campaign", type="custom", status="draft", created_by=owner_id)
|
|
db.add(campaign)
|
|
db.flush()
|
|
test = Test(technique_id=tech.id, name="T1059 test", state=TestState.draft, created_by=owner_id)
|
|
db.add(test)
|
|
db.flush()
|
|
db.add(CampaignTest(campaign_id=campaign.id, test_id=test.id, order_index=0))
|
|
db.commit()
|
|
db.refresh(campaign)
|
|
return campaign
|
|
|
|
|
|
def test_lead_can_submit_own_campaign(client, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = _post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "pending_approval"
|
|
|
|
|
|
def test_red_tech_cannot_submit(client, db, red_lead_user, red_tech_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
resp = _post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_tech_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_manager_can_approve_and_sets_start_date(client, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
_post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = _post(
|
|
client,
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
manager_headers,
|
|
json={"start_date": "2026-09-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "active"
|
|
assert body["start_date"] is not None
|
|
|
|
|
|
def test_lead_cannot_approve(client, db, red_lead_user, red_lead_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
_post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = _post(
|
|
client,
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
red_lead_headers,
|
|
json={"start_date": "2026-09-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_can_approve_as_manager_backup(client, db, red_lead_user, red_lead_headers, auth_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
_post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = _post(
|
|
client,
|
|
f"/api/v1/campaigns/{campaign.id}/approve",
|
|
auth_headers,
|
|
json={"start_date": "2026-09-01T00:00:00"},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_manager_can_reject_with_reason(client, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
_post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = _post(
|
|
client,
|
|
f"/api/v1/campaigns/{campaign.id}/reject",
|
|
manager_headers,
|
|
json={"reason": "Needs more detail"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "draft"
|
|
|
|
|
|
def test_manager_reject_without_reason_rejected_by_validation(client, db, red_lead_user, red_lead_headers, manager_headers):
|
|
campaign = _make_draft_campaign(db, red_lead_user.id)
|
|
_post(client, f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
|
|
|
|
resp = _post(
|
|
client,
|
|
f"/api/v1/campaigns/{campaign.id}/reject",
|
|
manager_headers,
|
|
json={"reason": ""},
|
|
)
|
|
assert resp.status_code == 400
|