Update molnix_status when run sync_molnix command.

api/management/commands/sync_molnix.py

@@ -10,7 +10,12 @@
 from api.molnix_utils import MolnixApi
 from deployments.models import MolnixTag, MolnixTagGroup, Personnel, PersonnelDeployment
 from main.sentry import SentryMonitor
-from notifications.models import SurgeAlert, SurgeAlertCategory, SurgeAlertType
+from notifications.models import (
+    SurgeAlert,
+    SurgeAlertCategory,
+    SurgeAlertStatus,
+    SurgeAlertType,
+)
 
 CRON_NAME = "sync_molnix"
 
@@ -41,6 +46,23 @@
 }
 
 
+def get_molnix_status(position_status):
+    """
+    A position_status of active should be shown as Open
+    A position_status of archived should be shown as Closed
+    A position_status of unfilled should be shown as Stood Down
+    If the position_status is non other than active, archived, unfilled then show Closed.
+    """
+    molnix_status_dict = {
+        "active": SurgeAlertStatus.OPEN,
+        "unfilled": SurgeAlertStatus.STOOD_DOWN,
+        "archived": SurgeAlertStatus.CLOSED,
+    }
+    if molnix_status_dict.get(position_status, None) is not None:
+        return molnix_status_dict.get(position_status)
+    return SurgeAlertStatus.CLOSED
+
+
 def prt(message_text, molnix_id, position_or_event_id=0, organization=""):
     warning_type = 0
     if message_text == "Position does not have a valid Emergency tag":
@@ -470,14 +492,14 @@ def sync_open_positions(molnix_positions, molnix_api, countries):
         # print(json.dumps(position, indent=2))
         go_alert.molnix_id = position["id"]
         go_alert.message = position["name"]
-        go_alert.molnix_status = position["status"]
+        go_alert.molnix_status = get_molnix_status(position["status"])
         go_alert.event = event
         go_alert.country = country
         go_alert.opens = get_datetime(position["opens"])
         go_alert.closes = get_datetime(position["closes"])
         go_alert.start = get_datetime(position["start"])
         go_alert.end = get_datetime(position["end"])
-        go_alert.is_active = position["status"] == "active"
+        go_alert.is_active = get_molnix_status(position["status"]) == SurgeAlertStatus.OPEN
         go_alert.save()
         add_tags_to_obj(go_alert, position["tags"])
         if created:
@@ -499,11 +521,11 @@ def sync_open_positions(molnix_positions, molnix_api, countries):
         if not position:
             warnings.append("Position id %d not found in Molnix API" % alert.molnix_id)
         if position and position["status"] == "unfilled":
-            alert.molnix_status = position["status"]
+            alert.molnix_status = SurgeAlertStatus.STOOD_DOWN
         if position and position["closes"]:
             alert.closes = get_datetime(position["closes"])
         if position and position["status"] == "archived":
-            alert.molnix_status = position["status"]
+            alert.molnix_status = SurgeAlertStatus.CLOSED
             alert.is_active = False
         else:
             alert.is_active = False

notifications/admin.py

@@ -23,8 +23,8 @@ def std(self, obj):
         "event__name",
     )
     readonly_fields = ("molnix_id", "is_stood_down")
-    list_display = ("__str__", "message", "start", "molnix_id", "molnix_status", "status", "std")
-    list_filter = ("molnix_status", "status", "is_stood_down")
+    list_display = ("__str__", "message", "start", "molnix_id", "molnix_status", "std")
+    list_filter = ("molnix_status", "is_stood_down")
 
 
 class SubscriptionAdmin(CompareVersionAdmin):

notifications/drf_views.py

@@ -62,7 +62,7 @@ class SurgeAlertViewset(viewsets.ReadOnlyModelViewSet):
     authentication_classes = (TokenAuthentication,)
     queryset = SurgeAlert.objects.prefetch_related("molnix_tags", "molnix_tags__groups").select_related("event", "country").all()
     filterset_class = SurgeAlertFilter
