diff --git a/cloud_chatops/src/custom_exceptions.py b/cloud_chatops/src/custom_exceptions.py index 327ccd13..b70ad511 100644 --- a/cloud_chatops/src/custom_exceptions.py +++ b/cloud_chatops/src/custom_exceptions.py @@ -1,25 +1,29 @@ """This module contains custom exceptions to handle errors for the Application.""" -class RepoNotFound(Exception): +class RepoNotFound(LookupError): """Error: The requested repository does not exist on GitHub.""" -class UnknownHTTPError(Exception): +class UnknownHTTPError(RuntimeError): """Error: The received HTTP response is unexpected.""" -class RepositoriesNotGiven(Exception): +class RepositoriesNotGiven(RuntimeError): """Error: repos.csv does not contain any repositories.""" -class TokensNotGiven(Exception): +class TokensNotGiven(RuntimeError): """Error: Token values are either empty or not given.""" -class UserMapNotGiven(Exception): +class UserMapNotGiven(RuntimeError): """Error: User map is empty.""" -class BadGitHubToken(Exception): +class BadGitHubToken(RuntimeError): """Error: GitHub REST Api token is invalid.""" + + +class ChannelNotFound(LookupError): + """Error: The channel was not found.""" diff --git a/cloud_chatops/src/get_github_prs.py b/cloud_chatops/src/get_github_prs.py index 744157e5..51d91448 100644 --- a/cloud_chatops/src/get_github_prs.py +++ b/cloud_chatops/src/get_github_prs.py @@ -1,15 +1,18 @@ """ This module handles the HTTP requests and formatting to the GitHub REST Api. -It will get all open pull requests in provided STFC owned repositories. +It will get all open pull requests in provided repositories. """ -from typing import List, Dict, Union +from typing import List, Dict import requests -from src.read_data import get_token -from src.custom_exceptions import RepoNotFound, UnknownHTTPError, BadGitHubToken +from read_data import get_token +from pr_dataclass import PrData +from custom_exceptions import RepoNotFound, UnknownHTTPError, BadGitHubToken class GetGitHubPRs: + # pylint: disable=R0903 + # Disabling this as in the future there is likely to be more public functions. """ This class handles getting the open PRs from the GitHub Rest API. """ @@ -22,29 +25,63 @@ def __init__(self, repos: List[str], owner: str): """ self.repos = repos self.owner = owner + self._http_handler = HTTPHandler() - def run(self) -> Dict[str, List]: + def run(self) -> List[PrData]: """ This method is the entry point to the class. - It runs the HTTP request methods and the formatting methods. + It runs the HTTP request methods and returns the responses. :return: The responses from the HTTP requests. """ - unformatted_responses = self.request_all_repos_http() - formatted_responses = self.format_http_responses(unformatted_responses) - return formatted_responses + responses = self._request_all_repos_http() + return self._parse_pr_to_dataclass(responses) - def request_all_repos_http(self) -> Dict[str, List[Dict]]: + def _request_all_repos_http(self) -> List[Dict]: """ This method starts a request for each repository and returns a list of those PRs. :return: A dictionary of repos and their PRs. """ - responses = {} + responses = [] for repo in self.repos: url = f"https://api.github.com/repos/{self.owner}/{repo}/pulls" - response = self.get_http_response(url) - responses[repo] = response + responses += self._http_handler.make_request(url) return responses + @staticmethod + def _parse_pr_to_dataclass(responses: List[Dict]) -> List[PrData]: + """ + This module converts the responses from the HTTP request into Dataclasses to be more easily handled. + :param responses: List of responses made from HTTP requests + :return: Responses in dataclasses + """ + responses_dataclasses = [] + for pr in responses: + responses_dataclasses.append( + PrData( + pr_title=f"{pr['title']} #{pr['number']}", + user=pr["user"]["login"], + url=pr["html_url"], + created_at=pr["created_at"], + draft=pr["draft"], + ) + ) + return responses_dataclasses + + +class HTTPHandler: + """ + This class makes the HTTP requests to the GitHub REST API. + """ + + def make_request(self, url: str) -> List[Dict]: + """ + This method gets the HTTP response from the URL given and returns the response as a list. + :param url: The URL to make the HTTP request to. + :return: List of PRs. + """ + response = self.get_http_response(url) + return [response] if isinstance(response, dict) else response + def get_http_response(self, url: str) -> List[Dict]: """ This method sends a HTTP request to the GitHub Rest API endpoint and returns all open PRs from that repository. @@ -53,57 +90,28 @@ def get_http_response(self, url: str) -> List[Dict]: """ headers = {"Authorization": "token " + get_token("GITHUB_TOKEN")} response = requests.get(url, headers=headers, timeout=60) - self.validate_response(response, url) + self._validate_response(response) return response.json() - def format_http_responses( - self, responses: Union[Dict[str, List], Dict[str, Dict]] - ) -> Dict[str, List]: - """ - This method checks the formats the responses from GitHub are in a consistent format. - :param responses: GitHub's HTTP responses. - :return: Dictionary of responses. - """ - culled_responses = self.remove_empty_response(responses) - for repo, response in culled_responses.items(): - if isinstance(response, dict): - responses[repo] = [response] - return responses - - @staticmethod - def remove_empty_response( - responses: Union[Dict[str, List], Dict[str, Dict]] - ) -> Union[Dict[str, List], Dict[str, Dict]]: - """ - This method removes all empty responses from the Dictionary. - An empty response is the result of no open pull requests. - :param responses: The responses to check. - :return: The responses with no empty values. - """ - to_remove = [] - for repo, response in responses.items(): - if not response: - to_remove.append(repo) - - for i in to_remove: - del responses[i] - return responses - @staticmethod - def validate_response(response: requests.get, url: str) -> None: + def _validate_response(response: requests.get) -> None: """ This method checks the status code of the HTTP response and handles exceptions accordingly. :param response: The response to check. - :param url: The url that was not found. """ - if response.status_code == 401: - raise BadGitHubToken( - "Your GitHub api token is invalid. Check that it hasn't expired." - ) - if response.status_code == 404: - raise RepoNotFound(f'The repo at the url "{url}" could not be found.') + match response.status_code: + case 200: + pass + case 401: + raise BadGitHubToken( + "Your GitHub api token is invalid. Check that it hasn't expired." + ) + case 404: + raise RepoNotFound( + f'The repository at the url "{response.url}" could not be found.' + ) - if response.status_code != 200: - raise UnknownHTTPError( - f"The HTTP response code is unknown and cannot be handled. Response: {response.status_code}" - ) + case _: + raise UnknownHTTPError( + f"The HTTP response code is unknown and cannot be handled. Response: {response.status_code}" + ) diff --git a/cloud_chatops/src/main.py b/cloud_chatops/src/main.py index de468d07..03d0059d 100644 --- a/cloud_chatops/src/main.py +++ b/cloud_chatops/src/main.py @@ -4,7 +4,7 @@ """ import asyncio -from src.slack_app import run_app +from slack_app import run_app def main(): diff --git a/cloud_chatops/src/online_notif.py b/cloud_chatops/src/online_notif.py index ddfa76f6..170f3d3c 100644 --- a/cloud_chatops/src/online_notif.py +++ b/cloud_chatops/src/online_notif.py @@ -1,7 +1,7 @@ """This module handles the Slack Application status notifications.""" from slack_sdk import WebClient -from src.read_data import get_token, get_maintainer +from read_data import get_token, get_maintainer def online_notif() -> None: diff --git a/cloud_chatops/src/pr_dataclass.py b/cloud_chatops/src/pr_dataclass.py index 78ca3c7d..29f6981d 100644 --- a/cloud_chatops/src/pr_dataclass.py +++ b/cloud_chatops/src/pr_dataclass.py @@ -2,19 +2,17 @@ This module declares the dataclass used to store PR information. This is preferred over dictionaries as dataclasses make code more readable. """ + from dataclasses import dataclass -# Disabling this Pylint error as the dataclass needs to hold more than 7 attributes. -# pylint: disable=R0902 + @dataclass class PrData: """Class holding information about a single pull request.""" + pr_title: str user: str url: str created_at: str - channel: str - thread_ts: str - mention: bool draft: bool old: bool = False diff --git a/cloud_chatops/src/pr_reminder.py b/cloud_chatops/src/pr_reminder.py index eec6f54b..f1b086db 100644 --- a/cloud_chatops/src/pr_reminder.py +++ b/cloud_chatops/src/pr_reminder.py @@ -1,175 +1,166 @@ """This module handles the posting of messages to Slack using the Slack SDK WebClient class.""" -from typing import List, Dict +from typing import List from datetime import datetime, timedelta from slack_sdk import WebClient from slack_sdk.errors import SlackApiError -from src.read_data import get_token, get_user_map, get_repos -from src.get_github_prs import GetGitHubPRs -from src.pr_dataclass import PrData +from read_data import get_token, get_user_map, get_repos +from get_github_prs import GetGitHubPRs +from pr_dataclass import PrData +from custom_exceptions import ChannelNotFound class PostPRsToSlack: + # pylint: disable=R0903 + # Disabling this as there only needs to be one entry point. """ This class handles the Slack posting. """ - - def __init__(self): - super().__init__() - self.repos = get_repos() - self.client = WebClient(token=get_token("SLACK_BOT_TOKEN")) + # The default channel is dev-chatops + def __init__(self, mention=False, channel="C06U37Y02R4"): + self.channel = channel + self.thread_ts = "" + self.mention = mention self.slack_ids = get_user_map() + self.message_builder = PRMessageBuilder(self.mention) + self.client = WebClient(token=get_token("SLACK_BOT_TOKEN")) self.prs = GetGitHubPRs(get_repos(), "stfc").run() - self.channel = "temp-chatops" - def run(self, mention, channel=None) -> None: + def run(self) -> None: """ - This method calls the functions to post the reminder message and subsequent PR messages. - :param channel: Changes the channel to post the messages to. - :param mention: To mention the users in Slack or just post their name. + This method sets class attributes then cals the reminder and thread post methods. """ - if channel: - self.channel = channel - reminder_message = self.post_reminder_message() - self.post_thread_messages(self.prs, reminder_message, mention) + self._post_reminder_message() + self._post_thread_messages(self.prs) - def post_reminder_message(self) -> WebClient.chat_postMessage: + def _post_reminder_message(self) -> None: """ This method posts the main reminder message to start the thread if PR notifications. - :return: The reminder message return object """ - reminder = self.client.chat_postMessage( + response = self.client.chat_postMessage( channel=self.channel, text="Here are the outstanding PRs as of today:", ) - return reminder + self.thread_ts = response.data["ts"] - def post_thread_messages( - self, - prs: Dict[str, List], - reminder_message: WebClient.chat_postMessage, - mention: bool, - ) -> None: + def _post_thread_messages(self, prs: List[PrData]) -> None: """ This method iterates through each PR and calls the post method for them. - :param mention: To mention the users or not :param prs: A list of PRs from GitHub - :param reminder_message: The reminder message object """ prs_found = False - for repo in prs.values(): - for pr in repo: - info = PrData( - pr_title=f"{pr['title']} #{pr['number']}", - user=pr["user"]["login"], - url=pr["html_url"], - created_at=pr["created_at"], - channel=reminder_message.data["channel"], - thread_ts=reminder_message.data["ts"], - mention=mention, - draft=pr["draft"], - ) - prs_found = True - checked_info = self.check_pr(info) - self.send_thread(checked_info) + for pr in prs: + prs_found = True + response = self._send_thread(pr) + self._send_thread_react(pr, response.data["ts"]) if not prs_found: - self.send_no_prs(reminder_message) - - def send_no_prs(self, reminder: WebClient.chat_postMessage) -> None: - """ - This method sends a message to the user that they have no PRs open. - This method is only called if no other PRs have been mentioned. - :param reminder: The thread message to send under. - """ - self.client.chat_postMessage( - text="There are no outstanding pull requests open.", - channel=reminder.data["channel"], - thread_ts=reminder.data["ts"], - unfurl_links=False, - ) - - def check_pr(self, info: PrData) -> PrData: - """ - This method validates certain information in the PR data such as who authored the PR and if it's old or not. - :param info: The information to validate. - :return: The validated information. - """ - if info.user not in self.slack_ids: - # If the PR author is not in the Slack ID mapping - # then we set the user to mention as David Fairbrother - # as the team lead to deal with this PR. - info.user = "U01JG0LKU3W" - else: - info.user = self.github_to_slack_username(info.user) - opened_date = datetime.fromisoformat(info.created_at).replace(tzinfo=None) - datetime_now = datetime.now().replace(tzinfo=None) - time_cutoff = datetime_now - timedelta(days=30 * 6) - if opened_date < time_cutoff: - info.old = True - del info.created_at - return info + self._send_no_prs_found() - def send_thread( - self, - data: PrData - ) -> None: + def _send_thread(self, pr_data: PrData) -> WebClient.chat_postMessage: """ - This method sends the thread message and prepares the reactions. - :param data: The PR data as a dataclass + This method sends the message and returns the response. + :param pr_data: The PR data as a dataclass + :return: The message response. """ - message = self.construct_message(data.pr_title, data.user, data.url, data.mention, data.draft, data.old) + message = self.message_builder.make_message(pr_data) response = self.client.chat_postMessage( - text=message, channel=data.channel, thread_ts=data.thread_ts, unfurl_links=False + text=message, + channel=self.channel, + thread_ts=self.thread_ts, + unfurl_links=False, ) assert response["ok"] - reactions = { - "old": data.old, - "draft": data.draft, - } - self.send_thread_react(data.channel, response.data["ts"], reactions) + return response - def send_thread_react(self, channel: str, ts: str, reactions: Dict) -> None: + def _send_thread_react(self, pr_data: PrData, message_ts: str) -> None: """ This method sends reactions to the PR message if necessary. + :param pr_data: The PR info + :param message_ts: The timestamp of the message to react to """ - mapping = { + reactions = { "old": "alarm_clock", "draft": "scroll", } - for react in reactions: - if reactions[react]: + for react, react_id in reactions.items(): + if getattr(pr_data, react): react_response = self.client.reactions_add( - channel=channel, name=mapping[react], timestamp=ts + channel=self.channel, name=react_id, timestamp=message_ts ) assert react_response["ok"] - def construct_message( - self, pr_title: str, user: str, url: str, mention: bool, draft: bool, old: bool - ) -> str: + def _send_no_prs_found(self) -> None: """ - This method constructs the PR message depending on if the PR is old and if the message should mention or not. - :param pr_title: The title of the PR. - :param user: The author of the PR. - :param url: The URL of the PR. - :param mention: To mention users or not. - :param draft: If the PR is a draft. - :param old: If the PR is older than 6 months. + This method sends a message to the chat that there are no PRs. + """ + self.client.chat_postMessage( + text="There are no outstanding pull requests open.", + channel=self.channel, + thread_ts=self.thread_ts, + unfurl_links=False, + ) + + def _set_channel_id(self, channel_name: str) -> None: + """ + This method will get the channel id from the channel name and set the attrribute to the class. + This is necessary as the chat.postMessage method takes name or id but reactions.add only takes id. + This method also acts as a channel verif + :param channel_name: The channel name to lookup :return: """ + channels = self.client.conversations_list(types="private_channel")["channels"] + channel_obj = next( + (channel for channel in channels if channel["name"] == channel_name), None + ) + if channel_obj: + self.channel = channel_obj["id"] + else: + raise ChannelNotFound( + f"The channel {channel_name} could not be found. Check the bot is a member of the channel.\n" + f' You can use "/invite @Cloud ChatOps" to invite the app to your channel.' + ) + + +class PRMessageBuilder: + """This class handles constructing the PR messages to be sent.""" + + # pylint: disable=R0903 + # Disabling this as there only needs to be one entry point. + # The default user here is David Fairbrother + def __init__(self, mention, default_user_id="U01JG0LKU3W"): + self.client = WebClient(token=get_token("SLACK_BOT_TOKEN")) + self.slack_ids = get_user_map() + self.mention = mention + self.default_user_id = default_user_id + + def make_message(self, pr_data: PrData) -> str: + """ + This method checks the pr data and makes a string message from it. + :param pr_data: The PR info + :return: The message to post + """ + checked_info = self._check_pr_info(pr_data) + return self._construct_string(checked_info) + + def _construct_string(self, pr_data: PrData) -> str: + """ + This method constructs the PR message depending on if the PR is old and if the message should mention or not. + :param pr_data: The data class containing the info about the PR. + :return: The message as a single string. + """ message = [] - if old: + if pr_data.old: message.append("*This PR is older than 6 months. Consider closing it:*") - message.append(f"Pull Request: <{url}|{pr_title}>") - if mention and not draft: - message.append(f"Author: <@{user}>") + message.append(f"Pull Request: <{pr_data.url}|{pr_data.pr_title}>") + if self.mention and not pr_data.draft: + message.append(f"Author: <@{pr_data.user}>") else: - name = self.get_real_name(user) + name = self._get_real_name(pr_data.user) message.append(f"Author: {name}") - return "\n".join(message) - def get_real_name(self, username: str) -> str: + def _get_real_name(self, username: str) -> str: """ This method uses the Slack client method to get the real name of a user and returns it. :param username: The user ID to look for @@ -181,12 +172,36 @@ def get_real_name(self, username: str) -> str: name = username return name - def github_to_slack_username(self, user: str) -> str: + def _github_to_slack_username(self, user: str) -> str: """ This method checks if we have a Slack id for the GitHub user and returns it. :param user: GitHub username to check for :return: Slack ID or GitHub username """ - if user in self.slack_ids: - return self.slack_ids[user] + if user not in self.slack_ids: + user = self.default_user_id + else: + user = self.slack_ids[user] return user + + @staticmethod + def _check_pr_age(time_created: str) -> bool: + """ + This method checks if the PR is older than 6 months. + :param time_created: The date the PR was created. + :return: PR older or not. + """ + opened_date = datetime.fromisoformat(time_created).replace(tzinfo=None) + datetime_now = datetime.now().replace(tzinfo=None) + time_cutoff = datetime_now - timedelta(days=30 * 6) + return opened_date < time_cutoff + + def _check_pr_info(self, info: PrData) -> PrData: + """ + This method validates certain information in the PR data such as who authored the PR and if it's old or not. + :param info: The information to validate. + :return: The validated information. + """ + info.user = self._github_to_slack_username(info.user) + info.old = self._check_pr_age(info.created_at) + return info diff --git a/cloud_chatops/src/read_data.py b/cloud_chatops/src/read_data.py index 74707b8f..3ed9e4b0 100644 --- a/cloud_chatops/src/read_data.py +++ b/cloud_chatops/src/read_data.py @@ -2,7 +2,7 @@ from typing import List, Dict import json -from src.custom_exceptions import RepositoriesNotGiven, UserMapNotGiven, TokensNotGiven +from custom_exceptions import RepositoriesNotGiven, UserMapNotGiven, TokensNotGiven def validate_required_files() -> None: diff --git a/cloud_chatops/src/slack_app.py b/cloud_chatops/src/slack_app.py index c0d50c59..539e0030 100644 --- a/cloud_chatops/src/slack_app.py +++ b/cloud_chatops/src/slack_app.py @@ -9,9 +9,9 @@ from slack_bolt.app.async_app import AsyncApp from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler import schedule -from src.pr_reminder import PostPRsToSlack -from src.read_data import get_token, validate_required_files -from src.online_notif import online_notif +from pr_reminder import PostPRsToSlack +from read_data import get_token, validate_required_files +from online_notif import online_notif logging.basicConfig(level=logging.DEBUG) @@ -23,11 +23,11 @@ async def schedule_jobs() -> None: This function schedules tasks for the async loop to run when the time is right. """ - def run_pr(channel, mention=False): + def run_pr(channel, mention=False) -> None: """ This is a placeholder function for the schedule to accept. """ - PostPRsToSlack().run(mention=mention, channel=channel) + PostPRsToSlack(mention=mention, channel=channel).run() schedule.every().monday.at("09:00").do( run_pr, mention=True, channel="pull-requests" diff --git a/cloud_chatops/tests/test_read_data.py b/cloud_chatops/tests/test_read_data.py index 82f9fe0e..c79eeca0 100644 --- a/cloud_chatops/tests/test_read_data.py +++ b/cloud_chatops/tests/test_read_data.py @@ -1,3 +1,5 @@ +"""This test file covers all tests for the read_data module.""" + from unittest.mock import patch, mock_open from src.read_data import get_token, get_repos, get_user_map, get_maintainer