Fixes #20: Correct validation retry logs and pending counts

netbox_scripts/ingestion.py

@@ -24,17 +24,20 @@
 
 from contextlib import contextmanager
 from datetime import timedelta
+from typing import NamedTuple
 
 from django.core.exceptions import ValidationError
 from django.db import router
 from django.utils import timezone
 from django.utils.translation import gettext as _
 
+from core.models import Job
+
 from . import branching
 from .choices import ActivationPolicyChoices, ProjectSourceTypeChoices, RevisionStatusChoices
 from .constants import ACTIVATABLE_REVISION_STATUSES, VALIDATION_LEASE_SECONDS
 from .jobs import ProjectScriptFileRefreshJob, RevisionValidationJob
-from .models import ScriptFile, ScriptProject
+from .models import ScriptFile, ScriptProject, ScriptProjectRevision
 from .permissions import refuse_unpermitted_selection_change, validate_script_file_permissions
 from .storage import config, service, store
 from .storage.exceptions import UnsafePathError
@@ -43,6 +46,7 @@
 from .utils import data_source_relative_path, source_path_to_dotted_name
 
 __all__ = (
+    'IngestedRevision',
     'check_upload_conflicts',
     'check_upload_preconditions',
     'current_source_tree',
@@ -60,6 +64,14 @@
 SOURCE_SUFFIX = '.py'
 
 
+class IngestedRevision(NamedTuple):
+    """The result of one ingestion call: the staging result plus any Job queued to process the revision."""
+
+    revision: ScriptProjectRevision
+    created: bool
+    processing_job: Job | None
+
+
 def uploaded_source_path(filename):
     """
     Return the canonical project-relative path one uploaded file is stored at.
@@ -175,8 +187,7 @@ def ingest_upload(project, *, filename, content, declare=True, activate_once=Fal
         files = current_source_tree(project)
         files[path] = bytes(content)
         staged = service.stage_revision(project, files, activate_once=activate_once)
-        queue_revision_processing(staged.revision)
-        return staged
+        return IngestedRevision(staged.revision, staged.created, queue_revision_processing(staged.revision))
 
 
 def ingest_data_source(project):
@@ -192,8 +203,7 @@ def ingest_data_source(project):
                 )
             )
         staged = service.stage_revision(project, _data_source_tree(project))
-        queue_revision_processing(staged.revision)
-        return staged
+        return IngestedRevision(staged.revision, staged.created, queue_revision_processing(staged.revision))
 
 
 def queue_revision_processing(revision):

netbox_scripts/jobs.py

@@ -20,7 +20,12 @@
 
 from . import activation, branching
 from .choices import ActivationPolicyChoices, MigrationStateChoices, RevisionStatusChoices
-from .constants import ACTIVATABLE_REVISION_STATUSES, STALLED_CLEANUP_GRACE_SECONDS, VALIDATION_JOB_TIMEOUT
+from .constants import (
+    ACTIVATABLE_REVISION_STATUSES,
+    PENDING_VERDICT_REVISION_STATUSES,
+    STALLED_CLEANUP_GRACE_SECONDS,
+    VALIDATION_JOB_TIMEOUT,
+)
 from .execution import RESOLUTION_FAILURES, ScriptNotExecutableError, run_script
 from .models import MigrationRun, NetBoxScript, ScriptProject, ScriptProjectRevision
 from .models.migration import migration_lock
@@ -299,9 +304,12 @@ def run(self, project_id=None, **kwargs):
                     )
                 )
         else:
-            self.logger.info(
-                _('Revision {digest} matches the source and another run owns it.').format(digest=revision.digest[:12])
+            template = (
+                _('Revision {digest} is queued for validation again, because its validation lease expired.')
+                if staged.processing_job is not None
+                else _('Revision {digest} matches the source and another run owns it.')
             )
+            self.logger.info(template.format(digest=revision.digest[:12]))
 
 
 class ProjectScriptFileRefreshJob(JobRunner):
@@ -364,7 +372,7 @@ def run(self, project_id=None, **kwargs):
                     )
                     return
                 staged = service.refresh_revision_script_files(source)
-                queue_revision_processing(staged.revision)
+                processing_job = queue_revision_processing(staged.revision)
         except (StorageError, StorageConfigurationError, OSError) as error:
             detail = _('Refreshing the script files of "{project}" failed and needs another run: {error}').format(
                 project=project, error=error
@@ -377,6 +385,13 @@ def run(self, project_id=None, **kwargs):
             self.logger.info(
                 _('Revision {digest} is staged and queued for validation.').format(digest=revision.digest[:12])
             )
+        elif revision.status == RevisionStatusChoices.VALIDATING:
+            template = (
+                _('Revision {digest} is queued for validation again, because its validation lease expired.')
+                if processing_job is not None
+                else _('Revision {digest} matches this selection and another run owns it.')
+            )
+            self.logger.info(template.format(digest=revision.digest[:12]))
         elif revision.pk == project.active_revision_id:
             self.logger.info(_('The selection has not changed, this project already serves it.'))
         else:
@@ -944,7 +959,9 @@ def run(self, **kwargs):
             status = result['revision_status']
             # Any status but materialized is one the revision already held when content addressing found it.
             # Whole sentences per branch: a translator cannot reorder a substituted fragment.
-            if status == RevisionStatusChoices.MATERIALIZED:
+            if status == RevisionStatusChoices.MATERIALIZED or (
+                status == RevisionStatusChoices.VALIDATING and result['processing_queued']
+            ):
                 template = (
                     _('Created project {key}, revision {revision_pk} is queued for validation.')
                     if result['created']
@@ -959,7 +976,7 @@ def run(self, **kwargs):
                 )
                 message = template.format(key=result['key'], revision_pk=result['revision_pk'], label=labels[status])
             self.logger.info(message)
-        pending = sum(1 for result in staged if result['revision_status'] == RevisionStatusChoices.MATERIALIZED)
+        pending = sum(1 for result in staged if result['revision_status'] in PENDING_VERDICT_REVISION_STATUSES)
         refused = len(results) - len(staged)
         self.logger.info(
             _(

netbox_scripts/migration/staging.py

@@ -43,6 +43,7 @@ def stage(proposed, modules):
                 'revision_pk': staged.revision.pk,
                 'revision_created': staged.created,
                 'revision_status': staged.revision.status,
+                'processing_queued': staged.processing_job is not None,
             }
         )
     return results
@@ -109,7 +110,7 @@ def _declare(project, members):
 
 
 def _stage_content(project, members):
-    """Stage the project's source and return the StagedRevision."""
+    """Stage the project's source and return the IngestedRevision."""
     if project.source_type == ProjectSourceTypeChoices.UPLOAD:
         member = members[0]
         source = legacy_source.read_source(member)

netbox_scripts/tests/migration/test_staging.py

@@ -1,5 +1,6 @@
 import hashlib
 import tempfile
+from datetime import timedelta
 from pathlib import Path
 from unittest import mock
 
@@ -18,6 +19,7 @@
     ProjectSourceTypeChoices,
     RevisionStatusChoices,
 )
