From 9f874b991d54748d0969d396180a3060ee0e0419 Mon Sep 17 00:00:00 2001 From: Martin Hauser Date: Wed, 28 Oct 2026 17:20:40 +0100 Subject: [PATCH] fix(scripts): Report requeued expired validations The reconciliation and Script File refresh Jobs now say when a validation whose lease expired was queued again, and migration staging counts every revision still awaiting a verdict. Fixes #20 --- netbox_scripts/ingestion.py | 20 +++++++--- netbox_scripts/jobs.py | 29 +++++++++++--- netbox_scripts/migration/staging.py | 3 +- .../tests/migration/test_staging.py | 40 +++++++++++++++++++ netbox_scripts/tests/test_ingestion.py | 8 +++- netbox_scripts/tests/test_reconciliation.py | 36 +++++++++++++++++ .../tests/test_script_file_refresh.py | 32 +++++++++++++++ 7 files changed, 154 insertions(+), 14 deletions(-) diff --git a/netbox_scripts/ingestion.py b/netbox_scripts/ingestion.py index 5b72793..581fab4 100644 --- a/netbox_scripts/ingestion.py +++ b/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): diff --git a/netbox_scripts/jobs.py b/netbox_scripts/jobs.py index a9a103f..97c29d0 100644 --- a/netbox_scripts/jobs.py +++ b/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( _( diff --git a/netbox_scripts/migration/staging.py b/netbox_scripts/migration/staging.py index 2579daf..052d16e 100644 --- a/netbox_scripts/migration/staging.py +++ b/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) diff --git a/netbox_scripts/tests/migration/test_staging.py b/netbox_scripts/tests/migration/test_staging.py index 34aa8b0..0b89cb1 100644 --- a/netbox_scripts/tests/migration/test_staging.py +++ b/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 = [] diff --git a/netbox_scripts/tests/test_ingestion.py b/netbox_scripts/tests/test_ingestion.py index b1ec4d4..edab104 100644 --- a/netbox_scripts/tests/test_ingestion.py +++ b/netbox_scripts/tests/test_ingestion.py @@ -248,11 +248,13 @@ 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) @@ -260,10 +262,12 @@ def test_a_validation_inside_its_lease_is_left_to_its_owner(self): 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 diff --git a/netbox_scripts/tests/test_reconciliation.py b/netbox_scripts/tests/test_reconciliation.py index be61039..6741034 100644 --- a/netbox_scripts/tests/test_reconciliation.py +++ b/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() diff --git a/netbox_scripts/tests/test_script_file_refresh.py b/netbox_scripts/tests/test_script_file_refresh.py index 6fa4916..97498fd 100644 --- a/netbox_scripts/tests/test_script_file_refresh.py +++ b/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)