diff --git a/tools/migrators/README.md b/tools/migrators/README.md index 460873ff..61881ebf 100644 --- a/tools/migrators/README.md +++ b/tools/migrators/README.md @@ -172,8 +172,9 @@ Configuration is done via environment variables passed to the docker container. The tool is capable of migrating user notification rules from PagerDuty to Grafana OnCall. Notification rules from the `"When a high-urgency incident is assigned to me..."` section in PagerDuty settings are -taken into account and will be migrated to default notification rules in Grafana OnCall for each user. Note that delays -between notification rules may be slightly different in Grafana OnCall, see [Limitations](#limitations) for more info. +taken into account and will be migrated to both default and important notification rules in Grafana OnCall +for each user. Note that delays between notification rules may be slightly different in Grafana OnCall, +see [Limitations](#limitations) for more info. When running the migration, existing notification rules in Grafana OnCall will be deleted for every affected user. diff --git a/tools/migrators/lib/pagerduty/resources/notification_rules.py b/tools/migrators/lib/pagerduty/resources/notification_rules.py index 58a49dd0..7f712d7c 100644 --- a/tools/migrators/lib/pagerduty/resources/notification_rules.py +++ b/tools/migrators/lib/pagerduty/resources/notification_rules.py @@ -27,21 +27,25 @@ def migrate_notification_rules(user: dict) -> None: rule for rule in user["notification_rules"] if rule["urgency"] == "high" ] - oncall_rules = transform_notification_rules( - notification_rules, user["oncall_user"]["id"] - ) + for important in (False, True): + oncall_rules = transform_notification_rules( + notification_rules, user["oncall_user"]["id"], important + ) - for rule in oncall_rules: - OnCallAPIClient.create("personal_notification_rules", rule) + for rule in oncall_rules: + OnCallAPIClient.create("personal_notification_rules", rule) - if oncall_rules: - # delete old notification rules if any new rules were created - for rule in user["oncall_user"]["notification_rules"]: - OnCallAPIClient.delete("personal_notification_rules/{}".format(rule["id"])) + if oncall_rules: + # delete old notification rules if any new rules were created + for rule in user["oncall_user"]["notification_rules"]: + if rule["important"] == important: + OnCallAPIClient.delete( + "personal_notification_rules/{}".format(rule["id"]) + ) def transform_notification_rules( - notification_rules: list[dict], user_id: str + notification_rules: list[dict], user_id: str, important: bool ) -> list[dict]: """ Transform PagerDuty user notification rules to Grafana OnCall personal notification rules. @@ -58,7 +62,9 @@ def transform_notification_rules( previous_delay = notification_rules[idx - 1]["start_delay_in_minutes"] delay -= previous_delay - oncall_notification_rules += transform_notification_rule(rule, delay, user_id) + oncall_notification_rules += transform_notification_rule( + rule, delay, user_id, important + ) oncall_notification_rules = remove_duplicate_rules_between_waits( oncall_notification_rules @@ -68,12 +74,12 @@ def transform_notification_rules( def transform_notification_rule( - notification_rule: dict, delay: int, user_id: str + notification_rule: dict, delay: int, user_id: str, important: bool ) -> list[dict]: contact_method_type = notification_rule["contact_method"]["type"] oncall_type = PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP[contact_method_type] - notify_rule = {"user_id": user_id, "type": oncall_type, "important": False} + notify_rule = {"user_id": user_id, "type": oncall_type, "important": important} if not delay: return [notify_rule] @@ -82,6 +88,6 @@ def transform_notification_rule( "user_id": user_id, "type": "wait", "duration": transform_wait_delay(delay), - "important": "False", + "important": important, } return [wait_rule, notify_rule] diff --git a/tools/migrators/lib/tests/pagerduty/test_migrate_notification_rules.py b/tools/migrators/lib/tests/pagerduty/test_migrate_notification_rules.py new file mode 100644 index 00000000..566d1614 --- /dev/null +++ b/tools/migrators/lib/tests/pagerduty/test_migrate_notification_rules.py @@ -0,0 +1,85 @@ +from unittest.mock import call, patch + +from lib.oncall.api_client import OnCallAPIClient +from lib.pagerduty.resources.notification_rules import migrate_notification_rules + + +@patch.object(OnCallAPIClient, "delete") +@patch.object(OnCallAPIClient, "create") +def test_migrate_notification_rules(api_client_create_mock, api_client_delete_mock): + migrate_notification_rules( + { + "notification_rules": [ + { + "contact_method": {"type": "sms_contact_method"}, + "start_delay_in_minutes": 0, + "urgency": "high", + }, + { + "contact_method": {"type": "push_notification_contact_method"}, + "start_delay_in_minutes": 5, + "urgency": "high", + }, + ], + "oncall_user": { + "id": "EXISTING_USER_ID", + "notification_rules": [ + {"id": "EXISTING_RULE_ID_1", "important": False}, + {"id": "EXISTING_RULE_ID_2", "important": True}, + ], + }, + } + ) + + assert api_client_create_mock.call_args_list == [ + call( + "personal_notification_rules", + { + "user_id": "EXISTING_USER_ID", + "type": "notify_by_sms", + "important": False, + }, + ), + call( + "personal_notification_rules", + { + "user_id": "EXISTING_USER_ID", + "type": "wait", + "duration": 300, + "important": False, + }, + ), + call( + "personal_notification_rules", + { + "user_id": "EXISTING_USER_ID", + "type": "notify_by_mobile_app", + "important": False, + }, + ), + call( + "personal_notification_rules", + {"user_id": "EXISTING_USER_ID", "type": "notify_by_sms", "important": True}, + ), + call( + "personal_notification_rules", + { + "user_id": "EXISTING_USER_ID", + "type": "wait", + "duration": 300, + "important": True, + }, + ), + call( + "personal_notification_rules", + { + "user_id": "EXISTING_USER_ID", + "type": "notify_by_mobile_app", + "important": True, + }, + ), + ] + assert api_client_delete_mock.call_args_list == [ + call("personal_notification_rules/EXISTING_RULE_ID_1"), + call("personal_notification_rules/EXISTING_RULE_ID_2"), + ]