jbi/jira/service.py

# This import is needed (as of Pyhon 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 concurrent
import json
import logging
from functools import lru_cache
from typing import Any, Iterable, Optional

import requests
from dockerflow import checks
from requests import exceptions as requests_exceptions

from jbi import Operation, environment
from jbi.bugzilla import models as bugzilla_models
from jbi.jira.utils import markdown_to_jira
from jbi.models import ActionContext

from .client import JiraClient, JiraCreateError

settings = environment.get_settings()

logger = logging.getLogger(__name__)

JIRA_REQUIRED_PERMISSIONS = {
    "ADD_COMMENTS",
    "CREATE_ISSUES",
    "DELETE_ISSUES",
    "EDIT_ISSUES",
}


class JiraService:
    """Used by action workflows to perform action-specific Jira tasks"""

    def __init__(self, client) -> None:
        self.client = client

    def fetch_visible_projects(self) -> list[str]:
        """Return list of projects that are visible with the configured Jira credentials"""

        projects = self.client.permitted_projects()
        return [project["key"] for project in projects]

    def get_issue(self, context: ActionContext, issue_key):
        """Return the Jira issue fields or `None` if not found."""
        logger.info("Getting issue %s", issue_key, extra=context.model_dump())
        try:
            response = self.client.get_issue(issue_key)
            logger.info(
                "Received issue %s",
                issue_key,
                extra={"response": response, **context.model_dump()},
            )
            return response
        except requests_exceptions.HTTPError as exc:
            if getattr(exc.response, "status_code", None) != 404:
                raise
            logger.error(
                "Could not read issue %s: %s",
                issue_key,
                exc,
                extra=context.model_dump(),
            )
            return None

    def create_jira_issue(
        self, context: ActionContext, description: str, issue_type: str
    ):
        """Create a Jira issue with basic fields in the project and return its key."""
        bug = context.bug
        fields: dict[str, Any] = {
            "summary": bug.summary,
            "issuetype": {"name": issue_type},
            "description": markdown_to_jira(
                description, max_length=context.action.parameters.jira_char_limit
            ),
            "project": {"key": context.jira.project},
        }
        logger.info(
            "Creating new Jira issue for Bug %s",
            bug.id,
            extra={"fields": fields, **context.model_dump()},
        )
        try:
            response = self.client.create_issue(fields=fields)
        except requests.HTTPError as exc:
            assert exc.response is not None
            try:
                response = exc.response.json()
            except json.JSONDecodeError:
                response = exc.response.text

            logger.exception(
                "Failed to create issue for Bug %s",
                bug.id,
                extra={"response": response, **context.model_dump()},
            )
            raise JiraCreateError(f"Failed to create issue for Bug {bug.id}") from exc

        # Jira response can be of the form: List or Dictionary
        # if a list is returned, get the first item
        issue_data = response[0] if isinstance(response, list) else response
        logger.info(
            "Jira issue %s created for Bug %s",
            issue_data["key"],
            bug.id,
            extra={"response": response, **context.model_dump()},
        )
        return issue_data

    def add_jira_comment(self, context: ActionContext):
        """Publish a comment on the specified Jira issue"""
        context = context.update(operation=Operation.COMMENT)
        commenter = context.event.user.login if context.event.user else "unknown"

        if context.event.target == "attachment":
            routing_key = (
                context.event.routing_key or ".modify"  # only to please type checking.
            )
            _, verb = routing_key.rsplit(".", 1)
            past_verb = {"modify": "modified"}.get(verb, f"{verb}d")
            formatted_comment = f"*{commenter}* {past_verb} an attachment"
        else:
            comment = context.bug.comment
            assert comment  # See jbi.steps.create_comment()
            assert comment.body  # Also see jbi.steps.create_comment()
            prefix = f"*{commenter}* commented: \n"
            formatted_comment = prefix + markdown_to_jira(
                comment.body,
                max_length=context.action.parameters.jira_char_limit - len(prefix),
            )

        issue_key = context.jira.issue
        jira_response = self.client.issue_add_comment(
            issue_key=issue_key,
            comment=formatted_comment,
        )
        logger.info(
            "User comment added to Jira issue %s",
            issue_key,
            extra=context.model_dump(),
        )
        return jira_response

    def add_jira_comments_for_changes(self, context: ActionContext):
        """Add comments on the specified Jira issue for each change of the event"""
        bug = context.bug
        event = context.event
        issue_key = context.jira.issue

        comments: list = []
        user = event.user.login if event.user else "unknown"
        for change in event.changes or []:
            if change.field in ["status", "resolution"]:
                comments.append(
                    {
                        "modified by": user,
                        "resolution": bug.resolution,
                        "status": bug.status,
                    }
                )
            if change.field in ["assigned_to", "assignee"]:
                comments.append({"assignee": bug.assigned_to})

        jira_response_comments = []
        for i, comment in enumerate(comments):
            logger.info(
                "Create comment #%s on Jira issue %s",
                i + 1,
                issue_key,
                extra=context.update(operation=Operation.COMMENT).model_dump(),
            )
            jira_response = self.client.issue_add_comment(
                issue_key=issue_key, comment=json.dumps(comment, indent=4)
            )
            jira_response_comments.append(jira_response)

        return jira_response_comments

    def delete_jira_issue_if_duplicate(
        self, context: ActionContext, latest_bug: bugzilla_models.Bug
    ):
        """Rollback the Jira issue creation if there is already a linked Jira issue
        on the Bugzilla ticket"""
        issue_key = context.jira.issue
        jira_key_in_bugzilla = latest_bug.extract_from_see_also(
            project_key=context.jira.project
        )
        _duplicate_creation_event = (
            jira_key_in_bugzilla is not None and issue_key != jira_key_in_bugzilla
        )
        if not _duplicate_creation_event:
            return None

        logger.warning(
            "Delete duplicated Jira issue %s from Bug %s",
            issue_key,
            context.bug.id,
            extra=context.update(operation=Operation.DELETE).model_dump(),
        )
        jira_response_delete = self.client.delete_issue(issue_id_or_key=issue_key)
        return jira_response_delete

    def add_link_to_bugzilla(self, context: ActionContext):
        """Add link to Bugzilla ticket in Jira issue"""
        bug = context.bug
        issue_key = context.jira.issue
        bugzilla_url = f"{settings.bugzilla_base_url}/show_bug.cgi?id={bug.id}"
        logger.info(
            "Link %r on Jira issue %s",
            bugzilla_url,
            issue_key,
            extra=context.update(operation=Operation.LINK).model_dump(),
        )
        icon_url = f"{settings.bugzilla_base_url}/favicon.ico"
        return self.client.create_or_update_issue_remote_links(
            issue_key=issue_key,
            link_url=bugzilla_url,
            title=bugzilla_url,
            icon_url=icon_url,
            icon_title=icon_url,
        )

    def clear_assignee(self, context: ActionContext):
        """Clear the assignee of the specified Jira issue."""
        issue_key = context.jira.issue
        logger.info("Clearing assignee", extra=context.model_dump())
        return self.client.update_issue_field(key=issue_key, fields={"assignee": None})

    def find_jira_user(self, context: ActionContext, email: str):
        """Lookup Jira users, raise an error if not exactly one found."""
        logger.info("Find Jira user with email %s", email, extra=context.model_dump())
        users = self.client.user_find_by_user_string(query=email)
        if len(users) != 1:
            raise ValueError(f"User {email} not found")
        return users[0]

    def assign_jira_user(self, context: ActionContext, email: str):
        """Set the assignee of the specified Jira issue, raise if fails."""
        issue_key = context.jira.issue
        assert issue_key  # Until we have more fine-grained typing of contexts

        jira_user = self.find_jira_user(context, email)
        jira_user_id = jira_user["accountId"]
        try:
            # There doesn't appear to be an easy way to verify that
            # this user can be assigned to this issue, so just try
            # and do it.
            return self.update_issue_field(
                context, "assignee", jira_user_id, wrap_value="accountId"
            )
        except (requests_exceptions.HTTPError, IOError) as exc:
            raise ValueError(
                f"Could not assign {jira_user_id} to issue {issue_key}"
            ) from exc

    def update_issue_field(
        self,
        context: ActionContext,
        field: str,
        value: Any,
        wrap_value: Optional[str] = None,
    ):
        bug = context.bug
        issue_key = context.jira.issue
        logger.info(
            f"Updating {field} of Jira issue %s to %s for Bug %s",
            issue_key,
            value,
            bug.id,
            extra=context.model_dump(),
        )
        fields: dict[str, Any] = {field: {wrap_value: value} if wrap_value else value}
        response = self.client.update_issue_field(key=issue_key, fields=fields)
        logger.info(
            f"Updated {field} of Jira issue %s to %s for Bug %s",
            issue_key,
            value,
            bug.id,
            extra={"response": response, **context.model_dump()},
        )
        return response

    def update_issue_status(self, context: ActionContext, jira_status: str):
        """Update the status of the Jira issue"""
        issue_key = context.jira.issue
        assert issue_key  # Until we have more fine-grained typing of contexts

        kwargs: dict[str, Any] = {}
        if jira_status == "Cancelled":
            kwargs["fields"] = {
                "resolution": {"name": "Invalid"},
            }
            kwargs["update"] = {
                "comment": [{"add": {"body": "Issue was cancelled."}}],
            }

        logger.info(
            "Updating Jira status to %s",
            jira_status,
            extra=context.model_dump(),
        )
        return self.client.set_issue_status(
            issue_key,
            jira_status,
            **kwargs,
        )

    def update_issue_summary(self, context: ActionContext):
        """Update's an issue's summary with the description of an incoming bug"""
        truncated_summary = context.bug.summary or ""

        if len(truncated_summary) > context.action.parameters.jira_char_limit:
            # Truncate on last word.
            truncated_summary = truncated_summary[
                : context.action.parameters.jira_char_limit
            ].rsplit(maxsplit=1)[0]

        return self.update_issue_field(
            context, field="summary", value=truncated_summary
        )

    def update_issue_resolution(self, context: ActionContext, jira_resolution: str):
        """Update the resolution of the Jira issue."""
        return self.update_issue_field(
            context,
            field="resolution",
            value=jira_resolution,
            wrap_value="name",
        )

    def update_issue_components(
        self,
        context: ActionContext,
        components: Iterable[str],
    ) -> tuple[Optional[dict], set]:
        """Attempt to add components to the specified issue

        Args:
            issue_key: key of the issues to add the components to
            project: the project key
            components: Component names to add to the issue

        Returns:
            The Jira response (if any), and any components that weren't added
            to the issue because they weren't available on the project
        """
        missing_components = set(components)
        jira_components = []

        all_project_components = self.client.get_project_components(
            context.jira.project
        )
        for comp in all_project_components:
            if comp["name"] in missing_components:
                jira_components.append({"id": comp["id"]})
                missing_components.remove(comp["name"])

        if not jira_components:
            return None, missing_components

        resp = self.update_issue_field(
            context, field="components", value=jira_components
        )
        return resp, missing_components

    def update_issue_labels(
        self, issue_key: str, add: Iterable[str], remove: Optional[Iterable[str]]
    ):
        """Update the labels for a specified issue

        Args:
            issue_key: key of the issues to modify the labels on
            add: labels to add
            remove (Optional): labels to remove

        Returns:
            The response from Jira
        """
        if not remove:
            remove = []

        updated_labels = [{"add": label} for label in add] + [
            {"remove": label} for label in remove
        ]
        return self.client.update_issue(
            issue_key=issue_key,
            update={"update": {"labels": updated_labels}},
        )

    def check_jira_connection(self):
        try:
            if self.client.get_server_info(True) is None:
                return [checks.Error("Login fails", id="login.fail")]
        except requests.RequestException:
            return [checks.Error("Could not connect to server", id="jira.server.down")]
        return []

    def check_jira_all_projects_are_visible(self, actions):
        # Do not bother executing the rest of checks if connection fails.
        if messages := self.check_jira_connection():
            return messages

        try:
            visible_projects = self.fetch_visible_projects()
        except requests.HTTPError:
            return [
                checks.Error(
                    "Error fetching visible Jira projects", id="jira.visible.error"
                )
            ]

        missing_projects = actions.configured_jira_projects_keys - set(visible_projects)
        if missing_projects:
            return [
                checks.Warning(
                    f"Jira projects {missing_projects} are not visible with configured credentials",
                    id="jira.projects.missing",
                )
            ]

        return []

    def check_jira_all_projects_have_permissions(self, actions):
        """Fetches and validates that required permissions exist for the configured projects"""

        # Do not bother executing the rest of checks if connection fails.
        if messages := self.check_jira_connection():
            return messages

        try:
            projects = self.client.permitted_projects(JIRA_REQUIRED_PERMISSIONS)
        except requests.HTTPError:
            return [
                checks.Error(
                    "Error fetching permitted Jira projects", id="jira.permitted.error"
                )
            ]

        projects_with_required_perms = {project["key"] for project in projects}
        missing_perms = (
            actions.configured_jira_projects_keys - projects_with_required_perms
        )
        if missing_perms:
            missing = ", ".join(missing_perms)
            return [
                checks.Warning(
                    f"Missing permissions for projects {missing}",
                    id="jira.permitted.missing",
                )
            ]

        return []

    def check_jira_all_project_custom_components_exist(self, actions):
        # Do not bother executing the rest of checks if connection fails.

        if messages := self.check_jira_connection():
            return messages

        results = []

        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(self._check_project_components, action): action
                for action in actions
                if action.parameters.jira_components.set_custom_components
            }
            for future in concurrent.futures.as_completed(futures):
                results.extend(future.result())
        return results

    def _check_project_components(self, action):
        project_key = action.parameters.jira_project_key
        specified_components = set(
            action.parameters.jira_components.set_custom_components
        )

        try:
            all_project_components = self.client.get_project_components(project_key)
        except requests.HTTPError:
            return [
                checks.Error(
                    f"Error checking project components for {project_key}",
                    id="jira.components.error",
                )
            ]

        try:
            all_components_names = set(comp["name"] for comp in all_project_components)
        except KeyError:
            return [
                checks.Error(
                    f"Unexpected get_project_components response for {action.whiteboard_tag}",
                    id="jira.components.parsing",
                )
            ]

        unknown = specified_components - all_components_names
        if unknown:
            return [
                checks.Warning(
                    f"Jira project {project_key} does not have components {unknown}",
                    id="jira.components.missing",
                )
            ]

        return []

    def check_jira_all_project_issue_types_exist(self, actions):
        # Do not bother executing the rest of checks if connection fails.

        if messages := self.check_jira_connection():
            return messages

        try:
            paginated_project_response = self.client.paginated_projects(
                expand="issueTypes", keys=sorted(actions.configured_jira_projects_keys)
            )
        except requests.RequestException:
            return [
                checks.Error(
                    "Couldn't fetch projects",
                    id="jira.projects.error",
                )
            ]

        projects = paginated_project_response["values"]
        issue_types_by_project = {
            project["key"]: {issue_type["name"] for issue_type in project["issueTypes"]}
            for project in projects
        }
        missing_issue_types_by_project = {}
        for action in actions:
            action_issue_types = set(action.parameters.issue_type_map.values())
            project_issue_types = issue_types_by_project.get(
                action.jira_project_key, set()
            )
            if missing_issue_types := action_issue_types - project_issue_types:
                missing_issue_types_by_project[action.jira_project_key] = (
                    missing_issue_types
                )
        if missing_issue_types_by_project:
            return [
                checks.Warning(
                    f"Jira projects {set(missing_issue_types_by_project.keys())} with missing issue types",
                    obj=missing_issue_types_by_project,
                    id="jira.types.missing",
                )
            ]
        return []

    def check_jira_pandoc_install(self):
        if markdown_to_jira("- Test") != "* Test":
            return [checks.Error("Pandoc conversion failed", id="jira.pandoc")]
        return []


@lru_cache(maxsize=1)
def get_service():
    """Get atlassian Jira Service"""
    client = JiraClient(
        url=settings.jira_base_url,
        username=settings.jira_username,
        password=settings.jira_api_key,  # package calls this param 'password' but actually expects an api key
        cloud=True,  # we run against an instance of Jira cloud
    )

    return JiraService(client=client)