fix: handle static DynamicGroups in job filter and repo sync

changes/1029.fixed

@@ -0,0 +1 @@
+Fixed job failure when a device is assigned to a static DynamicGroup linked to a GoldenConfigSetting.

nautobot_golden_config/jobs.py

@@ -76,7 +76,7 @@ def get_refreshed_repos(job_obj, repo_types, data=None):
     repository_records = set()
     for group in dynamic_groups:
         # Make sure the data(device qs) device exist in the dg first.
-        if data.filter(group.generate_query()).exists():
+        if data.filter(pk__in=group.members.values_list("pk", flat=True)).exists():
             for repo_type in repo_types:
                 repo = getattr(group.golden_config_setting, repo_type, None)
                 if repo:

nautobot_golden_config/tests/test_jobs.py

@@ -2,11 +2,14 @@
 
 from unittest.mock import MagicMock, patch
 
+from django.contrib.contenttypes.models import ContentType
 from nautobot.apps.testing import TransactionTestCase, create_job_result_and_run_job
 from nautobot.dcim.models import Device
-from nautobot.extras.models import JobLogEntry
+from nautobot.extras.choices import DynamicGroupTypeChoices
+from nautobot.extras.models import DynamicGroup, GitRepository, GraphQLQuery, JobLogEntry
 
 from nautobot_golden_config import jobs
