jbi/steps.py

"""
Collection of reusable action steps.

Each step takes an `ActionContext` and a list of arbitrary parameters.
"""

# This import is needed (as of Python 3.11) to enable type checking with modules
# imported under `TYPE_CHECKING`
# https://docs.python.org/3/whatsnew/3.7.html#pep-563-postponed-evaluation-of-annotations
# https://docs.python.org/3/whatsnew/3.11.html#pep-563-may-not-be-the-future
from __future__ import annotations

import logging
from enum import Enum, auto
from typing import TYPE_CHECKING, Iterable, Optional

from requests import exceptions as requests_exceptions

from jbi import Operation
from jbi.environment import get_settings


class StepStatus(Enum):
    """
    Options for the result of executing a step function:
        SUCCESS: The step succeeded at doing meaningful work
        INCOMPLETE: The step did not execute successfully, but it's an error we anticipated
        NOOP: The step executed successfully, but didn't have any meaningful work to do
    """

    SUCCESS = auto()
    INCOMPLETE = auto()
    NOOP = auto()


# https://docs.python.org/3.11/library/typing.html#typing.TYPE_CHECKING
if TYPE_CHECKING:
    from jbi.bugzilla.service import BugzillaService
    from jbi.jira import JiraService
    from jbi.models import ActionContext, ActionParams

    StepResult = tuple[StepStatus, ActionContext]

logger = logging.getLogger(__name__)


def create_comment(context: ActionContext, *, jira_service: JiraService) -> StepResult:
    """Create a Jira comment using `context.bug.comment`"""
    bug = context.bug

    if context.event.target == "comment":
        if bug.comment is None:
            logger.info(
                "No matching comment found in payload",
                extra=context.model_dump(),
            )
            return (StepStatus.NOOP, context)

        if not bug.comment.body:
            logger.info(
                "Comment message is empty",
                extra=context.model_dump(),
            )
            return (StepStatus.NOOP, context)

    jira_response = jira_service.add_jira_comment(context)
    context = context.append_responses(jira_response)
    return (StepStatus.SUCCESS, context)


def create_issue(
    context: ActionContext,
    *,
    parameters: ActionParams,
    jira_service: JiraService,
    bugzilla_service: BugzillaService,
) -> StepResult:
    """Create the Jira issue with the first comment as the description."""
    bug = context.bug
    issue_type = parameters.issue_type_map.get(bug.type or "", "Task")
    # In the payload of a bug creation, the `comment` field is `null`.
    description = bugzilla_service.get_description(bug.id)

    jira_create_response = jira_service.create_jira_issue(
        context, description, issue_type
    )
    issue_key = jira_create_response.get("key")

    context = context.update(
        jira=context.jira.update(issue=issue_key),
    )
    context = context.append_responses(jira_create_response)
    return (StepStatus.SUCCESS, context)


def add_link_to_jira(
    context: ActionContext, *, bugzilla_service: BugzillaService
) -> StepResult:
    """Add the URL to the Jira issue in the `see_also` field on the Bugzilla ticket"""
    settings = get_settings()
    jira_url = f"{settings.jira_base_url}browse/{context.jira.issue}"
    logger.info(
        "Link %r on Bug %s",
        jira_url,
        context.bug.id,
        extra=context.update(operation=Operation.LINK).model_dump(),
    )
    bugzilla_response = bugzilla_service.add_link_to_see_also(context.bug, jira_url)
    context = context.append_responses(bugzilla_response)
    return (StepStatus.SUCCESS, context)


def add_link_to_bugzilla(
    context: ActionContext, *, jira_service: JiraService
) -> StepResult:
    """Add the URL of the Bugzilla ticket to the links of the Jira issue"""
    jira_response = jira_service.add_link_to_bugzilla(context)
    context = context.append_responses(jira_response)
    return (StepStatus.SUCCESS, context)


def maybe_delete_duplicate(
    context: ActionContext,
    *,
    bugzilla_service: BugzillaService,
    jira_service: JiraService,
) -> StepResult:
    """
    In the time taken to create the Jira issue the bug may have been updated so
    re-retrieve it to ensure we have the latest data, and delete any duplicate
    if two Jira issues were created for the same Bugzilla ticket.
    """
    latest_bug = bugzilla_service.refresh_bug_data(context.bug)
    jira_response_delete = jira_service.delete_jira_issue_if_duplicate(
        context, latest_bug
    )
    if jira_response_delete:
        context = context.append_responses(jira_response_delete)
        return (StepStatus.SUCCESS, context)
    return (StepStatus.NOOP, context)


def update_issue_summary(
    context: ActionContext, *, jira_service: JiraService
) -> StepResult:
    """Update the Jira issue's summary if the linked bug is modified."""

    if "summary" not in context.event.changed_fields():
        return (StepStatus.NOOP, context)

    jira_response_update = jira_service.update_issue_summary(context)
    context = context.append_responses(jira_response_update)
    return (StepStatus.SUCCESS, context)


