feat(tests): remove Time Log, move Tempo sync to Phase Timeline
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

- Remove WorklogTimeline (manual time log) from test detail page
- TestPhaseTimeline now accepts testId, fetches its own worklogs,
  and shows Tempo sync status on the Red Team Execution row:
    • green badge if already synced (with worklog ID tooltip)
    • 'Sync to Tempo' button (blue) if not yet synced
- Add POST /tests/{id}/sync-tempo backend endpoint for manual sync:
  finds unsynced red_team_execution worklogs and pushes them to Tempo

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
kitos
2026-05-28 14:09:16 +02:00
parent 986e91a88a
commit d3baa9c032
5 changed files with 214 additions and 34 deletions

View File

@@ -649,3 +649,78 @@ def get_retest_chain(
}
for t in chain
]
# ---------------------------------------------------------------------------
# POST /tests/{id}/sync-tempo — manual Tempo sync for red execution worklog
# ---------------------------------------------------------------------------
@router.post("/{test_id}/sync-tempo")
def sync_tempo(
test_id: uuid.UUID,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Manually sync this test's red team execution worklog(s) to Tempo.
Useful when the automatic sync failed at phase completion (e.g. Tempo
was not yet configured). Only red_team_execution worklogs are eligible.
Already-synced worklogs are skipped. Returns a summary of what happened.
"""
from datetime import datetime as _dt
from app.models.worklog import Worklog
from app.services.tempo_service import auto_log_test_worklog
from app.services.test_crud_service import get_test_or_raise as _get
test = _get(db, test_id)
worklogs = (
db.query(Worklog)
.filter(
Worklog.entity_type == "test",
Worklog.entity_id == test_id,
Worklog.activity_type == "red_team_execution",
)
.all()
)
if not worklogs:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No red team execution worklog found for this test.",
)
results = []
for wl in worklogs:
if wl.tempo_synced:
results.append({"worklog_id": str(wl.id), "status": "already_synced"})
continue
try:
result = auto_log_test_worklog(
db=db,
test=test,
user=current_user,
activity_type=wl.activity_type,
duration_seconds=wl.duration_seconds,
)
if result and isinstance(result, dict):
wl.tempo_synced = _dt.utcnow()
wl.tempo_worklog_id = str(result.get("tempoWorklogId", ""))
db.commit()
results.append({"worklog_id": str(wl.id), "status": "synced"})
else:
results.append({
"worklog_id": str(wl.id),
"status": "skipped",
"detail": "Tempo not configured or conditions not met.",
})
except Exception as exc:
results.append({
"worklog_id": str(wl.id),
"status": "error",
"detail": str(exc),
})
return {"results": results}