Update test case

- Mock timezone.now

Author: susilnem

deploy/helm/ifrcgo-helm/values.yaml

@@ -111,8 +111,8 @@ cronjobs:
     schedule: '0 9 * * *'
   - command: 'ingest_country_plan_file'
     schedule: '1 0 * * *'
-  - command: 'update_alert_status'
-    schedule: '0 */12 * * *'
+  - command: 'update_surge_alert_status'
+    schedule: '1 */12 * * *'
 
 elasticsearch:
   enabled: true

notifications/management/commands/update_alert_status.py renamed to notifications/management/commands/update_surge_alert_status.py

@@ -28,4 +28,4 @@ def handle(self, *args, **options):
                 )
             )
         except Exception as e:
-            logger.error('Error while updating alerts status: %s' % str(e))
+            logger.error('Error while updating alerts status', exc_info=True)

notifications/migrations/0014_surgealert_status.py

@@ -1,4 +1,4 @@
-# Generated by Django 3.2.23 on 2024-02-13 05:20
+# Generated by Django 3.2.23 on 2024-02-13 09:43
 
 from django.db import migrations, models
 

notifications/models.py

@@ -23,11 +23,16 @@ class SurgeAlertCategory(models.IntegerChoices):
     SHELTER = 3, _('shelter')
     STAND_DOWN = 4, _('stand down')
 
+
 class SurgeAlertStatus(models.IntegerChoices):
+    '''
+    Note: Ordering value should be in order of Open, Stood Down, Closed to supported custom defined ordering logic
+    '''
     OPEN = 0, _('Open')
     STOOD_DOWN = 1, _('Stood Down')
     CLOSED = 2, _('Closed')
 
+
 class SurgeAlert(models.Model):
 
     atype = models.IntegerField(choices=SurgeAlertType.choices, verbose_name=_('alert type'), default=0)
@@ -58,19 +63,23 @@ class SurgeAlert(models.Model):
 
     # Don't set `auto_now_add` so we can modify it on save
     created_at = models.DateTimeField(verbose_name=_('created at'))
-    status = models.IntegerField(choices=SurgeAlertStatus.choices, verbose_name=_('alert status'), default=0)
+    status = models.IntegerField(choices=SurgeAlertStatus.choices, verbose_name=_('alert status'), default=SurgeAlertStatus.OPEN)
 
     class Meta:
         ordering = ['-created_at']
         verbose_name = _('Surge Alert')
         verbose_name_plural = _('Surge Alerts')
 
     def save(self, *args, **kwargs):
+        """
+        If the alert status is marked as stood_down, then the status is Stood Down.
+        If the closing timestamp (closes) is earlier than the current date, the status is displayed as Closed.
+        Otherwise, it is displayed as Open.
+        """
         # On save, if `created` is not set, make it the current time
         if (not self.id and not self.created_at) or (self.created_at > timezone.now()):
             self.created_at = timezone.now()
         self.is_stood_down = self.molnix_status == 'unfilled'
-        
         if self.is_stood_down:
             self.status = SurgeAlertStatus.STOOD_DOWN
         elif self.closes and self.closes < timezone.now():

notifications/tests.py

@@ -1,6 +1,6 @@
-import time
 from datetime import datetime, timedelta
 from django.utils import timezone
+from unittest.mock import patch
 from django.conf import settings
 from modeltranslation.utils import build_localized_fieldname
 
@@ -140,6 +140,63 @@ def _to_csv(*items):
         ))
         self.assertEqual(response['count'], 2)
 
