Aegis/backend/app/services/tempo_service.py

"""Tempo time-tracking integration service.

Authentication model
--------------------
Each user authenticates to Tempo with their own personal Tempo API token,
stored in ``user.tempo_api_token``.  This is different from the Jira API token.
Obtain a Tempo token at: Jira → Apps → Tempo → Settings → API Integration.

The global ``settings.TEMPO_ENABLED`` flag acts as a kill-switch.  When False,
all Tempo calls are silently skipped regardless of whether users have tokens.

What goes to Tempo
------------------
Both **red team execution** and **blue team evaluation** time are logged to
Tempo.  Red team time runs from when the red_tech clicks "Start" to "Done".
Blue team time runs from when the blue_tech clicks "Start Evaluation" (sets
blue_work_started_at) to when they submit, so it reflects actual working time
rather than queue time.
"""

# Import logging
import logging

# Import Optional from typing
from typing import Optional

# Import Session from sqlalchemy.orm
from sqlalchemy.orm import Session

# Import settings from app.config
from app.config import settings

# Import InvalidOperationError from app.domain.exceptions
from app.domain.exceptions import InvalidOperationError

# Import JiraLink, JiraLinkEntityType from app.models.jira_link
from app.models.jira_link import JiraLink, JiraLinkEntityType

# Import Test from app.models.test
from app.models.test import Test

# Import User from app.models.user
from app.models.user import User

# Assign logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)

def has_tempo_configured(user) -> bool:
    """Return True if *user* has a personal Tempo API token stored (legacy)."""
    return bool(getattr(user, "tempo_api_token", None))


def get_admin_tempo_token(db=None) -> Optional[str]:
    """Return the admin Tempo API token from system_configs or env-var fallback."""
    if db is not None:
        try:
            from app.models.system_config import SystemConfig
            row = db.query(SystemConfig).filter(
                SystemConfig.key == "tempo.admin_token"
            ).first()
            if row and row.value:
                return row.value
        except Exception:  # nosec B110
            pass
    return getattr(settings, "TEMPO_API_TOKEN", None) or None


def has_admin_tempo_configured(db=None) -> bool:
    """Return True if the admin Tempo token is configured."""
    return bool(get_admin_tempo_token(db))


def get_admin_tempo_client(db=None):
    """Return a Tempo API v4 client authenticated with the admin token."""
    token = get_admin_tempo_token(db)
    if not token:
        raise InvalidOperationError(
            "Admin Tempo token not configured. "
            "Set it in System Settings → Jira Configuration."
        )
    try:
        from tempoapiclient import client_v4 as tempo_client
        base_url = _get_tempo_base_url(db)
        return tempo_client.Tempo(auth_token=token, base_url=base_url)
    except ImportError:
        raise InvalidOperationError(
            "tempo-api-python-client is not installed. "
            "Run: pip install tempo-api-python-client"
        )


_TEMPO_DEFAULT_BASE_URL = "https://api.tempo.io/4"
_TEMPO_EU_BASE_URL = "https://api.eu.tempo.io/4"

# system_configs key for admin-configurable Tempo base URL
_TEMPO_BASE_URL_CONFIG_KEY = "tempo.base_url"


def _get_tempo_base_url(db=None) -> str:
    """Return the Tempo API base URL.

    Reads ``tempo.base_url`` from ``system_configs`` first (allows the admin
    to set the EU endpoint ``https://api.eu.tempo.io/4`` without redeploying).
    Falls back to ``settings.TEMPO_BASE_URL`` env-var, then the global default.
    """
    if db is not None:
        try:
            from app.models.system_config import SystemConfig
            row = db.query(SystemConfig).filter(
                SystemConfig.key == _TEMPO_BASE_URL_CONFIG_KEY
            ).first()
            if row and row.value:
                return row.value.rstrip("/")
        except Exception:  # nosec B110
            pass  # DB unavailable — fall through to defaults

    env_url = getattr(settings, "TEMPO_BASE_URL", None)
    return (env_url or _TEMPO_DEFAULT_BASE_URL).rstrip("/")


def get_user_tempo_client(user, db=None):
    """Return a Tempo API v4 client authenticated as *user*.

    Pass ``db`` to allow reading the admin-configured base URL from
    ``system_configs`` (needed for EU Tempo workspaces).

    Raises ``InvalidOperationError`` when the user has no token or the
    client library is not installed.
    """
    token = getattr(user, "tempo_api_token", None)
    if not token:
        raise InvalidOperationError(
            "No Tempo API token configured. "
            "Add it in Settings → Profile → Tempo Integration."
        )
    try:
        # Import client_v4 as tempo_client from tempoapiclient
        from tempoapiclient import client_v4 as tempo_client
        base_url = _get_tempo_base_url(db)
        logger.debug("Using Tempo base URL: %s", base_url)
        return tempo_client.Tempo(auth_token=token, base_url=base_url)
    except ImportError:
        # Raise InvalidOperationError
        raise InvalidOperationError(
            # Literal argument value
            "tempo-api-python-client is not installed. "
            "Run: pip install tempo-api-python-client"
        )


