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

Issue:
Add `fractal job` command
At the moment, we have a `fractal workflow job-status` command. In view of https://github.com/fractal-analytics-platform/fractal-server/issues/369, it would be useful to have a more specific `fractal job` command. This should include:
* `job show-status job_id`  -> the current `workflow job-status`
* `job download-logs job_id` -> related to https://github.com/fractal-analytics-platform/fractal-server/issues/369
* `job list` -> useful to find back the `job_id` in case it was not stored

Add `fractal job download-logs` command
Ref #369


Discussion / hints:

Fix patch:
diff --git a/fractal/cmd/__init__.py b/fractal/cmd/__init__.py
index 74d9e2693..600799354 100644
--- a/fractal/cmd/__init__.py
+++ b/fractal/cmd/__init__.py
@@ -11,6 +11,9 @@
 from ._dataset import dataset_delete_resource
 from ._dataset import dataset_edit
 from ._dataset import dataset_show
+from ._job import job_download_logs
+from ._job import job_list
+from ._job import job_status
 from ._project import project_add_dataset
 from ._project import project_create
 from ._project import project_list
@@ -26,7 +29,6 @@
 from ._workflow import workflow_edit_task
 from ._workflow import workflow_list
 from ._workflow import workflow_new
-from ._workflow import workflow_query_job_status
 from ._workflow import workflow_remove_task
 from ._workflow import workflow_show
 
@@ -166,8 +168,18 @@ async def workflow(
         iface = await workflow_remove_task(client, **kwargs)
     elif subcmd == "apply":
         iface = await workflow_apply(client, **kwargs)
-    elif subcmd == "job-status":
-        iface = await workflow_query_job_status(client, batch=batch, **kwargs)
+    return iface
+
+
+async def job(
+    client: AuthClient, subcmd: str, batch: bool = False, **kwargs
+) -> BaseInterface:
+    if subcmd == "list":
+        iface = await job_list(client, batch=batch, **kwargs)
+    elif subcmd == "status":
+        iface = await job_status(client, batch=batch, **kwargs)
+    elif subcmd == "download-logs":
+        iface = await job_download_logs(client, **kwargs)
     return iface
 
 
diff --git a/fractal/cmd/_job.py b/fractal/cmd/_job.py
new file mode 100644
index 000000000..0c4f5cb2d
--- /dev/null
+++ b/fractal/cmd/_job.py
@@ -0,0 +1,65 @@
+from pathlib import Path
+
+from ..authclient import AuthClient
+from ..common.schemas import ApplyWorkflowRead
+from ..config import settings
+from ..interface import BaseInterface
+from ..interface import PrintInterface
+from ..interface import RichJsonInterface
+from ..response import check_response
+
+
+async def job_status(
+    client: AuthClient,
+    job_id: int,
+    batch: bool = False,
+    do_not_separate_logs: bool = False,
+    **kwargs,
+) -> BaseInterface:
+    """
+    Query the status of a workflow-execution job
+    """
+
+    res = await client.get(f"{settings.BASE_URL}/job/{job_id}")
+    job = check_response(
+        res, expected_status_code=200, coerce=ApplyWorkflowRead
+    )
+    if batch:
+        return PrintInterface(retcode=0, data=job.status)
+    else:
+        data = job.sanitised_dict()
+        if do_not_separate_logs or (job.log is None):
+            return RichJsonInterface(retcode=0, data=data)
+        else:
+            log = data.pop("log")
+            extra_lines = ["\nThis is the job log:\n", log]
+            return RichJsonInterface(
+                retcode=0, data=data, extra_lines=extra_lines
+            )
+
+
+async def job_list(
+    client: AuthClient,
+    project_id: int,
+    batch: bool = False,
+    **kwargs,
+) -> BaseInterface:
+
+    raise NotImplementedError("job_list not implemented")
+
+
+async def job_download_logs(
+    client: AuthClient,
+    job_id: int,
+    output: str,
+    batch: bool = False,
+    **kwargs,
+) -> BaseInterface:
+
+    # Check that output folder does not already exist
+    if Path(output).exists():
+        return PrintInterface(
+            retcode=1, data=f"ERROR: {output} already exists"
+        )
+
+    raise NotImplementedError("job_download_logs not implemented")
diff --git a/fractal/parser.py b/fractal/parser.py
index 8e8ccb7dd..847e264ed 100644
--- a/fractal/parser.py
+++ b/fractal/parser.py
@@ -442,23 +442,54 @@
     help="Command to be run before starting a worker",
 )
 
