def METHOD_NAME():
    with pytest.raises(
        DagsterInvalidDefinitionError,
        match=(
            'cannot be downstream of dynamic output "dynamic_op:result" since input "a" maps to'
            " a node that is already downstream of another dynamic output"
        ),
    ):
        @graph
        def composed(a):
            dynamic_echo(a).map(echo)
        @job
        def _should_fail():
            dynamic_op().map(composed)