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

Issue:
Raise a warning upon import/export of a workflow with a custom task
Ref:
* https://github.com/fractal-analytics-platform/fractal-server/issues/729


Discussion / hints:
Note a task is "custom" if `task.owner is not None`
Re: export
* Call the workflow-export endpoint
* Check the workflow/task_list/element/task/owner attributes, and raise a warning if appropriate
* Continue with the command
Re: import
* Call the workflow-import endpoint
* Look at its response body, check the workflow/task_list/element/task/owner attributes, and raise a warning if appropriate
Example of the warning message upon workflow export:
```python
    # Emit a warning when exporting a workflow with custom tasks
    for wftask in workflow.task_list:
        if wftask.task.owner is not None:
            logging.warning(
                f"Custom tasks (like the one with id={wftask.task.id} and "
                f'source="{wftask.task.source}") are not meant to be '
                "portable; re-importing this workflow may not work as "
                "expected."
            )
```

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 379505173..435abc729 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,7 +11,7 @@
     * For `edit` commands, always prepend the new arguments with `new`, e.g. as in `task edit ... --new-version` (\#498).
     * Add `--type` optional argument to `fractal dataset new` command (\#479).
     * For `fractal workflow apply`:
-        * Ttransform `project_id` from option to positional argument (\#479).
+        * Transform `project_id` from option to positional argument (\#479).
         * Make `output_dataset_id` a required positional argument (\#483).
     * Add `--username/--new-username` to `fractal user` subcommands (\#493).
     * Remove `--private` option for `fractal task collect` (\#493).
@@ -20,6 +20,7 @@
     * Add `--pinned-dependency` argument to `task collect` (\#508).
     * Add `--args-schema` and `--args-schema-version` arguments to `task new` command (\#511).
     * Add `--new-args-schema` and `--new-args-schema-version` arguments to `task edit` command (\#511).
+    * Raise warning when importing/exporting workflows with custom tasks (\#513).
 * Package and repository:
     * Fix a bug in tests, by starting the fractal-server FastAPI app in a more standard way (\#481).
     * Require pydantic version to be `>=1.10.8` (\#486, \#490).
@@ -27,6 +28,7 @@
     * Improve handling of a `ConnectError` in the CI (\#502).
     * Remove arbitrary `kwargs` from internal functions (\#503).
 
+
 # 1.2.0
 
 * Align with [fractal-server 1.2.0](https://fractal-analytics-platform.github.io/fractal-server/changelog/#120) (\#472).
diff --git a/fractal/cmd/_workflow.py b/fractal/cmd/_workflow.py
index 29706b087..dc81b8f99 100644
--- a/fractal/cmd/_workflow.py
+++ b/fractal/cmd/_workflow.py
@@ -254,6 +254,7 @@ async def workflow_apply(
     apply_wf_read = check_response(
         res, expected_status_code=202, coerce=ApplyWorkflowRead
     )
+
     return RichJsonInterface(retcode=0, data=apply_wf_read.sanitised_dict())
 
 
@@ -270,6 +271,20 @@ async def workflow_import(
     wf_read = check_response(
         res, expected_status_code=201, coerce=WorkflowRead
     )
+
+    warnings = [
+        workflow_task.task.source
+        for workflow_task in wf_read.task_list
+        if workflow_task.task.owner
+    ]
+    if warnings:
+        sources_str = ", ".join([f"'{s}'" for s in warnings])
+        logging.warning(
+            "This workflow includes custom tasks (the ones with sources: "
+            f"{sources_str}), which are not meant to be portable; "
+            "importing this workflow may not work as expected."
+        )
+
     if batch:
         datastr = f"{wf_read.id}"
         for wftask in wf_read.task_list:
@@ -295,6 +310,22 @@ async def workflow_export(
     workflow = check_response(
         res, expected_status_code=200, coerce=WorkflowExport
     )
+
+    warnings = [
+        workflow_task.task.source
+        for workflow_task in workflow.task_list
+        if not workflow_task.task.source.startswith(
+            ("pip_local:", "pip_remote:")
+        )
+    ]
+    if warnings:
+        sources_str = ", ".join([f"'{s}'" for s in warnings])
+        logging.warning(
+            "This workflow includes custom tasks (the ones with sources: "
+            f"{sources_str}), which are not meant to be portable; "
+            "re-importing this workflow may not work as expected."
+        )
+
     with Path(json_file).open("w") as f:
         json.dump(workflow.dict(), f, indent=2)
     return PrintInterface(


Test patch:
diff --git a/tests/data/import-export/workflow.json b/tests/data/import-export/workflow.json
index 0b245c7ea..225ec7df2 100644
--- a/tests/data/import-export/workflow.json
+++ b/tests/data/import-export/workflow.json
@@ -7,13 +7,7 @@
                 "arg": "value"
             },
             "meta": {"key": "value"},
-            "order": 0,
             "task": {
-                "command": "/SOME/PATH/dummy0/venv/bin/python /SOME/PATH/dummy0/venv/lib/python3.8/site-packages/fractal_tasks_dummy/dummy.py",
-                "input_type": "Any",
-                "meta": {},
-                "name": "dummy",
-                "output_type": "Any",
                 "source": "pip_local:fractal_tasks_dummy:0.1.0:::dummy"
             }
         },
@@ -21,15 +15,14 @@
             "args": {
                 "message": "task 1"
             },
-            "order": 1,
             "task": {
-                "command": "/SOME/PATH/dummy0/venv/bin/python /SOME/PATH/dummy0/venv/lib/python3.8/site-packages/fractal_tasks_dummy/dummy_parallel.py",
-                "input_type": "Any",
-                "meta": {"parallelization_level": "index"},
-                "name": "dummy parallel",
-                "output_type": "Any",
                 "source": "pip_local:fractal_tasks_dummy:0.1.0:::dummy_parallel"
             }
+        },
+        {
+            "task": {
+                "source": "custom_source"
+            }
         }
     ]
 }
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index e8b4a94e6..2619f36cf 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -536,17 +536,31 @@ async def test_workflow_export(
     invoke,
     workflow_factory,
     tmp_path: Path,
+    task_factory,
+    caplog,
 ):
     NAME = "WorkFlow"
     wf = await workflow_factory(name=NAME)
     prj_id = wf.project_id
     wf_id = wf.id
     filename = str(tmp_path / "exported_wf.json")
+
+    task = await task_factory(owner="exact-lab")
+    res = await invoke(
+        f"workflow add-task {prj_id} {wf_id} --task-id {task.id}"
+    )
+    assert res.retcode == 0
+
     res = await invoke(
         f"workflow export {prj_id} {wf_id} --json-file {filename}"
     )
-    debug(res.data)
     assert res.retcode == 0
+    assert caplog.records[-1].msg == (
+        "This workflow includes custom tasks (the ones with sources: "
+        f"'{task.source}'), which are not meant to be portable; "
+        "re-importing this workflow may not work as expected."
+    )
+    debug(res.data)
     with open(filename, "r") as f:
         exported_wf = json.load(f)
         assert exported_wf["name"] == NAME
@@ -561,10 +575,9 @@ async def test_workflow_export(
 async def test_workflow_import(
     register_user,
     invoke,
-    workflow_factory,
-    tmp_path: Path,
     testdata_path: Path,
-    project_factory,
+    task_factory,
+    caplog,
 ):
     # collect tasks
     PACKAGE_NAME = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
@@ -585,6 +598,7 @@ async def test_workflow_import(
     assert res_pj.retcode == 0
     project_id = res_pj.data["id"]
 
+    await task_factory(source="custom_source", owner="exact-lab")
     # import workflow into project
     filename = str(testdata_path / "import-export/workflow.json")
     res = await invoke(
@@ -592,6 +606,12 @@ async def test_workflow_import(
     )
     debug(res.data)
     assert res.retcode == 0
+    assert caplog.records[-1].msg == (
+        "This workflow includes custom tasks (the ones with sources: "
+        "'custom_source'), which are not meant to be portable; "
+        "importing this workflow may not work as expected."
+    )
+
     imported_workflow = res.data
 
     # get the workflow from the server, and check that it is the same