Merge pull request #448 from ExaWorks/fix_some_tests

Files used by jobs need to be on a shared FS. This is typically the case

Author: hategan

docs/user_guide.rst

@@ -311,7 +311,7 @@ following example shows:
 .. literalinclude:: ../tests/user_guide/test_prelaunch.py
     :language: python
     :dedent: 4
-    :lines: 12-18
+    :lines: 13-19
 
 where the contents of ``pre_launch.sh`` is
 

tests/_test_tools.py

@@ -1,6 +1,10 @@
+import os
+import shutil
+import tempfile
+from contextlib import contextmanager
 from datetime import timedelta
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Union, Iterator
 
 from executor_test_params import ExecutorTestParams
 
@@ -53,3 +57,18 @@ def _get_executor_instance(ep: ExecutorTestParams, job: Optional[Job] = None) ->
         if ep.queue_name is not None:
             job.spec.attributes.queue_name = ep.queue_name
     return JobExecutor.get_instance(ep.executor, url=ep.url)
+
+
+@contextmanager
+def _deploy(path: Union[Path, str]) -> Iterator[Path]:
+    # Copies `path` to a directory assumed to be on a shared FS (~/.psij/test) and
+    # returns the resulting path
+    if isinstance(path, str):
+        path = Path(path)
+    with tempfile.NamedTemporaryFile(dir=Path.home() / '.psij' / 'test', delete=False) as df:
+        try:
+            df.close()
+            shutil.copyfile(path, df.name)
+            yield Path(df.name)
+        finally:
+            os.unlink(df.name)

tests/test_nodefile.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from _test_tools import assert_completed, _get_executor_instance
+from _test_tools import assert_completed, _get_executor_instance, _deploy
 from executor_test_params import ExecutorTestParams
 from psij import Job, JobSpec, ResourceSpecV1
 
@@ -14,17 +14,20 @@
 def test_nodefile(execparams: ExecutorTestParams) -> None:
     if execparams.executor in NOT_TESTED:
         pytest.skip('This test does not work with %s' % execparams.executor)
+    if execparams.launcher == 'single':
+        pytest.skip('This test does not work with the single launcher')
 
     my_path = os.path.dirname(os.path.realpath(__file__))
 
     N_PROC = 4
     with TemporaryDirectory(dir=Path.home() / '.psij' / 'test') as td:
         outp = Path(td, 'stdout.txt')
-        spec = JobSpec('/bin/bash', [os.path.join(my_path, 'test_nodefile.sh'), str(N_PROC)],
-                       stdout_path=outp)
-        job = Job(spec)
-        spec.resources = ResourceSpecV1(process_count=N_PROC)
-        ex = _get_executor_instance(execparams, job)
-        ex.submit(job)
-        status = job.wait()
-        assert_completed(job, status)
+        with _deploy(os.path.join(my_path, 'test_nodefile.sh')) as excp:
+            spec = JobSpec('/bin/bash', [str(excp), str(N_PROC)],
+                           stdout_path=outp)
+            job = Job(spec)
+            spec.resources = ResourceSpecV1(process_count=N_PROC)
+            ex = _get_executor_instance(execparams, job)
+            ex.submit(job)
+            status = job.wait()
+            assert_completed(job, status)

tests/user_guide/test_prelaunch.py

@@ -1,20 +1,21 @@
 import os
 from pathlib import Path
 
-from _test_tools import assert_completed
+from _test_tools import assert_completed, _deploy
 from psij import Job, JobSpec, JobExecutor
 
 
 def test_user_guide_pre_launch() -> None:
     script_dir = os.path.dirname(os.path.realpath(__file__))
 
-    # START
-    ex = JobExecutor.get_instance('local')
-    spec = JobSpec('/bin/bash', ['-c', 'module is-loaded test'])
-    spec.pre_launch = Path(script_dir) / 'pre_launch.sh'
+    with _deploy(Path(script_dir) / 'pre_launch.sh') as pre_launch_sh_path:
+        # START
+        ex = JobExecutor.get_instance('local')
+        spec = JobSpec('/bin/bash', ['-c', 'module is-loaded test'])
+        spec.pre_launch = pre_launch_sh_path
 
-    job = Job(spec)
-    ex.submit(job)
-    status = job.wait()
-    # END
-    assert_completed(job, status)
+        job = Job(spec)
+        ex.submit(job)
+        status = job.wait()
+        # END
+        assert_completed(job, status)