File indexing completed on 2026-11-09 10:47:38
0001 """Exercise both dispatcher routes through a real child harness process."""
0002 import importlib.util
0003 import json
0004 import os
0005 from pathlib import Path
0006 import tempfile
0007 import unittest
0008 from unittest.mock import patch
0009
0010 ROOT = Path(__file__).resolve().parents[1]
0011 module_spec = importlib.util.spec_from_file_location('es_dispatcher', ROOT / 'scripts/evgen_job_dispatcher.py')
0012 dispatcher = importlib.util.module_from_spec(module_spec)
0013 module_spec.loader.exec_module(dispatcher)
0014 submit_spec = importlib.util.spec_from_file_location('es_submitter', ROOT / 'scripts/submit-evgen-task.py')
0015 submitter = importlib.util.module_from_spec(submit_spec)
0016 submit_spec.loader.exec_module(submitter)
0017
0018
0019 class ProductionDispatcher(unittest.TestCase):
0020 def run_harness(self, production):
0021 with tempfile.TemporaryDirectory() as directory:
0022 sandbox = Path(directory) / 'sandbox'; sandbox.mkdir()
0023 harness = sandbox / dispatcher.PAYLOAD_SUBDIR / 'es/es_harness.py'
0024 harness.parent.mkdir(parents=True)
0025 harness.write_text('import json,sys\nfrom pathlib import Path\n'
0026 'p=Path(sys.argv[sys.argv.index("--summary")+1])\n'
0027 'p.parent.mkdir(parents=True,exist_ok=True)\n'
0028 'p.write_text(json.dumps({"args":sys.argv[1:]}))\n')
0029 proxy = Path(directory) / 'native-proxy'; proxy.write_text('fixture proxy')
0030 env = dict(os.environ, ES_PAYLOAD_IMAGE='image:26.07.3', ES_SLOTS='4',
0031 PILOT_X509_USER_PROXY=str(proxy),
0032 ES_OUTPUT_DATASET='/RECO/26.07.3/sample/10kevents')
0033 with patch.dict(os.environ, env, clear=True), \
0034 patch.object(dispatcher, 'write_job_report') as report:
0035 rc = dispatcher.run_event_service('sample', 'whole-task', str(sandbox),
0036 'test-channel', production=production)
0037 self.assertEqual(rc, 0)
0038 if production:
0039 self.assertEqual((sandbox / 'worker-proxy').read_text(), 'fixture proxy')
0040 self.assertEqual((sandbox / 'worker-proxy').stat().st_mode & 0o777, 0o600)
0041 summary = json.loads((sandbox.parent / dispatcher.ES_WORK_DIR / 'es_summary.json').read_text())
0042 return summary['args'], report.call_args.kwargs['extra']['es']
0043
0044 def test_production_keeps_normal_outputs_without_canary_lifetime(self):
0045 args, report = self.run_harness(True)
0046 self.assertFalse(any('CANARY_' in arg for arg in args))
0047 self.assertEqual(report['dataset'], 'epic:/RECO/26.07.3/sample/10kevents')
0048
0049 def test_canary_keeps_its_existing_expiring_output_route(self):
0050 args, report = self.run_harness(False)
0051 self.assertIn('CANARY_OUTPUT_DATASET=TEST/canary/whole-task', args)
0052 self.assertIn('CANARY_LIFETIME_S=604800', args)
0053 self.assertEqual(report['dataset'], 'epic:/TEST/canary/whole-task')
0054
0055 def test_production_sandbox_never_contains_the_live_proxy(self):
0056 with tempfile.TemporaryDirectory() as directory:
0057 root = Path(directory)
0058 proxy = root / 'live-proxy'; proxy.write_text('PRIVATE_PROXY_FIXTURE')
0059 payload = root / 'payload-source'; payload.mkdir()
0060 (payload / 'VERSION').write_text('test')
0061 spec = {'csvBase': 'whole-sample', 'csvRows': ['input,hepmc3.tree.root,10000,0000'],
0062 'env': {}, 'workerProxy': True, 'nEventsPerWorker': 250}
0063 with patch.object(submitter, '_reporting_env', return_value={}):
0064 work = Path(submitter._assemble_sandbox(spec, str(proxy), str(root / 'work'),
0065 payload_dir=str(payload)))
0066 sandbox = work / 'sandbox'
0067 self.assertIn('X509_USER_PROXY=worker-proxy',
0068 (sandbox / 'environment-whole-sample.sh').read_text())
0069 self.assertFalse((sandbox / proxy.name).exists())
0070 self.assertFalse(any(b'PRIVATE_PROXY_FIXTURE' in p.read_bytes()
0071 for p in sandbox.rglob('*') if p.is_file()))
