feat(enabler): audit snapshot

backend/core/migrations/0070_complianceassessmentsnapshot.py

@@ -0,0 +1,54 @@
+# Generated by Django 5.1.8 on 2025-04-25 19:06
+
+import django.db.models.deletion
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0069_auto_20250414_2023"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ComplianceAssessmentSnapshot",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                (
+                    "created_at",
+                    models.DateTimeField(auto_now_add=True, verbose_name="Created at"),
+                ),
+                (
+                    "updated_at",
+                    models.DateTimeField(auto_now=True, verbose_name="Updated at"),
+                ),
+                (
+                    "is_published",
+                    models.BooleanField(default=False, verbose_name="published"),
+                ),
+                ("format", models.IntegerField(default=1)),
+                ("revision", models.IntegerField(default=1)),
+                ("key_indicators", models.JSONField()),
+                ("data", models.JSONField()),
+                (
+                    "compliance_assessment",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="core.complianceassessment",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
+    ]

backend/core/models.py

@@ -3059,15 +3059,19 @@ class Meta:
         verbose_name = _("Compliance assessment")
         verbose_name_plural = _("Compliance assessments")
 
-    def upsert_daily_metrics(self):
+    def key_indicators(self):
         per_status = dict()
         per_result = dict()
         for item in self.get_requirements_status_count():
             per_status[item[1]] = item[0]
 
         for item in self.get_requirements_result_count():
             per_result[item[1]] = item[0]
-        total = RequirementAssessment.objects.filter(compliance_assessment=self).count()
+        total = (
+            RequirementAssessment.objects.filter(compliance_assessment=self)
+            .filter(requirement__assessable=True)
+            .count()
+        )
         data = {
             "reqs": {
                 "total": total,
@@ -3077,7 +3081,40 @@ def upsert_daily_metrics(self):
                 "score": self.get_global_score(),
             },
         }
+        return data
+
+    def create_snapshot(self):
+        logger.info(f"creating snapshot for {self.name}")
+        key_indicators = self.key_indicators()
+        requirements_assessments = []
+        for ra in RequirementAssessment.objects.filter(compliance_assessment=self):
+            # ).exclude(result=RequirementAssessment.Result.NOT_ASSESSED):#probably a risky optimization
+            entry = {
+                "id": str(ra.id),
+                "result": ra.result,
+                "status": ra.status,
+                "score": ra.score,
+                "doc_score": ra.documentation_score,
+            }
+            requirements_assessments.append(entry)
+        data = {"requirements_assessments": requirements_assessments}
+        ac = [
+            {"id": str(ac.id), "status": ac.status}
+            for ac in AppliedControl.objects.filter(
+                requirement_assessments__compliance_assessment=self
+            )
+        ]
+        if len(ac) > 0:
+            data.update({"applied_controls": ac})
+        snapshot = ComplianceAssessmentSnapshot.objects.create_snapshot(
+            compliance_assessment=self,
+            key_indicators=key_indicators,
+            data=data,
+        )
+        logger.info(f"created snapshot {snapshot.id} for {self.name}")
 
+    def upsert_daily_metrics(self):
+        data = self.key_indicators()
         HistoricalMetric.update_daily_metric(
             model=self.__class__.__name__, object_id=self.id, data=data
         )
@@ -4338,6 +4375,39 @@ class Meta:
         verbose_name_plural = "Task nodes"
 
 
+class SnapshotManager(models.Manager):
+    def create_snapshot(self, compliance_assessment, key_indicators, data, format=1):
+        latest = (
+            self.filter(compliance_assessment=compliance_assessment)
+            .order_by("-revision")
+            .first()
+        )
+
+        revision = 1
+        if latest:
+            revision = latest.revision + 1
+
+        return self.create(
+            compliance_assessment=compliance_assessment,
+            revision=revision,
+            key_indicators=key_indicators,
+            data=data,
+            format=format,
+        )
+
+
+class ComplianceAssessmentSnapshot(AbstractBaseModel):
+    format = models.IntegerField(default=1)
+    compliance_assessment = models.ForeignKey(
+        "ComplianceAssessment", on_delete=models.CASCADE
+    )
+    revision = models.IntegerField(default=1)
+    key_indicators = models.JSONField()
+    data = models.JSONField()
+
+    objects = SnapshotManager()
+
+
 common_exclude = ["created_at", "updated_at"]
 
 auditlog.register(