Commit 979d8f02 authored by Paul McCarthy's avatar Paul McCarthy 🚵
Browse files

New/adjusted unit tests for run(submit=True) and cmd/fslwrapper(submit=True)

parent 29cbf562
Pipeline #1866 failed with stages
in 16 minutes and 38 seconds
......@@ -15,6 +15,7 @@ try: from unittest import mock
# python 2
except ImportError: import mock
import six
import pytest
import fsl.utils.tempdir as tempdir
......@@ -22,7 +23,7 @@ from fsl.utils.platform import platform as fslplatform
import fsl.utils.run as run
import fsl.utils.fslsub as fslsub
from . import make_random_image, mockFSLDIR
from . import make_random_image, mockFSLDIR, touch
def test_run():
......@@ -142,16 +143,25 @@ def test_runfsl():
fslplatform.fsldir = old_fsldir
mock_fsl_sub = textwrap.dedent("""
#!/usr/bin/env bash
jid=12345
cmd=$1
name=`basename $cmd`
$cmd > "$name".o"$jid"
touch "$name".e"$jid"
echo $jid
exit 0
""").strip()
def mock_submit(cmd, **kwargs):
if isinstance(cmd, six.string_types):
name = cmd.split()[0]
else:
name = cmd[0]
name = op.basename(name)
jid = '12345'
output = run.run(cmd)
with open('{}.o{}'.format(name, jid), 'wt') as f:
f.write(output)
with open('{}.e{}'.format(name, jid), 'wt') as f:
for k, v in kwargs.items():
f.write('{}: {}\n'.format(k, v))
return (jid,)
def test_run_submit():
......@@ -167,16 +177,30 @@ def test_run_submit():
exit 0
""").strip()
with tempdir.tempdir(), mockFSLDIR():
with tempdir.tempdir(), \
mockFSLDIR(), \
mock.patch('fsl.utils.fslsub.submit', mock_submit):
mkexec(op.expandvars('$FSLDIR/bin/fsltest'), test_script)
mkexec(op.expandvars('$FSLDIR/bin/fsl_sub'), mock_fsl_sub)
jid = run.run('fsltest', submit=True)[0]
assert jid == '12345'
stdout, stderr = fslsub.output(jid, 'fsltest')
stdout, stderr = fslsub.output(jid)
assert stdout.strip() == 'test_script running'
assert stderr.strip() == ''
kwargs = {'name' : 'abcde', 'ram' : '4GB'}
jid = run.run('fsltest', submit=kwargs)[0]
assert jid == '12345'
stdout, stderr = fslsub.output(jid)
experr = '\n'.join(['{}: {}'.format(k, v) for k, v in kwargs.items()])
assert stdout.strip() == 'test_script running'
assert stderr.strip() == experr
......@@ -8,6 +8,10 @@
import os.path as op
import os
import shlex
import textwrap
try: from unittest import mock
except ImportError: import mock
import pytest
......@@ -16,10 +20,12 @@ import nibabel as nib
import fsl.utils.tempdir as tempdir
import fsl.utils.run as run
import fsl.utils.fslsub as fslsub
import fsl.wrappers.wrapperutils as wutils
from . import mockFSLDIR
from .test_run import mock_submit
def test_applyArgStyle():
......@@ -310,7 +316,6 @@ def test_cmdwrapper():
assert func(1, 2)[0] == 'func 1 2'
def test_fslwrapper():
@wutils.fslwrapper
def func(a, b):
......@@ -319,3 +324,73 @@ def test_fslwrapper():
with run.dryrun(), mockFSLDIR() as fsldir:
expected = '{} 1 2'.format(op.join(fsldir, 'bin', 'func'))
assert func(1, 2)[0] == expected
_test_script = textwrap.dedent("""
#!/usr/bin/env bash
echo "test_script running: $1 $2"
exit 0
""").strip()
def _test_script_func(a, b):
return ['test_script', str(a), str(b)]
def test_cmdwrapper_submit():
test_func = wutils.cmdwrapper(_test_script_func)
newpath = op.pathsep.join(('.', os.environ['PATH']))
with tempdir.tempdir(), \
mock.patch('fsl.utils.fslsub.submit', mock_submit), \
mock.patch.dict(os.environ, {'PATH' : newpath}):
with open('test_script', 'wt') as f:
f.write(_test_script)
os.chmod('test_script', 0o755)
jid = test_func(1, 2, submit=True)
assert jid == ('12345',)
stdout, stderr = fslsub.output('12345')
assert stdout.strip() == 'test_script running: 1 2'
assert stderr.strip() == ''
def test_fslwrapper_submit():
test_func = wutils.fslwrapper(_test_script_func)
with mockFSLDIR() as fsldir, \
mock.patch('fsl.utils.fslsub.submit', mock_submit):
test_file = op.join(fsldir, 'bin', 'test_script')
with open(test_file, 'wt') as f:
f.write(_test_script)
os.chmod(test_file, 0o755)
jid = test_func(1, 2, submit=True)
assert jid == ('12345',)
stdout, stderr = fslsub.output('12345')
assert stdout.strip() == 'test_script running: 1 2'
assert stderr.strip() == ''
kwargs = {'name' : 'abcde', 'ram' : '4GB'}
jid = test_func(1, 2, submit=kwargs)
assert jid == ('12345',)
stdout, stderr = fslsub.output('12345')
experr = '\n'.join(['{}: {}'.format(k, v) for k, v in kwargs.items()])
assert stdout.strip() == 'test_script running: 1 2'
assert stderr.strip() == experr
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment