diff --git a/.github/workflows/ci_code.yml b/.github/workflows/ci_code.yml index 4f883375..0ede9730 100644 --- a/.github/workflows/ci_code.yml +++ b/.github/workflows/ci_code.yml @@ -55,4 +55,5 @@ jobs: NOTIFY_TEST_KEY: ${{ secrets.NOTIFY_TEST_KEY }} NOTIFY_SRE_USER_NAME: ${{ secrets.NOTIFY_SRE_USER_NAME }} NOTIFY_SRE_CLIENT_SECRET: ${{ secrets.NOTIFY_SRE_CLIENT_SECRET }} - NOTIFY_OPS_CHANNEL_ID: ${{ secrets.NOTIFY_OPS_CHANNEL_ID }} \ No newline at end of file + NOTIFY_OPS_CHANNEL_ID: ${{ secrets.NOTIFY_OPS_CHANNEL_ID }} + SRE_BOT_EMAIL: ${{ secrets.SRE_BOT_EMAIL }} \ No newline at end of file diff --git a/app/integrations/google_drive.py b/app/integrations/google_drive.py index fc372f73..0c02a1e0 100644 --- a/app/integrations/google_drive.py +++ b/app/integrations/google_drive.py @@ -23,6 +23,7 @@ "https://www.googleapis.com/auth/drive.file", "https://www.googleapis.com/auth/docs", "https://www.googleapis.com/auth/spreadsheets", + "https://www.googleapis.com/auth/calendar", ] diff --git a/app/integrations/google_workspace/google_calendar.py b/app/integrations/google_workspace/google_calendar.py index e69de29b..aaf9779f 100644 --- a/app/integrations/google_workspace/google_calendar.py +++ b/app/integrations/google_workspace/google_calendar.py @@ -0,0 +1,163 @@ +import os +import logging +from datetime import datetime, timedelta + +import pytz +import json + +from integrations.google_workspace.google_service import ( + get_google_service, + handle_google_api_errors, +) + +# Get the email for the SRE bot +SRE_BOT_EMAIL = os.environ.get("SRE_BOT_EMAIL") + +# If modifying these scopes, delete the file token.json. +SCOPES = ["https://www.googleapis.com/auth/calendar"] + + +# Schedule a calendar event by finding the first available slot in the next 60 days that all participants are free in and book the event +@handle_google_api_errors +def schedule_event(event_details): + # initialize the google service + service = get_google_service( + "calendar", "v3", delegated_user_email=SRE_BOT_EMAIL, scopes=SCOPES + ) + + # Define the time range for the query + now = datetime.utcnow() + time_min = now.isoformat() + "Z" # 'Z' indicates UTC time + time_max = (now + timedelta(days=60)).isoformat() + "Z" + + # Construct the items array + items = [] + emails = json.loads(event_details).get("emails") + incident_name = json.loads(event_details).get("topic") + for email in emails: + email = email.strip() + items.append({"id": email}) + + # Construct the request body + freebusy_query = { + "timeMin": time_min, + "timeMax": time_max, + "items": items, + } + + # Execute the query to find all the busy times for all the participants + freebusy_result = service.freebusy().query(body=freebusy_query).execute() + + # return the first available slot to book the event + first_available_start, first_available_end = find_first_available_slot( + freebusy_result + ) + + # If there are no available slots, return None + if first_available_start is None and first_available_end is None: + logging.info("No available slots found") + return None + + # Crete the event in everyone's calendar + event_link = book_calendar_event( + service, first_available_start, first_available_end, emails, incident_name + ) + + return event_link + + +# Create a calendar event in everyone's calendar +def book_calendar_event(service, start, end, emails, incident_name): + # Build the attendees array + attendees = [] + for email in emails: + attendees.append({"email": email.strip()}) + + # Create the event + event = { + "summary": "Retro " + incident_name, + "description": "This is a retro meeting to discuss incident: " + incident_name, + "start": {"dateTime": start.isoformat()}, + "end": {"dateTime": end.isoformat()}, + "attendees": attendees, + "conferenceData": { + "createRequest": { + "requestId": f"{start.timestamp()}", # Unique ID per event to avoid collisions + "conferenceSolutionKey": { + "type": "hangoutsMeet" # This automatically generates a Google Meet link + }, + } + }, + "reminders": { + "useDefault": False, + "overrides": [ + {"method": "popup", "minutes": 10}, + ], + }, + } + + # call the google calendar API to create the event and send an email to all attendees + event = ( + service.events() + .insert( + calendarId="primary", body=event, conferenceDataVersion=1, sendUpdates="all" + ) + .execute() + ) + + # Return the link to the calendar event + return event.get("htmlLink") + + +# Function to use the freebusy response to find the first available spot in the next 60 days. We look for a 30 minute windows, 3 +# days in the future, ignoring weekends +def find_first_available_slot( + freebusy_response, duration_minutes=30, days_in_future=3, search_days_limit=60 +): + # EST timezone + est = pytz.timezone("US/Eastern") + + # Combine all busy times into a single list and sort them + busy_times = [] + for calendar in freebusy_response["calendars"].values(): + for busy_period in calendar["busy"]: + start = ( + datetime.strptime(busy_period["start"], "%Y-%m-%dT%H:%M:%SZ") + .replace(tzinfo=pytz.UTC) + .astimezone(est) + ) + end = ( + datetime.strptime(busy_period["end"], "%Y-%m-%dT%H:%M:%SZ") + .replace(tzinfo=pytz.UTC) + .astimezone(est) + ) + busy_times.append((start, end)) + busy_times.sort(key=lambda x: x[0]) + + for day_offset in range(days_in_future, days_in_future + search_days_limit): + # Calculate the start and end times of the search window for the current day + search_date = datetime.now(tz=est) + timedelta(days=day_offset) + + # Check if the day is Saturday (5) or Sunday (6) and skip it + if search_date.weekday() in [5, 6]: + continue + + search_start = search_date.replace( + hour=13, minute=0, second=0, microsecond=0 + ) # 1 PM EST + search_end = search_date.replace( + hour=15, minute=0, second=0, microsecond=0 + ) # 3 PM EST + + # Attempt to find an available slot within this day's search window + for current_time in ( + search_start + timedelta(minutes=i) for i in range(0, 121, duration_minutes) + ): + slot_end = current_time + timedelta(minutes=duration_minutes) + if all( + slot_end <= start or current_time >= end for start, end in busy_times + ): + if slot_end <= search_end: + return current_time, slot_end + + return None, None # No available slot found after searching the limit diff --git a/app/modules/incident/incident.py b/app/modules/incident/incident.py index f0bc8f01..bd1357d4 100644 --- a/app/modules/incident/incident.py +++ b/app/modules/incident/incident.py @@ -357,6 +357,9 @@ def submit(ack, view, say, body, client, logger): text = "Run `/sre incident close` to update the status of the incident document and incident spreadsheet to closed and to archive the channel" say(text=text, channel=channel_id) + text = "Run `/sre incident schedule` to let the SRE bot schedule a Retro Google calendar meeting for all participants." + say(text=text, channel=channel_id) + def generate_success_modal(body, channel_id, channel_name): locale = body["view"]["blocks"][0]["elements"][0]["value"] diff --git a/app/modules/incident/incident_helper.py b/app/modules/incident/incident_helper.py index 40d99c14..38122e7b 100644 --- a/app/modules/incident/incident_helper.py +++ b/app/modules/incident/incident_helper.py @@ -1,7 +1,8 @@ +import os import json import logging from integrations import google_drive -from integrations.google_workspace import google_docs +from integrations.google_workspace import google_docs, google_calendar from integrations.slack import channels as slack_channels from integrations.sentinel import log_to_sentinel @@ -20,6 +21,9 @@ \n `/sre incident roles` \n - manages roles in an incident channel \n - gérer les rôles dans un canal d'incident +\n `/sre incident schedule` +\n - schedules a 30 minute meeting a week into the future in everyone's calendars for the incident post mortem process. +\n - planifie une réunion de 30 minutes une semaine dans le futur dans les calendriers de tout le monde pour le processus de post-mortem de l'incident. \n `/sre incident close` \n - close the incident, archive the channel and update the incident spreadsheet and document \n - clôturer l'incident, archiver le canal et mettre à jour la feuille de calcul et le document de l'incident @@ -37,6 +41,8 @@ def register(bot): bot.action("delete_folder_metadata")(delete_folder_metadata) bot.action("archive_channel")(archive_channel_action) bot.view("view_save_incident_roles")(save_incident_roles) + bot.view("view_save_event")(save_incident_retro) + bot.action("confirm_click")(confirm_click) def handle_incident_command(args, client, body, respond, ack): @@ -59,6 +65,8 @@ def handle_incident_command(args, client, body, respond, ack): close_incident(client, body, ack) case "stale": stale_incidents(client, body, ack) + case "schedule": + schedule_incident_retro(client, body, ack) case _: respond( f"Unknown command: {action}. Type `/sre incident help` to see a list of commands." @@ -408,6 +416,177 @@ def stale_incidents(client, body, ack): client.views_update(view_id=placeholder_modal["view"]["id"], view=blocks) +# Function to be triggered when the /sre incident schedule command is called. This function brings up a modal window +# that explains how the event is scheduled and allows the user to schedule a retro meeting for the incident after the +# submit button is clicked. +def schedule_incident_retro(client, body, ack): + ack() + channel_id = body["channel_id"] + channel_name = body["channel_name"] + user_id = body["user_id"] + + # if we are not in an incident channel, then we need to display a message to the user that they need to use this command in an incident channel + if not channel_name.startswith("incident-"): + try: + response = client.chat_postEphemeral( + text=f"Channel {channel_name} is not an incident channel. Please use this command in an incident channel.", + channel=channel_id, + user=user_id, + ) + except Exception as e: + logging.error( + f"Could not post ephemeral message to user %s due to {e}.", + user_id, + ) + return + + # Get security group members. We want to exclude them from the list of people to schedule the event for + security_group_users = client.usergroups_users_list( + usergroup=os.getenv("SLACK_SECURITY_USER_GROUP_ID") + )["users"] + print("Security group users: ", security_group_users) + + # get all users in a channel + users = client.conversations_members(channel=channel_id)["members"] + + # Get the channel topic + channel_topic = client.conversations_info(channel=channel_id)["channel"]["topic"][ + "value" + ] + # If for some reason the channel topic is empty, set it to "Incident Retro" + if channel_topic == "": + channel_topic = "Incident Retro" + logging.warning("Channel topic is empty. Setting it to 'Incident Retro'") + + user_emails = [] + + # get the email addresses of all the users in the channel, except security group members and any apps/bots in the channel, since bots don't have an email address associated with them. + for user in users: + if user not in security_group_users: + response = client.users_info(user=user)["user"]["profile"] + # don't include bots in the list of users + if "bot_id" not in response: + user_emails.append(response["email"]) + + # convert the data to string so that we can send it as private metadata + data_to_send = json.dumps({"emails": user_emails, "topic": channel_topic}) + + blocks = { + "type": "modal", + "callback_id": "view_save_event", + "private_metadata": data_to_send, + "title": {"type": "plain_text", "text": "SRE - Schedule Retro 🗓️"}, + "submit": {"type": "plain_text", "text": "Schedule"}, + "blocks": ( + [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "*By clicking this button an event will be scheduled.* The following rules will be followed:", + }, + }, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "1. The event will be scheduled for the first available 30 minute timeslot starting 3 calendar days from now.", + }, + }, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "2. A proposed event will be added to everyone's calendar that is part of this channel (except Security team).", + }, + }, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "3. The retro will be scheduled only between 1:00pm and 3:00pm EDT to accomodate all time differences.", + }, + }, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "4. If no free time exists for the next 2 months, the event will not be scheduled.", + }, + }, + ] + ), + } + # Open the modal window + client.views_open( + trigger_id=body["trigger_id"], + view=blocks, + ) + + +# Function to create the calendar event and bring up a modal that contains a link to the event. If the event could not be scheduled, +# a message is displayed to the user that the event could not be scheduled. +def save_incident_retro(client, ack, body, view): + ack() + + # pass the data using the view["private_metadata"] to the schedule_event function + event_link = google_calendar.schedule_event(view["private_metadata"]) + # if we could not schedule the event, display a message to the user that the event could not be scheduled + if event_link is None: + blocks = { + "type": "modal", + "title": {"type": "plain_text", "text": "SRE - Schedule Retro 🗓️"}, + "blocks": ( + [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "*Could not schedule event - no free time was found!*", + }, + } + ] + ), + } + # if the event was scheduled successfully, display a message to the user that the event was scheduled and provide a link to the event + else: + blocks = { + "type": "modal", + "title": {"type": "plain_text", "text": "SRE - Schedule Retro 🗓️"}, + "blocks": ( + [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "*Successfully schduled calender event!*", + }, + "accessory": { + "type": "button", + "text": { + "type": "plain_text", + "text": "View Calendar Event", + }, + "value": "view_event", + "url": f"{event_link}", + "action_id": "confirm_click", + }, + }, + ] + ), + } + # Open the modal and log that the event was scheduled successfully + client.views_open(trigger_id=body["trigger_id"], view=blocks) + logging.info("Event has been scheduled successfully. Link: %s", event_link) + + +# We just need to handle the action here and record who clicked on it +def confirm_click(ack, body, client): + ack() + username = body["user"]["username"] + logging.info(f"User {username} viewed the calendar event.") + + def view_folder_metadata(client, body, ack): ack() folder_id = body["actions"][0]["value"] diff --git a/app/requirements.txt b/app/requirements.txt index 5fa49316..5d8ac209 100644 --- a/app/requirements.txt +++ b/app/requirements.txt @@ -9,6 +9,8 @@ geoip2==4.8.0 google-api-python-client==2.122.0 google-auth-httplib2==0.2.0 google-auth-oauthlib==0.8.0 +google-api-core==2.12.0 +google-auth==2.23.3 httpx==0.27.0 itsdangerous==2.1.2 Jinja2==3.1.3 diff --git a/app/tests/integrations/google_workspace/test_google_calendar.py b/app/tests/integrations/google_workspace/test_google_calendar.py new file mode 100644 index 00000000..737eb153 --- /dev/null +++ b/app/tests/integrations/google_workspace/test_google_calendar.py @@ -0,0 +1,319 @@ +"""Unit tests for google_calendar module.""" + +import json +from unittest.mock import patch, MagicMock +from datetime import datetime, timedelta +import pytest +import pytz +import integrations.google_workspace.google_calendar as google_calendar + +# Mocked dependencies +SRE_BOT_EMAIL = "sre-bot@cds-snc.ca" +SCOPES = ["https://www.googleapis.com/auth/calendar"] + + +# Fixture to mock the event details JSON string +@pytest.fixture +def event_details(): + return json.dumps( + { + "emails": ["user1@example.com", "user2@example.com"], + "topic": "Incident Response Meeting", + } + ) + + +# Fixture to mock the Google service object +@pytest.fixture +def google_service_mock(): + service = MagicMock() + service.freebusy().query().execute.return_value = "Mocked FreeBusy Query Result" + return service + + +# Fixture to mock the calendar service object +@pytest.fixture +def calendar_service_mock(): + # Mock for the Google Calendar service object + service_mock = MagicMock() + + # Properly set the return value for the execute method to return the expected dictionary directly + service_mock.events.return_value.insert.return_value.execute.return_value = { + "htmlLink": "https://calendar.google.com/event_link" + } + + return service_mock + + +# Fixture to mock the timezone +@pytest.fixture +def est_timezone(): + return pytz.timezone("US/Eastern") + + +# Fixture to mock the datetime.now() function +@pytest.fixture +def mock_datetime_now(est_timezone): + """Fixture to mock datetime.now() to return a specific time in the EST timezone.""" + # Mocking the specific date we want to consider as "now" + specific_now = datetime( + 2023, 4, 10, 10, 0, tzinfo=est_timezone + ) # Assuming April 10, 2023, is a Monday + + with patch( + "integrations.google_workspace.google_calendar.datetime" + ) as mock_datetime: + mock_datetime.now.return_value = specific_now + mock_datetime.strptime.side_effect = lambda *args, **kw: datetime.strptime( + *args, **kw + ) + yield mock_datetime + + +# Test out the schedule_event function is successful +@patch("integrations.google_workspace.google_calendar.get_google_service") +@patch("integrations.google_workspace.google_calendar.find_first_available_slot") +@patch("integrations.google_workspace.google_calendar.book_calendar_event") +def test_schedule_event_successful( + book_calendar_event_mock, + find_first_available_slot_mock, + get_google_service_mock, + event_details, + google_service_mock, +): + # Set up the mock return values + get_google_service_mock.return_value = google_service_mock + find_first_available_slot_mock.return_value = ( + datetime.utcnow().isoformat(), + (datetime.utcnow() + timedelta(hours=1)).isoformat(), + ) + book_calendar_event_mock.return_value = "https://calendar.link" + + # Call the function under test + event_link = google_calendar.schedule_event(event_details) + + # Assertions + get_google_service_mock.assert_called_once_with( + "calendar", "v3", delegated_user_email=SRE_BOT_EMAIL, scopes=SCOPES + ) + assert google_service_mock.freebusy().query().execute.call_count == 1 + find_first_available_slot_mock.assert_called_once() + book_calendar_event_mock.assert_called_once() + assert event_link == "https://calendar.link" + + +# Test out the schedule_event function when no available slots are found +@patch("integrations.google_workspace.google_calendar.get_google_service") +@patch("integrations.google_workspace.google_calendar.find_first_available_slot") +@patch("integrations.google_workspace.google_calendar.book_calendar_event") +def test_schedule_event_no_available_slots( + book_calendar_event_mock, + find_first_available_slot_mock, + get_google_service_mock, + event_details, + google_service_mock, +): + # Set up the mock return values + get_google_service_mock.return_value = google_service_mock + find_first_available_slot_mock.return_value = (None, None) + + # Call the function under test + event_link = google_calendar.schedule_event(event_details) + + # Assertions + assert event_link is None + book_calendar_event_mock.assert_not_called() + + +# Test out the book_calendar_event function is successful +def test_book_calendar_event_success(calendar_service_mock): + start = datetime.utcnow() + end = start + timedelta(hours=1) + emails = ["user1@example.com", "user2@example.com"] + incident_name = "Network Outage" + + event_link = google_calendar.book_calendar_event( + calendar_service_mock, start, end, emails, incident_name + ) + + assert event_link == "https://calendar.google.com/event_link" + calendar_service_mock.events().insert.assert_called_once() + assert ( + calendar_service_mock.events().insert.call_args[1]["body"]["summary"] + == "Retro Network Outage" + ) + assert calendar_service_mock.events().insert.call_args[1]["sendUpdates"] == "all" + + +# Test out the book_calendar_event function with empty emails +def test_book_calendar_event_with_empty_emails(calendar_service_mock): + start = datetime.utcnow() + end = start + timedelta(hours=1) + emails = [] + incident_name = "System Upgrade" + + event_link = google_calendar.book_calendar_event( + calendar_service_mock, start, end, emails, incident_name + ) + + assert event_link == "https://calendar.google.com/event_link" + assert ( + len(calendar_service_mock.events().insert.call_args[1]["body"]["attendees"]) + == 0 + ) + + +# Test out the book_calendar_event function with no conference data +def test_book_calendar_event_with_no_conference_data(calendar_service_mock): + start = datetime.utcnow() + end = start + timedelta(hours=1) + emails = ["user1@example.com"] + incident_name = "Database Migration" + + # Simulate behavior when conference data is not provided by the API response + calendar_service_mock.events().insert().execute.return_value = { + "htmlLink": "https://calendar.google.com/event_link", + "conferenceData": None, + } + + event_link = google_calendar.book_calendar_event( + calendar_service_mock, start, end, emails, incident_name + ) + + assert event_link == "https://calendar.google.com/event_link" + assert ( + "conferenceData" in calendar_service_mock.events().insert.call_args[1]["body"] + ) + + +# Test out the book_calendar_event function with no HTML link +def test_book_calendar_event_no_html_link(calendar_service_mock): + # Adjust the mock to not include 'htmlLink' in the response. + calendar_service_mock.events.return_value.insert.return_value.execute.return_value = ( + {} + ) + start = datetime.utcnow() + end = start + timedelta(hours=1) + emails = ["user@example.com"] + incident_name = "No Link Incident" + + event_link = google_calendar.book_calendar_event( + calendar_service_mock, start, end, emails, incident_name + ) + + # Assert that the function handles the missing 'htmlLink' gracefully. + assert event_link is None + + +# Test out the find_first_available_slot function on the first available weekday +def test_available_slot_on_first_weekday(mock_datetime_now, est_timezone): + freebusy_response = { + "calendars": { + "primary": { + "busy": [ + # Assuming the provided busy time is on April 10, 2023, from 1 PM to 1:30 PM UTC (9 AM to 9:30 AM EST) + { + "start": "2023-04-10T17:00:00Z", + "end": "2023-04-10T17:30:00Z", + } # Busy at 1 PM to 1:30 PM EST on April 10 + ] + } + } + } + + start, end = google_calendar.find_first_available_slot( + freebusy_response, duration_minutes=30, days_in_future=3, search_days_limit=60 + ) + + # Since April 10 is the "current" day and we start searching from 3 days in the future (April 13), + # we expect the function to find the first available slot on April 13 between 1 PM and 3 PM EST. + expected_start = datetime( + 2023, 4, 13, 13, 0, tzinfo=est_timezone + ) # Expected start time in EST + expected_end = datetime( + 2023, 4, 13, 13, 30, tzinfo=est_timezone + ) # Expected end time in EST + + assert ( + start == expected_start and end == expected_end + ), "The function should find the first available slot on April 13, 2023, from 1 PM to 1:30 PM EST." + + +# Test out the find_first_available_slot function when multiple busy days +def test_opening_exists_after_busy_days(mock_datetime_now, est_timezone): + freebusy_response = { + "calendars": { + "primary": { + "busy": [ + {"start": "2023-04-13T17:00:00Z", "end": "2023-04-13T19:00:00Z"}, + {"start": "2023-04-14T17:00:00Z", "end": "2023-04-14T19:00:00Z"}, + {"start": "2023-04-17T17:00:00Z", "end": "2023-04-17T19:00:00Z"}, + {"start": "2023-04-18T17:00:00Z", "end": "2023-04-18T19:00:00Z"}, + ] + } + } + } + + start, end = google_calendar.find_first_available_slot( + freebusy_response, duration_minutes=30, days_in_future=3, search_days_limit=60 + ) + + expected_start = datetime( + 2023, 4, 13, 14, 30, tzinfo=est_timezone + ) # Expected start time in EST + expected_end = datetime( + 2023, 4, 13, 15, 0, tzinfo=est_timezone + ) # Expected end time in EST + + assert ( + start == expected_start and end == expected_end + ), "Expected to find an available slot correctly." + + +# Test that weekends are skipped when searching for available slots +def test_skipping_weekends(mock_datetime_now, est_timezone): + freebusy_response = { + "calendars": { + "primary": { + "busy": [ + # Assuming weekdays are busy, but we expect the function to skip weekends automatically + ] + } + } + } + + # For this test, ensure the mocked 'now' falls before a weekend, and verify that the function skips to the next weekday + start, end = google_calendar.find_first_available_slot( + freebusy_response, duration_minutes=30, days_in_future=1, search_days_limit=60 + ) + + # Adjust these expected values based on the specific 'now' being mocked + expected_start = datetime(2023, 4, 11, 13, 0, tzinfo=est_timezone) + expected_end = datetime(2023, 4, 11, 13, 30, tzinfo=est_timezone) + + assert ( + start == expected_start and end == expected_end + ), "Expected to find an available slot after skipping the weekend" + + +# Test that no available slots are found within the search limit +def test_no_available_slots_within_search_limit(mock_datetime_now): + freebusy_response = { + "calendars": { + "primary": { + "busy": [ + # Simulate a scenario where every eligible day within the search window is fully booked + # For simplicity, let's assume a pattern that covers the search hours for the next 60 days + {"start": "2023-04-10T17:00:00Z", "end": "2023-08-13T19:00:00Z"}, + ] + } + } + } + + start, end = google_calendar.find_first_available_slot( + freebusy_response, duration_minutes=30, days_in_future=3, search_days_limit=60 + ) + + assert ( + start is None and end is None + ), "Expected no available slots within the search limit" diff --git a/app/tests/modules/incident/test_incident_helper.py b/app/tests/modules/incident/test_incident_helper.py index c1e9e0c2..f47af32e 100644 --- a/app/tests/modules/incident/test_incident_helper.py +++ b/app/tests/modules/incident/test_incident_helper.py @@ -1,11 +1,13 @@ import json - +import os from modules import incident_helper import logging from unittest.mock import ANY, MagicMock, patch +SLACK_SECURITY_USER_GROUP_ID = os.getenv("SLACK_SECURITY_USER_GROUP_ID") + def test_handle_incident_command_with_empty_args(): respond = MagicMock() @@ -771,3 +773,305 @@ def test_return_channel_name_empty_string(): def test_return_channel_name_prefix_only(): # Test the function with a string that is only the prefix. assert incident_helper.return_channel_name("incident-") == "#" + + +def test_schedule_incident_retro_successful_no_bots(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": ["U34333"]} + mock_client.conversations_members.return_value = {"members": ["U12345", "U67890"]} + mock_client.conversations_info.return_value = { + "channel": {"topic": {"value": "Retro Topic"}} + } + mock_client.users_info.side_effect = [ + {"user": {"profile": {"email": "user1@example.com"}}}, + {"user": {"profile": {"email": "user2@example.com"}}}, + ] + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + mock_ack.assert_called_once() + + # Verify the correct API calls were made + mock_client.usergroups_users_list.assert_called_once_with( + usergroup=SLACK_SECURITY_USER_GROUP_ID + ) + mock_client.conversations_members.assert_called_once_with(channel="C1234567890") + mock_client.conversations_info.assert_called_once_with(channel="C1234567890") + + # Check the users_info method was called correctly + calls = [call for call in mock_client.users_info.call_args_list] + assert ( + len(calls) == 2 + ) # Ensure we tried to fetch info for two users, one being a bot + + # Verify the modal payload contains the correct data + expected_data = json.dumps( + {"emails": ["user1@example.com", "user2@example.com"], "topic": "Retro Topic"} + ) + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +def test_schedule_incident_retro_successful_bots(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": ["U3333"]} + mock_client.conversations_members.return_value = { + "members": ["U12345", "U67890", "U54321"] + } + mock_client.conversations_info.return_value = { + "channel": {"topic": {"value": "Retro Topic"}} + } + mock_client.users_info.side_effect = [ + {"user": {"profile": {"email": "user1@example.com"}}}, + {"user": {"profile": {"email": "user2@example.com"}}}, + { + "user": {"profile": {"email": "user3@example.com", "bot_id": "B12345"}} + }, # This simulates a bot user + ] + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + mock_ack.assert_called_once() + + # Verify the correct API calls were made + mock_client.usergroups_users_list.assert_called_once_with( + usergroup=SLACK_SECURITY_USER_GROUP_ID + ) + mock_client.conversations_members.assert_called_once_with(channel="C1234567890") + mock_client.conversations_info.assert_called_once_with(channel="C1234567890") + + # Check the users_info method was called correctly + calls = [call for call in mock_client.users_info.call_args_list] + assert ( + len(calls) == 3 + ) # Ensure we tried to fetch info for three users, one being a bot + + # Verify the modal payload contains the correct data + expected_data = json.dumps( + {"emails": ["user1@example.com", "user2@example.com"], "topic": "Retro Topic"} + ) + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +def test_schedule_incident_retro_successful_security_group(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": ["U12345", "U444444"]} + mock_client.conversations_members.return_value = { + "members": ["U12345", "U67890", "U54321"] + } + mock_client.conversations_info.return_value = { + "channel": {"topic": {"value": "Retro Topic"}} + } + mock_client.users_info.side_effect = [ + {"user": {"profile": {"email": "user2@example.com"}}}, + { + "user": {"profile": {"email": "user3@example.com", "bot_id": "B12345"}} + }, # This simulates a bot user + ] + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + mock_ack.assert_called_once() + + # Verify the correct API calls were made + mock_client.usergroups_users_list.assert_called_once_with( + usergroup=SLACK_SECURITY_USER_GROUP_ID + ) + mock_client.conversations_members.assert_called_once_with(channel="C1234567890") + mock_client.conversations_info.assert_called_once_with(channel="C1234567890") + + # Check the users_info method was called correctly + calls = [call for call in mock_client.users_info.call_args_list] + assert ( + len(calls) == 2 + ) # Ensure we tried to fetch info for two users, minus the user being in the security group + + # Verify the modal payload contains the correct data + expected_data = json.dumps( + {"emails": ["user2@example.com"], "topic": "Retro Topic"} + ) + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +def test_schedule_incident_retro_successful_no_security_group(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": []} + mock_client.conversations_members.return_value = { + "members": ["U12345", "U67890", "U54321"] + } + mock_client.conversations_info.return_value = { + "channel": {"topic": {"value": "Retro Topic"}} + } + mock_client.users_info.side_effect = [ + {"user": {"profile": {"email": "user1@example.com"}}}, + {"user": {"profile": {"email": "user2@example.com"}}}, + { + "user": {"profile": {"email": "user3@example.com", "bot_id": "B12345"}} + }, # This simulates a bot user + ] + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + mock_ack.assert_called_once() + + # Verify the correct API calls were made + mock_client.usergroups_users_list.assert_called_once_with( + usergroup=SLACK_SECURITY_USER_GROUP_ID + ) + mock_client.conversations_members.assert_called_once_with(channel="C1234567890") + mock_client.conversations_info.assert_called_once_with(channel="C1234567890") + + # Check the users_info method was called correctly + calls = [call for call in mock_client.users_info.call_args_list] + assert ( + len(calls) == 3 + ) # Ensure we tried to fetch info for two users, minus the user being in the security group + + # Verify the modal payload contains the correct data + expected_data = json.dumps( + {"emails": ["user1@example.com", "user2@example.com"], "topic": "Retro Topic"} + ) + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +def test_schedule_incident_retro_with_no_users(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": ["U444444"]} + mock_client.conversations_info.return_value = { + "channel": {"topic": {"value": "Retro Topic"}} + } + mock_client.users_info.side_effect = [] + + # Adjust the mock to simulate no users in the channel + mock_client.conversations_members.return_value = {"members": []} + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + # construct the expected data object + expected_data = json.dumps({"emails": [], "topic": "Retro Topic"}) + # Assertions to validate behavior when no users are present in the channel + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +def test_schedule_incident_retro_with_no_topic(): + mock_client = MagicMock() + mock_ack = MagicMock() + mock_client.usergroups_users_list.return_value = {"users": ["U444444"]} + mock_client.conversations_info.return_value = {"channel": {"topic": {"value": ""}}} + mock_client.users_info.side_effect = [] + + # Adjust the mock to simulate no users in the channel + mock_client.conversations_members.return_value = {"members": []} + + body = { + "channel_id": "C1234567890", + "trigger_id": "T1234567890", + "channel_name": "incident-2024-01-12-test", + "user_id": "U12345", + } + + incident_helper.schedule_incident_retro(mock_client, body, mock_ack) + + # construct the expected data object and set the topic to a default one + expected_data = json.dumps({"emails": [], "topic": "Incident Retro"}) + # Assertions to validate behavior when no users are present in the channel + assert ( + mock_client.views_open.call_args[1]["view"]["private_metadata"] == expected_data + ) + + +@patch("integrations.google_workspace.google_calendar.schedule_event") +def test_save_incident_retro_success(schedule_event_mock): + mock_client = MagicMock() + mock_ack = MagicMock() + schedule_event_mock.return_value = "http://example.com/event" + body_mock = {"trigger_id": "some_trigger_id"} + view_mock_with_link = {"private_metadata": "event details for scheduling"} + + # Call the function + incident_helper.save_incident_retro( + mock_client, mock_ack, body_mock, view_mock_with_link + ) + + # Assertions + mock_ack.assert_called_once() # Ensure ack() was called + mock_client.views_open.assert_called_once() # Ensure the modal was opened + + # Verify the modal content for success + assert ( + mock_client.views_open.call_args[1]["view"]["blocks"][0]["text"]["text"] + == "*Successfully schduled calender event!*" + ) + + +@patch("integrations.google_workspace.google_calendar.schedule_event") +def test_save_incident_retro_failure(schedule_event_mock): + mock_client = MagicMock() + mock_ack = MagicMock() + schedule_event_mock.return_value = None + body_mock = {"trigger_id": "some_trigger_id"} + view_mock_with_link = {"private_metadata": "event details for scheduling"} + + # Call the function + incident_helper.save_incident_retro( + mock_client, mock_ack, body_mock, view_mock_with_link + ) + + # Assertions + mock_ack.assert_called_once() # Ensure ack() was called + mock_client.views_open.assert_called_once() # Ensure the modal was opened + + # Verify the modal content for success + assert ( + mock_client.views_open.call_args[1]["view"]["blocks"][0]["text"]["text"] + == "*Could not schedule event - no free time was found!*" + )