-    ordering_fields = ("created_at", "atype", "category", "event", "is_stood_down", "status", "opens")
+    ordering_fields = ("created_at", "atype", "category", "event", "is_stood_down", "molnix_status", "opens")
     search_fields = (
         "operation",
         "message",

notifications/factories.py

@@ -1,7 +1,7 @@
 import factory
 from factory import fuzzy
 
-from .models import SurgeAlert
+from .models import SurgeAlert, SurgeAlertStatus
 
 
 class SurgeAlertFactory(factory.django.DjangoModelFactory):
@@ -11,7 +11,7 @@ class Meta:
     message = fuzzy.FuzzyText(length=100)
     atype = fuzzy.FuzzyInteger(low=1)
     category = fuzzy.FuzzyInteger(low=1)
-    molnix_status = fuzzy.FuzzyChoice(choices=["active", "inactive"])
+    molnix_status = fuzzy.FuzzyChoice(choices=SurgeAlertStatus)
 
     @factory.post_generation
     def molnix_tags(self, create, extracted, **_):

notifications/migrations/0015_rename_molnix_status_surgealert_molnix_status_old.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.2.13 on 2024-07-16 08:46
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("notifications", "0014_surgealert_status"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="surgealert",
+            name="molnix_status",
+        ),
+        migrations.RenameField(
+            model_name="surgealert",
+            old_name="status",
+            new_name="molnix_status",
+        ),
+    ]

notifications/models.py

@@ -49,9 +49,6 @@ class SurgeAlert(models.Model):
     # ID in Molnix system, if parsed from Molnix.
     molnix_id = models.IntegerField(blank=True, null=True)
 
-    # Status field from Molnix - `unfilled` denotes Stood-Down
-    molnix_status = models.CharField(blank=True, null=True, max_length=32)
-
     # It depends on molnix_status. Check "save" method below.
     is_stood_down = models.BooleanField(verbose_name=_("is stood down?"), default=False)
     opens = models.DateTimeField(blank=True, null=True)
@@ -65,27 +62,18 @@ 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=SurgeAlertStatus.OPEN)
+    molnix_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):
-        """
-        A molnix_status of active should be shown as Open
-        A molnix_status of archived should be shown as Closed
-        A molnix_status of unfilled should be shown as Stood Down
-        """
         if (not self.id and not self.created_at) or (self.created_at > timezone.now()):
             self.created_at = timezone.now()
-        if self.molnix_status == "active":
-            self.status = SurgeAlertStatus.OPEN
-        elif self.molnix_status == "unfilled":
-            self.status = SurgeAlertStatus.STOOD_DOWN
-        else:
-            self.status = SurgeAlertStatus.CLOSED
         return super(SurgeAlert, self).save(*args, **kwargs)
 
     def __str__(self):

notifications/serializers.py

@@ -15,7 +15,7 @@ class SurgeAlertSerializer(ModelSerializer):
     event = SurgeEventSerializer()
     country = MiniCountrySerializer()
     atype_display = serializers.CharField(source="get_atype_display", read_only=True)
-    status_display = serializers.CharField(source="get_status_display", read_only=True)
+    molnix_status_display = serializers.CharField(source="get_molnix_status_display", read_only=True)
     category_display = serializers.CharField(source="get_category_display", read_only=True)
     molnix_tags = MolnixTagSerializer(many=True, read_only=True)
 
@@ -43,8 +43,8 @@ class Meta:
             "end",
             "is_active",
             "is_stood_down",
-            "status",
-            "status_display",
+            "molnix_status",
+            "molnix_status_display",
         )
 
 

notifications/tests.py

@@ -1,8 +1,6 @@
 from datetime import datetime, timedelta
-from unittest.mock import patch
 
 from django.conf import settings
-from django.core.management import call_command
 from django.utils import timezone
 from modeltranslation.utils import build_localized_fieldname
 
@@ -158,116 +156,42 @@ def test_surge_alert_status(self):
         molnix_tag_2 = MolnixTagFactory.create(name="L-FRA")
         molnix_tag_3 = MolnixTagFactory.create(name="AMER")
 
