diff --git a/netbox/core/tests/test_jobs.py b/netbox/core/tests/test_jobs.py
new file mode 100644
index 000000000..859b743bc
--- /dev/null
+++ b/netbox/core/tests/test_jobs.py
@@ -0,0 +1,367 @@
+import uuid
+from datetime import timedelta
+from unittest.mock import MagicMock, patch
+
+import requests
+from django.conf import settings
+from django.test import TestCase, override_settings
+from django.utils import timezone
+
+from core.choices import DataSourceStatusChoices
+from core.jobs import SyncDataSourceJob, SystemHousekeepingJob
+from core.models import DataFile, DataSource, Job
+
+
+def _make_runner(cls, **job_attrs):
+    """
+    Build a JobRunner without going through ``__init__``.
+
+    ``JobRunner.__init__`` attaches a ``JobLogHandler`` to a module-level
+    singleton logger, so calling it once per test would accumulate handlers
+    across the suite. Bypass it and stub the logger directly.
+    """
+    runner = cls.__new__(cls)
+    runner.job = MagicMock(**job_attrs)
+    runner.logger = MagicMock()
+    return runner
+
+
+class HousekeepingRunnerMixin:
+    """Provides a `_runner()` helper that builds a SystemHousekeepingJob with a mock job."""
+
+    @staticmethod
+    def _runner():
+        return _make_runner(SystemHousekeepingJob)
+
+
+class SyncDataSourceJobTestCase(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.datasource = DataSource.objects.create(
+            name='Test Source',
+            type='local',
+            source_url='/tmp/test',
+        )
+
+    def test_enqueue_sets_datasource_status_to_queued(self):
+        job = MagicMock()
+        job.object = self.datasource
+
+        with patch('core.models.Job.enqueue', return_value=job):
+            result = SyncDataSourceJob.enqueue(instance=self.datasource)
+
+        self.assertIs(result, job)
+        # Verify both the in-memory assignment (`datasource.status = ...`) and the
+        # persisted update (`DataSource.objects.filter(pk=...).update(...)`).
+        self.assertEqual(self.datasource.status, DataSourceStatusChoices.QUEUED)
+        self.datasource.refresh_from_db()
+        self.assertEqual(self.datasource.status, DataSourceStatusChoices.QUEUED)
+
+    def test_enqueue_without_object_is_noop(self):
+        # Baseline: the datasource starts at NEW. (Captures setUpTestData drift.)
+        self.assertEqual(self.datasource.status, DataSourceStatusChoices.NEW)
+
+        job = MagicMock()
+        job.object = None
+
+        with (
+            patch('core.models.Job.enqueue', return_value=job),
+            patch(
+                'core.jobs.DataSource.objects.filter',
+                wraps=DataSource.objects.filter,
+            ) as filter_,
+        ):
+            result = SyncDataSourceJob.enqueue()
+
+        self.assertIs(result, job)
+        # Intent: the `if datasource := job.object` branch was skipped, so no
+        # filter().update() call was even attempted.
+        filter_.assert_not_called()
+        # Outcome: no side effect on any existing DataSource.
+        self.datasource.refresh_from_db()
+        self.assertEqual(self.datasource.status, DataSourceStatusChoices.NEW)
+
+    def test_run_syncs_datasource_and_updates_search_cache(self):
+        datafile = DataFile.objects.create(
+            source=self.datasource,
+            path='test.txt',
+            last_updated=timezone.now(),
+            size=4,
+            hash='0' * 64,
+            data=b'test',
+        )
+        runner = _make_runner(SyncDataSourceJob, object_id=self.datasource.pk)
+
+        with (
+            patch('core.models.DataSource.sync') as sync,
+            patch('core.jobs.search_backend.cache') as cache,
+        ):
+            runner.run()
+
+        sync.assert_called_once_with()
+        cache.assert_called_once()
+        # The cache argument should iterate over the datasource's data files.
+        cache_arg = cache.call_args.args[0]
+        self.assertEqual(list(cache_arg), [datafile])
+
+    def test_run_marks_datasource_failed_and_reraises_on_sync_error(self):
+        runner = _make_runner(SyncDataSourceJob, object_id=self.datasource.pk)
+
+        with (
+            patch('core.models.DataSource.sync', side_effect=RuntimeError('boom')),
+            patch('core.jobs.search_backend.cache') as cache,
+        ):
+            with self.assertRaisesMessage(RuntimeError, 'boom'):
+                runner.run()
+
+        self.datasource.refresh_from_db()
+        self.assertEqual(self.datasource.status, DataSourceStatusChoices.FAILED)
+        cache.assert_not_called()
+
+    def test_run_raises_when_datasource_no_longer_exists(self):
+        # If the DataSource was deleted between enqueue and run, the initial lookup
+        # raises DoesNotExist; the framework will surface it as a job error.
+        runner = _make_runner(SyncDataSourceJob, object_id=99_999_999)
+
+        with self.assertRaises(DataSource.DoesNotExist):
+            runner.run()
+
+
+class SystemHousekeepingRunTestCase(HousekeepingRunnerMixin, TestCase):
+    SUBMETHODS = (
+        'send_census_report',
+        'clear_expired_sessions',
+        'prune_changelog',
+        'delete_expired_jobs',
+        'check_for_new_releases',
+    )
+
+    @override_settings(DEBUG=True)
+    def test_run_skips_when_debug_is_enabled(self):
+        # DEBUG is checked before sys.argv; sys.argv is irrelevant here.
+        runner = self._runner()
+        patches = {name: patch.object(SystemHousekeepingJob, name) for name in self.SUBMETHODS}
+        mocks = {name: p.start() for name, p in patches.items()}
+        self.addCleanup(lambda: [p.stop() for p in patches.values()])
+
+        runner.run()
+
+        for mock in mocks.values():
+            mock.assert_not_called()
+
+    @override_settings(DEBUG=False)
+    def test_run_skips_during_test_invocation(self):
+        runner = self._runner()
+        patches = {name: patch.object(SystemHousekeepingJob, name) for name in self.SUBMETHODS}
+        mocks = {name: p.start() for name, p in patches.items()}
+        self.addCleanup(lambda: [p.stop() for p in patches.values()])
+
+        with patch('core.jobs.sys.argv', ['manage.py', 'test']):
+            runner.run()
+
+        for mock in mocks.values():
+            mock.assert_not_called()
+
+    @override_settings(DEBUG=False)
+    def test_run_executes_all_housekeeping_tasks(self):
+        runner = self._runner()
+        patches = {name: patch.object(SystemHousekeepingJob, name) for name in self.SUBMETHODS}
+        mocks = {name: p.start() for name, p in patches.items()}
+        self.addCleanup(lambda: [p.stop() for p in patches.values()])
+
+        with patch('core.jobs.sys.argv', ['netbox']):
+            runner.run()
+
+        for mock in mocks.values():
+            mock.assert_called_once_with()
+
+
+class SendCensusReportTestCase(HousekeepingRunnerMixin, TestCase):
+    @override_settings(ISOLATED_DEPLOYMENT=True)
+    def test_send_census_report_skips_when_isolated_deployment(self):
+        with patch('core.jobs.requests.get') as get:
+            self._runner().send_census_report()
+        get.assert_not_called()
+
+    @override_settings(ISOLATED_DEPLOYMENT=False, CENSUS_REPORTING_ENABLED=False)
+    def test_send_census_report_skips_when_reporting_disabled(self):
+        with patch('core.jobs.requests.get') as get:
+            self._runner().send_census_report()
+        get.assert_not_called()
+
+    @override_settings(
+        ISOLATED_DEPLOYMENT=False,
+        CENSUS_REPORTING_ENABLED=True,
+        CENSUS_URL='https://census.example/',
+        DEPLOYMENT_ID='abc123',
+    )
+    def test_send_census_report_sends_expected_payload(self):
+        with (
+            patch('core.jobs.requests.get') as get,
+            patch('core.jobs.resolve_proxies', return_value={'https': 'proxy'}) as resolve,
+        ):
+            self._runner().send_census_report()
+
+        resolve.assert_called_once_with(url='https://census.example/')
+        get.assert_called_once()
+        kwargs = get.call_args.kwargs
+        self.assertEqual(kwargs['url'], 'https://census.example/')
+        self.assertEqual(kwargs['timeout'], 3)
+        self.assertEqual(kwargs['proxies'], {'https': 'proxy'})
+        self.assertEqual(kwargs['params']['deployment_id'], 'abc123')
+        self.assertIn('version', kwargs['params'])
+        self.assertIn('python_version', kwargs['params'])
+
+    @override_settings(
+        ISOLATED_DEPLOYMENT=False,
+        CENSUS_REPORTING_ENABLED=True,
+        CENSUS_URL='https://census.example/',
+    )
+    def test_send_census_report_swallows_request_exception(self):
+        with (
+            patch('core.jobs.requests.get', side_effect=requests.RequestException('down')),
+            patch('core.jobs.resolve_proxies', return_value={}),
+        ):
+            self._runner().send_census_report()  # must not raise
+
+
+class ClearExpiredSessionsTestCase(HousekeepingRunnerMixin, TestCase):
+    def test_clear_expired_sessions_calls_session_store(self):
+        engine = MagicMock()
+        with patch('core.jobs.import_module', return_value=engine) as import_module:
+            self._runner().clear_expired_sessions()
+
+        import_module.assert_called_once_with(settings.SESSION_ENGINE)
+        engine.SessionStore.clear_expired.assert_called_once_with()
+
+    def test_clear_expired_sessions_handles_not_implemented(self):
+        engine = MagicMock()
+        engine.SessionStore.clear_expired.side_effect = NotImplementedError
+        runner = self._runner()
+
+        with patch('core.jobs.import_module', return_value=engine):
+            runner.clear_expired_sessions()  # must not raise
+
+        runner.logger.warning.assert_called_once()
+        self.assertIn(
+            'does not support',
+            runner.logger.warning.call_args.args[0],
+        )
+
+
+class DeleteExpiredJobsTestCase(HousekeepingRunnerMixin, TestCase):
+    def test_delete_expired_jobs_skips_when_retention_unset(self):
+        old_job = Job.objects.create(name='old', job_id=uuid.uuid4())
+        Job.objects.filter(pk=old_job.pk).update(created=timezone.now() - timedelta(days=365))
+
+        with patch('core.jobs.Config') as config_cls:
+            config_cls.return_value.JOB_RETENTION = 0
+            self._runner().delete_expired_jobs()
+
+        self.assertTrue(Job.objects.filter(pk=old_job.pk).exists())
+
+    def test_delete_expired_jobs_deletes_only_jobs_older_than_retention(self):
+        old_job = Job.objects.create(name='old', job_id=uuid.uuid4())
+        recent_job = Job.objects.create(name='recent', job_id=uuid.uuid4())
+        Job.objects.filter(pk=old_job.pk).update(created=timezone.now() - timedelta(days=30))
+        Job.objects.filter(pk=recent_job.pk).update(created=timezone.now() - timedelta(hours=1))
+
+        with patch('core.jobs.Config') as config_cls:
+            config_cls.return_value.JOB_RETENTION = 7
+            self._runner().delete_expired_jobs()
+
+        self.assertFalse(Job.objects.filter(pk=old_job.pk).exists())
+        self.assertTrue(Job.objects.filter(pk=recent_job.pk).exists())
+
+
+class CheckForNewReleasesTestCase(HousekeepingRunnerMixin, TestCase):
+    @override_settings(ISOLATED_DEPLOYMENT=True)
+    def test_check_for_new_releases_skips_when_isolated(self):
+        with patch('core.jobs.requests.get') as get:
+            self._runner().check_for_new_releases()
+        get.assert_not_called()
+
+    @override_settings(ISOLATED_DEPLOYMENT=False, RELEASE_CHECK_URL=None)
+    def test_check_for_new_releases_skips_when_url_unset(self):
+        with patch('core.jobs.requests.get') as get:
+            self._runner().check_for_new_releases()
+        get.assert_not_called()
+
+    @override_settings(ISOLATED_DEPLOYMENT=False, RELEASE_CHECK_URL='https://api.example/')
+    def test_check_for_new_releases_handles_request_exception(self):
+        with (
+            patch('core.jobs.requests.get', side_effect=requests.RequestException('down')),
+            patch('core.jobs.cache') as cache,
+            patch('core.jobs.resolve_proxies', return_value={}),
+        ):
+            self._runner().check_for_new_releases()
+
+        cache.set.assert_not_called()
+
+    @override_settings(ISOLATED_DEPLOYMENT=False, RELEASE_CHECK_URL='https://api.example/')
+    def test_check_for_new_releases_caches_latest_stable_release(self):
+        response = MagicMock()
+        response.json.return_value = [
+            {'tag_name': 'v4.5.0', 'html_url': 'https://example/4.5.0'},
+            {'tag_name': 'v4.6.0', 'html_url': 'https://example/4.6.0'},
+            {'tag_name': 'v4.7.0-rc1', 'html_url': 'https://example/rc', 'prerelease': True},
+            {'tag_name': 'v4.7.0-dev', 'html_url': 'https://example/dev', 'devrelease': True},
+            {'html_url': 'https://example/no-tag'},
+        ]
+
+        with (
+            patch('core.jobs.requests.get', return_value=response) as get,
+            patch('core.jobs.cache.set') as cache_set,
+            patch('core.jobs.resolve_proxies', return_value={'http': 'proxy'}) as resolve,
+        ):
+            self._runner().check_for_new_releases()
+
+        # HTTP request: URL, GitHub API Accept header, resolved proxies.
+        resolve.assert_called_once_with(url='https://api.example/')
+        get.assert_called_once_with(
+            url='https://api.example/',
+            headers={'Accept': 'application/vnd.github.v3+json'},
+            proxies={'http': 'proxy'},
+        )
+        response.raise_for_status.assert_called_once_with()
+
+        cache_set.assert_called_once()
+        # Accept either positional or keyword form for cache.set(key, value, ttl).
+        call = cache_set.call_args
+        bound = {**dict(zip(('key', 'value', 'timeout'), call.args)), **call.kwargs}
+        self.assertEqual(bound['key'], 'latest_release')
+        self.assertIsNone(bound['timeout'])
+        latest_version, latest_url = bound['value']
+        self.assertEqual(str(latest_version), '4.6.0')
+        self.assertEqual(latest_url, 'https://example/4.6.0')
+
+
+class PruneChangelogTestCase(HousekeepingRunnerMixin, TestCase):
+    def test_prune_changelog_skips_when_retention_unset(self):
+        with (
+            patch('core.jobs.Config') as config_cls,
+            patch('core.jobs.ObjectChange') as object_change,
+        ):
+            config_cls.return_value.CHANGELOG_RETENTION = None
+            self._runner().prune_changelog()
+
+        object_change.objects.filter.assert_not_called()
+
+    def test_prune_changelog_uses_strict_cutoff_filter(self):
+        # Implementation pin: prune_changelog must use time__lt (strict less-than) so a
+        # record exactly at the cutoff is retained. End-to-end behavior of the prune is
+        # covered by ChangelogPruneRetentionTestCase in core/tests/test_changelog.py.
+        with (
+            patch('core.jobs.Config') as config_cls,
+            patch('core.jobs.ObjectChange') as object_change,
+            patch('core.jobs.timezone') as tz,
+        ):
+            config_cls.return_value.CHANGELOG_RETENTION = 7
+            config_cls.return_value.CHANGELOG_RETAIN_CREATE_LAST_UPDATE = False
+            tz.now.return_value = timezone.datetime(2026, 1, 8, tzinfo=timezone.get_current_timezone())
+            object_change.objects.filter.return_value.delete.return_value = (0, {})
+
+            self._runner().prune_changelog()
+
+        expected_cutoff = timezone.datetime(2026, 1, 1, tzinfo=timezone.get_current_timezone())
+        object_change.objects.filter.assert_called_once_with(time__lt=expected_cutoff)
diff --git a/netbox/extras/tests/test_jobs.py b/netbox/extras/tests/test_jobs.py
new file mode 100644
index 000000000..6906ebb69
--- /dev/null
+++ b/netbox/extras/tests/test_jobs.py
@@ -0,0 +1,329 @@
+from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.db import DEFAULT_DB_ALIAS
+from django.test import TestCase
+
+from extras.jobs import ScriptJob
+from utilities.exceptions import AbortScript
+
+
+def _make_runner(**job_attrs):
+    """
+    Build a ScriptJob without going through ``__init__``.
+
+    ``JobRunner.__init__`` attaches a ``JobLogHandler`` to a module-level
+    singleton logger; instantiating one per test would accumulate handlers.
+    """
+    runner = ScriptJob.__new__(ScriptJob)
+    runner.job = MagicMock(**job_attrs)
+    runner.logger = MagicMock()
+    return runner
+
+
+class DummyScript:
+    """A minimal stand-in for a Script implementation."""
+
+    full_name = 'tests.DummyScript'
+
+    def __init__(self, run_result=None, run_exception=None, failed=False):
+        self._run_result = run_result
+        self._run_exception = run_exception
+        self.failed = failed
+        self.output = None
+        self.request = None
+        self.run = MagicMock(side_effect=self._run_impl)
+        self.log_info = MagicMock()
+        self.log_failure = MagicMock()
+        self.get_job_data = MagicMock(return_value={'status': 'done'})
+
+    def _run_impl(self, data, commit):
+        if self._run_exception is not None:
+            raise self._run_exception
+        return self._run_result
+
+
+class RunScriptTestCase(TestCase):
+    def test_run_script_success_commit_true_sets_output_and_job_data(self):
+        runner = _make_runner()
+        script = DummyScript(run_result='hello')
+
+        runner.run_script(script, request=None, data={'k': 'v'}, commit=True)
+
+        self.assertEqual(script.output, 'hello')
+        script.run.assert_called_once_with({'k': 'v'}, True)
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_commit_false_rolls_back_without_raising(self):
+        runner = _make_runner()
+        script = DummyScript(run_result='hello')
+
+        runner.run_script(script, request=None, data={}, commit=False)
+
+        script.run.assert_called_once_with({}, False)
+        # Rollback path: log_info() is called (with the translated revert message);
+        # job.data is still populated. Don't assert exact wording.
+        script.log_info.assert_called_once()
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_commit_false_logs_warning_when_script_failed(self):
+        runner = _make_runner()
+        script = DummyScript(run_result='hello', failed=True)
+
+        with patch('extras.jobs.logging.getLogger') as get_logger:
+            logger = get_logger.return_value
+            runner.run_script(script, request=None, data={}, commit=False)
+
+        logger.warning.assert_any_call('Script failed')
+
+    def test_run_script_abort_script_logs_failure_and_reraises(self):
+        # Non-report scripts call `script.log_failure(msg)` positionally; report-style
+        # scripts use `log_failure(message=msg)`. See `test_run_script_abort_script_uses_report_log_failure_signature`.
+        runner = _make_runner()
+        script = DummyScript(run_exception=AbortScript('nope'))
+
+        with self.assertRaises(AbortScript):
+            runner.run_script(script, request=None, data={}, commit=True)
+
+        # Outcome assertions: AbortScript re-raised, failure logged on the script,
+        # rollback path traversed (log_info called), job.data populated.
+        script.log_failure.assert_called_once()
+        # Non-report path uses the positional signature; verify both that there is
+        # a positional arg and no `message=` kwarg.
+        self.assertTrue(script.log_failure.call_args.args)
+        self.assertFalse(script.log_failure.call_args.kwargs)
+        script.log_info.assert_called_once()
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_abort_script_uses_report_log_failure_signature(self):
+        runner = _make_runner()
+        script = DummyScript(run_exception=AbortScript('reportfail'))
+
+        with (
+            patch('extras.jobs.is_report', return_value=True),
+            self.assertRaises(AbortScript),
+        ):
+            runner.run_script(script, request=None, data={}, commit=True)
+
+        # For reports, log_failure is called with the keyword-form signature.
+        script.log_failure.assert_called_once()
+        self.assertIn('message', script.log_failure.call_args.kwargs)
+
+    def test_run_script_general_exception_logs_traceback_and_reraises(self):
+        runner = _make_runner()
+        script = DummyScript(run_exception=RuntimeError('boom'))
+
+        with self.assertRaisesMessage(RuntimeError, 'boom'):
+            runner.run_script(script, request=None, data={}, commit=True)
+
+        script.log_failure.assert_called_once()
+        # The failure message wraps a traceback; assert on the structural marker
+        # (Traceback header) rather than translated copy.
+        message = script.log_failure.call_args.kwargs['message']
+        self.assertIn('Traceback', message)
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_sends_clear_events_when_request_is_present_and_error_occurs(self):
+        runner = _make_runner()
+        script = DummyScript(run_exception=RuntimeError('boom'))
+        request = MagicMock()
+
+        with patch('extras.jobs.clear_events.send') as send:
+            with self.assertRaises(RuntimeError):
+                runner.run_script(script, request=request, data={}, commit=True)
+
+        send.assert_called_once_with(request)
+
+    def test_run_script_default_db_uses_single_atomic(self):
+        # Symmetric with test_run_script_enters_secondary_atomic_when_changelog_db_differs:
+        # when the changelog DB is the default, only a single atomic() block is opened.
+        runner = _make_runner()
+        script = DummyScript(run_result='ok')
+        atomic_calls = []
+
+        @contextmanager
+        def fake_atomic(using):
+            atomic_calls.append(using)
+            yield
+
+        with (
+            patch('extras.jobs.router.db_for_write', return_value=DEFAULT_DB_ALIAS),
+            patch('extras.jobs.transaction.atomic', side_effect=fake_atomic),
+        ):
+            runner.run_script(script, request=None, data={}, commit=True)
+
+        self.assertEqual(atomic_calls, [DEFAULT_DB_ALIAS])
+        script.run.assert_called_once_with({}, True)
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_enters_secondary_atomic_when_changelog_db_differs(self):
+        # Verifies that the code enters two nested atomic() context managers when
+        # the changelog DB differs from the default. Does not exercise real rollback
+        # semantics (transaction.atomic is patched).
+        runner = _make_runner()
+        script = DummyScript(run_result='ok')
+
+        atomic_calls = []
+
+        @contextmanager
+        def fake_atomic(using):
+            atomic_calls.append(using)
+            yield
+
+        with (
+            patch('extras.jobs.router.db_for_write', return_value='changelog_db'),
+            patch('extras.jobs.transaction.atomic', side_effect=fake_atomic),
+        ):
+            runner.run_script(script, request=None, data={}, commit=True)
+
+        self.assertEqual(atomic_calls, [DEFAULT_DB_ALIAS, 'changelog_db'])
+        script.run.assert_called_once_with({}, True)
+        self.assertEqual(runner.job.data, {'status': 'done'})
+
+    def test_run_script_enters_secondary_atomic_when_commit_false(self):
+        # Mirror of the previous test for the commit=False branch on the secondary
+        # DB path. Verifies both atomic blocks are entered; does not exercise true
+        # rollback semantics (transaction.atomic is patched).
+        runner = _make_runner()
+        script = DummyScript(run_result='ok')
+        atomic_calls = []
+
+        @contextmanager
+        def fake_atomic(using):
+            atomic_calls.append(using)
+            yield
+
+        with (
+            patch('extras.jobs.router.db_for_write', return_value='changelog_db'),
+            patch('extras.jobs.transaction.atomic', side_effect=fake_atomic),
+        ):
+            runner.run_script(script, request=None, data={}, commit=False)
+
+        self.assertEqual(atomic_calls, [DEFAULT_DB_ALIAS, 'changelog_db'])
+        script.run.assert_called_once_with({}, False)
+        script.log_info.assert_called_once()
+
+
+class ScriptJobRunTestCase(TestCase):
+    @staticmethod
+    def _runner():
+        return _make_runner(object_id=1)
+
+    @staticmethod
+    def _script_model(script_instance):
+        return SimpleNamespace(pk=1, python_class=MagicMock(return_value=script_instance))
+
+    def test_run_loads_script_model_and_instantiates_python_class(self):
+        runner = self._runner()
+        script_instance = DummyScript()
+        script_model = self._script_model(script_instance)
+
+        with (
+            patch('extras.jobs.ScriptModel.objects.get', return_value=script_model) as get_script_model,
+            patch.object(ScriptJob, 'run_script') as run_script,
+            patch.dict('netbox.registry.registry', {'request_processors': []}, clear=False),
+        ):
+            runner.run(data={'k': 'v'}, commit=True)
+
+        get_script_model.assert_called_once_with(pk=1)
+        script_model.python_class.assert_called_once_with()
+        run_script.assert_called_once()
+        args, _ = run_script.call_args
+        self.assertIs(args[0], script_instance)
+        self.assertEqual(args[2], {'k': 'v'})  # data
+        self.assertIs(args[3], True)  # commit
+
+    def test_run_merges_request_files_into_data(self):
+        runner = self._runner()
+        script_instance = DummyScript()
+        script_model = self._script_model(script_instance)
+        request = MagicMock(FILES={'upload': 'fileobj'}, id='req-1')
+        data = {'k': 'v'}
+
+        with (
+            patch('extras.jobs.ScriptModel.objects.get', return_value=script_model),
+            patch.object(ScriptJob, 'run_script') as run_script,
+            patch.dict('netbox.registry.registry', {'request_processors': []}, clear=False),
+        ):
+            runner.run(data=data, request=request, commit=True)
+
+        passed_data = run_script.call_args.args[2]
+        self.assertEqual(passed_data['k'], 'v')
+        self.assertEqual(passed_data['upload'], 'fileobj')
+
+    def test_run_sets_script_request_when_request_is_present(self):
+        runner = self._runner()
+        script_instance = DummyScript()
+        script_model = self._script_model(script_instance)
+        request = MagicMock(FILES={}, id='req-1')
+
+        with (
+            patch('extras.jobs.ScriptModel.objects.get', return_value=script_model),
+            patch.object(ScriptJob, 'run_script'),
+            patch.dict('netbox.registry.registry', {'request_processors': []}, clear=False),
+        ):
+            runner.run(data={}, request=request, commit=True)
+
+        self.assertIs(script_instance.request, request)
+
+    def _processor_factories(self, entered):
+        def ctx_a(request):
+            @contextmanager
+            def _ctx():
+                entered.append('proc_a')
+                yield
+
+            return _ctx()
+
+        def ctx_event(request):
+            @contextmanager
+            def _ctx():
+                entered.append('event_tracking')
+                yield
+
+            return _ctx()
+
+        return ctx_a, ctx_event
+
+    def test_run_uses_request_processors_when_commit_true(self):
+        runner = self._runner()
+        script_instance = DummyScript()
+        script_model = self._script_model(script_instance)
+        entered = []
+        ctx_a, ctx_event = self._processor_factories(entered)
+
+        with (
+            patch('extras.jobs.ScriptModel.objects.get', return_value=script_model),
+            patch.object(ScriptJob, 'run_script'),
+            patch('extras.jobs.event_tracking', new=ctx_event),
+            patch.dict(
+                'netbox.registry.registry',
+                {'request_processors': [ctx_a, ctx_event]},
+                clear=False,
+            ),
+        ):
+            runner.run(data={}, commit=True)
+
+        self.assertEqual(entered, ['proc_a', 'event_tracking'])
+
+    def test_run_skips_event_tracking_when_commit_false(self):
+        runner = self._runner()
+        script_instance = DummyScript()
+        script_model = self._script_model(script_instance)
+        entered = []
+        ctx_a, ctx_event = self._processor_factories(entered)
+
+        with (
+            patch('extras.jobs.ScriptModel.objects.get', return_value=script_model),
+            patch.object(ScriptJob, 'run_script'),
+            patch('extras.jobs.event_tracking', new=ctx_event),
+            patch.dict(
+                'netbox.registry.registry',
+                {'request_processors': [ctx_a, ctx_event]},
+                clear=False,
+            ),
+        ):
+            runner.run(data={}, commit=False)
+
+        self.assertEqual(entered, ['proc_a'])