Kitos
9b98f60a9a
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
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""Jira integration service — wraps atlassian-python-api for Jira REST calls."""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
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__)
|
|
|
|
_jira_client = None
|
|
|
|
|
|
def get_jira_client():
|
|
"""Return a lazily-initialised Jira client, or raise if disabled."""
|
|
global _jira_client
|
|
if not settings.JIRA_ENABLED:
|
|
raise InvalidOperationError("Jira integration is not enabled")
|
|
if _jira_client is None:
|
|
from atlassian import Jira
|
|
|
|
_jira_client = Jira(
|
|
url=settings.JIRA_URL,
|
|
username=settings.JIRA_USERNAME,
|
|
password=settings.JIRA_API_TOKEN,
|
|
cloud=settings.JIRA_IS_CLOUD,
|
|
)
|
|
return _jira_client
|
|
|
|
|
|
def search_jira_issues(query: str, max_results: int = 10) -> list[dict]:
|
|
"""Search Jira issues by JQL or free text."""
|
|
jira = get_jira_client()
|
|
jql = query if "=" in query or "~" in query else f'summary ~ "{query}"'
|
|
results = jira.jql(jql, limit=max_results)
|
|
return [
|
|
{
|
|
"issue_key": issue["key"],
|
|
"summary": issue["fields"]["summary"],
|
|
"status": issue["fields"]["status"]["name"],
|
|
"assignee": (issue["fields"].get("assignee") or {}).get("displayName"),
|
|
"priority": (issue["fields"].get("priority") or {}).get("name"),
|
|
}
|
|
for issue in results.get("issues", [])
|
|
]
|
|
|
|
|
|
def create_jira_issue(
|
|
project_key: str,
|
|
summary: str,
|
|
description: str,
|
|
issue_type: str = "Task",
|
|
labels: Optional[list[str]] = None,
|
|
custom_fields: Optional[dict] = None,
|
|
) -> dict:
|
|
"""Create a Jira issue and return its key + id."""
|
|
jira = get_jira_client()
|
|
fields: dict = {
|
|
"project": {"key": project_key},
|
|
"summary": summary,
|
|
"description": description,
|
|
"issuetype": {"name": issue_type},
|
|
}
|
|
if labels:
|
|
fields["labels"] = labels
|
|
if custom_fields:
|
|
fields.update(custom_fields)
|
|
|
|
result = jira.issue_create(fields=fields)
|
|
return {"issue_key": result["key"], "issue_id": result["id"]}
|
|
|
|
|
|
def sync_jira_to_aegis(db: Session, link: JiraLink) -> None:
|
|
"""Pull current status from Jira into the local link record."""
|
|
jira = get_jira_client()
|
|
issue = jira.issue(link.jira_issue_key)
|
|
fields = issue.get("fields", {})
|
|
link.jira_status = fields.get("status", {}).get("name")
|
|
link.jira_priority = (fields.get("priority") or {}).get("name")
|
|
link.jira_assignee = (fields.get("assignee") or {}).get("displayName")
|
|
link.jira_story_points = str(fields.get("customfield_10016", ""))
|
|
link.last_synced_at = datetime.utcnow()
|
|
db.flush()
|
|
|
|
|
|
def sync_aegis_to_jira(db: Session, link: JiraLink, entity_data: dict) -> None:
|
|
"""Push an Aegis status update as a Jira comment."""
|
|
jira = get_jira_client()
|
|
comment_body = _build_sync_comment(entity_data)
|
|
jira.issue_add_comment(link.jira_issue_key, comment_body)
|
|
link.last_synced_at = datetime.utcnow()
|
|
db.flush()
|
|
|
|
|
|
def _build_sync_comment(data: dict) -> str:
|
|
"""Build a formatted Jira comment from entity data."""
|
|
lines = ["h3. Aegis Sync Update", ""]
|
|
for key, value in data.items():
|
|
lines.append(f"*{key}:* {value}")
|
|
lines.append(f"\n_Synced at {datetime.utcnow().isoformat()}_")
|
|
return "\n".join(lines)
|