-        alert1 = SurgeAlertFactory.create(
+        SurgeAlertFactory.create(
             message="CEA Coordinator, Floods, Atlantis",
             country=country_1,
+            molnix_status=SurgeAlertStatus.OPEN,
             molnix_tags=[molnix_tag_1, molnix_tag_2],
-            molnix_status="active",
             opens=timezone.now() - timedelta(days=2),
             closes=timezone.now() + timedelta(days=5),
         )
-        alert2 = SurgeAlertFactory.create(
+        SurgeAlertFactory.create(
             message="WASH Coordinator, Earthquake, Neptunus",
             country=country_2,
-            molnix_status="archived",
+            molnix_status=SurgeAlertStatus.CLOSED,
             molnix_tags=[molnix_tag_1, molnix_tag_3],
             opens=timezone.now() - timedelta(days=2),
             closes=timezone.now() - timedelta(days=1),
         )
-        alert3 = SurgeAlertFactory.create(
+        SurgeAlertFactory.create(
             message="New One",
             country=country_2,
-            molnix_status="unfilled",
+            molnix_status=SurgeAlertStatus.STOOD_DOWN,
             molnix_tags=[molnix_tag_1, molnix_tag_3],
         )
 
-        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}))
+        response = _fetch(dict({"molnix_status": SurgeAlertStatus.OPEN}))
+        print("RES", response)
 
         self.assertEqual(response["count"], 1)
-        self.assertEqual(response["results"][0]["status"], SurgeAlertStatus.OPEN)
+        self.assertEqual(response["results"][0]["molnix_status"], SurgeAlertStatus.OPEN)
 
-        response = _fetch(dict({"status": SurgeAlertStatus.CLOSED}))
+        response = _fetch(dict({"molnix_status": SurgeAlertStatus.CLOSED}))
         self.assertEqual(response["count"], 1)
-        self.assertEqual(response["results"][0]["status"], SurgeAlertStatus.CLOSED)
+        self.assertEqual(response["results"][0]["molnix_status"], SurgeAlertStatus.CLOSED)
 
-        response = _fetch(dict({"status": SurgeAlertStatus.STOOD_DOWN}))
+        response = _fetch(dict({"molnix_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)
-        country_1 = CountryFactory.create(iso3="NPP", region=region_1)
-        country_2 = CountryFactory.create(iso3="CTT", 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")
-
-        # Override the original save method to create dummy data without restriction
-        original_save = SurgeAlert.save
-
-        def mocked_save(self, *args, **kwargs):
-            original_save(self, *args, **kwargs)
-
-        SurgeAlert.save = mocked_save
-
-        alert1 = SurgeAlertFactory.create(
-            message="CEA Coordinator, Floods, Atlantis",
-            country=country_1,
-            molnix_status="unfilled",
-            molnix_tags=[molnix_tag_1, molnix_tag_2],
-            opens=timezone.now() - timedelta(days=2),
-            closes=timezone.now() + timedelta(seconds=5),
-            status=SurgeAlertStatus.CLOSED,
-        )
-
-        alert2 = SurgeAlertFactory.create(
-            message="WASH Coordinator, Earthquake, Neptunus",
-            country=country_2,
-            molnix_status="unfilled",
-            molnix_tags=[molnix_tag_1, molnix_tag_3],
-            opens=timezone.now() - timedelta(days=2),
-            closes=timezone.now() - timedelta(days=1),
-            status=SurgeAlertStatus.STOOD_DOWN,
-        )
-
-        alert3 = SurgeAlertFactory.create(
-            message="WASH Coordinator, Earthquake, Neptunus",
-            country=country_2,
-            molnix_status="archived",
-            molnix_tags=[molnix_tag_1, molnix_tag_2],
-            opens=timezone.now() - timedelta(days=4),
-            closes=timezone.now() - timedelta(days=1),
-            status=SurgeAlertStatus.OPEN,
-        )
-
-        alert4 = SurgeAlertFactory.create(
-            message="WASH Coordinator, Earthquake, Neptunus",
-            country=country_2,
-            molnix_status="active",
-            molnix_tags=[molnix_tag_1, molnix_tag_2, molnix_tag_3],
-            opens=timezone.now() - timedelta(days=3),
-            closes=timezone.now() - timedelta(days=1),
-            status=SurgeAlertStatus.OPEN,
-        )
-
-        # Restore the original save method after the test
-        SurgeAlert.save = original_save
-
-        with patch("django.utils.timezone.now") as mock_now:
-            mock_now.return_value = datetime.now() + timedelta(days=1)
-            call_command("update_surge_alert_status")
-
-        self.assertEqual(alert1.status, SurgeAlertStatus.STOOD_DOWN)
-        self.assertEqual(alert2.status, SurgeAlertStatus.STOOD_DOWN)
-        self.assertEqual(alert3.status, SurgeAlertStatus.CLOSED)
-        self.assertEqual(alert4.status, SurgeAlertStatus.OPEN)
+        self.assertEqual(response["results"][0]["molnix_status"], SurgeAlertStatus.STOOD_DOWN)