# Define function log_worklog
def log_worklog(
    user,
    jira_issue_id: int,
    # Entry: author_account_id
    author_account_id: str,
    # Entry: date
    date: str,
    # Entry: time_spent_seconds
    time_spent_seconds: int,
    # Entry: description
    description: str,
    db=None,
) -> dict:
    """Create a worklog entry in Tempo using *user*'s personal token.

    Note: tempoapiclient raises SystemExit (not Exception) on API errors, so
    we intercept BaseException and re-raise as RuntimeError to keep it non-fatal.
    """
    tempo = get_user_tempo_client(user, db=db)
    try:
        return tempo.create_worklog(
            accountId=author_account_id,
            issueId=jira_issue_id,
            dateFrom=date,
            timeSpentSeconds=time_spent_seconds,
            description=description,
        )
    except Exception:
        raise
    except BaseException as exc:
        # tempoapiclient raises SystemExit on HTTP errors (e.g. 400 Bad Request).
        # SystemExit is a BaseException, not Exception, so convert it so callers
        # can catch it with the usual `except Exception` pattern.
        raise RuntimeError(f"Tempo API error: {exc}") from exc


def get_tempo_client():
    """Raise InvalidOperationError if Tempo integration is not enabled.

    Use ``get_user_tempo_client(user, db)`` to obtain a per-user authenticated
    client.  This function exists primarily to give tests a surface for checking
    the enabled state without needing a user context.
    """
    if not settings.TEMPO_ENABLED:
        raise InvalidOperationError("Tempo integration is not enabled")
    raise InvalidOperationError(
        "Use get_user_tempo_client(user) to get a user-specific Tempo client"
    )


def auto_log_test_worklog(
    db: Session,
    test: Test,
    user: User,
    activity_type: str,
    duration_seconds: Optional[int] = None,
) -> Optional[dict]:
    """Log time to Tempo for the given test if conditions are met.

    Uses the admin Tempo token (preferred) so regular users need no personal
    token.  Falls back to the user's personal token when no admin token is set.
    Both ``red_team_execution`` and ``blue_team_evaluation`` are logged.

    ``duration_seconds``, when provided, is used as-is.  When omitted, the
    duration is computed from the test's phase start timestamp.

    Returns the Tempo worklog response dict, or ``None`` if skipped.
    Completely non-fatal — errors are logged and swallowed.
    """
    # Only recognised activity types
    if activity_type not in ("red_team_execution", "blue_team_evaluation"):
        logger.debug("Skipping Tempo sync for activity_type=%s", activity_type)
        return None

    # Global kill-switch
    if not settings.TEMPO_ENABLED:
        return None

    # Compute duration from test timestamps when not supplied by the caller
    if duration_seconds is None:
        from datetime import datetime as _dt
        if activity_type == "blue_team_evaluation":
            started = getattr(test, "blue_work_started_at", None) or getattr(test, "blue_started_at", None)
        else:
            started = getattr(test, "red_started_at", None)
        if started is None:
            logger.debug("No start timestamp on test %s for %s; skipping Tempo worklog", test.id, activity_type)
            return None
        ended = getattr(test, "updated_at", None) or _dt.utcnow()
        duration_seconds = max(int((ended - started).total_seconds()), 0)

    if duration_seconds <= 0:
        logger.debug("Skipping Tempo sync for test %s: duration=%ds", test.id, duration_seconds)
        return None

    # Round UP to the nearest whole minute; enforce ≥ 60 s
    import math
    duration_seconds = max(60, math.ceil(duration_seconds / 60) * 60)

    # user.jira_account_id is required for Tempo worklog attribution
    jira_account_id = (getattr(user, "jira_account_id", "") or "").strip()
    if not jira_account_id:
        logger.debug(
            "User %s has no jira_account_id; skipping Tempo worklog",
            getattr(user, "username", user),
        )
        return None

    # Need a Jira link with a numeric issue ID
    link = (
        db.query(JiraLink)
        .filter(
            JiraLink.entity_id == test.id,
            JiraLink.entity_type == JiraLinkEntityType.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

    # Use the phase start timestamp for the worklog date
    if activity_type == "blue_team_evaluation":
        work_date = (
            (test.blue_work_started_at or test.blue_started_at or test.created_at)
            .strftime("%Y-%m-%d")
        )
        description = f"[Aegis] Blue Team evaluation: {test.name}"
    else:
        work_date = (
            (test.red_started_at or getattr(test, "updated_at", None) or test.created_at)
            .strftime("%Y-%m-%d")
        )
        description = f"[Aegis] Red Team execution: {test.name}"

    try:
        # Admin token preferred; fall back to user's personal token for legacy setups
        admin_token = get_admin_tempo_token(db)
        if admin_token:
            from tempoapiclient import client_v4 as tempo_client
            base_url = _get_tempo_base_url(db)
            tempo = tempo_client.Tempo(auth_token=admin_token, base_url=base_url)
        elif has_tempo_configured(user):
            tempo = get_user_tempo_client(user, db=db)
        else:
            logger.debug(
                "No Tempo credentials available; skipping worklog for test %s", test.id
            )
            return None

        try:
            result = tempo.create_worklog(
                accountId=jira_account_id,
                issueId=int(link.jira_issue_id),
                dateFrom=work_date,
                timeSpentSeconds=duration_seconds,
                description=description,
            )
        except BaseException as exc:
            raise RuntimeError(f"Tempo API error: {exc}") from exc

        logger.info(
            "Tempo worklog created for test %s by user %s: %ds on %s",
            test.id, getattr(user, "username", user), duration_seconds, work_date,
        )
        return result
    except Exception as e:
        logger.warning(
            "Tempo worklog failed for test %s (user %s): %s",
            test.id, getattr(user, "username", user), e, exc_info=True,
        )
        return None