fix(api): add attack paths scan DB defaults (#11830)

Co-authored-by: Josema Camacho <josema@prowler.com>

api/CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable changes to the **Prowler API** are documented in this file.
 
+## [1.33.1] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- Attack Paths: Scan rows now have database defaults for `is_migrated` and `sink_backend` so `scan-perform-scheduled` inserts survive deploy skew [(#11826)](https://github.com/prowler-cloud/prowler/pull/11826)
+
+---
+
 ## [1.33.0] (Prowler v5.32.0)
 
 ### 🚀 Added

api/src/backend/api/migrations/0097_attack_paths_scan_db_defaults.py

@@ -0,0 +1,25 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0096_attack_paths_scan_is_migrated"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="attackpathsscan",
+            name="is_migrated",
+            field=models.BooleanField(db_default=False, default=False),
+        ),
+        migrations.AlterField(
+            model_name="attackpathsscan",
+            name="sink_backend",
+            field=models.CharField(
+                choices=[("neo4j", "Neo4j"), ("neptune", "Neptune")],
+                db_default="neo4j",
+                default="neo4j",
+                max_length=16,
+            ),
+        ),
+    ]

api/src/backend/api/models.py

@@ -814,9 +814,10 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
     # still using the previous graph shape. Query catalog selection uses this
     # flag; physical read routing uses sink_backend below.
     # TODO: drop after Neptune cutover
-    is_migrated = models.BooleanField(default=False)
+    is_migrated = models.BooleanField(default=False, db_default=False)
     sink_backend = models.CharField(
         choices=SinkBackendChoices.choices,
+        db_default=SinkBackendChoices.NEO4J,
         default=SinkBackendChoices.NEO4J,
         max_length=16,
     )

api/src/backend/tasks/tests/test_attack_paths_scan.py

@@ -2,8 +2,10 @@ from contextlib import nullcontext
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from unittest.mock import MagicMock, call, patch
+from uuid import uuid4
 
 import pytest
+from api.db_utils import rls_transaction
 from api.models import (
     AttackPathsScan,
     Finding,
@@ -15,6 +17,7 @@ from api.models import (
     StatusChoices,
     Task,
 )
+from django.db import DEFAULT_DB_ALIAS
 from django_celery_results.models import TaskResult
 from prowler.lib.check.models import Severity
 from tasks.jobs.attack_paths import findings as findings_module
@@ -2244,6 +2247,58 @@ class TestInternetAnalysis:
 class TestAttackPathsDbUtilsGraphDataReady:
     """Tests for db_utils functions related to graph_data_ready lifecycle."""
 
+    def test_database_defaults_allow_legacy_insert_without_cutover_columns(
+        self, tenants_fixture, providers_fixture, scans_fixture
+    ):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+        provider.provider = Provider.ProviderChoices.AWS
+        provider.save()
+        scan = scans_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        attack_paths_scan_id = uuid4()
+        now = datetime.now(tz=UTC)
+
+        with rls_transaction(str(tenant.id), using=DEFAULT_DB_ALIAS) as cursor:
+            cursor.execute(
+                """
+                INSERT INTO attack_paths_scans (
+                    id,
+                    inserted_at,
+                    updated_at,
+                    state,
+                    progress,
+                    graph_data_ready,
+                    started_at,
+                    tenant_id,
+                    provider_id,
+                    scan_id
+                )
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                """,
+                [
+                    attack_paths_scan_id,
+                    now,
+                    now,
+                    StateChoices.SCHEDULED,
+                    0,
+                    False,
+                    now,
+                    tenant.id,
+                    provider.id,
+                    scan.id,
+                ],
+            )
+
+            attack_paths_scan = AttackPathsScan.objects.get(id=attack_paths_scan_id)
+
+        assert attack_paths_scan.is_migrated is False
+        assert (
+            attack_paths_scan.sink_backend == AttackPathsScan.SinkBackendChoices.NEO4J
+        )
+
     def test_create_attack_paths_scan_first_scan_defaults_to_false(
         self, tenants_fixture, providers_fixture, scans_fixture
     ):