-# workflow job-status
-workflow_job_status_parser = workflow_subparsers.add_parser(
-    "job-status",
-    help="Query workflow-execution status",
+
+# JOB GROUP
+
+job_parser = subparsers_main.add_parser("job", help="job commands")
+job_subparsers = job_parser.add_subparsers(
+    title="Valid subcommand", dest="subcmd", required=True
+)
+
+# job list
+job_list_parser = job_subparsers.add_parser(
+    "list", help="List jobs for given project"
+)
+job_list_parser.add_argument(
+    "project_id",
+    help="Project ID",
+)
+
+# job status
+job_status_parser = job_subparsers.add_parser(
+    "status",
+    help="Query status of workflow-execution job",
     argument_default=ap.SUPPRESS,
 )
-workflow_job_status_parser.add_argument(
-    "--job-id",
+job_status_parser.add_argument(
+    "job_id",
     help="Id of the job",
 )
-workflow_job_status_parser.add_argument(
+job_status_parser.add_argument(
     "--do-not-separate-logs",
     dest="do_not_separate_logs",
     help="Show the job logs in the main output, instead of a separate field",
     action="store_true",
 )
 
+# job download-logs
+job_download_logs_parser = job_subparsers.add_parser(
+    "download-logs",
+    help="Download full folder of workflow-execution job",
+)
+job_download_logs_parser.add_argument(
+    "job_id",
+    help="Id of the job",
+)
+job_download_logs_parser.add_argument(
+    "--output",
+    help="Path of the output folder",
+)
+
 
 # VERSION GROUP
 version_parser = subparsers_main.add_parser(


Test patch:
diff --git a/tests/fixtures_testserver.py b/tests/fixtures_testserver.py
index f39bed57a..840282807 100644
--- a/tests/fixtures_testserver.py
+++ b/tests/fixtures_testserver.py
@@ -151,6 +151,29 @@ async def _workflow_factory(**wf_args_override):
     return _workflow_factory
 
 
+@pytest.fixture
+async def job_factory(db):
+    from fractal_server.app.models.run import ApplyWorkflow
+
+    async def _job_factory(**job_args_override):
+        job_args = dict(
+            project_id=1,
+            input_dataset_id=1,
+            output_dataset_id=2,
+            workflow_id=1,
+            overwrite_input=False,
+            worker_init="WORKER_INIT string",
+        )
+        job_args.update(job_args_override)
+        j = ApplyWorkflow(**job_args)
+        db.add(j)
+        await db.commit()
+        await db.refresh(j)
+        return j
+
+    return _job_factory
+
+
 @pytest.fixture
 async def user_factory(client, testserver):
     async def __register_user(email: str, password: str, slurm_user: str):
diff --git a/tests/test_job.py b/tests/test_job.py
new file mode 100644
index 000000000..b236d5103
--- /dev/null
+++ b/tests/test_job.py
@@ -0,0 +1,167 @@
+from pathlib import Path
+
+import pytest  # noqa F401
+from devtools import debug
+
+
+LOG = "Here are some logs"
+
+
+@pytest.mark.parametrize("status", ["done", "failed"])
+async def test_job_status(
+    register_user,
+    invoke,
+    tmp_path: Path,
+    status: str,
+    workflow_factory,
+    job_factory,
+):
+    """
+    GIVEN a job entry in the database
+    WHEN calling the `job status` command with multiple options
+    THEN the client response has the expected status and log attributes
+    """
+
+    # Create mock Workflow/ApplyWorkflow objects
+    wf = await workflow_factory()
+    debug(wf)
+    log = None if status == "done" else LOG
+    workflow_path = tmp_path / f"workflow_{wf.id}"
+    workflow_path.mkdir()
+    job = await job_factory(
+        working_dir=workflow_path.as_posix(),
+        worfklow_id=wf.id,
+        status=status,
+        log=log,
+    )
+    debug(job)
+
+    # Check `job status` output
+    cmd = f"job status {job.id}"
+    debug(cmd)
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    assert res.data["status"] == status
+    debug(res.data)
+    if status == "done":
+        assert res.data["log"] is None
+    elif status == "failed":
+        assert "log" not in res.data
+        assert LOG in res.extra_lines
+        res.show()
+
+    # Check `job status` output with --batch
+    cmd = f"--batch job status {job.id}"
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    assert res.data == status
+
+    # Check `job status` output with `--do-not-separate-logs`
+    if status == "failed":
+        cmd = f"job status {job.id} --do-not-separate-logs"
+        res = await invoke(cmd)
+        debug(res.data)
+        assert res.retcode == 0
+        assert res.data["status"] == status
+        debug(res.data)
+        assert res.data["log"] == LOG
+
+
+@pytest.mark.xfail(reason="Missing endpoint server side")
+async def test_job_list(
+    register_user,
+    invoke,
+    tmp_path: Path,
+    project_factory,
+    workflow_factory,
+    job_factory,
+):
+    """
+    GIVEN several job entries in the database
+    WHEN calling the `job list ` command
+    THEN the client response lists the jobs associated to the project
+    """
+
+    # Create mock Project/Workflow/ApplyWorkflow objects
+    pj = await project_factory()
+    project_id = pj.id
+    wf_1 = await workflow_factory(project_id=project_id)
+    wf_2 = await workflow_factory(project_id=project_id)
+    wd_1 = tmp_path / f"workflow_{wf_1.id}"
+    wd_2 = tmp_path / f"workflow_{wf_2.id}"
+    job1 = await job_factory(
+        working_dir=wd_1.as_posix(),
+        worfklow_id=wf_1.id,
+        status="running",
+    )
+    job2 = await job_factory(
+        working_dir=wd_2.as_posix(),
+        worfklow_id=wf_2.id,
+        status="done",
+    )
+    debug(job1)
+    debug(job2)
+
+    # Check `job status` output
+    cmd = f"job list {project_id}"
+    debug(cmd)
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    debug(res.data)
+
+    # FIXME add assertions
+    # FIXME also test --batch option
+
+
+@pytest.mark.xfail(reason="Missing endpoint server side")
+async def test_job_download_logs(
+    register_user,
+    invoke,
+    tmp_path: Path,
+    workflow_factory,
+    job_factory,
+):
+    """
+    GIVEN a job entry in the database
+    WHEN calling the `job download-logs` command
+    THEN FIXME
+    """
+
+    # Create mock Workflow/ApplyWorkflow objects
+    wf = await workflow_factory()
+    wd = tmp_path / f"workflow_{wf.id}"
+    job = await job_factory(
+        working_dir=wd.as_posix(),
+        worfklow_id=wf.id,
+        status="running",
+    )
+
+    # Write something in a logfile within the workflow directory
+    LOGFILE = "log.txt"
+    wd.mkdir()
+    with (wd / LOGFILE).open("w") as f:
+        f.write(LOG)
+
+    # Check that download-logs fails if the output folder already exists
+    output_fail = tmp_path / "output_dir_for_logs_fail"
+    output_fail.mkdir()
+    cmd = f"job download-logs {job.project_id} --output {str(output_fail)}"
+    debug(cmd)
+    res = await invoke(cmd)
+    assert res.retcode == 1
+
+    # Check standard `job download-logs` output
+    output = tmp_path / "output_dir_for_logs"
+    cmd = f"job download-logs {job.project_id} --output {str(output)}"
+    debug(cmd)
+    res = await invoke(cmd)
+    assert res.retcode == 0
+    debug(res.data)
+
+    # Check that the logfile is there and has the right content
+    logfile = output / LOGFILE
+    assert logfile.exists()
+    with logfile.open("r") as f:
+        contents = f.read()
+    # FIXME: review the following assertion
+    assert contents == LOG
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index 1baf96dd0..a35cb79b8 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -315,8 +315,10 @@ async def test_workflow_apply(
     assert res.retcode == 0
     assert res.data["status"] == "submitted"
 
+    # TODO: add an assertion about the output, instead of calling `job status`
+
     # Check that job completed successfully
-    cmd = f"workflow job-status --job-id {job_id}"
+    cmd = f"job status {job_id}"
     starting_time = time.perf_counter()
     debug(cmd)
     while True:
@@ -329,12 +331,6 @@ async def test_workflow_apply(
     debug(res.data)
     assert res.data["history"][0] == TASK_NAME
 
-    # Test output with --batch
-    cmd = f"--batch workflow job-status --job-id {job_id}"
-    res = await invoke(cmd)
-    assert res.retcode == 0
-    assert res.data == "done"
-
     # Prepare and run a workflow with a failing task
     args_file = str(tmp_path / "args.json")
     with open(args_file, "w") as f:
@@ -354,7 +350,7 @@ async def test_workflow_apply(
     job_id = res.data["id"]
 
     # Verify that status is failed, and that there is a log
-    cmd = f"workflow job-status --job-id {job_id} --do-not-separate-logs"
+    cmd = f"job status {job_id} --do-not-separate-logs"
     starting_time = time.perf_counter()
     while True:
         res = await invoke(cmd)
@@ -368,13 +364,3 @@ async def test_workflow_apply(
     assert res.data["log"] is not None
     # Note: the failing task is not added to the history
     assert len(res.data["history"]) > 0
-
-    # Test job-status default (without --do-not-separate-logs)
-    cmd = f"workflow job-status --job-id {job_id}"
-    res = await invoke(cmd)
-    assert res.retcode == 0
-    debug(res.data)
-    debug(res.extra_lines)
-    assert "log" not in res.data
-    assert res.extra_lines
-    res.show()