Add test to check that some things in spec/attrs really make it to the

submit script for batch scheduler executors.

Author: hategan

tests/test_executor.py

@@ -1,11 +1,16 @@
+import secrets
 import uuid
 from pathlib import Path
+from typing import List, Callable, Dict
 
-from psij import SubmitException, Job, JobSpec, JobState
-from tempfile import TemporaryDirectory
+import pytest
+
+from psij import SubmitException, Job, JobSpec, JobState, JobExecutor, JobAttributes, ResourceSpecV1
+from tempfile import TemporaryDirectory, TemporaryFile
 
 from executor_test_params import ExecutorTestParams
 from _test_tools import _get_executor_instance, _get_timeout, assert_completed, _make_test_dir
+from psij.executors.batch.batch_scheduler_executor import BatchSchedulerExecutor
 
 
 def test_simple_job(execparams: ExecutorTestParams) -> None:
@@ -161,3 +166,58 @@ def test_list(execparams: ExecutorTestParams) -> None:
     assert job.native_id is not None
     ids = ex.list()
     assert job.native_id in ids
+
+
+def _get_batch_executors() -> List[str]:
+    r = []
+    for name in JobExecutor.get_executor_names():
+        try:
+            ex = JobExecutor.get_instance(name)
+            if isinstance(ex, BatchSchedulerExecutor):
+                r.append(name)
+        except Exception:
+            pass
+    return r
+
+
+def _check_str_attrs(ex: BatchSchedulerExecutor, job: Job, names: List[str],
+                     l: Callable[[str, str], None]) -> None:
+    for name in names:
+        tok = secrets.token_hex()
+        l(name, tok)
+        with TemporaryFile(mode='w+') as f:
+            ex.generate_submit_script(job, ex._create_script_context(job), f)
+            f.seek(0)
+            contents = f.read()
+            if contents.find(tok) == -1:
+                print('Failed to find "%s" in:\n%s' % (tok, contents))
+                pytest.fail('Script generation failed for %s' % name)
+
+
+_PREFIX_TR = {'pbspro': 'pbs'}
+def _get_attr_prefix(exec_name: str) -> str:
+    if exec_name in _PREFIX_TR:
+        return _PREFIX_TR[exec_name]
+    else:
+        return exec_name
+
+
+@pytest.mark.parametrize('exec_name', _get_batch_executors())
+def test_submit_script_generation(exec_name: str) -> None:
+    ex = JobExecutor.get_instance(exec_name)
+    assert isinstance(ex, BatchSchedulerExecutor)
+
+    c_attrs: Dict[str, object] = {}
+    attrs = JobAttributes(custom_attributes=c_attrs)
+    res = ResourceSpecV1()
+    spec = JobSpec(resources=res, attributes=attrs)
+    spec.launcher = 'single'
+    job = Job(spec=spec)
+
+    prefix = _get_attr_prefix(exec_name)
+    _check_str_attrs(ex, job, ['executable', 'directory'],
+                     lambda k, v: setattr(spec, k, v))
+    _check_str_attrs(ex, job, ['queue_name', 'project_name', 'reservation_id'],
+                     lambda k, v: setattr(attrs, k, v))
+    _check_str_attrs(ex, job, [prefix + '.cust_attr1', prefix + '.cust_attr2'],
+                     lambda k, v: c_attrs.__setitem__(k, v))