Repository: fractal-analytics-platform/fractal-client (language: python)

Issue:
Also modify `meta` in `fractal workflow add-task`
I'm trying to run the cellpose_segmentation task on the FMI setup which has no gpu (and therefore, we don't have the gpu specified as a possible option of executors).

I used the new ability to edit the executor of a workflow task to cpu-low, which is available. But the workflow still fails early on with this error message:

(redacted error message)
```
ERROR:    Exception in ASGI application
 in recursive_task_assembly
    raise ValueError(msg)
ValueError: this_task.executor='gpu' is not in data_flow_kernel.executors.keys()=dict_keys(['cpu-low', 'cpu-mid', 'cpu-high', '_parsl_internal'])
```

When checking the workflow, the relevant workflow task looks like this:

```
   {
      "workflow_id": 2,
      "task_id": 5,
      "order": 5,
      "meta": null,
      "args": {
        "args": {
          "labeling_channel": "A01_C01",
          "labeling_level": 2,
          "ROI_table_name": "well_ROI_table"
        },
        "meta": {
          "executor": "cpu-low"
        }
      },
      "id": 16,
      "task": {
        "name": "Cellpose Segmentation",
        "command": "/opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRACTAL_ROOT/.fractal/fractal-tasks-core0.4.6/venv/bin/python /opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRAC
TAL_ROOT/.fractal/fractal-tasks-core0.4.6/venv/lib/python3.8/site-packages/fractal_tasks_core/cellpose_segmentation.py",
        "source": "pip:fractal-tasks-core==0.4.6",
        "input_type": "zarr",
        "output_type": "zarr",
        "default_args": {
          "labeling_channel": "A01_C01"
        },
        "meta": {
          "executor": "gpu",
          "parallelization_level": "image"
        },
        "id": 5
      }
    },

```

It appears the `recursive_task_assembly` of parsl uses the default executor, not the one provided to the workflow task. If this is just a parsl issue, then the new slurm executors should solve it. But would be good to have a test in place to check that a task that defaults to have a gpu executor can be used in a setup without gpu executors when the executor is overwritten in the workflow task to something that is available.
Add `--batch` option to `workflow_add_task`
It should return a set of IDs (especially including the one of the new WorkflowTask), so that it can be used in scripts.


