feat(campaigns): timeline endpoint, admin-only manual activation, remove lead-set dates

2026-07-03 13:30:21 +02:00
parent 6186c246a4
commit 959def2f49
4 changed files with 59 additions and 21 deletions
@@ -56,6 +56,10 @@ from app.services.campaign_crud_service import (
    get_campaign_history as crud_get_history,
 )

+from app.services.campaign_crud_service import (
+    get_campaign_timeline as crud_get_timeline,
+)
+
 # Import  from app.services.campaign_crud_service
 from app.services.campaign_crud_service import (
    get_campaign_progress_data as crud_get_progress,
@@ -149,7 +153,6 @@ class CampaignCreate(BaseModel):
    tags: Optional[list[str]] = Field(default_factory=list)
    # Assign scheduled_at = None
    scheduled_at: Optional[str] = None
-    start_date: Optional[str] = None  # ISO date — campaign won't activate before this


 # Define class CampaignUpdate
@@ -168,7 +171,6 @@ class CampaignUpdate(BaseModel):
    tags: Optional[list[str]] = None
    # Assign scheduled_at = None
    scheduled_at: Optional[str] = None
-    start_date: Optional[str] = None  # ISO date — can be updated while still in draft


 # Define class AddTestPayload
@@ -327,7 +329,6 @@ def create_campaign(
            tags=payload.tags,
            # Keyword argument: scheduled_at
            scheduled_at=payload.scheduled_at,
-            start_date=payload.start_date,
        )
        campaign_id = result["id"]
        log_action(
@@ -786,8 +787,8 @@ def activate_campaign(
    campaign_id: str,
    force: bool = Query(False, description="Activate even if start_date is in the future"),
    db: Session = Depends(get_db),
-    # Entry: current_user
-    current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
+    # admin passes automatically via require_any_role's built-in bypass — do not add "admin" here
+    current_user: User = Depends(require_any_role("admin")),
 ):
    """Activate a campaign, moving it from draft to active.

@@ -796,7 +797,7 @@ def activate_campaign(
    activates immediately regardless of start_date.
    """
    from fastapi import HTTPException
-    campaign_obj = db.query(Campaign).filter(Campaign.id == campaign_id).first()
+    campaign_obj = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first()
    if campaign_obj and campaign_obj.start_date and not force:
        now = datetime.utcnow()
        if campaign_obj.start_date > now:
@@ -967,7 +968,7 @@ def get_campaign_progress_endpoint(
 # ---------------------------------------------------------------------------

 class GenerateFromActorPayload(BaseModel):
-    start_date: Optional[str] = None  # ISO date YYYY-MM-DD
+    pass


@router.post("/from-threat-actor/{actor_id}", status_code=201)
@@ -993,14 +994,10 @@ def generate_campaign_from_actor(
    Returns:
        dict: Serialised representation of the newly generated campaign.
    """
-    start_date_parsed = (
-        datetime.fromisoformat(payload.start_date) if payload.start_date else None
-    )
    campaign = generate_campaign_from_threat_actor(
        db,
        uuid.UUID(actor_id),
        current_user,
-        start_date=start_date_parsed,
    )

    # Open context manager
@@ -1130,6 +1127,20 @@ def get_campaign_history(
    return crud_get_history(db, campaign_id)


+# ---------------------------------------------------------------------------
+# GET /campaigns/{id}/timeline — Audit-log history for this campaign
+# ---------------------------------------------------------------------------
+
+@router.get("/{campaign_id}/timeline")
+def get_campaign_timeline_endpoint(
+    campaign_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> list:
+    """Return the chronological audit-log history for a campaign."""
+    return crud_get_timeline(db, campaign_id)
+
+
 # ---------------------------------------------------------------------------
 # GET /campaigns/{id}/timing-summary — Aggregated timing across campaign tests
 # ---------------------------------------------------------------------------
@@ -281,7 +281,6 @@ def create_campaign(
    tags: Optional[list[str]] = None,
    # Entry: scheduled_at
    scheduled_at: Optional[str] = None,
-    start_date: Optional[str] = None,
 ) -> dict:
    """Create a new campaign. Does not commit; caller commits."""
    # Assign campaign = Campaign(
@@ -302,7 +301,6 @@ def create_campaign(
        created_by=creator_id,
        # Keyword argument: scheduled_at
        scheduled_at=datetime.fromisoformat(scheduled_at) if scheduled_at else None,
-        start_date=datetime.fromisoformat(start_date) if start_date else None,
    )
    # Stage new record(s) for database insertion
    db.add(campaign)
@@ -459,8 +457,6 @@ def update_campaign(
    if "scheduled_at" in fields and fields["scheduled_at"]:
        # Assign fields["scheduled_at"] = datetime.fromisoformat(fields["scheduled_at"])
        fields["scheduled_at"] = datetime.fromisoformat(fields["scheduled_at"])
-    if "start_date" in fields and fields["start_date"]:
-        fields["start_date"] = datetime.fromisoformat(fields["start_date"])

    # Iterate over fields.items()
    for field, value in fields.items():
@@ -679,7 +675,7 @@ def activate_campaign(db: Session, campaign_id: str) -> Campaign:
    Does not commit; caller commits.
    """
    # Assign campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
-    campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
+    campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first()
    # Check: not campaign
    if not campaign:
        # Raise EntityNotFoundError
@@ -691,7 +687,7 @@ def activate_campaign(db: Session, campaign_id: str) -> Campaign:
        raise BusinessRuleViolation("Only draft campaigns can be activated")

    # Assign test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign_...
-    test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign_id).count()
+    test_count = db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).count()
    # Check: test_count == 0
    if test_count == 0:
        # Raise BusinessRuleViolation
@@ -8,7 +8,6 @@ threat actors, and progress calculation.
 import logging
 import uuid
 from datetime import datetime
-from typing import Optional

 # Import Session from sqlalchemy.orm
 from sqlalchemy.orm import Session
@@ -179,8 +178,6 @@ def generate_campaign_from_threat_actor(
    actor_id: uuid.UUID,
    # Entry: user
    user: User,
-    *,
-    start_date: Optional[datetime] = None,
 ) -> Campaign:
    """Auto-generate a campaign from a threat actor's uncovered techniques.

@@ -238,7 +235,6 @@ def generate_campaign_from_threat_actor(
        created_by=user.id,
        # Keyword argument: tags
        tags=[actor.name, "auto-generated"],
-        start_date=start_date,
    )
    # Stage new record(s) for database insertion
    db.add(campaign)
@@ -254,3 +254,38 @@ def test_direct_add_test_blocked_on_active_campaign_via_router(api, db, red_lead
        json={"test_id": str(new_test.id)},
    )
    assert resp.status_code == 400
+
+
+def test_get_campaign_timeline(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    api("post", f"/api/v1/campaigns/{campaign.id}/submit", red_lead_headers)
+
+    resp = api("get", f"/api/v1/campaigns/{campaign.id}/timeline", red_lead_headers)
+    assert resp.status_code == 200
+    actions = [e["action"] for e in resp.json()]
+    assert "submit_campaign_for_approval" in actions
+
+
+def test_lead_cannot_directly_activate_draft_campaign(api, db, red_lead_user, red_lead_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    resp = api("post", f"/api/v1/campaigns/{campaign.id}/activate", red_lead_headers)
+    assert resp.status_code == 403
+
+
+def test_admin_can_still_directly_activate_as_override(api, db, red_lead_user, red_lead_headers, auth_headers):
+    campaign = _make_draft_campaign(db, red_lead_user.id)
+    resp = api("post", f"/api/v1/campaigns/{campaign.id}/activate", auth_headers)
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "active"
+
+
+def test_create_campaign_payload_has_no_start_date_field(api, db, red_lead_headers):
+    resp = api(
+        "post",
+        "/api/v1/campaigns",
+        red_lead_headers,
+        json={"name": "No date campaign", "start_date": "2026-01-01"},
+    )
+    assert resp.status_code == 201
+    # start_date silently ignored — only the manager can ever set it, via /approve
+    assert resp.json()["start_date"] is None