+from nautobot_golden_config.models import GoldenConfigSetting
 from nautobot_golden_config.tests.conftest import (
     create_device,
     create_orphan_device,
@@ -620,3 +623,50 @@ def test_run_all_job_multiple_repos_both_disabled(self, mock_ensure_git_reposito
 
         log_entries = JobLogEntry.objects.filter(job_result=job_result, grouping="GC Repo Commit and Push")
         self.assertEqual(log_entries.count(), 0)
+
+
+@patch("nautobot_golden_config.nornir_plays.config_backup.run_backup", MagicMock(return_value="foo"))
+@patch.object(jobs, "ensure_git_repository")
+class GCReposStaticGroupTestCase(TransactionTestCase):
+    """Test that jobs succeed when a device belongs to a static DynamicGroup."""
+
+    databases = ("default", "job_logs")
+
+    def setUp(self) -> None:
+        """Setup test data reproducing issue #1029: static group alongside dynamic filter group."""
+        self.device = create_device(name="static-device")
+        dgs_gc_settings_and_job_repo_objects()
+
+        ct_device = ContentType.objects.get_for_model(Device)
+        static_group = DynamicGroup.objects.create(
+            name="static-dg",
+            content_type=ct_device,
+            group_type=DynamicGroupTypeChoices.TYPE_STATIC,
+        )
+        static_group.add_members([self.device])
+
+        GoldenConfigSetting.objects.create(
+            name="static_group_setting",
+            slug="static_group_setting",
+            weight=2000,
+            backup_path_template="test/backup",
+            intended_path_template="test/intended",
+            jinja_path_template="{{jinja_path}}",
+            backup_test_connectivity=True,
+            dynamic_group=static_group,
+            sot_agg_query=GraphQLQuery.objects.get(name="GC-SoTAgg-Query-1"),
+            backup_repository=GitRepository.objects.get(name="test-backup-repo-1"),
+        )
+        super().setUp()
+
+    @patch("nautobot_golden_config.utilities.constant.ENABLE_BACKUP", True)
+    def test_backup_job_with_static_group_does_not_crash(self, mock_ensure_git_repository):
+        """Backup job must not raise RuntimeError when device is in a static group."""
+        mock_ensure_git_repository.return_value = True
+        job_result = create_job_result_and_run_job(
+            module="nautobot_golden_config.jobs",
+            name="BackupJob",
+            device=Device.objects.filter(name=self.device.name),
+        )
+        log_entries = JobLogEntry.objects.filter(job_result=job_result, grouping="GC After Run")
+        self.assertEqual(log_entries.first().message, "Finished the Backup Configurations job execution.")

nautobot_golden_config/tests/test_utilities/test_helpers.py

@@ -8,6 +8,7 @@
 from django.test import TestCase
 from jinja2 import exceptions as jinja_errors
 from nautobot.dcim.models import Device, Location, LocationType, Platform
+from nautobot.extras.choices import DynamicGroupTypeChoices
 from nautobot.extras.management import populate_status_choices
 from nautobot.extras.models import DynamicGroup, GitRepository, GraphQLQuery, Status, Tag
 from nornir_nautobot.exceptions import NornirNautobotException
@@ -318,3 +319,53 @@ def test_device_to_settings_map_gc_weight_change(self):
         # Regenerate the device to settings map to ensure it is up to date.
         temp_device_to_settings_map = get_device_to_settings_map(queryset=Device.objects.all())
         self.assertEqual(temp_device_to_settings_map[test_device.id], self.test_settings_a)
+
+
+class HelpersTestStaticGroup(TestCase):
+    """Test get_job_filter with a static DynamicGroup in scope."""
+
+    def setUp(self):
+        """Set up a static DynamicGroup and a GoldenConfigSetting using it."""
+        GitRepository.objects.all().delete()
+        create_helper_repo(name="backup-static-test", provides="backupconfigs")
+        create_helper_repo(name="intended-static-test", provides="intendedconfigs")
+        create_helper_repo(name="jinja-static-test", provides="jinjatemplate")
+
+        GoldenConfigSetting.objects.all().delete()
+
+        content_type = ContentType.objects.get(app_label="dcim", model="device")
+        graphql_query = GraphQLQuery.objects.create(
+            name="static-group-test-query",
+            query="query ($device_id: ID!) { device(id: $device_id) { name } }",
+        )
+
+        populate_status_choices()
+        self.device = create_device(name="static-device")
+
+        static_group = DynamicGroup.objects.create(
+            name="static-type-dg",
+            content_type=content_type,
+            group_type=DynamicGroupTypeChoices.TYPE_STATIC,
+        )
+        static_group.add_members([self.device])
+        GoldenConfigSetting.objects.create(
+            name="static_group_setting",
+            slug="static_group_setting",
+            weight=2000,
+            backup_repository=GitRepository.objects.get(name="backup-static-test"),
+            intended_repository=GitRepository.objects.get(name="intended-static-test"),
+            jinja_repository=GitRepository.objects.get(name="jinja-static-test"),
+            dynamic_group=static_group,
+            sot_agg_query=graphql_query,
+        )
+
+    def test_get_job_filter_with_static_group_does_not_raise(self):
+        """Verify get_job_filter does not raise when a GoldenConfigSetting uses a static group."""
+        result = get_job_filter()
+        self.assertIn(self.device, result)
+
+    def test_device_to_settings_map_with_static_group(self):
+        """Verify get_device_to_settings_map resolves a device in a static DynamicGroup to its setting."""
+        result = get_device_to_settings_map(queryset=Device.objects.all())
+        static_setting = GoldenConfigSetting.objects.get(name="static_group_setting")
+        self.assertEqual(result[self.device.id], static_setting)

nautobot_golden_config/utilities/helper.py

@@ -6,7 +6,7 @@
 
 from django.conf import settings
 from django.contrib import messages
-from django.db.models import OuterRef, Q, Subquery
+from django.db.models import Q
 from django.template import engines
 from django.urls import reverse
 from django.utils.html import format_html
@@ -76,7 +76,7 @@ def get_job_filter(data=None):
         dynamic_group__filter__iexact="{}", dynamic_group__group_type=DynamicGroupTypeChoices.TYPE_DYNAMIC_FILTER
     ).exists():
         for obj in models.GoldenConfigSetting.objects.all():
-            raw_qs = raw_qs | obj.dynamic_group.generate_query()
+            raw_qs = raw_qs | Q(pk__in=obj.dynamic_group.members.values_list("pk", flat=True))
 
     base_qs = Device.objects.filter(raw_qs)
 
@@ -176,20 +176,12 @@ def render_jinja_template(obj, logger, template):
 def get_device_to_settings_map(queryset):
     """Helper function to map heightest weighted GC settings to devices."""
     update_dynamic_groups_cache()
-    annotated_queryset = queryset.all().annotate(
-        gc_settings=Subquery(
-            models.GoldenConfigSetting.objects.filter(
-                dynamic_group__static_group_associations__associated_object_id=OuterRef("id"),
-                dynamic_group__static_group_associations__associated_object_type__app_label="dcim",
-                dynamic_group__static_group_associations__associated_object_type__model="device",
-            )
-            .order_by("-weight")
-            # [:1] is a ORM/DB "limit 1" query, not a python slice.
-            .values("id")[:1]
-        )
-    )
-    gcs = {gc.id: gc for gc in models.GoldenConfigSetting.objects.all()}
-    return {device.id: gcs[device.gc_settings] for device in annotated_queryset}
+    device_to_settings = {}
+    for device in queryset.all():
+        setting = models.GoldenConfigSetting.objects.get_for_device(device)
+        if setting is not None:
+            device_to_settings[device.pk] = setting
+    return device_to_settings
 
 
 def get_json_config(config):