Codelab: WDL assertion statements

WDL/Tree.py

@@ -234,6 +234,14 @@ def _workflow_node_dependencies(self) -> Iterable[str]:
         yield from _expr_workflow_node_dependencies(self.expr)
 
 
+class Assertion(Decl):
+    message: str
+
+    def __init__(self, pos: SourcePosition, expr: Expr.Base) -> None:
+        super().__init__(pos, Type.Boolean(), f"_assert_L{pos.line}C{pos.column}", expr)
+        self.message = f"assertion failed: {str(expr)} ({pos.uri} Ln {pos.line} Col {pos.column})"
+
+
 class Task(SourceNode):
     """
     WDL Task

WDL/_grammar.py

@@ -289,11 +289,11 @@
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 
 workflow: "workflow" CNAME "{" workflow_element* "}"
-?workflow_element: input_decls | any_decl | call | scatter | conditional | workflow_outputs | meta_section
+?workflow_element: input_decls | any_decl | call | scatter | conditional | workflow_outputs | meta_section | assertion
 
 scatter: "scatter" "(" CNAME "in" expr ")" "{" inner_workflow_element* "}"
 conditional: "if" "(" expr ")" "{" inner_workflow_element* "}"
-?inner_workflow_element: any_decl | call | scatter | conditional
+?inner_workflow_element: any_decl | call | scatter | conditional | assertion
 
 call: "call" namespaced_ident ("after" CNAME)* _call_body? -> call
     | "call" namespaced_ident "as" CNAME ("after" CNAME)* _call_body? -> call_as
@@ -314,6 +314,7 @@
              | meta_section
              | runtime_section
              | any_decl -> noninput_decl
+             | assertion -> noninput_decl
 
 tasks: task*
 
@@ -351,6 +352,8 @@
 
 struct: "struct" CNAME "{" unbound_decl* "}"
 
+assertion: "assert" expr
+
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 // type
 ///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -483,7 +486,7 @@
 %ignore COMMENT
 """
 keywords["development"] = set(
-    "Array Directory File Float Int Map None Pair String alias as call command else false if import input left meta object output parameter_meta right runtime scatter struct task then true workflow".split(
+    "Array Directory File Float Int Map None Pair String alias as assert call command else false if import input left meta object output parameter_meta right runtime scatter struct task then true workflow".split(
         " "
     )
 )

WDL/_parser.py

@@ -325,6 +325,9 @@ def decl(self, items, meta):
             self._sp(meta), items[0], items[1].value, (items[2] if len(items) > 2 else None)
         )
 
+    def assertion(self, items, meta):
+        return Tree.Assertion(self._sp(meta), items[0])
+
     def input_decls(self, items, meta):
         return {"inputs": items}
 

WDL/runtime/task.py

@@ -418,6 +418,8 @@ def map_paths(fn: Union[Value.File, Value.Directory]) -> str:
         vj = json.dumps(v.json)
         logger.info(_("eval", name=decl.name, value=(v.json if len(vj) < 4096 else "(((large)))")))
         container_env = container_env.bind(decl.name, v)
+        if isinstance(decl, Tree.Assertion) and not v.value:
+            raise Error.RuntimeError(decl.message)
 
     return container_env
 

WDL/runtime/workflow.py

@@ -367,6 +367,8 @@ def _do_job(
                 else:
                     assert job.node.type.optional
                     v = Value.Null()
+                if isinstance(job.node, Tree.Assertion) and not v.value:
+                    raise Error.RuntimeError(job.node.message)
             return Env.Bindings(Env.Binding(job.node.name, v))
 
         if isinstance(job.node, WorkflowOutputs):

tests/test_7runner.py

@@ -1032,3 +1032,46 @@ def test1(self):
 set123
 """,
         )
+class TestAssert(RunnerTestCase):
+    task1 = R"""
+    version development
+    task div {
+        input {
+            Int numerator
+            Int denominator
+        }
+        assert denominator != 0
+        command {
+            expr ~{numerator} / ~{denominator}
+        }
+        output {
+            Int quotient = read_int(stdout())
+        }
+    }
+    """
+
+    def test_positive(self):
+        outputs = self._run(self.task1, {"numerator": 7, "denominator": 2})
+        self.assertEqual(outputs["quotient"], 3)
+
+    def test_negative(self):
+        self._run(self.task1, {"numerator": 7, "denominator": 0}, expected_exception=WDL.Error.RuntimeError)
+
+    wf1 = R"""
+    version development
+    workflow div {
+        input {
+            Int numerator
+            Int denominator
+        }
+        assert denominator != 0
+        output {
+            Int quotient = numerator / denominator
+        }
+    }
+    """
+
+    def test_workflow(self):
+        outputs = self._run(self.wf1, {"numerator": 7, "denominator": 2})
+        self.assertEqual(outputs["quotient"], 3)
+        self._run(self.wf1, {"numerator": 7, "denominator": 0}, expected_exception=WDL.Error.RuntimeError)