Add script to delete empty history entries from LOS, Links, and Devices

src/meshapi/management/commands/dedupe_history_entries.py

@@ -0,0 +1,57 @@
+import logging
+from argparse import ArgumentParser
+from typing import Any
+
+from django.core.management.base import BaseCommand
+from django.db import models, transaction
+
+from meshapi.models.devices.device import Device
+from meshapi.models.link import Link
+from meshapi.models.los import LOS
+
+
+class Command(BaseCommand):
+    help = "Deletes duplicate history entries on history tables"
+
+    def add_arguments(self, parser: ArgumentParser) -> None:
+        pass
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        self.deduplicate_history(Link.objects.all())
+        self.deduplicate_history(Device.objects.all())
+        self.deduplicate_history(LOS.objects.all())
+
+    def deduplicate_history(self, model_objects: models.QuerySet) -> None:
+        for m in model_objects:
+            logging.info(f"{m.id}, {m}")
+            # Delete history from each object. Atomic block makes it go faster
+            with transaction.atomic():
+                try:
+                    history = m.history.all()
+                    # If there's no history, bail
+                    if not history:
+                        continue
+                    # This is the history object that last changed something
+                    last_meaningful_history = None
+                    for h in history:
+                        # Grab the first history object we come across
+                        if not last_meaningful_history:
+                            last_meaningful_history = h
+                            continue
+                        delta = last_meaningful_history.diff_against(
+                            h, foreign_keys_are_objs=False  # This makes foreign keys show up as UUIDs
+                        )
+                        # If there were fields that changed meaningfully, then
+                        # track that by updating last_meaningful_history and
+                        # keep going
+                        if delta.changes or delta.changed_fields:
+                            logging.info(f"Preserving Change: {delta}")
+                            last_meaningful_history = h
+                            continue
+
+                        # Otherwise, delete the history object
+                        h.delete()
+
+                except Exception as e:
+                    logging.exception(f"Could not get history for this link: {e}")
+                    raise e

src/meshapi/tests/test_dedupe_history_entries.py

@@ -0,0 +1,95 @@
+from datetime import datetime, date, timezone
+from django.core import management
+from uuid import UUID
+from django.test import TestCase
+from simple_history.models import HistoricalRecords
+
+from meshapi.models.devices.device import Device
+from meshapi.models.link import Link
+from meshapi.models.node import Node
+
+
+class TestDedupeHistoryEntries(TestCase):
+    def setUp(self) -> None:
+        self.node1 = Node(
+            network_number=1234,
+            status=Node.NodeStatus.ACTIVE,
+            type=Node.NodeType.STANDARD,
+            latitude=0,
+            longitude=0,
+        )
+        self.node1.save()
+
+        self.node2 = Node(
+            network_number=5678,
+            status=Node.NodeStatus.ACTIVE,
+            type=Node.NodeType.STANDARD,
+            latitude=0,
+            longitude=0,
+        )
+        self.node2.save()
+
+        self.node3 = Node(
+            network_number=9012,
+            status=Node.NodeStatus.ACTIVE,
+            type=Node.NodeType.STANDARD,
+            latitude=0,
+            longitude=0,
+        )
+        self.node3.save()
+
+        self.device1 = Device(
+            node=self.node1,
+            status=Device.DeviceStatus.ACTIVE,
+            name="nycmesh-1234-dev1",
+        )
+        self.device1.save()
+
+        self.device2 = Device(
+            node=self.node2,
+            status=Device.DeviceStatus.ACTIVE,
+            name="nycmesh-5678-dev2",
+        )
+        self.device2.save()
+
+        self.device3 = Device(
+            node=self.node3,
+            status=Device.DeviceStatus.ACTIVE,
+            name="nycmesh-9012-dev3",
+        )
+        # Save the device multiple times to see if we can get some bad history
+        self.device3.save()
+        self.device3.save()
+        self.device3.save()
+
+        self.link = Link(
+            from_device=self.device1,
+            to_device=self.device2,
+            status=Link.LinkStatus.ACTIVE,
+            type=Link.LinkType.FIVE_GHZ,
+            uisp_id="fake-uisp-uuid",
+        )
+
+        # Save the link multiple times to see if we can get some bad history
+        self.link.save()
+        self.link.save()
+        self.link.save()
+        self.link.save()
+
+
+    def test_deduplicate_history(self):
+        # Ensure link has 4 entries
+        self.assertEqual(4, len(self.link.history.all()))
+
+        # Ensure device3 has 3
+        self.assertEqual(3, len(self.device3.history.all()))
+
+        # etc
+        self.assertEqual(1, len(self.device2.history.all()))
+
+        management.call_command("dedupe_history_entries")
+        self.assertEqual(1, len(self.link.history.all()))
+        self.assertEqual(1, len(self.device3.history.all()))
+        self.assertEqual(1, len(self.device2.history.all()))
+
+