cli: implement share-status command

Adds a new command to the CLI to retrieve whom a workflow is shared with.

Closes reanahub#686

docs/cmd_list.txt

@@ -33,6 +33,7 @@ Workflow execution commands:
 Workflow sharing commands:
   share-add     Share a workflow with other users (read-only).
   share-remove  Unshare a workflow.
+  share-status  Show with whom a workflow is shared.
 
 Workspace interactive commands:
   close  Close an interactive session.

reana_client/api/client.py

@@ -1365,3 +1365,38 @@ def unshare_workflow(workflow, user_email_to_unshare_with, access_token):
             f"Message: {e.response.json()['message']}"
         )
         raise Exception(e.response.json()["message"])
+
+
+def get_workflow_sharing_status(workflow, access_token):
+    """Get the share status of a workflow.
+
+    :param workflow: name or id of the workflow.
+    :param access_token: access token of the current user.
+
+    :return: a dictionary containing the ``workflow_id``, ``workflow_name``, and
+             a ``sharing_status`` key with the result of the operation.
+    """
+    try:
+        (
+            response,
+            http_response,
+        ) = current_rs_api_client.api.get_workflow_share_status(
+            workflow_id_or_name=workflow,
+            access_token=access_token,
+        ).result()
+
+        if http_response.status_code == 200:
+            return response
+        else:
+            raise Exception(
+                "Expected status code 200 but replied with "
+                f"{http_response.status_code}"
+            )
+
+    except HTTPError as e:
+        logging.debug(
+            "Workflow sharing status could not be retrieved: "
+            f"\nStatus: {e.response.status_code}\nReason: {e.response.reason}\n"
+            f"Message: {e.response.json()['message']}"
+        )
+        raise Exception(e.response.json()["message"])

reana_client/cli/workflow.py

@@ -1605,3 +1605,75 @@ def share_workflow_remove(ctx, workflow, access_token, users):  # noqa D412
 
     else:
         display_message(f"Cannot find workflow {workflow}", msg_type="error")
+
+
+@workflow_sharing_group.command("share-status")
+@check_connection
+@add_workflow_option
+@add_access_token_options
+@click.option(
+    "--format",
+    "_format",
+    multiple=True,
+    default=None,
+    help="Format output according to column titles or column "
+    "values. Use <columm_name>=<column_value> format.",
+)
+@click.option(
+    "--json",
+    "output_format",
+    flag_value="json",
+    default=None,
+    help="Get output in JSON format.",
+)
+@click.pass_context
+def share_workflow_status(
+    ctx, workflow, _format, output_format, access_token
+):  # noqa D412
+    """Show with whom a workflow is shared.
+
+    The `share-status` command allows for checking with whom a workflow is
+    shared.
+
+    Example:
+
+    $ reana-client share-status -w myanalysis.42
+    """
+    from reana_client.api.client import get_workflow_sharing_status
+
+    try:
+        sharing_status = get_workflow_sharing_status(workflow, access_token)
+
+        if sharing_status:
+            shared_with = sharing_status.get("shared_with", [])
+
+            if shared_with:
+                headers = ["user_email", "valid_until"]
+                data = [
+                    [
+                        entry["user_email"],
+                        entry["valid_until"]
+                        if entry["valid_until"] is not None
+                        else "-",
+                    ]
+                    for entry in shared_with
+                ]
+
+                display_formatted_output(data, headers, _format, output_format)
+            else:
+                display_message(
+                    f"Workflow {workflow} is not shared with anyone.", msg_type="info"
+                )
+        else:
+            display_message(
+                f"Workflow {workflow} is not shared with anyone.", msg_type="info"
+            )
+    except Exception as e:
+        logging.debug(traceback.format_exc())
+        logging.debug(str(e))
+        display_message(
+            "An error occurred while checking workflow sharing status:\n{}".format(
+                str(e)
+            ),
+            msg_type="error",
+        )

tests/test_cli_workflows.py

@@ -1084,3 +1084,36 @@ def test_share_remove_workflow():
             )
             assert result.exit_code == 0
             assert response["message"] in result.output
+
+
+def test_share_status_workflow():
+    """Test share-status workflows."""
+    status_code = 200
+    response = {
+        "message": "is not shared with anyone",
+        "workflow_id": "string",
+        "workflow_name": "string",
+    }
+    env = {"REANA_SERVER_URL": "localhost"}
+    mock_http_response, mock_response = Mock(), Mock()
+    mock_http_response.status_code = status_code
+    mock_response = response
+    reana_token = "000000"
+    runner = CliRunner(env=env)
+    with runner.isolation():
+        with patch(
+            "reana_client.api.client.current_rs_api_client",
+            make_mock_api_client("reana-server")(mock_response, mock_http_response),
+        ):
+            result = runner.invoke(
+                cli,
+                [
+                    "share-status",
+                    "-t",
+                    reana_token,
+                    "--workflow",
+                    "test-workflow.1",
+                ],
+            )
+            assert result.exit_code == 0
+            assert response["message"] in result.output