tests: relocate all notification plugin tests

gapitio · Jun 11, 2024 · 2197b79 · 2197b79
1 parent fcc1360
commit 2197b79
Show file tree

Hide file tree

Showing 3 changed files with 218 additions and 150 deletions.
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -64,8 +64,7 @@ jobs:
           SENDGRID_SENDER: ${{ secrets.SENDGRID_SENDER }}
           SENDGRID_RECEIVER: ${{ secrets.SENDGRID_RECEIVER }}
         run: |
-          pytest --cov=alerta tests/test_notification_rule.py::NotificationRuleTestCase::test_twilio_sms_channel \
-            tests/test_notification_rule.py::NotificationRuleTestCase::test_sendgrid_channel
+          pytest --cov=alerta tests/test_notification_rule_plugin.py
       - uses: act10ns/slack@v2
         with:
           status: ${{ job.status }}
@@ -127,8 +126,7 @@ jobs:
           SENDGRID_SENDER: ${{ secrets.SENDGRID_SENDER }}
           SENDGRID_RECEIVER: ${{ secrets.SENDGRID_RECEIVER }}
         run: |
-          pytest --cov=alerta tests/test_notification_rule.py::NotificationRuleTestCase::test_twilio_sms_channel \
-            tests/test_notification_rule.py::NotificationRuleTestCase::test_sendgrid_channel
+          pytest --cov=alerta tests/test_notification_rule_plugin.py
       - uses: act10ns/slack@v2
         with:
           status: ${{ job.status }}

diff --git a/tests/test_notification_rule.py b/tests/test_notification_rule.py
@@ -1,6 +1,5 @@
 import json
 import logging
-import os
 import unittest
 from datetime import datetime, timedelta
 
@@ -15,21 +14,13 @@ def get_id(object: dict):
     return object['id']
 
 
-def get_delay_id(object: dict):
-    return {'rule_id': object['notification_rule_id'], 'alert_id': object['alert_id']}
-
-
-def get_history_id(object: dict):
-    return {'rule_id': object['rule'], 'alert_id': object['alert']}
-
-
 class NotificationRuleTestCase(unittest.TestCase):
     def setUp(self) -> None:
         test_config = {
             'TESTING': True,
             'AUTH_REQUIRED': True,
             'CUSTOMER_VIEWS': True,
-            'PLUGINS': ['notification_rule'],
+            'PLUGINS': [],
         }
         self.app = create_app(test_config)
         self.client = self.app.test_client()
@@ -1011,139 +1002,4 @@ def test_advanced_severity(self):
         self.assertIn(notification_rule, active_notification_rules)
         self.assertIn(to_critical_major_from_all_rule, active_notification_rules)
         self.assertNotIn(to_normal_from_critical_major_rule, active_notification_rules)
