diff --git a/src/django_admin_runner/migrations/0011_scheduledcommand_django_admin_runner_scheduledcommand_source_valid.py b/src/django_admin_runner/migrations/0011_scheduledcommand_django_admin_runner_scheduledcommand_source_valid.py new file mode 100644 index 0000000..2a3df4e --- /dev/null +++ b/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", + ), + ), + ] diff --git a/src/django_admin_runner/models.py b/src/django_admin_runner/models.py index 3561643..2479394 100644 --- a/src/django_admin_runner/models.py +++ b/src/django_admin_runner/models.py @@ -31,9 +31,13 @@ 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, @@ -41,13 +45,15 @@ class Status(models.TextChoices): ) 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: diff --git a/tests/test_migrations.py b/tests/test_migrations.py index a27d547..6106e39 100644 --- a/tests/test_migrations.py +++ b/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) diff --git a/tests/test_schedules.py b/tests/test_schedules.py index f606303..433d643 100644 --- a/tests/test_schedules.py +++ b/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="")