more debugging of singularity

Author: tclose

pydra/design/shell.py

@@ -345,6 +345,9 @@ def make(
             klass = None
             input_helps, output_helps = {}, {}
 
+            if isinstance(wrapped, list):
+                wrapped = " ".join(wrapped)
+
             executable, inferred_inputs, inferred_outputs = parse_command_line_template(
                 wrapped,
                 inputs=inputs,

pydra/engine/submitter.py

@@ -211,16 +211,20 @@ def Split(defn: TaskDef, output_types: dict):
 
             task_def = Split(defn=task_def, output_types=output_types)
 
+            environment = None
         elif task_def._combiner:
             raise ValueError(
                 f"Task {self} is marked for combining, but not splitting. "
                 "Use the `split` method to split the task before combining."
             )
+        else:
+            environment = self.environment
+
         task = Task(
             task_def,
             submitter=self,
             name="main",
-            environment=self.environment,
+            environment=environment,
             hooks=hooks,
         )
         try:

pydra/engine/tests/test_environments.py

@@ -34,18 +34,18 @@ def newcache(x):
 
     shelly_job = Task(
         definition=shelly,
-        submitter=Submitter(cache_dir=newcache("shelly-task")),
-        name="shelly",
+        submitter=Submitter(cache_dir=newcache("native-task")),
+        name="native",
     )
     env_outputs = Native().execute(shelly_job)
 
-    outputs = shelly(cache_dir=newcache("shelly-exec"))
+    outputs = shelly(cache_dir=newcache("native-exec"))
     assert drop_stderr(env_outputs) == drop_stderr(attrs_values(outputs))
 
-    outputs = shelly(environment=Native(), cache_dir=newcache("shelly-call"))
+    outputs = shelly(environment=Native(), cache_dir=newcache("native-call"))
     assert env_outputs == attrs_values(outputs)
 
-    with Submitter(cache_dir=newcache("shelly-submitter"), environment=Native()) as sub:
+    with Submitter(cache_dir=newcache("native-submitter"), environment=Native()) as sub:
         result = sub(shelly)
     assert drop_stderr(env_outputs) == drop_stderr(attrs_values(result.outputs))
 
@@ -66,15 +66,15 @@ def newcache(x):
 
     shelly_job = Task(
         definition=shelly,
-        submitter=Submitter(cache_dir=newcache("shelly")),
-        name="shelly",
+        submitter=Submitter(cache_dir=newcache("docker")),
+        name="docker",
     )
     outputs_dict = docker.execute(shelly_job)
 
-    with Submitter(cache_dir=newcache("shelly_sub"), environment=docker) as sub:
+    with Submitter(cache_dir=newcache("docker_sub"), environment=docker) as sub:
         result = sub(shelly)
 
-    outputs = shelly(environment=docker, cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=docker, cache_dir=newcache("docker_call"))
     # If busybox isn't found locally, then the stderr will have the download progress from
     # the Docker auto-pull in it
     for key in ["stdout", "return_code"]:
@@ -106,19 +106,19 @@ def newcache(x):
     shelly = shell.define(cmd)()
     shelly_job = Task(
         definition=shelly,
-        submitter=Submitter(cache_dir=newcache("shelly")),
-        name="shelly",
+        submitter=Submitter(cache_dir=newcache("docker")),
+        name="docker",
     )
     assert shelly.cmdline == cmd
     outputs_dict = docker.execute(shelly_job)
 
     with Submitter(
-        worker="cf", cache_dir=newcache("shelly_sub"), environment=docker
+        worker="cf", cache_dir=newcache("docker_sub"), environment=docker
     ) as sub:
         result = sub(shelly)
     assert outputs_dict == attrs_values(result.outputs)
 
-    outputs = shelly(cache_dir=newcache("shelly_call"), environment=docker)
+    outputs = shelly(cache_dir=newcache("docker_call"), environment=docker)
     assert outputs_dict == attrs_values(outputs)
 
 
@@ -136,17 +136,17 @@ def newcache(x):
     shelly = Shelly()
     shelly_job = Task(
         definition=shelly,
-        submitter=Submitter(cache_dir=newcache("shelly")),
-        name="shelly",
+        submitter=Submitter(cache_dir=newcache("singu")),
+        name="singu",
     )
     assert shelly.cmdline == cmd
     outputs_dict = sing.execute(shelly_job)
 
-    with Submitter(cache_dir=newcache("shelly_sub"), environment=sing) as sub:
+    with Submitter(cache_dir=newcache("singu_sub"), environment=sing) as sub:
         results = sub(shelly)
     assert drop_stderr(outputs_dict) == drop_stderr(attrs_values(results.outputs))
 
-    outputs = shelly(environment=sing, cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=sing, cache_dir=newcache("singu_call"))
     assert drop_stderr(outputs_dict) == drop_stderr(attrs_values(outputs))
 
 
@@ -164,19 +164,19 @@ def newcache(x):
     shelly = Shelly()
     shelly_job = Task(
         definition=shelly,
-        submitter=Submitter(cache_dir=newcache("shelly")),
-        name="shelly",
+        submitter=Submitter(cache_dir=newcache("singu")),
+        name="singu",
     )
     assert shelly.cmdline == cmd
     outputs_dict = sing.execute(shelly_job)
 
     with Submitter(
-        worker=plugin, environment=sing, cache_dir=newcache("shelly_sub")
+        worker=plugin, environment=sing, cache_dir=newcache("singu_sub")
     ) as sub:
         results = sub(shelly)
     assert drop_stderr(outputs_dict) == drop_stderr(attrs_values(results.outputs))
 
-    outputs = shelly(environment=sing, cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=sing, cache_dir=newcache("singu_call"))
     # singularity gives info about cashed image in stderr
     assert drop_stderr(outputs_dict) == drop_stderr(attrs_values(outputs))
 
@@ -218,14 +218,14 @@ def newcache(x):
         f.write("hello ")
 
     shelly = shelly_with_input_factory(filename=filename, executable="cat")
-    shelly_job = make_job(shelly, tmp_path, "shelly")
+    shelly_job = make_job(shelly, tmp_path, "native")
     outputs_dict = Native().execute(shelly_job)
 
-    with Submitter(environment=Native(), cache_dir=newcache("shelly_sub")) as sub:
+    with Submitter(environment=Native(), cache_dir=newcache("native_sub")) as sub:
         results = sub(shelly)
     assert outputs_dict == attrs_values(results.outputs)
 
-    outputs = shelly(environment=Native(), cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=Native(), cache_dir=newcache("native_call"))
     assert outputs_dict == attrs_values(outputs)
 
 
@@ -247,12 +247,12 @@ def newcache(x):
     filename = [filename_1, filename_2]
 
     shelly = shelly_with_input_factory(filename=None, executable="cat")
-    with Submitter(environment=Native(), cache_dir=newcache("shelly")) as sub:
+    with Submitter(environment=Native(), cache_dir=newcache("native")) as sub:
         results = sub(shelly.split(file=filename))
     assert [s.strip() for s in results.outputs.stdout] == ["hello", "hi"]
 
     outputs = shelly.split(file=filename)(
-        environment=Native(), cache_dir=newcache("shelly_call")
+        environment=Native(), cache_dir=newcache("native_call")
     )
     assert [s.strip() for s in outputs.stdout] == ["hello", "hi"]
 
@@ -273,14 +273,14 @@ def newcache(x):
         f.write("hello ")
 
     shelly = shelly_with_input_factory(filename=filename, executable="cat")
-    outputs_dict = docker.execute(make_job(shelly, tmp_path, "shelly"))
+    outputs_dict = docker.execute(make_job(shelly, tmp_path, "docker"))
 
     with Submitter(environment=docker, cache_dir=newcache("shell_sub")) as sub:
         results = sub(shelly)
 
     assert outputs_dict == attrs_values(results.outputs)
 
-    outputs = shelly(environment=docker, cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=docker, cache_dir=newcache("docker_call"))
     assert outputs_dict == attrs_values(outputs)
 
 
@@ -300,18 +300,18 @@ def newcache(x):
         f.write("hello ")
 
     shelly = shelly_with_input_factory(filename=filename, executable="cat")
-    shelly_job = make_job(shelly, tmp_path, "shelly_job")
+    shelly_job = make_job(shelly, tmp_path, "docker_job")
     outputs_dict = docker.execute(shelly_job)
 
     with Submitter(
-        environment=docker, cache_dir=newcache("shelly_sub"), worker=plugin
+        environment=docker, cache_dir=newcache("docker_sub"), worker=plugin
     ) as sub:
         results = sub(shelly)
     with Submitter(worker=plugin) as sub:
         results = sub(shelly)
     assert outputs_dict == attrs_values(results.outputs)
 
-    outputs = shelly(environment=docker, cache_dir=newcache("shelly_call"))
+    outputs = shelly(environment=docker, cache_dir=newcache("docker_call"))
     assert outputs_dict == attrs_values(outputs)
 
 
@@ -338,13 +338,13 @@ def newcache(x):
 
     shelly = shelly_with_input_factory(filename=None, executable="cat")
 
-    with Submitter(environment=docker, cache_dir=newcache("shelly_sub")) as sub:
+    with Submitter(environment=docker, cache_dir=newcache("docker_sub")) as sub:
         results = sub(shelly.split(file=filename))
 
     assert [s.strip() for s in results.outputs.stdout] == ["hello", "hi"]
 
     outputs = shelly.split(file=filename)(
-        environment=docker, cache_dir=newcache("shelly_call")
+        environment=docker, cache_dir=newcache("docker_call")
     )
     assert [s.strip() for s in outputs.stdout] == ["hello", "hi"]
 
@@ -391,16 +391,16 @@ def newcache(x):
 
     # execute does not create the cashedir, so this part will fail,
     # but I guess we don't want to use it this way anyway
-    # shelly = create_shelly_outputfile(tempdir=tmp_path, filename=filename, name="shelly")
+    # shelly = create_shelly_outputfile(tempdir=tmp_path, filename=filename, name="native")
     # outputs_dict = Native().execute(shelly)
 
     shelly = shelly_outputfile_factory(filename=filename)
 
-    with Submitter(environment=Native(), cache_dir=newcache("shelly_sub")) as sub:
+    with Submitter(environment=Native(), cache_dir=newcache("native_sub")) as sub:
         result = sub(shelly)
     assert Path(result.outputs.file_copy) == result.output_dir / "file_copy.txt"
 
-    call_cache = newcache("shelly_call")
+    call_cache = newcache("native_call")
 
     outputs = shelly(environment=Native(), cache_dir=call_cache)
     assert Path(outputs.file_copy) == call_cache / shelly._checksum / "file_copy.txt"
@@ -424,15 +424,15 @@ def newcache(x):
     filename = [filename_1, filename_2]
 
     shelly = shelly_outputfile_factory(filename=None)
-    with Submitter(environment=Native(), cache_dir=newcache("shelly")) as sub:
+    with Submitter(environment=Native(), cache_dir=newcache("native")) as sub:
         results = sub(shelly.split(file_orig=filename))
 
     assert [f.name for f in results.outputs.file_copy] == [
         "file_1_copy.txt",
         "file_2_copy.txt",
     ]
 
-    call_cache = newcache("shelly_call")
+    call_cache = newcache("native_call")
 
     outputs = shelly.split(file_orig=filename)(
         environment=Native(), cache_dir=call_cache
@@ -457,7 +457,7 @@ def newcache(x):
 
     shelly = shelly_outputfile_factory(filename=filename)
 
-    with Submitter(environment=docker_env, cache_dir=newcache("shelly")) as sub:
+    with Submitter(environment=docker_env, cache_dir=newcache("docker")) as sub:
         results = sub(shelly)
     assert results.outputs.file_copy == File(results.output_dir / "file_copy.txt")
 
@@ -485,7 +485,7 @@ def newcache(x):
 
     shelly = shelly_outputfile_factory(filename=None)
 
-    with Submitter(environment=docker_env, cache_dir=newcache("shelly_sub")) as sub:
+    with Submitter(environment=docker_env, cache_dir=newcache("docker_sub")) as sub:
         results = sub(shelly.split(file_orig=filename))
     assert [f.name for f in results.outputs.file_copy] == [
         "file_1_copy.txt",