def add_jira_comments_for_changes(
    context: ActionContext, *, jira_service: JiraService
) -> StepResult:
    """Add a Jira comment for each field (assignee, status, resolution) change on
    the Bugzilla ticket."""
    comments_responses = jira_service.add_jira_comments_for_changes(context)
    context.append_responses(comments_responses)
    return (StepStatus.SUCCESS, context)


def maybe_assign_jira_user(
    context: ActionContext, *, jira_service: JiraService
) -> StepResult:
    """Assign the user on the Jira issue, based on the Bugzilla assignee email.

    It will attempt to assign the Jira issue the same person as the bug is assigned to. This relies on
    the user using the same email address in both Bugzilla and Jira. If the user does not exist in Jira
    then the assignee is cleared from the Jira issue. The Jira account that JBI uses requires the "Browse
    users and groups" global permission in order to set the assignee.
    """
    event = context.event
    bug = context.bug

    if context.operation == Operation.CREATE:
        if not bug.is_assigned():
            return (StepStatus.NOOP, context)

        try:
            resp = jira_service.assign_jira_user(context, bug.assigned_to)  # type: ignore
            context.append_responses(resp)
            return (StepStatus.SUCCESS, context)
        except ValueError as exc:
            logger.info(str(exc), extra=context.model_dump())
            return (StepStatus.INCOMPLETE, context)

    if context.operation == Operation.UPDATE:
        if "assigned_to" not in event.changed_fields():
            return (StepStatus.SUCCESS, context)

        if not bug.is_assigned():
            resp = jira_service.clear_assignee(context)
        else:
            try:
                resp = jira_service.assign_jira_user(context, bug.assigned_to)  # type: ignore
            except ValueError as exc:
                logger.info(str(exc), extra=context.model_dump())
                # If that failed then just fall back to clearing the assignee.
                resp = jira_service.clear_assignee(context)
        context.append_responses(resp)
        return (StepStatus.SUCCESS, context)

    return (StepStatus.NOOP, context)


def _maybe_update_issue_mapped_field(
    source_field: str,
    context: ActionContext,
    parameters: ActionParams,
    jira_service: JiraService,
    wrap_value: Optional[str] = None,
) -> StepResult:
    source_value = getattr(context.bug, source_field, None) or ""
    target_field = getattr(parameters, f"jira_{source_field}_field")
    target_value = getattr(parameters, f"{source_field}_map").get(source_value)

    # If field is empty on create, or update is about another field, then nothing to do.
    if (context.operation == Operation.CREATE and source_value in ["", "---"]) or (
        context.operation == Operation.UPDATE
        and source_field not in context.event.changed_fields()
    ):
        return (StepStatus.NOOP, context)

    if target_value is None:
        logger.info(
            f"Bug {source_field} %r was not in the {source_field} map.",
            source_value,
            extra=context.update(
                operation=Operation.IGNORE,
            ).model_dump(),
        )
        return (StepStatus.INCOMPLETE, context)

    resp = jira_service.update_issue_field(
        context,
        target_field,
        target_value,
        wrap_value,
    )
    context.append_responses(resp)
    return (StepStatus.SUCCESS, context)


