feat(campaigns): campaign timing panel with Red/Blue aggregated metrics
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

Backend: GET /campaigns/{id}/timing-summary
  Aggregates timing across all campaign tests:
  - red_execution_secs: red_started_at → blue_started_at (minus paused)
  - blue_queue_secs:    blue_started_at → blue_work_started_at
  - blue_evaluation_secs: blue_work_started_at → validated (minus paused)
  - total_secs: sum of all three phases
  Returns totals + per-test breakdown sorted by total time desc.

Frontend: new CampaignTimingPanel component replaces WorklogTimeline
  - 4 summary cards: Red Execution / Blue Queue / Blue Evaluation / Total
  - Stacked horizontal bar showing time distribution
  - Per-test breakdown with individual mini-bars and phase durations
  - Shows 'No tests started yet' when no timing data available

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
kitos
2026-06-02 11:06:42 +02:00
parent a518c06653
commit 2b41b191bd
4 changed files with 318 additions and 3 deletions

View File

@@ -6,6 +6,7 @@ test ordering, progress tracking, and threat actor integration.
import logging
import uuid
from datetime import datetime
from typing import Optional
from fastapi import APIRouter, Depends, Query
@@ -15,6 +16,8 @@ from pydantic import BaseModel, Field
from app.database import get_db
from app.dependencies.auth import get_current_user, require_any_role
from app.models.user import User
from app.models.campaign import Campaign, CampaignTest
from app.models.test import Test
from app.services.campaign_service import generate_campaign_from_threat_actor
from app.services.campaign_crud_service import (
add_test_to_campaign as crud_add_test,
@@ -454,3 +457,97 @@ def get_campaign_history(
):
"""List all child campaigns (execution history) of a recurring campaign."""
return crud_get_history(db, campaign_id)
# ---------------------------------------------------------------------------
# GET /campaigns/{id}/timing-summary — Aggregated timing across campaign tests
# ---------------------------------------------------------------------------
def _seconds_between(start: datetime | None, end: datetime | None) -> int:
"""Return elapsed seconds between two datetimes; 0 if either is None."""
if not start or not end:
return 0
diff = (end - start).total_seconds()
return max(0, int(diff))
@router.get("/{campaign_id}/timing-summary")
def get_campaign_timing_summary(
campaign_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Return aggregated Red/Blue timing metrics for all tests in a campaign.
For each test we calculate:
- red_execution_secs : red_started_at → blue_started_at (minus red_paused_seconds)
- blue_queue_secs : blue_started_at → blue_work_started_at (waiting for Blue pick-up)
- blue_evaluation_secs: blue_work_started_at → first validation timestamp (minus blue_paused_seconds)
- total_secs : sum of the three phases
Returns totals + per-test breakdown.
"""
# Load campaign
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Campaign not found")
# Load all tests for this campaign
test_ids = [
ct.test_id
for ct in db.query(CampaignTest).filter(CampaignTest.campaign_id == campaign.id).all()
]
tests = db.query(Test).filter(Test.id.in_(test_ids)).all() if test_ids else []
breakdown = []
total_red = 0
total_queue = 0
total_blue = 0
for t in tests:
# Red execution: from start-execution to submit-to-blue, minus paused time
red_secs = max(
0,
_seconds_between(t.red_started_at, t.blue_started_at) - (t.red_paused_seconds or 0),
)
# Blue queue: from receiving the test to actually starting evaluation
queue_secs = _seconds_between(t.blue_started_at, t.blue_work_started_at)
# Blue evaluation: from starting evaluation to first validation, minus paused time
eval_end = t.red_validated_at or t.blue_validated_at
blue_secs = max(
0,
_seconds_between(t.blue_work_started_at, eval_end) - (t.blue_paused_seconds or 0),
)
total_red += red_secs
total_queue += queue_secs
total_blue += blue_secs
breakdown.append({
"test_id": str(t.id),
"test_name": t.name,
"state": t.state.value if t.state else None,
"red_execution_secs": red_secs,
"blue_queue_secs": queue_secs,
"blue_evaluation_secs": blue_secs,
"total_secs": red_secs + queue_secs + blue_secs,
"has_timing": bool(t.red_started_at),
})
total_secs = total_red + total_queue + total_blue
return {
"campaign_id": campaign_id,
"campaign_name": campaign.name,
"tests_total": len(tests),
"tests_with_timing": sum(1 for b in breakdown if b["has_timing"]),
"red_execution_secs": total_red,
"blue_queue_secs": total_queue,
"blue_evaluation_secs": total_blue,
"total_secs": total_secs,
"breakdown": sorted(breakdown, key=lambda x: -(x["total_secs"])),
}