swf-monitor/tests/test_native_es_worker_proxy.py

0001 import importlib.util
0002 import json
0003 import os
0004 import subprocess
0005 import sys
0006 from pathlib import Path
0007 import tempfile
0008 import unittest
0009 from unittest.mock import patch
0010
0011 ROOT = Path(__file__).resolve().parents[1]
0012
0013
0014 def load(name, file):
0015     spec = importlib.util.spec_from_file_location(name, ROOT / 'scripts' / file)
0016     module = importlib.util.module_from_spec(spec)
0017     spec.loader.exec_module(module)
0018     return module
0019
0020
0021 dispatcher = load('es_dispatcher', 'evgen_job_dispatcher.py')
0022 submitter = load('es_submitter', 'submit-evgen-task.py')
0023
0024
0025 class NativeWorkerProxy(unittest.TestCase):
0026     def test_credential_appears_only_on_worker_and_canary_outputs_stay_expiring(self):
0027         with tempfile.TemporaryDirectory() as directory:
0028             root = Path(directory)
0029             proxy = root / 'native-proxy'; proxy.write_text('PRIVATE_PROXY_FIXTURE')
0030             payload = root / 'payload'; (payload / 'es').mkdir(parents=True)
0031             (payload / 'VERSION').write_text('test')
0032             (payload / 'es/es_harness.py').write_text('import json,sys\nfrom pathlib import Path\n'
0033                 'p=Path(sys.argv[sys.argv.index("--summary")+1]);p.parent.mkdir(parents=True,exist_ok=True)\n'
0034                 'p.write_text(json.dumps({"args":sys.argv[1:]}))\n')
0035             spec = {'csvBase': 'native-test', 'csvRows': ['input,steer,20,0000'],
0036                     'env': {}, 'workerProxy': True, 'nEventsPerWorker': 5}
0037             with patch.object(submitter, '_reporting_env', return_value={}):
0038                 work = Path(submitter._assemble_sandbox(spec, str(proxy), str(root / 'work'),
0039                                                         payload_dir=str(payload)))
0040             sandbox = work / 'sandbox'
0041             self.assertFalse(any(b'PRIVATE_PROXY_FIXTURE' in p.read_bytes()
0042                                  for p in sandbox.rglob('*') if p.is_file()))
0043             self.assertIn('X509_USER_PROXY=worker-proxy',
0044                           (sandbox / 'environment-native-test.sh').read_text())
0045             with patch.dict(os.environ, {'ES_WORKER_PROXY': '1', 'PILOT_X509_USER_PROXY': str(proxy),
0046                                           'ES_PAYLOAD_IMAGE': 'image', 'ES_SLOTS': '4'}, clear=True), \
0047                     patch.object(dispatcher, 'write_job_report'):
0048                 rc = dispatcher.run_event_service('native-test', 'native-test', str(sandbox), 'channel')
0049             self.assertEqual(rc, 0)
0050             self.assertEqual((sandbox / 'worker-proxy').read_text(), 'PRIVATE_PROXY_FIXTURE')
0051             self.assertEqual((sandbox / 'worker-proxy').stat().st_mode & 0o777, 0o600)
0052             # The close starts with --cleanenv, then sources the environment
0053             # file: both credentials must resolve in that new process.
0054             check = subprocess.run(
0055                 ['/bin/bash', '-c',
0056                  'source environment-native-test.sh\nexec "$1" -c "$2"',
0057                  'close-test', sys.executable,
0058                  'import os,json;print(json.dumps({k:os.path.realpath(os.environ[k]) '
0059                  'for k in ["X509_USER_PROXY","PILOT_X509_USER_PROXY"]}))'],
0060                 cwd=sandbox, env={'PATH': os.environ['PATH']},
0061                 capture_output=True, text=True, check=True)
0062             self.assertEqual(json.loads(check.stdout), {
0063                 'X509_USER_PROXY': str(sandbox / 'worker-proxy'),
0064                 'PILOT_X509_USER_PROXY': str(sandbox / 'worker-proxy')})
0065             args = json.loads((work / 'es_work/es_summary.json').read_text())['args']
0066             self.assertIn('CANARY_OUTPUT_DATASET=TEST/canary/native-test', args)
0067             self.assertIn('CANARY_LIFETIME_S=604800', args)
0068
0069     def test_missing_native_credential_refuses_without_a_fallback(self):
0070         with patch.dict(os.environ, {'ES_WORKER_PROXY': '1', 'ES_PAYLOAD_IMAGE': 'image'}, clear=True):
0071             self.assertEqual(dispatcher.run_event_service('test', 'test', '/not-used', 'channel'), 2)