File indexing completed on 2026-11-09 10:47:31
0001 import importlib.util
0002 import json
0003 from datetime import datetime, timedelta
0004 from pathlib import Path
0005 from tempfile import TemporaryDirectory
0006 from unittest.mock import patch
0007
0008 from django.test import SimpleTestCase
0009
0010 from monitor_app.worker_admission import outcome
0011
0012
0013 class WorkerAdmissionTests(SimpleTestCase):
0014 def test_known_prejob_exit_is_declined_but_signal_alone_is_not(self):
0015 self.assertEqual(outcome('BNL_OSG_PanDA_1', 'failed', 73), 'declined')
0016 self.assertEqual(outcome('another_queue', 'failed', 73), 'failed')
0017 self.assertEqual(outcome('BNL_OSG_PanDA_1', 'failed', 143), 'failed')
0018 self.assertEqual(outcome('BNL_OSG_PanDA_1', 'cancelled', 143), 'cancelled')
0019
0020 def reporter(self):
0021 path = Path(__file__).resolve().parents[3] / 'scripts' / 'harvester-reporter.py'
0022 spec = importlib.util.spec_from_file_location('admission_reporter_test', path)
0023 module = importlib.util.module_from_spec(spec)
0024 spec.loader.exec_module(module)
0025 return module
0026
0027 def test_multi_image_capability_requires_fresh_active_allocator(self):
0028 reporter = self.reporter()
0029 with TemporaryDirectory() as tmp:
0030 root = Path(tmp)
0031 policy = root / 'image-policy.json'
0032 state = root / 'allocation-state.json'
0033 template = root / 'pilot.sdf'
0034 policy.write_text(json.dumps({'images': ['/cvmfs/one', '/cvmfs/three'], 'queue': 'OSG'}))
0035 template.write_text('environment = "EPICPROD_NATIVE_IMAGE=/cvmfs/one"\n'
0036 '+EpicprodImagePolicyFile = "' + str(policy) + '"\n')
0037 record = {'mode': 'active', 'images': ['/cvmfs/one', '/cvmfs/three'],
0038 'queue': 'OSG', 'observed_at': '2026-10-24T00:00:00Z',
0039 'valid_until': (datetime.utcnow() + timedelta(minutes=5)).strftime('%Y-%m-%dT%H:%M:%SZ')}
0040 state.write_text(json.dumps(record))
0041 self.assertEqual(reporter.pilot_dispatch(str(template))['images'], ['/cvmfs/one', '/cvmfs/three'])
0042 record['valid_until'] = '2020-01-01T00:00:00Z'
0043 state.write_text(json.dumps(record))
0044 self.assertEqual(reporter.pilot_dispatch(str(template))['mode'], 'unknown')
0045
0046 def test_per_job_dispatch_requires_a_launcher_that_implements_it(self):
0047 reporter = self.reporter()
0048 with TemporaryDirectory() as tmp:
0049 root = Path(tmp)
0050 launcher = root / 'epic-native-pilot.sh'
0051 template = root / 'pilot.sdf'
0052 template.write_text('environment = "EPICPROD_NATIVE_IMAGE=/cvmfs/bootstrap '
0053 'EPICPROD_NATIVE_DISPATCH=per-job EPICPROD_NATIVE_NESTING=on"\n'
0054 '+EpicprodNativeMode = "per-job"\n'
0055 'transfer_input_files = ' + str(launcher) + '\n')
0056 launcher.write_text('old launcher without dispatch support')
0057 self.assertEqual(reporter.pilot_dispatch(str(template))['mode'], 'unknown')
0058 launcher.write_text('case ${EPICPROD_NATIVE_DISPATCH:-bound} in\n'
0059 'per-job) unset PILOT_CONTAINER_NAME;;\nesac\n')
0060 result = reporter.pilot_dispatch(str(template))
0061 self.assertEqual(result['mode'], 'any')
0062 self.assertEqual(result['bootstrap_images'], ['/cvmfs/bootstrap'])
0063 self.assertEqual(result['images'], [])
0064 template.write_text(template.read_text().replace('NESTING=on', 'NESTING=off'))
0065 self.assertEqual(reporter.pilot_dispatch(str(template))['mode'], 'unknown')
0066
0067 def test_startup_evidence_includes_actual_slot_and_budget(self):
0068 reporter = self.reporter()
0069 with TemporaryDirectory() as tmp:
0070 stem = Path(tmp) / 'grid.123.0'
0071 stem.with_suffix('.0.err').write_text('ePIC runtime policy: PanDA queue=BNL_OSG_PanDA_1; execution queue=ComputeCanada-Fir; minimum=9000 s; source=default\nglidein life 4134 s is under the 9000 s a job needs; no job taken\n')
0072 stem.with_suffix('.0.out').write_text('ePIC native pilot: image /image, host fc20557, time, nesting off\n')
0073 stem.with_suffix('.0.log').write_text('Message from starter on slot1_2@glidein_123@fc20557:\nTimeSlotBusy (s) : 81\nTimeExecute (s) : 2\n')
0074 result = reporter.admission_evidence('https://logs/grid.123.0.log', [('https://logs', tmp)])
0075 self.assertEqual(result['execution_queue'], 'ComputeCanada-Fir')
0076 self.assertEqual(result['slot'], 'slot1_2@glidein_123@fc20557')
0077 self.assertEqual(result['remaining_s'], 4134)
0078 self.assertEqual(result['required_s'], 9000)
0079 self.assertEqual((result['busy_s'], result['execute_s']), (81, 2))
0080
0081 def test_expected_refusal_is_not_an_operational_error_reason(self):
0082 reporter = self.reporter()
0083 row = ['1', '123.0', 'BNL_OSG_PanDA_1', 'failed', 'completed',
0084 '2026-10-24 01:00:00', 'returned non-zero 73', None, None, None, '73']
0085 with patch.object(reporter, 'query', return_value=[row]), \
0086 patch.object(reporter, 'log_roots', return_value=[]), \
0087 patch.object(reporter, 'condor_reason') as read:
0088 report = reporter.not_finished({}, '2026-10-24 01:01:00', {})
0089 self.assertEqual(report['workers'], 0)
0090 self.assertEqual(report['sites'], {})
0091 read.assert_not_called()
