From add29ae363b80085186fc6b11efec2e6c313550c Mon Sep 17 00:00:00 2001 From: Michael Pruitt Date: Wed, 22 Apr 2026 16:31:58 -0500 Subject: [PATCH] Remove unused jira-escalation command and SlackClient bot integration Neither the jira-escalation CLI command nor the SlackClient bot token integration were wired up to any CI job or Prow step. Removing the dead code: jira_escalation command, escalation logic, SlackClient class, and associated tests. Co-Authored-By: Claude Sonnet 4.6 --- src/cli.py | 3 - src/commands/jira_escalation.py | 99 ----- src/escalation/__init__.py | 0 src/escalation/jira_escalation.py | 373 ----------------- src/objects/slack_base.py | 90 ---- .../escalation/test_jira_escalation.py | 387 ------------------ .../objects/slack/test_slack_client.py | 51 --- 7 files changed, 1003 deletions(-) delete mode 100644 src/commands/jira_escalation.py delete mode 100644 src/escalation/__init__.py delete mode 100644 src/escalation/jira_escalation.py delete mode 100644 src/objects/slack_base.py delete mode 100644 tests/unittests/functions/escalation/test_jira_escalation.py delete mode 100644 tests/unittests/objects/slack/test_slack_client.py diff --git a/src/cli.py b/src/cli.py index 8a07cd92..d4bed8c9 100644 --- a/src/cli.py +++ b/src/cli.py @@ -7,8 +7,6 @@ from src.commands.jira_config_gen import jira_config_gen from src.commands.report import report -from src.commands.jira_escalation import jira_escalation - @click.group() @click.option( @@ -25,7 +23,6 @@ def main(ctx: Context, pdb: bool) -> None: main.add_command(report) main.add_command(jira_config_gen) -main.add_command(jira_escalation) if __name__ == "__main__": should_raise = False diff --git a/src/commands/jira_escalation.py b/src/commands/jira_escalation.py deleted file mode 100644 index 8acb293f..00000000 --- a/src/commands/jira_escalation.py +++ /dev/null @@ -1,99 +0,0 @@ -import json -import click -from click import Context -from src.objects.jira_base import Jira -from src.escalation.jira_escalation import Jira_Escalation -from src.objects.slack_base import SlackClient - - -@click.option( - "--jira-config-path", - help="The path to the jira configuration file", - default="/tmp/jira.config", - type=click.Path(exists=True), -) -@click.option( - "--pdb", - help="Drop to `ipdb` shell on exception", - is_flag=True, -) -@click.option( - "--slack-bot-token", - help="Slack bot token for slack api, if not provide it will look for environment var SLACK_BOT_TOKEN", - required=False, -) -@click.option( - "--slack-channel", - help="Slack channel name to send notifications", - required=True, -) -@click.option( - "--default-labels", - help="List of default labels for filtering jira issues", - required=True, - multiple=True, -) -@click.option( - "--additional-labels", - help="List of additional labels for additional filtering jira issues", - required=True, - multiple=True, -) -@click.option( - "--default-jira-project", - help="Name of default jira project", - required=True, -) -@click.option( - "--team-slack-handle", - help="slack user group or team handle required to notify team members in slack", - required=True, -) -@click.option( - "--team-manager-email", - help="to notify team manager for escalation", - required=True, -) -@click.option( - "--reporter-email", - help="email of the person monitoring the results", - required=False, -) -@click.command("jira-escalation") -@click.pass_context -def jira_escalation( - ctx: Context, - jira_config_path: str, - pdb: bool, - slack_bot_token: str, - slack_channel: str, - default_labels: list[str], - additional_labels: list[str], - default_jira_project: str, - team_slack_handle: str, - team_manager_email: str, - reporter_email: str, -) -> None: - """Manages jira escalation""" - ctx.obj["PDB"] = pdb - - # Build Objects - jira_connection = Jira(jira_config_path=jira_config_path) - slack_client = SlackClient(slack_bot_token) - with open(jira_config_path) as jira_config_file: - jira_config = json.load(jira_config_file) - - base_issue_url = jira_config.get("url") - - Jira_Escalation( - jira=jira_connection, - slack_client=slack_client, - slack_channel=slack_channel, - default_labels=default_labels, - additional_labels=additional_labels, - default_jira_project=default_jira_project, - team_slack_handle=team_slack_handle, - team_manager_email=team_manager_email, - reporter_email=reporter_email, - base_issue_url=base_issue_url, - ) diff --git a/src/escalation/__init__.py b/src/escalation/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/src/escalation/jira_escalation.py b/src/escalation/jira_escalation.py deleted file mode 100644 index 16dd2362..00000000 --- a/src/escalation/jira_escalation.py +++ /dev/null @@ -1,373 +0,0 @@ -from datetime import timezone, datetime -import re -from typing import Optional -from jira import Issue -from src.objects.jira_adf import adf_doc -from src.objects.jira_adf import adf_mention -from src.objects.jira_adf import description_to_plain_text_for_search -from src.objects.jira_adf import inline_text -from src.objects.jira_adf import paragraph -from src.objects.jira_adf import plain_text_to_adf_doc -from src.objects.jira_base import Jira -from src.objects.slack_base import SlackClient -from simple_logger.logger import get_logger - - -LOGGER = get_logger(name=__name__) - - -class Jira_Escalation: - def __init__( - self, - jira: Jira, - slack_client: SlackClient, - slack_channel: str, - default_labels: list[str], - additional_labels: list[str], - default_jira_project: str, - team_slack_handle: str, - team_manager_email: str, - reporter_email: str, - base_issue_url: str, - ): - """ - builds Jira Escalation object, used to escalate issues with relevant escalation action - Args: - jira (Jira): a jira object to authenticate with jira - slack_client (Slack_client): slack object to authenticate with slack - slack_channel (str): slack channel name string - default_labels (list[str]): list of default labels to filter out jira search query - additional_labels (list[str]): list of additional labels to add additional filters - default_jira_project (str): jira project to search issues - team_slack_handle (str): slack handle for the user group - team_manager_email (str): email string used for copying lead contact - reporter_email (Optional[str]): email string to notify reporter - base_issue_url (str): jira server url - """ - self.default_labels = default_labels - self.additional_labels = additional_labels - self.default_project = default_jira_project - self.slack_client = slack_client - self.jira = jira - self.team_slack_handle = team_slack_handle - self.team_manager_email = team_manager_email - self.reporter_email = reporter_email if reporter_email else None - self.slack_channel = slack_channel - self.base_issue_url = base_issue_url - - issues_with_no_assignee = [] - - # build query and process ocp lp firewatch tickets in NOACK and ACK status - jira_query = "" - for status in ["NO ACK", "ACK"]: - jira_query = f'Project = {self.default_project} AND status in("{status}") ' - - jira_query = self.add_labels_to_jira_query(jira_query) - - LOGGER.info(f"Jira query for issues in {status}:\n{jira_query}") - - issues_with_no_assignee += self.process_issues(jira_query) - - # process PQE tickets - pqe_jira_query = ( - f"Project != {self.default_project} AND status not in('Resolved','Blocked','Closed','Backlog','Done')" - ) - - pqe_jira_query = self.add_labels_to_jira_query(pqe_jira_query) - LOGGER.info(f"Jira query for PQE issues:\n{pqe_jira_query}") - issues_with_no_assignee += self.process_issues(pqe_jira_query) - - # Escalation module summarizes all the issues with no assignee in one message. - message = "No assignee found for these issues , please do the necessary follow up : \n " - count = 0 - for issue in issues_with_no_assignee: - LOGGER.info(f" item in issue : {issue}") - count = count + 1 - - issue_url = f"<{self.base_issue_url}/browse/{issue}|{issue}>" - message += f"\n {count}. {issue_url} \n" - LOGGER.info(f"message to reporter : {message}") - if len(issues_with_no_assignee) > 0: - self.send_slack_notification(self.slack_channel, message) - - def process_issues(self, jira_query: str) -> list[str]: - """ - Process jira issues for escalation based on escalation criteria - Args: - jira_query (string): JQL query string to get the issues - - Returns: - list : issues with no assignee - """ - - issues_list = self.jira.search_issues(jql_query=jira_query) - LOGGER.info(f"Matching Jira issues : {issues_list}") - issues_with_no_assignee = [] - for issue_id in issues_list: - jira_issue = self.jira.get_issue_by_id_or_key_with_changelog(issue_id) - LOGGER.info(f"Starting to process Jira issue: {jira_issue.key}") - - if jira_issue.fields.assignee is None: - LOGGER.info(f" No assignee found for this jira issue : {jira_issue.key}") - issues_with_no_assignee.append(jira_issue.key) - continue - - # store required jira field values - comments = jira_issue.fields.comment.comments - assignee = jira_issue.fields.assignee - assignee_account_id = self.get_user_account_id(assignee) - assignee_email = self.get_user_email(assignee) - - current_time = datetime.now(timezone.utc) - - # check if there's a record for status change, if yes then assign that date - status_changed_date = None - if jira_issue.fields.status.name == "ACK": - status_changed_date = self.get_latest_status_change_date(jira_issue=jira_issue) - - # get the date of assignee's latest comment (compare by accountId for Cloud compatibility) - assignee_comments = [c for c in comments if self.get_user_account_id(c.author) == assignee_account_id] - assignee_comment_created_date = None - if assignee_comments: - latest_comment = max(assignee_comments, key=lambda c: self.parse_jira_datetime(c.updated)) - assignee_comment_created_date = self.parse_jira_datetime(latest_comment.updated) - - if assignee_comment_created_date: - last_updated_time = assignee_comment_created_date - LOGGER.info(f"assignee comment date: {last_updated_time}") - elif status_changed_date: - last_updated_time = status_changed_date - LOGGER.info(f"status change date: {last_updated_time}") - else: - last_updated_time = self.parse_jira_datetime(jira_issue.fields.updated) - LOGGER.info(f"last updated change date: {last_updated_time}") - - days_since_update = (current_time - last_updated_time).days - desc_text = description_to_plain_text_for_search(jira_issue.fields.description) - prow_job_name = self.extract_prow_job_name(desc_text) - prow_job_url = self.extract_prow_job_link(desc_text) - issue_url = f"<{self.base_issue_url}/browse/{jira_issue.key}|{jira_issue.key}>" - - # check if issue is updated within corresponding escalation periods and send relevant notifications - LOGGER.info(f" last update: {days_since_update}") - self.escalate_issues( - prow_job_name=prow_job_name, - prow_job_url=prow_job_url, - days_since_update=days_since_update, - jira_issue=jira_issue, - issue_url=issue_url, - assignee_email=assignee_email, - ) - - return issues_with_no_assignee - - def send_slack_notification( - self, slack_channel: str, message: str, email: Optional[str] = None, cc_user_email: Optional[str] = None - ) -> None: - """ - Sends slack notification to provided slack channel or email - Args: - slack_channel (str): name string for slack channel to post message - message (str): message string for posting message to slack channel - email (Optional[str]): email to tag user on slack - cc_user_email (Optional[str]): email to copy a user in slack message - """ - prefix_text = "" - append_text = "" - if email: - user_display_name = self.slack_client.get_slack_username(email) - prefix_text += f"Hi <@{user_display_name}>, \n" - - if cc_user_email: - cc_user_display_name = self.slack_client.get_slack_username(cc_user_email) - append_text += f"\n cc: <@{cc_user_display_name}> \n" - - if not email and not cc_user_email: - usergroup_id = self.slack_client.get_slack_usergroup(self.team_slack_handle) - if usergroup_id: - message = f" \n {message}" - - if prefix_text: - message = prefix_text + message - if append_text: - message = message + append_text - self.slack_client.send_notification(channel=slack_channel, text=message) - - def add_labels_to_jira_query(self, jira_query: str) -> str: - """ - Build JQL query string based on labels - Args: - jira_query (str): query string without any labels - - Returns: - jira_query (str): processed jira query string after addinng conditions to the labels - """ - - if self.additional_labels: - formatted_additional_labels = '","'.join(self.additional_labels) - LOGGER.info(f"Using additional labels: {formatted_additional_labels}") - jira_query += f' AND (labels IN("{formatted_additional_labels}"))' - - if self.default_labels: - default_condition = " AND".join([f'labels = "{label}"' for label in self.default_labels]) - jira_query += f" AND ({default_condition})" - - return jira_query - - @staticmethod - def extract_prow_job_link(jira_description_text: str) -> Optional[str]: - """ - Get the prow job url from jira issue description field - Args: - jira_description_text (str): text string from jira description field - - Returns: - Optional[str]: prow job url string extracted if found else None - """ - if not isinstance(jira_description_text, str): - raise TypeError("description must be str; use description_to_plain_text_for_search for ADF") - try: - url_pattern = r"https://prow\.ci\.openshift\.org/\S+" - match = re.search(url_pattern, jira_description_text) - return match.group(0).rstrip("]") if match else None - except Exception: - return None - - @staticmethod - def extract_prow_job_name(jira_description_text: str) -> Optional[str]: - """ - Get the prow job name from jira issue description field - Args: - jira_description_text (str): text string from jira description field - - Returns: - Optional[str]: prow job name string extracted if found else None - """ - - if not isinstance(jira_description_text, str): - raise TypeError("description must be str; use description_to_plain_text_for_search for ADF") - try: - pattern = r"periodic-ci-[^\|]+" - match = re.search(pattern, jira_description_text) - return match.group(0) if match else None - except Exception: - return None - - @staticmethod - def parse_jira_datetime(jira_timestamp_str: str) -> datetime: - parsed = datetime.strptime(jira_timestamp_str, "%Y-%m-%dT%H:%M:%S.%f%z") - - return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) - - def get_latest_status_change_date(self, jira_issue: Issue) -> Optional[datetime]: - """ - Get the prow job name from jira issue description field - Args: - jira_issue (Issue): jira_issue with chagelog - - Returns: - datetime: datetime value containing status change date - """ - changed_date = None - change_log = jira_issue.changelog - - status_changes = [] - for history in change_log.histories: - for item in history.items: - if item.field.lower() == "status": - changed_date = self.parse_jira_datetime(history.created) - status_changes.append(changed_date) - - LOGGER.info(f"Jira status changed recently on {max(status_changes)}") - return max(status_changes) if status_changes else None - - @staticmethod - def get_user_account_id(user: Optional[object]) -> Optional[str]: - """ - Extract accountId from a Jira user object. - - In Jira Cloud, users are identified by accountId rather than username. - This method safely extracts the accountId, falling back to None if unavailable. - - Args: - user: A Jira user object (e.g., issue.fields.assignee, comment.author) - - Returns: - The user's accountId if available, None otherwise. - """ - if user is None: - return None - return getattr(user, "accountId", None) - - @staticmethod - def get_user_email(user: Optional[object]) -> Optional[str]: - """ - Extract emailAddress from a Jira user object. - - In Jira Cloud, email visibility may be restricted by privacy settings. - This method safely extracts the email, returning None if unavailable. - - Args: - user: A Jira user object (e.g., issue.fields.assignee, comment.author) - - Returns: - The user's email address if available, None otherwise. - """ - if user is None: - return None - return getattr(user, "emailAddress", None) - - def escalate_issues( - self, - jira_issue: Issue, - prow_job_name: Optional[str], - prow_job_url: Optional[str], - days_since_update: int, - issue_url: str, - assignee_email: Optional[str], - ) -> None: - """ - Performs escalation check and sends relevant notification. - - Args: - jira_issue (Issue): jira issue to escalate - prow_job_name (str): job name used to build escalation message - prow_job_url (str): job url used to build escalation message - days_since_update (datetime): days passed between current date and jira issue update date - issue_url (str): url used to build escalation message - assignee_email (Optional[str]): Email for assignee to send escalation message. - May be None in Jira Cloud due to privacy settings. - """ - base_message = ( - f"please provide an update on issue : {issue_url} \n Prow Job Link : <{prow_job_url}|{prow_job_name}>" - ) - - if days_since_update > 4: - message = f"\n 4 or more days since last update, {base_message} " - LOGGER.info(f" escalation slack message : {message}") - self.send_slack_notification( - self.slack_channel, message, email=assignee_email, cc_user_email=self.team_manager_email - ) - - elif 2 < days_since_update <= 4: - message = f"\n 2 or more days since last update, {base_message}" - LOGGER.info(f" escalation slack message : {message}") - self.send_slack_notification(self.slack_channel, message, email=assignee_email) - - elif 1 < days_since_update <= 2: - LOGGER.info( - f"\n 1 or more days since last comment, please add comment if there is any updates on the issue: {jira_issue.key}" - ) - assignee_account_id = self.get_user_account_id(jira_issue.fields.assignee) - comment_body = ( - adf_doc( - paragraph( - adf_mention(assignee_account_id, "@"), - inline_text(", please provide update for this issue."), - ), - ) - if assignee_account_id - else plain_text_to_adf_doc("Please provide update for this issue.") - ) - LOGGER.info("escalation comment prepared for %s", jira_issue.key) - self.jira.comment(issue_id=jira_issue.key, comment=comment_body) diff --git a/src/objects/slack_base.py b/src/objects/slack_base.py deleted file mode 100644 index 86f181a1..00000000 --- a/src/objects/slack_base.py +++ /dev/null @@ -1,90 +0,0 @@ -import os -from typing import Optional -from slack_sdk import WebClient -from slack_sdk.errors import SlackApiError - -from simple_logger.logger import get_logger - -LOGGER = get_logger(name=__name__) - - -class SlackClient: - """ - A client for interacting with the Slack API. - Attributes: - client (WebClient): An instance of the Slack WebClient initialized with the slack bot token - """ - - def __init__(self, token: str) -> None: - """ - Constructs the slack object with the provided slack bot token - - Args: - token (str. optional): slack bot token , if not provided it defaults to SLACK_BOT_TOKEN environment variable. - """ - self.logger = LOGGER - - self.token = os.getenv("SLACK_BOT_TOKEN") if token is None else token - - if self.token is None: - raise ValueError("A bot token must be provided or set in the 'SLACK_BOT_TOKEN ' environment variable") - self.client = WebClient(token=self.token) - - def get_slack_username(self, email: str) -> Optional[str]: - """ - Look up a slack user by their email address. - - Args: - email (str): Email of the slack user to look up - - Returns: - str: A string containing user information if found, else None. - """ - try: - response = self.client.users_lookupByEmail(email=email) - user = response["user"] - return user["profile"]["display_name"] - except SlackApiError as e: - LOGGER.error(f"Error looking up user {e.response['error']}") - return None - - def send_notification(self, channel: str, text: str) -> None: - """ - Posts a message to specified slack channel - - Args: - channel (str): channel to post slack message - text (str): message text to post - - Raises: - SlackApiError: if an error occurs while sending the message - """ - try: - LOGGER.info(f"attempting to send slack message in channel #{channel}: {text}") - self.client.chat_postMessage(channel=channel, text=text) - - except SlackApiError as e: - LOGGER.error(f"Error posting slack message: {e.response['error']}") - - def get_slack_usergroup(self, group_name: str) -> Optional[str]: - """ - Look up a slack usergroup by their group name. - - Args: - group_name (str): string containing slack team handle or usergroup name - - Returns: - Optional[str]: A string containing usergroup id if found, else None. - """ - try: - response = self.client.usergroups_list() - for group in response["usergroups"]: - if group["name"] == group_name: - LOGGER.info(f"user group : {group['name']}") - usergroup_id = group["id"] - return usergroup_id - LOGGER.info("\n Slack user group not found \n") - return None - except SlackApiError as e: - LOGGER.error(f"Error looking up user {e.response['error']}") - return None diff --git a/tests/unittests/functions/escalation/test_jira_escalation.py b/tests/unittests/functions/escalation/test_jira_escalation.py deleted file mode 100644 index 22ce878d..00000000 --- a/tests/unittests/functions/escalation/test_jira_escalation.py +++ /dev/null @@ -1,387 +0,0 @@ -import pytest - -from unittest.mock import MagicMock, patch -from src.escalation.jira_escalation import Jira_Escalation, description_to_plain_text_for_search -from datetime import datetime, timedelta, timezone - -from simple_logger.logger import get_logger - - -LOGGER = get_logger(name=__name__) - - -FIXED_NOW = datetime(2025, 5, 1, tzinfo=timezone.utc) - -PROW_WIKI_LINE = "* Prow Job Link *: [periodic-ci-test-job-name #11234567|https://prow.ci.openshift.org/some/log/path]" - -PROW_DESCRIPTION_ADF = { - "type": "doc", - "version": 1, - "content": [ - { - "type": "heading", - "attrs": {"level": 4}, - "content": [{"type": "text", "text": "Job"}], - }, - { - "type": "paragraph", - "content": [ - {"type": "text", "text": PROW_WIKI_LINE}, - ], - }, - ], -} - - -class TestDescriptionToPlainTextForSearch: - def test_passes_through_plain_string(self): - assert description_to_plain_text_for_search(PROW_WIKI_LINE) == PROW_WIKI_LINE - - def test_flattens_adf_doc_to_searchable_text(self): - flat = description_to_plain_text_for_search(PROW_DESCRIPTION_ADF) - assert PROW_WIKI_LINE in flat - assert "periodic-ci-test-job-name" in flat - assert "https://prow.ci.openshift.org/some/log/path" in flat - - def test_extract_prow_job_name_after_flattening_adf_description(self): - flat = description_to_plain_text_for_search(PROW_DESCRIPTION_ADF) - assert Jira_Escalation.extract_prow_job_name(flat) == "periodic-ci-test-job-name #11234567" - - def test_extract_prow_job_link_after_flattening_adf_description(self): - flat = description_to_plain_text_for_search(PROW_DESCRIPTION_ADF) - assert Jira_Escalation.extract_prow_job_link(flat) == "https://prow.ci.openshift.org/some/log/path" - - def test_extractors_fail_on_raw_adf_dict_without_flattening(self): - with pytest.raises(TypeError): - Jira_Escalation.extract_prow_job_name(PROW_DESCRIPTION_ADF) - with pytest.raises(TypeError): - Jira_Escalation.extract_prow_job_link(PROW_DESCRIPTION_ADF) - - -@pytest.fixture -def fake_current_date(): - with patch("src.escalation.jira_escalation.datetime") as mock_datetime: - mock_datetime.now.return_value = FIXED_NOW - mock_datetime.strptime = datetime.strptime - yield mock_datetime - - -@pytest.fixture -def escalation_setup(): - mock_jira = MagicMock() - mock_slack = MagicMock() - return mock_jira, mock_slack - - -@pytest.fixture -def setup_jira_escalation(escalation_setup, fake_current_date): - """fixture to setup jira escalation instance""" - mock_jira_client, mock_slack_client = escalation_setup - - jira_escalation = Jira_Escalation( - jira=mock_jira_client, - slack_client=mock_slack_client, - slack_channel="test-channel", - default_labels=["test-lp"], - additional_labels=["test-labbel-a", "test-label-b"], - default_jira_project="test-project", - team_slack_handle="team-user-group", - team_manager_email="test-email-1@exmaple.com", - reporter_email="watcher@example.com", - base_issue_url="https://issues.stage.test.com", - ) - jira_escalation.send_slack_notification = MagicMock() - return jira_escalation, mock_jira_client, mock_slack_client - - -def create_mock_issue( - key: str, - assignee_email: str = None, - updated_days_ago: int = None, - comment_days_ago: int = None, - status_change_days_ago: int = None, - now: datetime = FIXED_NOW, -): - """Helper to create mock jira issue""" - - def days_ago(days) -> str: - days = (now - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S.%f%z") - return days - - fake_jira_issue = MagicMock() - fake_jira_issue.key = key - fake_jira_issue.fields = MagicMock() - - # Generate a unique accountId based on email (simulates Jira Cloud behavior) - assignee_account_id = f"accountId:{assignee_email}" if assignee_email else None - - # if assignee value is provided then update the mock field with actual value - if assignee_email: - assignee = MagicMock() - assignee.name = assignee_email.split("@")[0] - assignee.emailAddress = assignee_email - assignee.accountId = assignee_account_id - fake_jira_issue.fields.assignee = assignee - else: - fake_jira_issue.fields.assignee = None - - fake_jira_issue.fields.updated = days_ago(updated_days_ago) - - # Update the issue content with the tested job link - fake_jira_issue.fields.description = ( - "Prow Job link : [periodic-ci-test-job-openshift-fips|https://prow.ci.openshift.org/view/gs/some-log-path ]" - ) - - # Decides whether issue comments will be set or empty - comments_obj = MagicMock() - mock_comment = [] - if comment_days_ago is not None: - comment = MagicMock() - comment.author.emailAddress = assignee_email - comment.author.accountId = assignee_account_id - comment.updated = days_ago(comment_days_ago) - mock_comment.append(comment) - comments_obj.comments = mock_comment - fake_jira_issue.fields.comment = comments_obj - - # If issue supposed to be acknowledged by the assignee, update related fields - mock_changelog = MagicMock() - if status_change_days_ago is not None: - history = MagicMock() - history.created = days_ago(status_change_days_ago) - item = MagicMock() - item.field = "status" - item.toString = "ACK" - history.items = [item] - mock_changelog.histories = [history] - else: - mock_changelog.histories = [] - fake_jira_issue.changelog = mock_changelog - - return fake_jira_issue - - -def test_add_labels_to_jira_query(setup_jira_escalation): - escalation, _, _ = setup_jira_escalation - jira_query = 'Project = test-project AND status in("ACK") ' - query = escalation.add_labels_to_jira_query(jira_query) - expected_query = 'Project = test-project AND status in("ACK") AND (labels IN("test-labbel-a","test-label-b")) AND (labels = "test-lp")' - assert query == expected_query - - -class TestGetUserAccountId: - """Tests for get_user_account_id static method.""" - - def test_returns_account_id_when_present(self): - user = MagicMock() - user.accountId = "5b10ac8d82e05b22cc7d4ef5" # pragma: allowlist secret - assert Jira_Escalation.get_user_account_id(user) == "5b10ac8d82e05b22cc7d4ef5" # pragma: allowlist secret - - def test_returns_none_when_user_is_none(self): - assert Jira_Escalation.get_user_account_id(None) is None - - def test_returns_none_when_account_id_missing(self): - user = MagicMock(spec=[]) # Empty spec means no attributes - assert Jira_Escalation.get_user_account_id(user) is None - - -class TestGetUserEmail: - """Tests for get_user_email static method.""" - - def test_returns_email_when_present(self): - user = MagicMock() - user.emailAddress = "user@example.com" - assert Jira_Escalation.get_user_email(user) == "user@example.com" - - def test_returns_none_when_user_is_none(self): - assert Jira_Escalation.get_user_email(None) is None - - def test_returns_none_when_email_missing(self): - user = MagicMock(spec=[]) # Empty spec means no attributes - assert Jira_Escalation.get_user_email(user) is None - - def test_returns_none_when_email_is_none(self): - """Jira Cloud may return None for emailAddress due to privacy settings.""" - user = MagicMock() - user.emailAddress = None - assert Jira_Escalation.get_user_email(user) is None - - -@pytest.mark.parametrize( - "assignee_email, comment_days_ago, status_changed_days_ago, updated_days_ago, expect_cc_user, expect_add_jira_comment, expect_notify_assignee_in_slack", - [ - ("user@example.com", None, None, 1, False, False, False), - ("user@example.com", 5, None, 5, True, False, False), - ("user@example.com", 5, 5, 5, True, False, False), - ("user@example.com", 3, 3, 3, False, False, True), - ("user@example.com", 3, None, 3, False, False, True), - ("user@example.com", 3, None, 0, False, False, True), - ("user@example.com", 2, None, 2, False, True, False), - ("user@example.com", None, 5, 2, False, True, False), - ("user@example.com", 3, 1, 5, False, False, True), - ("user@example.com", 3, 5, 4, False, False, True), - ("user@example.com", 1, 1, 1, False, False, False), - (None, None, None, 1, False, False, False), - ], -) -def test_process_issues( - setup_jira_escalation, - assignee_email, - comment_days_ago, - status_changed_days_ago, - updated_days_ago, - expect_cc_user, - expect_add_jira_comment, - expect_notify_assignee_in_slack, -): - """ - Covers scenarios where assignee is present, and combination of when was the last time assignees - commented, status was changed, or when was the jira issue updated. - All jira issues will have `updated_days_ago` value, but that doesn't ensure if that update is due - to assignee comment or status change. The `process_issues` functionality prioritizes the assignee's comment, - followed by the status change and the last update date. - """ - fake_jira_issue = create_mock_issue( - key="Issue-1", - assignee_email=assignee_email, - comment_days_ago=comment_days_ago, - status_change_days_ago=status_changed_days_ago, - updated_days_ago=updated_days_ago, - ) - escalation, mock_jira, _ = setup_jira_escalation - - mock_jira.search_issues.return_value = [MagicMock(key="Issue-1")] - mock_jira.get_issue_by_id_or_key_with_changelog.return_value = fake_jira_issue - escalation.process_issues(jira_query="project = test-project") - - if assignee_email is None: - escalation.send_slack_notification.assert_not_called() - - if expect_cc_user: - escalation.send_slack_notification.assert_called_once() - args, kwargs = escalation.send_slack_notification.call_args - assert "test-email-1@exmaple.com" in kwargs["cc_user_email"] - - elif expect_add_jira_comment: - mock_jira.comment.assert_called_once() - kwargs = mock_jira.comment.call_args.kwargs - comment = kwargs["comment"] - assert isinstance(comment, dict) - mention = comment["content"][0]["content"][0] - assert mention["type"] == "mention" - assert mention["attrs"]["id"] == f"accountId:{assignee_email}" - - elif expect_notify_assignee_in_slack: - escalation.send_slack_notification.assert_called_once() - args, kwargs = escalation.send_slack_notification.call_args - assert assignee_email in kwargs or kwargs.get("email") - assert "cc_user_email" not in kwargs - else: - escalation.send_slack_notification.assert_not_called() - - -@pytest.mark.parametrize( - "description, expected_job_name", - [ - ( - "* Prow Job Link *: [periodic-ci-test-job-name #11234567|https://prow.ci.openshift.org/some/log/path]", - "periodic-ci-test-job-name #11234567", - ), - ("Random description without job name", None), - ], -) -def test_extract_job_name(description, expected_job_name, setup_jira_escalation): - escalation_instance, _, _ = setup_jira_escalation - - job_name = escalation_instance.extract_prow_job_name(description) - assert job_name == expected_job_name - - -@pytest.mark.parametrize( - "description, expected_job_url", - [ - ( - "* Prow Job Link *: [periodic-ci-test-job-name #11234567|https://prow.ci.openshift.org/some/log/path]", - "https://prow.ci.openshift.org/some/log/path", - ), - ("Random description without job name", None), - ], -) -def test_extract_prow_job_url(description, expected_job_url, setup_jira_escalation): - escalation_instance, _, _ = setup_jira_escalation - - job_name = escalation_instance.extract_prow_job_link(description) - assert job_name == expected_job_url - - -@pytest.mark.parametrize( - "query1_issues, query2_issues,query3_issues, assignees, updated_days_ago, expected_keys, expect_notifcation", - [ - (["ISSUE-1"], ["ISSUE-2"], ["ISSUE-3"], [None, None, None], 0, ["ISSUE-1", "ISSUE-2", "ISSUE-3"], True), - (["ISSUE-1"], ["ISSUE-2"], ["ISSUE-3"], [None, None, None], 0, ["ISSUE-1", "ISSUE-2", "ISSUE-3"], True), - (["ISSUE-1"], ["ISSUE-2"], [], [None, None], 2, ["ISSUE-1", "ISSUE-2"], True), - (["ISSUE-1"], [], [], [None, None, None], 1, ["ISSUE-1"], True), - ([], [], [], [], 5, [], False), - ], -) -def test_issues_with_no_assignee( - query1_issues, - query2_issues, - query3_issues, - assignees, - updated_days_ago, - expected_keys, - expect_notifcation, - escalation_setup, -): - mock_jira_client, mock_slack_client = escalation_setup - - all_keys = query1_issues + query2_issues + query3_issues - issues_by_key = {} - - for key, assignee in zip(all_keys, assignees): - issue = create_mock_issue( - key=key, - assignee_email=assignee if assignee else None, - comment_days_ago=None, - status_change_days_ago=None, - updated_days_ago=updated_days_ago, - ) - issues_by_key[key] = issue - - mock_jira_client.search_issues.side_effect = [ - [k for k in query1_issues], - [k for k in query2_issues], - [k for k in query3_issues], - ] - - def get_issue_by_id_or_key_with_changelog(key, expand=None): - return issues_by_key[key] - - mock_jira_client.get_issue_by_id_or_key_with_changelog.side_effect = get_issue_by_id_or_key_with_changelog - - # Validate message content, it must contain "No assignee", jira issue ids (`expected_keys`), and issue url for all the issues - with patch.object(Jira_Escalation, "send_slack_notification") as mock_notify: - Jira_Escalation( - jira=mock_jira_client, - slack_client=mock_slack_client, - slack_channel="test-channel", - default_labels=["test-lp"], - additional_labels=["test-labbel-a", "test-label-b"], - default_jira_project="test-project", - team_slack_handle="team-user-group", - team_manager_email="test-email-1@exmaple.com", - reporter_email="watcher@example.com", - base_issue_url="https://issues.stage.test.com", - ) - - if expect_notifcation: - mock_notify.assert_called_once() - args, kwargs = mock_notify.call_args - message = args[1] if args else kwargs.get("", "message") - - assert "No assignee" in message - for key in expected_keys: - assert f"https://issues.stage.test.com/browse/{key}" in message - else: - mock_notify.send_slack_notification.assert_not_called() diff --git a/tests/unittests/objects/slack/test_slack_client.py b/tests/unittests/objects/slack/test_slack_client.py deleted file mode 100644 index abe8e5f0..00000000 --- a/tests/unittests/objects/slack/test_slack_client.py +++ /dev/null @@ -1,51 +0,0 @@ -import pytest -from unittest.mock import MagicMock, patch -from slack_sdk.errors import SlackApiError -from slack_sdk import WebClient -from src.objects.slack_base import SlackClient - - -@pytest.fixture -def slack_client(): - """ - fixture to create slack client instance - """ - return SlackClient(token="test-token") - - -@patch.object(WebClient, "users_lookupByEmail") -def test_get_slack_username_success(mock_users_lookup, slack_client): - mock_users_lookup.return_value = {"user": {"profile": {"display_name": "test", "email": "test@example.com"}}} - username = slack_client.get_slack_username("test@example.com") - - assert username == "test" - mock_users_lookup.assert_called_once_with(email="test@example.com") - - -@patch.object(WebClient, "users_lookupByEmail") -def test_get_slack_username_failure(mock_users_lookup, slack_client): - mock_users_lookup.side_effect = SlackApiError( - "User not found", response=MagicMock(status=404, data={"error": "User not found"}) - ) - username = slack_client.get_slack_username("nonexistentemail@example.com") - - assert username is None - mock_users_lookup.assert_called_once_with(email="nonexistentemail@example.com") - - -@patch.object(WebClient, "chat_postMessage") -def test_send_notification_success(mock_chat_post, slack_client): - slack_client.send_notification("#test-channel", "test message") - - mock_chat_post.assert_called_once_with(channel="#test-channel", text="test message") - - -@patch.object(WebClient, "chat_postMessage") -def test_send_notification_failure(mock_chat_post, slack_client): - mock_chat_post.side_effect = SlackApiError( - "channel not found", response=MagicMock(status=404, sata={"error": "channel_not_found"}) - ) - - slack_client.send_notification("#invalid-channel", "test message") - - mock_chat_post.assert_called_once_with(channel="#invalid-channel", text="test message")