def maybe_update_issue_priority(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue priority
    """
    return _maybe_update_issue_mapped_field(
        "priority", context, parameters, jira_service, wrap_value="name"
    )


def maybe_update_issue_resolution(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue status
    https://support.atlassian.com/jira-cloud-administration/docs/what-are-issue-statuses-priorities-and-resolutions/
    """
    return _maybe_update_issue_mapped_field(
        "resolution", context, parameters, jira_service, wrap_value="name"
    )


def maybe_update_issue_severity(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue severity
    """
    return _maybe_update_issue_mapped_field(
        "severity", context, parameters, jira_service, wrap_value="value"
    )


def maybe_update_issue_points(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue story points
    """
    return _maybe_update_issue_mapped_field(
        "cf_fx_points", context, parameters, jira_service
    )


def maybe_update_issue_status(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue resolution
    https://support.atlassian.com/jira-cloud-administration/docs/what-are-issue-statuses-priorities-and-resolutions/
    """
    bz_status = context.bug.resolution or context.bug.status
    jira_status = parameters.status_map.get(bz_status or "")

    if jira_status is None:
        logger.info(
            "Bug status %r was not in the status map.",
            bz_status,
            extra=context.update(
                operation=Operation.IGNORE,
            ).model_dump(),
        )
        return (StepStatus.INCOMPLETE, context)

    if context.operation == Operation.CREATE:
        resp = jira_service.update_issue_status(context, jira_status)
        context.append_responses(resp)
        return (StepStatus.SUCCESS, context)

    if context.operation == Operation.UPDATE:
        changed_fields = context.event.changed_fields()

        if "status" in changed_fields or "resolution" in changed_fields:
            resp = jira_service.update_issue_status(context, jira_status)
            context.append_responses(resp)
            return (StepStatus.SUCCESS, context)

    return (StepStatus.NOOP, context)


def maybe_update_components(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Update the Jira issue components
    """
    candidate_components = set(parameters.jira_components.set_custom_components)
    if context.bug.component and parameters.jira_components.use_bug_component:
        candidate_components.add(context.bug.component)
    if context.bug.product and parameters.jira_components.use_bug_product:
        candidate_components.add(context.bug.product)
    if (
        context.bug.product_component
        and parameters.jira_components.use_bug_component_with_product_prefix
    ):
        candidate_components.add(context.bug.product_component)

    if not candidate_components:
        # no components to update
        return (StepStatus.NOOP, context)

    # Although we previously introspected the project components, we
    # still have to catch any potential 400 error response here, because
    # the `components` field may not be on the create / update issue.

    if not context.jira.issue:
        raise ValueError("Jira issue unset in Action Context")

    try:
        resp, missing_components = jira_service.update_issue_components(
            context=context,
            components=candidate_components,
        )
    except requests_exceptions.HTTPError as exc:
        if getattr(exc.response, "status_code", None) != 400:
            raise
        # If `components` is not a valid field on create/update screens,
        # then warn developers and ignore the error.
        logger.error(
            f"Could not set components on issue {context.jira.issue}: %s",
            str(exc),
            extra=context.model_dump(),
        )
        context.append_responses(exc.response)
        return (StepStatus.INCOMPLETE, context)

    if missing_components:
        logger.warning(
            "Could not find components '%s' in project",
            ",".join(sorted(missing_components)),
            extra=context.model_dump(),
        )
        return (StepStatus.INCOMPLETE, context)

    context.append_responses(resp)
    return (StepStatus.SUCCESS, context)


def _whiteboard_as_labels(labels_brackets: str, whiteboard: Optional[str]) -> list[str]:
    """Split the whiteboard string into a list of labels"""
    splitted = whiteboard.replace("[", "").split("]") if whiteboard else []
    stripped = [x.strip() for x in splitted if x not in ["", " "]]
    # Jira labels can't contain a " ", convert to "."
    nospace = [wb.replace(" ", ".") for wb in stripped]
    with_brackets = [f"[{wb}]" for wb in nospace]

    if labels_brackets == "yes":
        labels = with_brackets
    elif labels_brackets == "both":
        labels = nospace + with_brackets
    else:
        labels = nospace

    return ["bugzilla"] + labels


def _build_labels_update(
    labels_brackets, added, removed=None
) -> tuple[list[str], list[str]]:
    # We don't bother detecting if label was already there.
    additions = _whiteboard_as_labels(labels_brackets, added)
    removals = []
    if removed:
        before = _whiteboard_as_labels(labels_brackets, removed)
        removals = sorted(
            set(before).difference(set(additions))
        )  # sorted for unit testing
    return additions, removals


def sync_whiteboard_labels(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Set whiteboard tags as labels on the Jira issue.
    """
    # On update of whiteboard field, add/remove corresponding labels
    if context.event.changes:
        changes_by_field = {change.field: change for change in context.event.changes}
        if change := changes_by_field.get("whiteboard"):
            additions, removals = _build_labels_update(
                added=change.added,
                removed=change.removed,
                labels_brackets=parameters.labels_brackets,
            )
        else:
            return (StepStatus.NOOP, context)
    else:
        # On creation, just add them all.
        additions, removals = _build_labels_update(
            added=context.bug.whiteboard, labels_brackets=parameters.labels_brackets
        )

    return _update_issue_labels(context, jira_service, additions, removals)


def sync_keywords_labels(
    context: ActionContext, *, parameters: ActionParams, jira_service: JiraService
) -> StepResult:
    """
    Set keywords as labels on the Jira issue.
    """
    if context.event.changes:
        changes_by_field = {change.field: change for change in context.event.changes}
        if change := changes_by_field.get("keywords"):
            additions = [x.strip() for x in change.added.split(",")]
            removed = [x.strip() for x in change.removed.split(",")]
            removals = sorted(
                set(removed).difference(set(additions))
            )  # sorted for unit testing
        else:
            return (StepStatus.NOOP, context)
    else:
        # On creation, just add them all.
        additions = context.bug.keywords or []
        removals = []

    return _update_issue_labels(context, jira_service, additions, removals)


def _update_issue_labels(
    context: ActionContext,
    jira_service: JiraService,
    additions: Iterable[str],
    removals: Iterable[str],
) -> StepResult:
    if not context.jira.issue:
        raise ValueError("Jira issue unset in Action Context")
    try:
        resp = jira_service.update_issue_labels(
            issue_key=context.jira.issue, add=additions, remove=removals
        )
    except requests_exceptions.HTTPError as exc:
        if getattr(exc.response, "status_code", None) != 400:
            raise
        # If `labels` is not a valid field in this project, then warn developers
        # and ignore the error.
        logger.error(
            f"Could not set labels on issue {context.jira.issue}: %s",
            str(exc),
            extra=context.model_dump(),
        )
        context.append_responses(exc.response)
        return (StepStatus.INCOMPLETE, context)

    context.append_responses(resp)
    return (StepStatus.SUCCESS, context)