+from netbox_scripts.constants import VALIDATION_LEASE_SECONDS
 from netbox_scripts.jobs import MigrationStagingJob, RevisionValidationJob
 from netbox_scripts.migration import plan, source, staging
 from netbox_scripts.models import (
@@ -253,6 +255,44 @@ def test_the_job_reports_a_revision_that_needs_no_validation(self):
         self.assertIn('is Valid.', messages)
         self.assertIn('0 awaiting a verdict', messages)
 
+    def test_the_job_reports_a_revision_still_validating_as_awaiting_a_verdict(self):
+        revisions = {result['key']: result['revision_pk'] for result in self.stage_all()}
+        uploaded = self.project_for(ProjectSourceTypeChoices.UPLOAD)
+        synced = self.project_for(ProjectSourceTypeChoices.DATA_SOURCE)
+        # The expired claim goes to the uploaded Project, since the reconciliation tests cover the Data Source path.
+        expired = timezone.now() - timedelta(seconds=2 * VALIDATION_LEASE_SECONDS)
+        for project, started in ((uploaded, expired), (synced, timezone.now())):
+            ScriptProjectRevision.objects.filter(pk=revisions[project.key]).update(
+                status=RevisionStatusChoices.VALIDATING, validation_started=started
+            )
+
+        job = MigrationStagingJob.enqueue(immediate=True)
+        job.refresh_from_db()
+
+        messages = ' '.join(entry['message'] for entry in job.log_entries)
+        self.assertIn(
+            f'Reused project {uploaded.key}, revision {revisions[uploaded.key]} is queued for validation.', messages
+        )
+        self.assertIn(f'Reused project {synced.key}, revision {revisions[synced.key]} is Validating.', messages)
+        self.assertIn('2 awaiting a verdict', messages)
+        queued = {result['key']: result['processing_queued'] for result in job.data['projects']}
+        self.assertEqual(queued, {uploaded.key: True, synced.key: False})
+
+    def test_the_job_reports_a_valid_revision_queued_for_activation_by_its_status(self):
+        self.stage_and_validate()
+        uploaded = self.project_for(ProjectSourceTypeChoices.UPLOAD)
+        # A refused activate-once promotion leaves the request pending, so staging queues the activation.
+        ScriptProject.objects.filter(pk=uploaded.pk).update(source_activation_pending=True)
+
+        job = MigrationStagingJob.enqueue(immediate=True)
+        job.refresh_from_db()
+
+        result = next(result for result in job.data['projects'] if result['key'] == uploaded.key)
+        messages = ' '.join(entry['message'] for entry in job.log_entries)
+        self.assertTrue(result['processing_queued'])
+        self.assertIn(f'Reused project {uploaded.key}, revision {result["revision_pk"]} is Valid.', messages)
+        self.assertIn('0 awaiting a verdict', messages)
+
     def test_the_job_opens_its_run_under_the_lock(self):
         # Watched from inside the lock: no run exists at the moment it is taken.
         held = []

netbox_scripts/tests/test_ingestion.py

@@ -248,22 +248,26 @@ def test_a_validation_whose_lease_expired_is_queued_again(self):
             validation_started=timezone.now() - timedelta(seconds=2 * VALIDATION_LEASE_SECONDS),
         )
         self.enqueued.reset_mock()