Discussion / hints:
Yes, this is a bug. In principle we should be loading the `WorkflowTask.executor` property (see https://github.com/fractal-analytics-platform/fractal-server/blob/main/fractal_server/app/runner/_parsl/__init__.py#L133), which is correctly set to use `"executor": "cpu-low"` in your case.

I cannot say whether the new backend fixes it, but for sure a test is due.
When we define a project&workflow (for server `1.0.0b1` and slurm backend) as in
```bash
LABEL="2"

###############################################################################
# THINGS TO BE CHANGED BY THE USER
# Adapt to settings to where you run the example
BASE_FOLDER_EXAMPLE=`pwd`/..
###############################################################################

# Set useful variables
PRJ_NAME="proj-$LABEL"
DS_IN_NAME="input-ds-$LABEL"
DS_OUT_NAME="output-ds-$LABEL"
WF_NAME="Workflow $LABEL"

# Set cache path and remove any previous file from there
export FRACTAL_CACHE_PATH=`pwd`/".cache"
rm -rv ${FRACTAL_CACHE_PATH}

# Define/initialize empty project folder and temporary file
PROJ_DIR=`pwd`/tmp_${LABEL}
rm -r $PROJ_DIR
mkdir $PROJ_DIR

###############################################################################
# IMPORTANT: modify the following lines so that they point to absolute paths
INPUT_PATH=$BASE_FOLDER_EXAMPLE/images/10.5281_zenodo.7059515
OUTPUT_PATH=${PROJ_DIR}/output
###############################################################################

# Create project
OUTPUT=`fractal --batch project new $PRJ_NAME $PROJ_DIR`
PRJ_ID=`echo $OUTPUT | cut -d ' ' -f1`
DS_IN_ID=`echo $OUTPUT | cut -d ' ' -f2`
echo "PRJ_ID: $PRJ_ID"
echo "DS_IN_ID: $DS_IN_ID"

# Update dataset name/type, and add a resource
fractal dataset edit --name "$DS_IN_NAME" -t image --read-only $PRJ_ID $DS_IN_ID
fractal dataset add-resource -g "*.png" $PRJ_ID $DS_IN_ID $INPUT_PATH

# Add output dataset, and add a resource to it
DS_OUT_ID=`fractal --batch project add-dataset $PRJ_ID "$DS_OUT_NAME"`
echo "DS_OUT_ID: $DS_IN_ID"

fractal dataset edit -t zarr --read-write $PRJ_ID $DS_OUT_ID
fractal dataset add-resource -g "*.zarr" $PRJ_ID $DS_OUT_ID $OUTPUT_PATH

# Create workflow
WF_ID=`fractal --batch workflow new "$WF_NAME" $PRJ_ID`
echo "WF_ID: $WF_ID"

# Mapping of tasks to their IDs (until we add the by-name addressing)
#    "Create OME-ZARR structure": 1,
#    "Yokogawa to Zarr": 2,
#    "Replicate Zarr structure": 3,
#    "Maximum Intensity Projection": 4,
#    "Cellpose Segmentation": 5,
#    "Measurement": 6,
#    "Illumination correction": 7,
#    "Napari workflows wrapper": 8,
#    "Create OME-ZARR structure (multiplexing)": 9

# Add tasks to workflow
fractal workflow add-task $WF_ID 1 --args-file Parameters/args_create_ome_zarr.json
fractal workflow add-task $WF_ID 2
fractal workflow add-task $WF_ID 3
fractal workflow add-task $WF_ID 4
fractal workflow add-task $WF_ID 5 --args-file Parameters/args_cellpose_segmentation.json

# Look at the current workflows
fractal workflow show $WF_ID
echo
```

and then modify a workflow task as
```
$ fractal workflow edit-task 1 5  --json-file Parameters/cellpose_segmentation.json
```
where
```
$ cat Parameters/cellpose_segmentation.json 
{
   "args": {
        "labeling_level": 2
    },
    "meta": {
        "executor": "cpu-low"
    }
}
```
and if we then apply the workflow with `#fractal workflow apply -o $DS_OUT_ID -p $PRJ_ID $WF_ID $DS_IN_ID
`, then the task is submitted to the correct executor. This we can confirm by looking at the `partition` field in the submission script
```
#!/bin/sh
#SBATCH --output=/net/nfs4/pelkmanslab-fileserver-common/data/homes/fractal/fractal-demos/examples-v1/server/artifacts/workflow_000001_job_000001/slurmpy.stdout.%j.log
#SBATCH --partition=main
#SBATCH --mem=1g
#SBATCH --nodes=1
#SBATCH --job-name Cellpose_Segmentation
export CFUT_DIR=/net/nfs4/pelkmanslab-fileserver-common/data/homes/fractal/fractal-demos/examples-v1/server/artifacts/workflow_000001_job_000001
srun /data/homes/fractal/.conda/envs/fractal-server-v1/bin/python -m cfut.remote FMwWTeSpyEyLiXQBH7ZxIG2xf8EfmcSx
```

This suggests that the current version does not have this bug any more.
Hmm, I still run into this when I tried it slightly differently:

```
fractal workflow add-task $WF_ID 5 --args-file Parameters/args_cellpose_segmentation.json
```

I used the example 01, then used this as an args file (Parameters/args_cellpose_segmentation.json):
```
{
    "args": {
         "labeling_channel": "A01_C01",
         "labeling_level": 3,
         "ROI_table_name": "well_ROI_table",
         "label_name": "nuclei"
     },
     "meta": {
         "executor": "cpu-low"
     }
 }
```

I would expect this to work. The workflow task then looks like this:
```
   {
      "workflow_id": 1,
      "task_id": 5,
      "order": 4,
      "meta": null,
      "args": {
        "args": {
          "labeling_channel": "A01_C01",
          "labeling_level": 3,
          "ROI_table_name": "well_ROI_table",
          "label_name": "nuclei"
        },
        "meta": {
          "executor": "cpu-low"
        }
      },
      "id": 5,
      "task": {
        "name": "Cellpose Segmentation",
        "command": "/opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRACTAL_ROOT/.fractal/fractal-tasks-core0.5.1/venv/bin/python /opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRACTAL_ROOT/.fractal/fractal-tasks-core0.5.1/venv/lib/python3.8/site-packages/fractal_tasks_core/cellpose_segmentation.py",
        "source": "pip:fractal-tasks-core==0.5.1",
        "input_type": "zarr",
        "output_type": "zarr",
        "default_args": {
          "labeling_channel": "A01_C01"
        },
        "meta": {
          "executor": "gpu",
          "parallelization_level": "image"
        },
        "id": 5
      }
    }

```

I then get this error:

```
ERROR:    Exception in ASGI application
 in set_slurm_config
    config = config_dict[task.executor]
KeyError: 'gpu'

in set_slurm_config
    raise SlurmConfigError(f"Configuration not found: {task.executor}")
fractal_server.app.runner._slurm.SlurmConfigError: Configuration not found: gpu

```
Should it make a difference whether we have the executor in the parameter of the task when it's added or when it's modified? Why did you add it with the parameters and then modify it with the same parameters above?
I then manually did the `fractal workflow apply -o 2 -p 1 1 1 --worker-init "$WORKER_INIT"` and now my workflow task looks like this:

```
{
  "workflow_id": 1,
  "task_id": 5,
  "order": 4,
  "meta": {
    "executor": "cpu-low"
  },
  "args": {
    "args": {
      "labeling_channel": "A01_C01",
      "labeling_level": 3,
      "ROI_table_name": "well_ROI_table",
      "label_name": "nuclei"
    },
    "meta": {
      "executor": "cpu-low"
    },
    "labeling_channel": "A01_C01",
    "labeling_level": 3,
    "ROI_table_name": "well_ROI_table",
    "label_name": "nuclei"
  },
  "id": 5,
  "task": {
    "name": "Cellpose Segmentation",
    "command": "/opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRACTAL_ROOT/.fractal/fractal-tasks-core0.5.1/venv/bin/python /opt/fmi/gliberal/fractal/fractal-demos/examples-v1/server/FRACTAL_ROOT/.fractal/fractal-tasks-core0.5.1/venv/lib/python3.8/site-packages/fractal_tasks_core/cellpose_segmentation.py",
    "source": "pip:fractal-tasks-core==0.5.1",
    "input_type": "zarr",
    "output_type": "zarr",
    "default_args": {
      "labeling_channel": "A01_C01"
    },
    "meta": {
      "executor": "gpu",
      "parallelization_level": "image"
    },
    "id": 5
  }
}
```

Now it works. And now I understand what broke before: It put the args & meta into an "args" keyword when I used it in add-task.

So, the old way of adding both args & meta as part of adding a task doesn't work anymore then, correct? Is there a way to still do it in one as part of fractal workflow add-task?
Yes, there is a difference (maybe counterintuitive one, but there is one). Let's look at the help messages:

```
$ fractal workflow add-task -h
usage: fractal workflow add-task [-h] [--order ORDER] [--args-file ARGS_FILE] id task_id
positional arguments:
  id                    Workflow id
  task_id               Task id
optional arguments:
  -h, --help            show this help message and exit
  --order ORDER         Order of this task within the workflow's task list
  --args-file ARGS_FILE
                        Path to a json serialised file containing the arguments ovverrides of the task

$ fractal workflow edit-task -h
usage: fractal workflow edit-task [-h] [--json-file JSON_FILE] id workflow_task_id
positional arguments:
  id                    Workflow id
  workflow_task_id      Workflow task Id, the Id of a task inside the list of tasks
optional arguments:
  -h, --help            show this help message and exit
  --json-file JSON_FILE
                        Path to a json serialised file containing the attribute ovverrides of the task
```

`ARGS_FILE` and `JSON_FILE` have different meanings. The former is only for `args`, while the latter is both for `args` and `meta`. The first one should have no special key, while the second should have `args` and/or `meta` keys.

In fact in your logs you see:
```
"args": {
        "args": {
          "labeling_channel": "A01_C01",
          "labeling_level": 3,
          "ROI_table_name": "well_ROI_table",
          "label_name": "nuclei"
        },
```
which is incorrect. This would mean that the task has an argument called `args` which takes value `{"labeling_channel": "A01_C01", ...}`.

---

TL;DR
In the current version, the `add-task` can only modify the `args`, and not `meta`. To get it work with the current version, you should only write the args in `ARGS_FILE`, namely:
```JSON
{
          "labeling_channel": "A01_C01",
          "labeling_level": 3,
          "ROI_table_name": "well_ROI_table",
          "label_name": "nuclei"
        }
```

We  can then make things more uniform
Ok, thanks!
Renamed and move to client repo
Any preference about the required interface?
Right now it is
```
fractal workflow add-task ... --args-file /some/args/file.json
fractal workflow edit-task ... --json-file /some/json/file.json
```
with `/some/args/file.json` having no fractal-specific keywords, while `/some/json/file.json` must have `args` and/or `meta` keys.

Let's aim for a uniform interface of the two commands. Options:
1. They both take a `json-file`-like argument.
2. They both take two separate arguments (`args-file` and `meta-file`).
I think we'll mostly just change the args, not the meta. Thus, going with option 2 and having a meta json file we can provide for the few edge-cases where we want to modify it sounds good to me. Also keeps the args & meta separated well.

Then, it could look like this
```
fractal workflow add-task ... --args-file /some/args/file.json --meta-file /some/meta/file.json 
``` 

With both files being optional of course


Fix patch:
diff --git a/fractal/cmd/_workflow.py b/fractal/cmd/_workflow.py
index 0154ebcbe..557e07b8f 100644
--- a/fractal/cmd/_workflow.py
+++ b/fractal/cmd/_workflow.py
@@ -5,6 +5,7 @@
 
 from ..authclient import AuthClient
 from ..common.schemas import ApplyWorkflowCreate
+from ..common.schemas import ApplyWorkflowRead
 from ..common.schemas import WorkflowCreate
 from ..common.schemas import WorkflowRead
 from ..common.schemas import WorkflowTaskCreate
@@ -87,6 +88,7 @@ async def workflow_add_task(
     task_id_or_name: str,
     order: int = None,
     args_file: Optional[str] = None,
+    meta_file: Optional[str] = None,
     **kwargs,
 ) -> RichJsonInterface:
 
@@ -100,16 +102,20 @@ async def workflow_add_task(
         with Path(args_file).open("r") as f:
             args = json.load(f)
             workflow_task.args = args
+    if meta_file:
+        with Path(meta_file).open("r") as f:
+            meta = json.load(f)
+            workflow_task.meta = meta
 
     res = await client.post(
         f"{settings.BASE_URL}/workflow/{id}/add-task/",
         json=workflow_task.dict(),
     )
-    workflow_task = check_response(
+    workflow = check_response(
         res, expected_status_code=201, coerce=WorkflowRead
     )
 
-    return RichJsonInterface(retcode=0, data=workflow_task.dict())
+    return RichJsonInterface(retcode=0, data=workflow.dict())
 
 
 async def workflow_edit_task(
@@ -117,12 +123,30 @@ async def workflow_edit_task(
     *,
     id: int,
     workflow_task_id: int,
-    json_file: str,
+    args_file: Optional[str] = None,
+    meta_file: Optional[str] = None,
     **kwargs,
 ) -> RichJsonInterface:
 
-    with Path(json_file).open("r") as f:
-        payload = json.load(f)
+    # Check that at least one of args_file or meta_file was given (note: it
+    # would be reasonable to check it in the parser, but we are not aware of a
+    # method within argparse).
+    if not (args_file or meta_file):
+        raise ValueError(
+            "At least one of {args_file,meta_file} arguments is " "required"
+        )
+
+    # Combine args/meta payload
+    payload = {}
+    if args_file:
+        with Path(args_file).open("r") as f:
+            args = json.load(f)
+            payload["args"] = args
+    if meta_file:
+        with Path(meta_file).open("r") as f:
+            meta = json.load(f)
+            payload["meta"] = meta
+
     payload_update = WorkflowTaskUpdate(**payload)
     res = await client.patch(
         f"{settings.BASE_URL}/workflow/{id}/edit-task/{workflow_task_id}",
@@ -194,6 +218,7 @@ async def workflow_apply(
     res = await client.post(
         f"{settings.BASE_URL}/project/apply/", json=apply_wf_create.dict()
     )
-    # TODO check output
-
-    return RichJsonInterface(retcode=0, data=res.json())
+    apply_wf_read = check_response(
+        res, expected_status_code=202, coerce=ApplyWorkflowRead
+    )
+    return RichJsonInterface(retcode=0, data=apply_wf_read.json())
diff --git a/fractal/parser.py b/fractal/parser.py
index 9219a9f5a..5b4d40bc9 100644
--- a/fractal/parser.py
+++ b/fractal/parser.py
@@ -351,6 +351,14 @@
         "ovverrides of the task"
     ),
 )
+workflow_add_task_parser.add_argument(
+    "--meta-file",
+    help=(
+        "Path to a json serialised file containing the meta"
+        "ovverrides of the task"
+    ),
+)
+
 
 # workflow edit task
 workflow_edit_task_parser = workflow_subparsers.add_parser(
@@ -365,9 +373,16 @@
     help="Workflow task Id, the Id of a task inside the list of tasks",
 )
 workflow_edit_task_parser.add_argument(
-    "--json-file",
+    "--args-file",
+    help=(
+        "Path to a json serialised file containing the arguments "
+        "ovverrides of the task"
+    ),
+)
+workflow_edit_task_parser.add_argument(
+    "--meta-file",
     help=(
-        "Path to a json serialised file containing the attribute "
+        "Path to a json serialised file containing the meta"
         "ovverrides of the task"
     ),
 )


Test patch:
diff --git a/tests/test_task.py b/tests/test_task.py
index f97a3928f..8f26f9e48 100644
--- a/tests/test_task.py
+++ b/tests/test_task.py
@@ -80,68 +80,6 @@ async def test_repeated_task_collection(register_user, invoke, testdata_path):
     assert res0.data["data"]["info"] == "Already installed"
 
 
-async def test_task_apply(register_user, invoke, testdata_path):
-    """
-    GIVEN a project and a nontrivial workflow
-    WHEN the client requests to apply the workflow to the project
-    THEN the workflow is scheduled and executed, and the artifacts created
-    """
-    PACKAGE_NAME = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
-    WORKFLOW_NAME = "mywf"
-    DATASET_NAME = "myds"
-
-    res0 = await invoke(f"task collect {PACKAGE_NAME}")
-    debug(res0)
-    res0.show()
-    venv_path = res0.data["data"]["venv_path"]
-    debug(venv_path)
-    state_id = res0.data["id"]
-
-    while True:
-        res1 = await invoke(f"task check-collection {state_id}")
-        if res1.data["data"]["status"] == "OK":
-            debug(res1.data)
-            break
-        await asyncio.sleep(1)
-
-    res = await invoke("project new testproject prjpath")
-    debug(res)
-    assert res.retcode == 0
-
-    prj = res.data
-    prj_id = prj["id"]
-    input_dataset_id = prj["dataset_list"][0]["id"]
-
-    res = await invoke(f"project add-dataset {prj_id} {DATASET_NAME}")
-    assert res.retcode == 0
-    debug(res.data)
-    output_dataset_id = res.data["id"]
-
-    res = await invoke(
-        f"dataset add-resource {prj_id} {output_dataset_id} "
-        f"{testdata_path} -g 'out.json'"
-    )
-
-    res = await invoke(f"workflow new {WORKFLOW_NAME} {prj_id}")
-    workflow = res.data
-    workflow_id = workflow["id"]
-
-    TASK_ID = 1
-    res = await invoke(f"workflow add-task {workflow_id} {TASK_ID}")
-    assert res.retcode == 0
-
-    cmd = (
-        f"workflow apply {workflow_id} {input_dataset_id} "
-        f"-o {output_dataset_id} -p {prj['id']}"
-    )
-    debug(cmd)
-    res = await invoke(cmd)
-    debug(res.data)
-    assert res.retcode == 0
-
-    # TODO: verify outout
-
-
 @pytest.mark.xfail
 async def test_edit_task(register_user, invoke, clear_task_cache):
     # TODO:
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index 57635fcf4..98bde1d1a 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -1,3 +1,4 @@
+import asyncio
 import json
 import logging
 from pathlib import Path
@@ -88,17 +89,25 @@ async def test_add_task(
     wf = await workflow_factory()
     t = await task_factory()
 
-    custom_args = dict(custom="args")
+    ARGS = {"arg": "arg_value"}
+    META = {"executor": "some-executor"}
     args_file = tmp_path / "args_file.json"
     with args_file.open("w") as f:
-        json.dump(custom_args, f)
+        json.dump(ARGS, f)
+    meta_file = tmp_path / "meta_file.json"
+    with meta_file.open("w") as f:
+        json.dump(META, f)
 
-    cmd = f"workflow add-task {wf.id} {t.id} --args-file {args_file}"
+    cmd = (
+        f"workflow add-task {wf.id} {t.id} "
+        f"--args-file {args_file} --meta-file {meta_file}"
+    )
     debug(cmd)
     res = await invoke(cmd)
     assert res.retcode == 0
     debug(res.data)
-    assert res.data["task_list"][0]["args"] == custom_args
+    assert res.data["task_list"][0]["args"] == ARGS
+    assert res.data["task_list"][0]["meta"] == META
 
 
 async def test_add_task_by_name(
@@ -188,47 +197,110 @@ async def test_edit_workflow_task(
     assert res.retcode == 0
 
     # New arguments to be used
-    payload = dict(
-        args={"some_arg": "some_value"}, meta={"executor": "cpu-low"}
-    )
+    ARGS = {"some_arg": "some_value"}
+    META = {"executor": "cpu-low"}
 
-    json_file = tmp_path / "payload.json"
-    with json_file.open("w") as f:
-        json.dump(payload, f)
+    args_file = tmp_path / "args_file.json"
+    with args_file.open("w") as f:
+        json.dump(ARGS, f)
+    meta_file = tmp_path / "meta_file.json"
+    with meta_file.open("w") as f:
+        json.dump(META, f)
 
     # Edit workflow task
     debug(res.data)
     workflow_task_id = res.data["task_list"][0]["id"]
     cmd = (
         f"workflow edit-task {wf.id} {workflow_task_id} "
-        f"--json-file {json_file}"
+        f"--args-file {args_file} --meta-file {meta_file}"
     )
     debug(cmd)
     res = await invoke(cmd)
     assert res.retcode == 0
-    assert res.data["args"] == payload["args"]
-    assert res.data["meta"] == payload["meta"]
+    assert res.data["args"] == ARGS
+    assert res.data["meta"] == META
 
     # Check that also the workflow in the db was correctly updated
     res = await invoke(f"workflow show {wf.id}")
     assert res.retcode == 0
     debug(res.data)
-    assert res.data["task_list"][0]["args"] == payload["args"]
-    assert res.data["task_list"][0]["meta"] == payload["meta"]
+    assert res.data["task_list"][0]["args"] == ARGS
+    assert res.data["task_list"][0]["meta"] == META
 
     # Check if the correct error is raised where parallelization_level
     # is set
-    payload_error = dict(meta={"parallelization_level": "XXX"})
-
-    json_file = tmp_path / "payload_error.json"
-    with json_file.open("w") as f:
-        json.dump(payload_error, f)
+    META_err = {"parallelization_level": "XXX"}
+    meta_file = tmp_path / "meta_file.json"
+    with meta_file.open("w") as f:
+        json.dump(META_err, f)
 
     workflow_task_id = res.data["task_list"][0]["id"]
     cmd = (
         f"workflow edit-task {wf.id} {workflow_task_id} "
-        f"--json-file {json_file}"
+        f"--meta-file {meta_file}"
     )
     debug(cmd)
     with pytest.raises(ValueError):
         res = await invoke(cmd)
+
+
+async def test_workflow_apply(register_user, invoke, testdata_path):
+    """
+    GIVEN a project and a nontrivial workflow
+    WHEN the client requests to apply the workflow to the project
+    THEN the workflow is scheduled and executed, and the artifacts created
+    """
+    PACKAGE_NAME = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
+    WORKFLOW_NAME = "mywf"
+    DATASET_NAME = "myds"
+
+    res0 = await invoke(f"task collect {PACKAGE_NAME}")
+    debug(res0)
+    res0.show()
+    venv_path = res0.data["data"]["venv_path"]
+    debug(venv_path)
+    state_id = res0.data["id"]
+
+    while True:
+        res1 = await invoke(f"task check-collection {state_id}")
+        if res1.data["data"]["status"] == "OK":
+            debug(res1.data)
+            break
+        await asyncio.sleep(1)
+
+    res = await invoke("project new testproject prjpath")
+    debug(res)
+    assert res.retcode == 0
+
+    prj = res.data
+    prj_id = prj["id"]
+    input_dataset_id = prj["dataset_list"][0]["id"]
+
+    res = await invoke(f"project add-dataset {prj_id} {DATASET_NAME}")
+    assert res.retcode == 0
+    debug(res.data)
+    output_dataset_id = res.data["id"]
+
+    res = await invoke(
+        f"dataset add-resource {prj_id} {output_dataset_id} "
+        f"{testdata_path} -g 'out.json'"
+    )
+
+    res = await invoke(f"workflow new {WORKFLOW_NAME} {prj_id}")
+    workflow = res.data
+    workflow_id = workflow["id"]
+
+    TASK_ID = 1
+    res = await invoke(f"workflow add-task {workflow_id} {TASK_ID}")
+    assert res.retcode == 0
+
+    cmd = (
+        f"workflow apply {workflow_id} {input_dataset_id} "
+        f"-o {output_dataset_id} -p {prj['id']}"
+    )
+    debug(cmd)
+    res = await invoke(cmd)
+    debug(res.data)
+    assert res.retcode == 0
+
+    # TODO: verify outout