afterjobs: make global publisher variables available (#49702)

tests/test_ctl.py

@@ -23,6 +23,7 @@ from wcs.ctl.wipe_data import CmdWipeData
 from wcs.ctl.management.commands.runscript import Command as CmdRunScript
 from wcs.ctl.management.commands.trigger_jumps import select_and_jump_formdata
 from wcs.ctl.delete_tenant import CmdDeleteTenant
+from wcs.workflows import WorkflowStatusItem
 from wcs.sql import get_connection_and_cursor, cleanup_connection
 
 from utilities import create_temporary_pub, clean_temporary_pub
@@ -410,7 +411,7 @@ def test_shell():
 
 class TestAfterJob(AfterJob):
     def execute(self):
-        pass
+        self.foo = WorkflowStatusItem().compute('{{ global_title|default:"FAIL" }}')
 
 
 def test_runjob(pub):
@@ -419,11 +420,19 @@ def test_runjob(pub):
     with pytest.raises(CommandError):
         call_command('runjob', '--domain=example.net', '--job-id=%s' % 'invalid')
 
+    pub.load_site_options()
+    if not pub.site_options.has_section('variables'):
+        pub.site_options.add_section('variables')
+    pub.site_options.set('variables', 'global_title', 'HELLO')
+    with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:
+        pub.site_options.write(fd)
+
     job = TestAfterJob(label='test')
     job.store()
     assert AfterJob.get(job.id).status == 'registered'
     call_command('runjob', '--domain=example.net', '--job-id=%s' % job.id)
     assert AfterJob.get(job.id).status == 'completed'
+    assert AfterJob.get(job.id).foo == 'HELLO'
 
 
 def test_ctl_print_help(capsys):

wcs/ctl/management/commands/runjob.py

@@ -16,6 +16,7 @@
 
 import sys
 
+from quixote import get_publisher
 from django.core.management import CommandError
 
 from wcs.qommon.http_response import AfterJob
@@ -32,6 +33,7 @@ class Command(TenantCommand):
     def handle(self, *args, **options):
         domain = options.pop('domain')
         self.init_tenant_publisher(domain)
+        get_publisher().substitutions.feed(get_publisher())
         try:
             job = AfterJob.get(options['job_id'])
         except KeyError: