(MIDRC-1276) Add pre-filtering in funnel query for listing tasks

docs/openapi.yaml

@@ -98,6 +98,17 @@ paths:
     get:
       description: List the user's GA4GH TES tasks
       operationId: list_tasks
+      parameters:
+      - description: 'If true, retrieves all the TES tasks you may have access to,
+          instead of just your own. Default: false'
+        in: query
+        name: all
+        required: false
+        schema:
+          description: 'If true, retrieves all the TES tasks you may have access to,
+            instead of just your own. Default: false'
+          title: All
+          type: string
       responses:
         '200':
           content:
@@ -107,6 +118,12 @@ paths:
                 title: Response List Tasks
                 type: object
           description: Successful Response
+        '422':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
+          description: Validation Error
       security:
       - HTTPBearer: []
       summary: List Tasks

gen3workflow/routes/ga4gh_tes.py

@@ -8,7 +8,7 @@
 import json
 import re
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from gen3authz.client.arborist.errors import ArboristError
 from starlette.status import (
     HTTP_200_OK,
@@ -37,6 +37,13 @@
 TAGS_HIDDEN_FROM_USER.remove("_AUTHZ")
 
 
+def get_authz_string_for_user(user_id: str) -> str:
+    """
+    Get the authz_resource string for a user
+    """
+    return f"/services/workflow/gen3-workflow/tasks/{user_id}/TASK_ID_PLACEHOLDER"
+
+
 async def get_request_body(request: Request) -> dict:
     """
     Extract the body from a FastAPI request
@@ -161,9 +168,7 @@ async def create_task(request: Request, auth=Depends(Auth)) -> dict:
         )
         logger.error(err_msg)
         raise HTTPException(HTTP_400_BAD_REQUEST, err_msg)
-    authz_resource = (
-        f"/services/workflow/gen3-workflow/tasks/{user_id}/TASK_ID_PLACEHOLDER"
-    )
+    authz_resource = get_authz_string_for_user(user_id)
     body["tags"]["_AUTHZ"] = authz_resource
 
     if config["EKS_CLUSTER_NAME"]:
@@ -238,7 +243,14 @@ def apply_view_to_task(view: str, task: dict) -> dict:
 
 @router.get("/tasks", status_code=HTTP_200_OK)
 @router.get("/tasks/", status_code=HTTP_200_OK, include_in_schema=False)
-async def list_tasks(request: Request, auth=Depends(Auth)) -> dict:
+async def list_tasks(
+    request: Request,
+    auth=Depends(Auth),
+    all: str = Query(
+        None,
+        description="If true, retrieves all the TES tasks you may have access to, instead of just your own. Default: false",
+    ),
+) -> dict:
     """
     List the user's GA4GH TES tasks
     """
@@ -262,7 +274,18 @@ async def list_tasks(request: Request, auth=Depends(Auth)) -> dict:
     requested_view = query_params.get("view")
     query_params["view"] = "FULL"
 
-    # get all the tasks, regardless of access
+    if all is None:
+        query_params["tag_key"] = "_AUTHZ"
+        # get the user_id and construct an authz value
+        token_claims = await auth.get_token_claims()
+        user_id = token_claims.get("sub")
+        if not user_id:
+            err_msg = "No user sub in token"
+            logger.error(err_msg)
+            raise HTTPException(HTTP_401_UNAUTHORIZED, err_msg)
+        authz_resource = get_authz_string_for_user(user_id)
+        query_params["tag_value"] = authz_resource
+
     url = f"{config['TES_SERVER_URL']}/tasks"
     res = await make_tes_server_request(
         request.app.async_client, "get", url, params=query_params
@@ -293,6 +316,7 @@ async def list_tasks(request: Request, auth=Depends(Auth)) -> dict:
         raise HTTPException(e.code, e.message)
 
     # filter out tasks the current user does not have access to
+    # this should be redundant if all is None but is included in case server-side filtering fails
     listed_tasks["tasks"] = [
         apply_view_to_task(requested_view, task)
         for task in listed_tasks.get("tasks", [])

tests/test_ga4gh_tes.py

@@ -5,6 +5,7 @@
 import pytest_asyncio
 
 from gen3workflow.config import config
+from gen3workflow.routes.ga4gh_tes import get_authz_string_for_user
 from tests.conftest import (
     mock_arborist_request,
     mock_tes_server_request,
@@ -476,26 +477,43 @@ async def test_create_task_optimized_node_scheduling(
     )
 
 
+def test_get_authz_string():
+    """
+    Test that the get_authz_string_for_user returns the correct format.
+    """
+    user_id = TEST_USER_ID
+    assert get_authz_string_for_user(user_id) == (
+        f"/services/workflow/gen3-workflow/tasks/{user_id}/TASK_ID_PLACEHOLDER"
+    )
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("client", client_parameters, indirect=True)
+@pytest.mark.parametrize("get_all", [False, True])
 @pytest.mark.parametrize("view", ["BASIC", "MINIMAL", "FULL", None])
-async def test_list_tasks(client, access_token_patcher, view, trailing_slash):
+async def test_list_tasks(client, access_token_patcher, get_all, view, trailing_slash):
     """
     Calls to `GET /ga4gh/tes/v1/tasks` should be forwarded to the TES server, and any
     unsupported query params should be filtered out. Tasks the user does not have access
     to should be filtered out.
     When the TES server returns an error, gen3-workflow should return it as well.
     """
-    url = f"/ga4gh/tes/v1/tasks?state=COMPLETE&unsupported_param=value{'/' if trailing_slash else ''}"
+    url = f"/ga4gh/tes/v1/tasks{'/' if trailing_slash else ''}?state=COMPLETE&unsupported_param=value"
     if view:
         url += f"&view={view}"
+    if get_all:
+        url += "&all"
     res = await client.get(url, headers={"Authorization": f"bearer {TEST_USER_TOKEN}"})
 
     # the call to the TES server always has `view=FULL` so we get the _AUTHZ tag
+    query_params = {"state": "COMPLETE", "view": "FULL"}
+    if not get_all:
+        query_params["tag_key"] = "_AUTHZ"
+        query_params["tag_value"] = get_authz_string_for_user(TEST_USER_ID)
     mock_tes_server_request.assert_called_once_with(
         method="GET",
         path="/tasks",
-        query_params={"state": "COMPLETE", "view": "FULL"},
+        query_params=query_params,
         body="",
         status_code=client.tes_resp_code,
     )
@@ -548,6 +566,40 @@ async def test_list_tasks(client, access_token_patcher, view, trailing_slash):
         )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "client",
+    [
+        pytest.param({"tes_resp_code": 200}, id="no user"),
+    ],
+    indirect=True,
+)
+@pytest.mark.usefixtures("access_token_patcher")
+@pytest.mark.parametrize("access_token_patcher", [{"user_id": None}], indirect=True)
+@pytest.mark.parametrize("method", ["get", "post"])
+async def test_tasks_error_no_user(
+    client, access_token_patcher, method, trailing_slash
+):
+    """
+    Calls to `GET|POST /ga4gh/tes/v1/tasks` should return an error when the user_id
+    from authz is None. TES server should not be called.
+    """
+    url = f"/ga4gh/tes/v1/tasks{'/' if trailing_slash else ''}"
+    if method == "get":
+        res = await client.get(
+            url, headers={"Authorization": f"bearer {TEST_USER_TOKEN}"}
+        )
+    if method == "post":
+        res = await client.post(
+            url,
+            json={"name": "test-task"},
+            headers={"Authorization": f"bearer {TEST_USER_TOKEN}"},
+        )
+    assert res.status_code == 401, res.text
+    assert res.json() == {"detail": "No user sub in token"}
+    mock_tes_server_request.assert_not_called()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("client", client_parameters, indirect=True)
 async def test_delete_task(client, access_token_patcher, trailing_slash):