+        self.enqueued.return_value = mock.sentinel.processing_job
 
-        ingest_upload(self.project, filename='deploy.py', content=SCRIPT)
+        staged = ingest_upload(self.project, filename='deploy.py', content=SCRIPT)
 
         self.enqueued.assert_called_once()
         self.assertEqual(self.enqueued.call_args.args[0].pk, first.revision.pk)
+        self.assertIs(staged.processing_job, mock.sentinel.processing_job)
 
     def test_a_validation_inside_its_lease_is_left_to_its_owner(self):
         first = ingest_upload(self.project, filename='deploy.py', content=SCRIPT)
         ScriptProjectRevision.objects.filter(pk=first.revision.pk).update(
             status=RevisionStatusChoices.VALIDATING, validation_started=timezone.now()
         )
         self.enqueued.reset_mock()
+        self.enqueued.return_value = mock.sentinel.processing_job
 
-        ingest_upload(self.project, filename='deploy.py', content=SCRIPT)
+        staged = ingest_upload(self.project, filename='deploy.py', content=SCRIPT)
 
         self.enqueued.assert_not_called()
+        self.assertIsNone(staged.processing_job)
 
     def test_a_nested_path_is_preserved_at_this_layer(self):
         # Flattening to a basename happens in Django's uploaded-file handling, so it binds the

netbox_scripts/tests/test_reconciliation.py

@@ -1,10 +1,12 @@
 import shutil
 import tempfile
 import uuid
+from datetime import timedelta
 from pathlib import Path
 from unittest import mock
 
 from django.test import TestCase, override_settings
+from django.utils import timezone
 
 from core.choices import DataSourceStatusChoices, JobStatusChoices
 from core.exceptions import JobFailed
@@ -17,6 +19,7 @@
     ProjectSourceTypeChoices,
     RevisionStatusChoices,
 )
