Enforce ScheduledCommand.source choices at database level

src/django_admin_runner/migrations/0011_scheduledcommand_django_admin_runner_scheduledcommand_source_valid.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_admin_runner", "0010_commandexecution_label"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="scheduledcommand",
+            constraint=models.CheckConstraint(
+                condition=models.Q(source__in=["code", "admin"]),
+                name="django_admin_runner_scheduledcommand_source_valid",
+            ),
+        ),
+    ]

src/django_admin_runner/models.py

@@ -31,23 +31,29 @@ class Status(models.TextChoices):
         TIMEOUT = "TIMEOUT", "Timed out"
 
     command_name = models.CharField(max_length=200)
-    # Optional label: typed on the run form for manual runs, copied from
-    # the schedule's label for scheduled runs.
-    label = models.CharField(max_length=200, blank=True)
+    label = models.CharField(
+        max_length=200,
+        blank=True,
+        default="",
+        help_text="Optional label: set on the run form (manual runs) "
+        "or copied from the schedule (scheduled runs).",
+    )
     status = models.CharField(
         max_length=20,
         choices=Status.choices,
         default=Status.PENDING,
     )
     result_html = models.TextField(blank=True)
     kwargs = models.JSONField(default=dict)
-    # Set when a (graceful) stop was requested from the admin; the worker
-    # checks this flag on its output-flush heartbeat and on SIGTERM.
-    stop_requested = models.BooleanField(default=False)  # type: ignore[assignment]
-    # OS PID of the process running the command, recorded when the
-    # execution transitions to RUNNING — enables PID-targeted signals
-    # (django-q2 force kill).
-    worker_pid = models.IntegerField(null=True, blank=True)
+    stop_requested = models.BooleanField(
+        default=False,  # type: ignore[assignment]
+        help_text="Set when a graceful stop was requested from the admin.",
+    )
+    worker_pid = models.IntegerField(
+        null=True,
+        blank=True,
+        help_text="OS PID of the process running the command, recorded at start.",
+    )
     schedule = models.ForeignKey(
         "ScheduledCommand",
         null=True,
@@ -212,6 +218,10 @@ class Meta:
                 condition=models.Q(kind__in=["cron", "interval", "clocked"]),
                 name="django_admin_runner_scheduledcommand_kind_valid",
             ),
+            models.CheckConstraint(
+                condition=models.Q(source__in=["code", "admin"]),
+                name="django_admin_runner_scheduledcommand_source_valid",
+            ),
         ]
 
     def __str__(self) -> str:

tests/test_migrations.py

@@ -17,7 +17,7 @@ def _restore_latest(self):
         # Always end at the latest migration so other tests see full state.
         executor = MigrationExecutor(connection)
         executor.loader.build_graph()
-        executor.migrate([(APP, MIG_0006)])
+        executor.migrate(executor.loader.graph.leaf_nodes())
 
     def _migrate_to(self, name):
         executor = MigrationExecutor(connection)

tests/test_schedules.py

@@ -8,6 +8,7 @@
 
 import pytest
 from django.core.exceptions import ValidationError
+from django.db import IntegrityError
 from django.test import Client, override_settings
 from django.utils.timezone import localtime
 
@@ -960,3 +961,28 @@ def test_execution_shows_schedule_traceability(self, admin_client, active_comman
             f"/admin/django_admin_runner/commandexecution/{execution.pk}/change/"
         )
         assert response.status_code == 200
+
+
+@pytest.mark.django_db(transaction=True)
+class TestScheduledCommandConstraints:
+    """DB-level enforcement of ScheduledCommand field choices."""
+
+    def make_schedule(self, **kwargs):
+        return ScheduledCommand.objects.create(
+            command_name="simple_command",
+            kind=kwargs.pop("kind", "cron"),
+            cron=kwargs.pop("cron", "15 4 * * *"),
+            **kwargs,
+        )
+
+    def test_invalid_source_rejected_at_db_level(self, active_command):
+        with pytest.raises(IntegrityError):
+            self.make_schedule(source="bogus")
+
+    def test_valid_sources_accepted(self, active_command):
+        assert self.make_schedule(source="code").source == "code"
+        assert self.make_schedule(source="admin").source == "admin"
+
+    def test_invalid_kind_rejected_at_db_level(self, active_command):
+        with pytest.raises(IntegrityError):
+            self.make_schedule(kind="bogus", cron="")