feat(phase-36): automatic Tempo time tracking via workflow buttons + fix campaign test management

- Add red_started_at/blue_started_at timing fields to Test model with Alembic migration

- Modify workflow transitions to auto-create integrity-hashed worklogs: Start Execution records red_started_at, Submit to Blue Team stops Red timer and creates worklog then starts Blue timer, Submit for Review stops Blue timer and creates worklog

- Auto-sync worklogs to Tempo when test has a Jira link

- Add LiveTimer component showing real-time elapsed counter during active phases

- Clear timing fields on test reopen

- Fix campaign test management: replace broken navigate-to-tests flow with AddTestToCampaignModal that lets users search and add existing tests directly from the campaign detail page
This commit is contained in:
2026-02-17 16:59:19 +01:00
parent 005a09b42f
commit febf460580
10 changed files with 461 additions and 5 deletions

View File

@@ -0,0 +1,32 @@
"""add_phase_timing_fields
Revision ID: b021phasetiming
Revises: b020jiraworklogs
Create Date: 2026-02-17 18:00:00.000000
Add red_started_at and blue_started_at columns to the tests table
so that automatic worklogs can record real elapsed time per phase.
"""
from alembic import op
revision = "b021phasetiming"
down_revision = "b020jiraworklogs"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.execute("""
ALTER TABLE tests
ADD COLUMN IF NOT EXISTS red_started_at TIMESTAMP,
ADD COLUMN IF NOT EXISTS blue_started_at TIMESTAMP;
""")
def downgrade() -> None:
op.execute("""
ALTER TABLE tests
DROP COLUMN IF EXISTS red_started_at,
DROP COLUMN IF EXISTS blue_started_at;
""")

View File

@@ -49,6 +49,10 @@ class Test(Base):
blue_validation_status = Column(String, nullable=True) # pending / approved / rejected
blue_validation_notes = Column(Text, nullable=True)
# ── Phase timing fields (for automatic Tempo worklogs) ──────────
red_started_at = Column(DateTime, nullable=True)
blue_started_at = Column(DateTime, nullable=True)
# ── Remediation fields ───────────────────────────────────────────
remediation_steps = Column(Text, nullable=True)
remediation_status = Column(String, nullable=True) # pending / in_progress / completed / not_applicable

View File

@@ -137,6 +137,10 @@ class TestOut(BaseModel):
blue_validation_status: str | None = None
blue_validation_notes: str | None = None
# Phase timing fields (for Tempo worklogs)
red_started_at: datetime | None = None
blue_started_at: datetime | None = None
# Remediation fields
remediation_steps: str | None = None
remediation_status: str | None = None

View File

@@ -89,8 +89,26 @@ def auto_log_test_worklog(
def _calculate_duration(test, activity_type: str) -> int:
"""Estimate duration in seconds based on test timestamps and activity type."""
"""Calculate real duration in seconds from the phase timing fields.
Uses the actual start/end timestamps recorded by the workflow buttons,
so the data cannot be falsified.
"""
from datetime import datetime
now = datetime.utcnow()
if activity_type == "red_team_execution" and test.red_started_at:
delta = now - test.red_started_at
return max(int(delta.total_seconds()), 1)
if activity_type == "blue_team_evaluation" and test.blue_started_at:
delta = now - test.blue_started_at
return max(int(delta.total_seconds()), 1)
# Fallback for legacy activity types
if activity_type == "execution" and test.execution_date and test.created_at:
delta = test.execution_date - test.created_at
return max(int(delta.total_seconds()), 0)
return 3600 # default 1 hour if no timestamps available
return 0

View File

@@ -113,12 +113,15 @@ def start_execution(db: Session, test: Test, user: User) -> Test:
"""Move from ``draft`` → ``red_executing``.
Typically called by a **red_tech** when they begin the attack.
Starts the Red Team timer by recording ``red_started_at``.
"""
now = datetime.utcnow()
test = transition_state(
db, test, TestState.red_executing, user,
action_name="start_execution",
)
test.execution_date = datetime.utcnow()
test.execution_date = now
test.red_started_at = now
db.commit()
return test
@@ -127,11 +130,29 @@ def submit_red_evidence(db: Session, test: Test, user: User) -> Test:
"""Move from ``red_executing`` → ``blue_evaluating``.
Called by **red_tech** once they have finished documenting the attack.
Stops the Red Team timer and creates an automatic worklog.
Starts the Blue Team timer by recording ``blue_started_at``.
"""
now = datetime.utcnow()
test = transition_state(
db, test, TestState.blue_evaluating, user,
action_name="submit_red_evidence",
)
# Create automatic worklog for Red Team phase
_create_phase_worklog(
db,
test=test,
user=user,
phase_started_at=test.red_started_at,
phase_ended_at=now,
activity_type="red_team_execution",
description=f"Red Team execution: {test.name}",
)
# Start Blue Team timer
test.blue_started_at = now
db.commit()
return test
@@ -140,15 +161,83 @@ def submit_blue_evidence(db: Session, test: Test, user: User) -> Test:
"""Move from ``blue_evaluating`` → ``in_review``.
Called by **blue_tech** once they have finished documenting detection.
Stops the Blue Team timer and creates an automatic worklog.
"""
now = datetime.utcnow()
test = transition_state(
db, test, TestState.in_review, user,
action_name="submit_blue_evidence",
)
# Create automatic worklog for Blue Team phase
_create_phase_worklog(
db,
test=test,
user=user,
phase_started_at=test.blue_started_at,
phase_ended_at=now,
activity_type="blue_team_evaluation",
description=f"Blue Team evaluation: {test.name}",
)
db.commit()
return test
def _create_phase_worklog(
db: Session,
*,
test: Test,
user: User,
phase_started_at: datetime | None,
phase_ended_at: datetime,
activity_type: str,
description: str,
) -> None:
"""Create an automatic, integrity-hashed worklog for a completed phase.
Also triggers Tempo sync if the test has a Jira link.
"""
if not phase_started_at:
logger.warning(
"No phase start timestamp for test %s (%s), skipping worklog",
test.id, activity_type,
)
return
duration_seconds = max(int((phase_ended_at - phase_started_at).total_seconds()), 1)
try:
from app.services.worklog_service import create_worklog
wl = create_worklog(
db,
entity_type="test",
entity_id=test.id,
user_id=user.id,
activity_type=activity_type,
started_at=phase_started_at,
ended_at=phase_ended_at,
duration_seconds=duration_seconds,
description=description,
)
logger.info(
"Auto-worklog created for test %s: %s, %ds (worklog %s)",
test.id, activity_type, duration_seconds, wl.id,
)
# Sync to Tempo if enabled
try:
from app.services.tempo_service import auto_log_test_worklog
auto_log_test_worklog(db, test, user, activity_type)
except Exception as e:
logger.warning("Tempo sync failed for worklog: %s", e, exc_info=True)
except Exception as e:
logger.error("Failed to create auto-worklog for test %s: %s", test.id, e, exc_info=True)
def validate_as_red_lead(
db: Session,
test: Test,
@@ -428,5 +517,9 @@ def reopen_test(db: Session, test: Test, user: User) -> Test:
test.blue_validated_at = None
test.blue_validation_notes = None
# Clear phase timing fields
test.red_started_at = None
test.blue_started_at = None
db.commit()
return test