- fix format problems

- add test examples
- copy_artifact supports for sort

Signed-off-by: zjgemi <[email protected]>

Author: zjgemi

CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## 1.1.12
+
+- fix format problems
+- add test examples
+- copy_artifact supports for sort
+
 ## 1.1.11
 
 ### Added

VERSION

@@ -1 +1 @@
-1.1.11
+1.1.12

examples/test_async.py

@@ -1,28 +1,29 @@
-from dflow import (
-    ShellOPTemplate,
-    InputArtifact,
-    OutputArtifact,
-    Workflow,
-    Step
-)
 import time
 
+from dflow import (InputArtifact, OutputArtifact, ShellOPTemplate, Step,
+                   Workflow)
+
 if __name__ == "__main__":
     hello = ShellOPTemplate(name='Hello',
-            image="alpine:latest",
-            script="sleep 60 && echo Hello > /tmp/bar.txt")
+                            image="alpine:latest",
+                            script="sleep 60 && echo Hello > /tmp/bar.txt")
     hello.outputs.artifacts = {"bar": OutputArtifact(path="/tmp/bar.txt")}
 
     echo = ShellOPTemplate(name='Echo',
-                image="alpine:latest",
-                script="while [ 1 ]; do echo 'waiting...'; if [ -f /tmp/foo.txt ]; then cat /tmp/foo.txt; break; fi; sleep 1; done")
+                           image="alpine:latest",
+                           script="while [ 1 ]; do echo 'waiting...';"
+                           " if [ -f /tmp/foo.txt ]; then cat /tmp/foo.txt;"
+                           " break; fi; sleep 1; done")
     echo.inputs.artifacts = {"foo": InputArtifact(path="/tmp/foo.txt")}
 
     wf = Workflow(name="async")
     step0 = Step(name="hello", template=hello)
-    # This step will give output artifact "bar" which contains "Hello" after 60 seconds
-    step1 = Step(name="echo", template=echo, artifacts={"foo": step0.outputs.artifacts["bar"].pvc()})
-    # This step will wait the last step to finish and then print its output artifact
+    # This step will give output artifact "bar" which contains "Hello" after
+    # 60 seconds
+    step1 = Step(name="echo", template=echo, artifacts={
+                 "foo": step0.outputs.artifacts["bar"].pvc()})
+    # This step will wait the last step to finish and then print its output
+    # artifact
     wf.add([step0, step1])
     wf.submit()
 

examples/test_big_parameter.py

@@ -1,43 +1,38 @@
 import time
-from dflow import (
-    Workflow,
-    Step,
-    Steps,
-    InputParameter,
-    OutputParameter
-)
-from dflow.python import (
-    PythonOPTemplate,
-    OP,
-    OPIO,
-    OPIOSign,
-    BigParameter
-)
+
+from dflow import InputParameter, OutputParameter, Step, Steps, Workflow
+from dflow.python import (OP, OPIO, BigParameter, OPIOSign, PythonOPTemplate,
+                          upload_packages)
+
+if "__file__" in locals():
+    upload_packages.append(__file__)
+
 
 class Hello:
     def __init__(self, msg):
         self.msg = msg
 
+
 class Duplicate(OP):
     def __init__(self):
         pass
 
     @classmethod
     def get_input_sign(cls):
         return OPIOSign({
-            'foo' : BigParameter(Hello)
+            'foo': BigParameter(Hello)
         })
 
     @classmethod
     def get_output_sign(cls):
         return OPIOSign({
-            'foo' : BigParameter(Hello)
+            'foo': BigParameter(Hello)
         })
 
     @OP.exec_sign_check
     def execute(
             self,
-            op_in : OPIO,
+            op_in: OPIO,
     ) -> OPIO:
         foo = op_in["foo"]
         print(foo.msg)
@@ -47,32 +42,35 @@ def execute(
         })
         return op_out
 
-if __name__ == "__main__":
+
+def test_big_parameter():
     wf = Workflow(name="big-param")
 
     steps = Steps(name="hello-steps")
     steps.inputs.parameters["foo"] = InputParameter()
     steps.outputs.parameters["foo"] = OutputParameter()
 
     step1 = Step(
-        name="step1", 
+        name="step1",
         template=PythonOPTemplate(Duplicate, image="python:3.8"),
         parameters={"foo": steps.inputs.parameters["foo"]},
         key="step1"
     )
     steps.add(step1)
 
     step2 = Step(
-        name="step2", 
+        name="step2",
         template=PythonOPTemplate(Duplicate, image="python:3.8"),
         parameters={"foo": step1.outputs.parameters["foo"]},
         key="step2"
     )
     steps.add(step2)
 
-    steps.outputs.parameters["foo"].value_from_parameter = step2.outputs.parameters["foo"]
+    steps.outputs.parameters["foo"].value_from_parameter = \
+        step2.outputs.parameters["foo"]
 
-    big_step = Step(name="big-step", template=steps, parameters={"foo": Hello("hello")})
+    big_step = Step(name="big-step", template=steps,
+                    parameters={"foo": Hello("hello")})
     wf.add(big_step)
     wf.submit()
 