+from netbox_scripts.constants import VALIDATION_LEASE_SECONDS
 from netbox_scripts.jobs import ProjectReconciliationJob, ProjectScriptFileRefreshJob, RevisionValidationJob
 from netbox_scripts.models import (
     NetBoxScript,
@@ -167,6 +170,39 @@ def test_a_second_job_over_an_unchanged_directory_does_nothing(self):
         # second job resolves to is the one an operator has yet to act on.
         self.assertIn('waiting for an operator', str(job.log_entries))
 
+    def test_a_validation_whose_lease_expired_is_reported_as_queued_again(self):
+        self.run_job()
+        revision = ScriptProjectRevision.objects.get(project=self.project)
+        ScriptProjectRevision.objects.filter(pk=revision.pk).update(
+            status=RevisionStatusChoices.VALIDATING,
+            validation_started=timezone.now() - timedelta(seconds=2 * VALIDATION_LEASE_SECONDS),
+        )
+        self.validated.reset_mock()
+        self.validated.return_value = mock.sentinel.processing_job
+
+        job = self.run_job()
+
+        self.validated.assert_called_once_with(revision)
+        self.assertIn(
+            f'Revision {revision.digest[:12]} is queued for validation again, because its validation lease expired.',
+            str(job.log_entries),
+        )
+
+    def test_a_validation_inside_its_lease_is_reported_as_owned(self):
+        self.run_job()
+        revision = ScriptProjectRevision.objects.get(project=self.project)
+        ScriptProjectRevision.objects.filter(pk=revision.pk).update(
+            status=RevisionStatusChoices.VALIDATING, validation_started=timezone.now()
+        )
+        self.validated.reset_mock()
+
+        job = self.run_job()
+
+        self.validated.assert_not_called()
+        self.assertIn(
+            f'Revision {revision.digest[:12]} matches the source and another run owns it.', str(job.log_entries)
+        )
+
     def test_a_project_deleted_before_the_job_runs_completes_with_nothing_to_do(self):
         deleted_pk = self.project.pk
         self.project.delete()

netbox_scripts/tests/test_script_file_refresh.py

@@ -1,13 +1,16 @@
 import shutil
 import tempfile
+from datetime import timedelta
 from pathlib import Path
 from unittest import mock
 
 from django.test import TestCase, override_settings
+from django.utils import timezone
 
 from core.choices import JobStatusChoices
 from netbox_scripts import branching
 from netbox_scripts.choices import ActivationPolicyChoices, RevisionStatusChoices
+from netbox_scripts.constants import VALIDATION_LEASE_SECONDS
 from netbox_scripts.forms import ScriptProjectScriptFilesForm
 from netbox_scripts.ingestion import ingest_upload
 from netbox_scripts.jobs import ProjectScriptFileRefreshJob, RevisionValidationJob
@@ -167,6 +170,35 @@ def test_an_unchanged_selection_retries_a_revision_still_awaiting_a_verdict(self
         self.run_job()
         self.validated.assert_called_once()
 
+    def test_an_unchanged_selection_reports_an_expired_validation_as_queued_again(self):
+        revision = self.project.current_revision
+        ScriptProjectRevision.objects.filter(pk=revision.pk).update(
+            status=RevisionStatusChoices.VALIDATING,
+            validation_started=timezone.now() - timedelta(seconds=2 * VALIDATION_LEASE_SECONDS),
+        )
+        self.validated.return_value = mock.sentinel.processing_job
+
+        job = self.run_job()
+
+        self.validated.assert_called_once_with(revision)
+        self.assertIn(
+            f'Revision {revision.digest[:12]} is queued for validation again, because its validation lease expired.',
+            str(job.log_entries),
+        )
+
+    def test_an_unchanged_selection_reports_a_live_validation_as_owned(self):
+        revision = self.project.current_revision
+        ScriptProjectRevision.objects.filter(pk=revision.pk).update(
+            status=RevisionStatusChoices.VALIDATING, validation_started=timezone.now()
+        )
+
+        job = self.run_job()
+
+        self.validated.assert_not_called()
+        self.assertIn(
+            f'Revision {revision.digest[:12]} matches this selection and another run owns it.', str(job.log_entries)
+        )
+
     def test_a_project_with_no_stored_content_completes_with_nothing_to_do(self):
         empty = ScriptProject.objects.create(name='Empty', key='empty')
         job = ProjectScriptFileRefreshJob.enqueue(immediate=True, project_id=empty.pk)