+    def test_surge_alert_status(self):
+        region_1, region_2 = RegionFactory.create_batch(2)
+        country_1 = CountryFactory.create(iso3='ATL', region=region_1)
+        country_2 = CountryFactory.create(iso3='NPT', region=region_2)
+
+        molnix_tag_1 = MolnixTagFactory.create(name='OP-6700')
+        molnix_tag_2 = MolnixTagFactory.create(name='L-FRA')
+        molnix_tag_3 = MolnixTagFactory.create(name='AMER')
+
+        alert1 = SurgeAlertFactory.create(
+            message='CEA Coordinator, Floods, Atlantis',
+            country=country_1,
+            molnix_tags=[molnix_tag_1, molnix_tag_2],
+            opens=timezone.now() - timedelta(days=2),
+            closes=timezone.now() + timedelta(days=5)
+        )
+        alert2 = SurgeAlertFactory.create(
+            message='WASH Coordinator, Earthquake, Neptunus',
+            country=country_2,
+            molnix_tags=[molnix_tag_1, molnix_tag_3],
+            opens=timezone.now() - timedelta(days=2),
+            closes=timezone.now() - timedelta(days=1)
+        )
+        alert3 = SurgeAlertFactory.create(
+            message='New One',
+            country=country_2,
+            molnix_tags=[molnix_tag_1, molnix_tag_3],
+            molnix_status='unfilled',
+        )
+
+        self.assertEqual(alert1.status, SurgeAlertStatus.OPEN)
+        self.assertEqual(alert2.status, SurgeAlertStatus.CLOSED)
+        self.assertEqual(alert3.status, SurgeAlertStatus.STOOD_DOWN)
+
+        def _fetch(filters):
+            return self.client.get('/api/v2/surge_alert/', filters).json()
+
+        response = _fetch(dict({
+            'status': SurgeAlertStatus.OPEN
+        }))
+
+        self.assertEqual(response['count'], 1)
+        self.assertEqual(response['results'][0]['status'], SurgeAlertStatus.OPEN)
+
+        response = _fetch(dict({
+            'status': SurgeAlertStatus.CLOSED
+        }))
+        self.assertEqual(response['count'], 1)
+        self.assertEqual(response['results'][0]['status'], SurgeAlertStatus.CLOSED)
+
+        response = _fetch(dict({
+            'status': SurgeAlertStatus.STOOD_DOWN
+        }))
+        self.assertEqual(response['count'], 1)
+        self.assertEqual(response['results'][0]['status'], SurgeAlertStatus.STOOD_DOWN)
+
+
 class SurgeAlertTestCase(APITestCase):
     def test_update_alert_status_command(self):
         region_1, region_2 = RegionFactory.create_batch(2)
@@ -154,25 +211,25 @@ def test_update_alert_status_command(self):
             message='CEA Coordinator, Floods, Atlantis',
             country=country_1,
             molnix_tags=[molnix_tag_1, molnix_tag_2],
-            opens = timezone.now() - timedelta(days=2),
-            closes = timezone.now() + timedelta(seconds=5)
+            opens=timezone.now() - timedelta(days=2),
+            closes=timezone.now() + timedelta(seconds=5)
         )
         alert2 = SurgeAlertFactory.create(
             message='WASH Coordinator, Earthquake, Neptunus',
             country=country_2,
             molnix_tags=[molnix_tag_1, molnix_tag_3],
-            opens = timezone.now() - timedelta(days=2),
-            closes = timezone.now() - timedelta(days=1)
-
+            opens=timezone.now() - timedelta(days=2),
+            closes=timezone.now() - timedelta(days=1)
         )
         self.assertEqual(alert1.status, SurgeAlertStatus.OPEN)
         self.assertEqual(alert2.status, SurgeAlertStatus.CLOSED)
 
-        time.sleep(10)
-        call_command('update_alert_status')
+        with patch('django.utils.timezone.now') as mock_now:
+            mock_now.return_value = datetime.now() + timedelta(days=1)
+            call_command('update_surge_alert_status')
 
         alert1.refresh_from_db()
         alert2.refresh_from_db()
 
         self.assertEqual(alert1.status, SurgeAlertStatus.CLOSED)
-        self.assertEqual(alert2.status, SurgeAlertStatus.CLOSED)
+        self.assertEqual(alert2.status, SurgeAlertStatus.CLOSED)