@@ -88,3 +86,7 @@ def execute(
     wf = Workflow(name="big-param-resubmit")
     wf.add(big_step)
     wf.submit(reuse_step=[step])
+
+
+if __name__ == "__main__":
+    test_big_parameter()

examples/test_conditional_outputs.py

@@ -1,20 +1,14 @@
-from dflow import (
-    Workflow,
-    Step,
-    Steps,
-    if_expression,
-    Outputs,
-    OutputArtifact,
-    OutputParameter
-)
-from dflow.python import (
-    PythonOPTemplate,
-    OP,
-    OPIO,
-    OPIOSign,
-    Artifact
-)
 import random
+import time
+
+from dflow import (OutputArtifact, OutputParameter, Outputs, Step, Steps,
+                   Workflow, if_expression)
+from dflow.python import (OP, OPIO, Artifact, OPIOSign, PythonOPTemplate,
+                          upload_packages)
+
+if "__file__" in locals():
+    upload_packages.append(__file__)
+
 
 class Random(OP):
     @classmethod
@@ -34,7 +28,7 @@ def get_output_sign(cls):
     @OP.exec_sign_check
     def execute(
             self,
-            op_in : OPIO,
+            op_in: OPIO,
     ) -> OPIO:
         open("foo.txt", "w").write("head")
         open("bar.txt", "w").write("tail")
@@ -50,24 +44,36 @@ def execute(
             "bar": "bar.txt"
         })
 
-if __name__ == "__main__":
-    steps = Steps("conditional-steps", outputs=Outputs(parameters={"msg": OutputParameter()},
-            artifacts={"res": OutputArtifact()}))
 
-    random = Step(
-        name="random", 
+def test_conditional_outputs():
+    steps = Steps("conditional-steps", outputs=Outputs(
+        parameters={"msg": OutputParameter()},
+        artifacts={"res": OutputArtifact()}))
+
+    random_step = Step(
+        name="random",
         template=PythonOPTemplate(Random, image="python:3.8")
     )
-    steps.add(random)
+    steps.add(random_step)
 
     steps.outputs.parameters["msg"].value_from_expression = if_expression(
-            _if=random.outputs.parameters["is_head"] == True,
-            _then=random.outputs.parameters["msg1"], _else=random.outputs.parameters["msg2"])
+        _if=random_step.outputs.parameters["is_head"],
+        _then=random_step.outputs.parameters["msg1"],
+        _else=random_step.outputs.parameters["msg2"])
 
     steps.outputs.artifacts["res"].from_expression = if_expression(
-            _if=random.outputs.parameters["is_head"] == True,
-            _then=random.outputs.artifacts["foo"], _else=random.outputs.artifacts["bar"])
+        _if=random_step.outputs.parameters["is_head"],
+        _then=random_step.outputs.artifacts["foo"],
+        _else=random_step.outputs.artifacts["bar"])
 
     wf = Workflow(name="conditional", steps=steps)
 
     wf.submit()
+    while wf.query_status() in ["Pending", "Running"]:
+        time.sleep(1)
+
+    assert(wf.query_status() == "Succeeded")
+
+
+if __name__ == "__main__":
+    test_conditional_outputs()

examples/test_dag.py

@@ -1,36 +1,33 @@
-from dflow import (
-    ShellOPTemplate,
-    InputParameter,
-    OutputParameter,
-    InputArtifact,
-    OutputArtifact,
-    Workflow,
-    Task,
-    DAG,
-)
+from dflow import (DAG, InputArtifact, InputParameter, OutputArtifact,
+                   OutputParameter, ShellOPTemplate, Task, Workflow)
 
 if __name__ == "__main__":
-    hello = ShellOPTemplate(name='Hello',
-            image="alpine:latest",
-            script="echo Hello > /tmp/bar.txt && echo 1 > /tmp/result.txt")
-    hello.outputs.parameters = {"msg": OutputParameter(value_from_path="/tmp/result.txt")}
+    hello = ShellOPTemplate(
+        name='Hello',
+        image="alpine:latest",
+        script="echo Hello > /tmp/bar.txt && echo 1 > /tmp/result.txt")
+    hello.outputs.parameters = {"msg": OutputParameter(
+        value_from_path="/tmp/result.txt")}
     hello.outputs.artifacts = {"bar": OutputArtifact(path="/tmp/bar.txt")}
 
-    duplicate = ShellOPTemplate(name='Duplicate',
-                image="alpine:latest",
-                script="cat /tmp/foo.txt /tmp/foo.txt > /tmp/bar.txt && echo $(({{inputs.parameters.msg}}*2)) > /tmp/result.txt")
+    duplicate = ShellOPTemplate(
+        name='Duplicate',
+        image="alpine:latest",
+        script="cat /tmp/foo.txt /tmp/foo.txt > /tmp/bar.txt && "
+        "echo $(({{inputs.parameters.msg}}*2)) > /tmp/result.txt")
     duplicate.inputs.parameters = {"msg": InputParameter()}
-    duplicate.outputs.parameters = {"msg": OutputParameter(value_from_path="/tmp/result.txt")}
+    duplicate.outputs.parameters = {"msg": OutputParameter(
+        value_from_path="/tmp/result.txt")}
     duplicate.inputs.artifacts = {"foo": InputArtifact(path="/tmp/foo.txt")}
     duplicate.outputs.artifacts = {"bar": OutputArtifact(path="/tmp/bar.txt")}
 
     dag = DAG()
     hello0 = Task(name="hello0", template=hello)
     dag.add(hello0)
     hello1 = Task(name="hello1",
-            template=duplicate,
-            parameters={"msg": hello0.outputs.parameters["msg"]},
-            artifacts={"foo": hello0.outputs.artifacts["bar"]})
+                  template=duplicate,
+                  parameters={"msg": hello0.outputs.parameters["msg"]},
+                  artifacts={"foo": hello0.outputs.artifacts["bar"]})
     dag.add(hello1)
 
     wf = Workflow(name="dag", dag=dag)