feat(phase-35): Jira + Tempo integration with internal worklogs
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
Full Jira/Tempo pipeline: link Aegis entities to Jira issues, auto-sync
status hourly, log time internally with integrity hashing, and optionally
push worklogs to Tempo.
- 1.1 JiraLink model + Worklog model: Alembic migration b020 with indexes,
enums (jiralinkentitytype, jirasyncdirection), and integrity_hash column
- 1.2 Jira service: atlassian-python-api wrapper with lazy singleton client,
search/create/sync operations, feature-flagged via JIRA_ENABLED
- 1.3 Jira router: CRUD endpoints for /jira/links, /jira/search,
/jira/create-issue with audit logging and entity-to-issue auto-creation
- 1.4 Tempo service: worklog push via tempo-api-python-client, auto-log from
test completions when TEMPO_ENABLED, graceful fallback on failure
- 1.5 Worklog service + router: immutable internal time records with SHA-256
integrity hash, CRUD at /worklogs, /worklogs/{id}/verify endpoint
- 1.6 Frontend: JiraLinkPanel component (search, link, sync, unlink) and
WorklogTimeline component (timeline view, manual log form) integrated into
TestDetailPage sidebar, CampaignDetailPage grid, TechniqueDetailPage
- 1.7 Jira sync job: APScheduler hourly job syncs all links from Jira,
registered in background scheduler alongside existing jobs
This commit is contained in:
96
backend/app/services/tempo_service.py
Normal file
96
backend/app/services/tempo_service.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""Tempo time-tracking integration service."""
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import settings
|
||||
from app.domain.exceptions import InvalidOperationError
|
||||
from app.models.jira_link import JiraLink
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_tempo_client():
|
||||
"""Return a Tempo API client, or raise if disabled."""
|
||||
if not settings.TEMPO_ENABLED:
|
||||
raise InvalidOperationError("Tempo integration is not enabled")
|
||||
try:
|
||||
from tempoapiclient import client_v4 as tempo_client
|
||||
|
||||
return tempo_client.Tempo(auth_token=settings.TEMPO_API_TOKEN)
|
||||
except ImportError:
|
||||
raise InvalidOperationError(
|
||||
"tempo-api-python-client is not installed. "
|
||||
"Install it with: pip install tempo-api-python-client"
|
||||
)
|
||||
|
||||
|
||||
def log_worklog(
|
||||
jira_issue_id: int,
|
||||
author_account_id: str,
|
||||
date: str,
|
||||
time_spent_seconds: int,
|
||||
description: str,
|
||||
) -> dict:
|
||||
"""Create a worklog entry in Tempo."""
|
||||
tempo = get_tempo_client()
|
||||
worklog = tempo.create_worklog(
|
||||
accountId=author_account_id,
|
||||
issueId=jira_issue_id,
|
||||
dateFrom=date,
|
||||
timeSpentSeconds=time_spent_seconds,
|
||||
description=description,
|
||||
)
|
||||
return worklog
|
||||
|
||||
|
||||
def auto_log_test_worklog(
|
||||
db: Session,
|
||||
test,
|
||||
user,
|
||||
activity_type: str,
|
||||
) -> Optional[dict]:
|
||||
"""If the test has a Jira link, log time to Tempo automatically.
|
||||
|
||||
Returns the Tempo worklog response, or None if skipped.
|
||||
"""
|
||||
if not settings.TEMPO_ENABLED:
|
||||
return None
|
||||
|
||||
link = (
|
||||
db.query(JiraLink)
|
||||
.filter(JiraLink.entity_id == test.id, JiraLink.entity_type == "test")
|
||||
.first()
|
||||
)
|
||||
|
||||
if not link or not link.jira_issue_id:
|
||||
logger.debug("No Jira link for test %s, skipping Tempo worklog", test.id)
|
||||
return None
|
||||
|
||||
duration = _calculate_duration(test, activity_type)
|
||||
if duration <= 0:
|
||||
return None
|
||||
|
||||
try:
|
||||
result = log_worklog(
|
||||
jira_issue_id=int(link.jira_issue_id),
|
||||
author_account_id=getattr(user, "jira_account_id", "") or "",
|
||||
date=test.created_at.strftime("%Y-%m-%d"),
|
||||
time_spent_seconds=duration,
|
||||
description=f"[Aegis] {activity_type}: {test.name}",
|
||||
)
|
||||
logger.info("Tempo worklog created for test %s, %ds", test.id, duration)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.warning("Tempo worklog failed for test %s: %s", test.id, e, exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def _calculate_duration(test, activity_type: str) -> int:
|
||||
"""Estimate duration in seconds based on test timestamps and activity type."""
|
||||
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
|
||||
Reference in New Issue
Block a user