-        self.assertIn(simple_rule, active_notification_rules)
-
-    def test_delayed_notifications(self):
-        notification_rule = {
-            'environment': 'Production',
-            'channelId': 'SMS_Channel',
-            'service': ['Core'],
-            'receivers': [],
-        }
-
-        delayed_notification_rule = {
-            'environment': 'Production',
-            'channelId': 'SMS_Channel',
-            'service': ['Core'],
-            'receivers': [],
-            'delayTime': '1 second'
-        }
-
-        self.channel_id = self.create_api_obj('/notificationchannels', self.sms_channel, self.headers)['id']
-        data = self.create_api_obj('/notificationrules', notification_rule, self.headers)
-        notification_rule_id = data['id']
-
-        delayed_notification_rule = self.create_api_obj('/notificationrules', delayed_notification_rule, self.headers)
-        delayed_notification_rule_id = delayed_notification_rule['id']
-        self.assertEqual(delayed_notification_rule['notificationRule']['delayTime'], 1)
-
-        # new alert should activate notification_rule
-        data = self.create_api_obj('/alert', self.prod_alert, self.headers)
-        alert_id = data['id']
-        start = datetime.now()
-        active_notification_rules = self.create_api_obj('/notificationrules/active', data['alert'], self.headers, 200)['notificationRules']
-        self.assertIn(
-            notification_rule_id,
-            map(get_id, active_notification_rules),
-        )
-        delayed_rules = self.get_api_obj('/notificationdelay', self.headers)['notificationDelays']
-        self.assertNotEqual(delayed_rules, [])
-        self.assertIn({'rule_id': delayed_notification_rule_id, 'alert_id': alert_id}, map(get_delay_id, delayed_rules))
-        delayed_data = self.get_api_obj('/notificationdelay/fire', self.headers)
-        self.assertEqual(delayed_data['notifications'], [])
-        while len(delayed_data['notifications']) == 0:
-            delayed_data = self.get_api_obj('/notificationdelay/fire', self.headers)
-        self.assertTrue(datetime.now() - start >= timedelta(seconds=1))
-        self.assertIn({'rule_id': delayed_notification_rule_id, 'alert_id': alert_id}, map(get_delay_id, delayed_data['notifications']))
-
-    def test_twilio_sms_channel(self):
-        try:
-            twilio_config = {
-                'token': os.environ['TWILIO_TOKEN'],
-                'sid': os.environ['TWILIO_SID'],
-                'sender': os.environ['TWILIO_SENDER'],
-                'receiver': os.environ['TWILIO_RECEIVER']
-            }
-        except KeyError:
-            self.skipTest('Missing required twilio environment')
-        notification_rule = {
-            'environment': 'Production',
-            'channelId': 'sms',
-            'service': ['Core'],
-            'receivers': [twilio_config['receiver']],
-        }
-
-        channel = {
-            'id': 'sms',
-            'sender': twilio_config['sender'],
-            'type': 'twilio_sms',
-            'apiToken': twilio_config['token'],
-            'apiSid': twilio_config['sid']
-        }
-
-        self.create_api_obj('/notificationchannels', channel, self.headers)
-        data = self.create_api_obj('/notificationrules', notification_rule, self.headers)
-        notification_rule_id = data['id']
-
-        data = self.create_api_obj('/alert', self.prod_alert, self.headers)
-        alert_id = data['id']
-        active_notification_rules = self.create_api_obj('/notificationrules/active', data['alert'], self.headers, 200)['notificationRules']
-        self.assertIn(
-            notification_rule_id,
-            map(get_id, active_notification_rules),
-        )
-        history = self.get_api_obj('notificationhistory', self.headers)['notificationHistory']
-
-        while len(history) == 0:
-            history = self.get_api_obj('notificationhistory', self.headers)['notificationHistory']
-
-        self.assertIn(
-            {'rule_id': notification_rule_id, 'alert_id': alert_id},
-            map(get_history_id, history)
-        )
-        self.assertTrue(history[0]['sent'])
-
-    def test_sendgrid_channel(self):
-        try:
-            sendgrid_config = {
-                'token': os.environ['SENDGRID_TOKEN'],
-                'sender': os.environ['SENDGRID_SENDER'],
-                'receiver': os.environ['SENDGRID_RECEIVER']
-            }
-        except KeyError:
-            self.skipTest('Missing required twilio environment')
-        notification_rule = {
-            'environment': 'Production',
-            'channelId': 'sms',
-            'service': ['Core'],
-            'receivers': [sendgrid_config['receiver']],
-        }
-
-        channel = {
-            'id': 'sms',
-            'sender': sendgrid_config['sender'],
-            'type': 'sendgrid',
-            'apiToken': sendgrid_config['token'],
-        }
-
-        self.create_api_obj('/notificationchannels', channel, self.headers)
-        data = self.create_api_obj('/notificationrules', notification_rule, self.headers)
-        notification_rule_id = data['id']
-
-        data = self.create_api_obj('/alert', self.prod_alert, self.headers)
-        alert_id = data['id']
-        active_notification_rules = self.create_api_obj('/notificationrules/active', data['alert'], self.headers, 200)['notificationRules']
-        self.assertIn(
-            notification_rule_id,
-            map(get_id, active_notification_rules),
-        )
-        history = self.get_api_obj('notificationhistory', self.headers)['notificationHistory']
-
-        while len(history) == 0:
-            history = self.get_api_obj('notificationhistory', self.headers)['notificationHistory']
-
-        self.assertIn(
-            {'rule_id': notification_rule_id, 'alert_id': alert_id},
-            map(get_history_id, history)
-        )
-        self.assertTrue(history[0]['sent'])
+        self.assertIn(simple_rule, active_notification_rules)