Fix errors to be able to fetch soft deleted workflows that only exist in the DB (#1162)

* Add unit tests for fetching process with a soft deleted workflow

* Change get_auth_callbacks to work with optional workflow param

* Convert values to boolean for retryAllowed and resumeAllowed for graphql process user_permissions

* Bumpversion to 4.6.2

---------

Co-authored-by: Peter Boers <[email protected]>

Author: tjeerddie

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.6.1
+current_version = 4.6.2
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(rc(?P<build>\d+))?

orchestrator/__init__.py

@@ -13,7 +13,7 @@
 
 """This is the orchestrator workflow engine."""
 
-__version__ = "4.6.1"
+__version__ = "4.6.2"
 
 
 from structlog import get_logger

orchestrator/api/api_v1/endpoints/processes.py

@@ -101,7 +101,7 @@ def get_steps_to_evaluate_for_rbac(pstat: ProcessStat) -> StepList:
     return StepList(past_steps >> first(remaining_steps))
 
 
-def get_auth_callbacks(steps: StepList, workflow: Workflow) -> tuple[Authorizer | None, Authorizer | None]:
+def get_auth_callbacks(steps: StepList, workflow: Workflow | None) -> tuple[Authorizer | None, Authorizer | None]:
     """Iterate over workflow and prior steps to determine correct authorization callbacks for the current step.
 
     It's safest to always iterate through the steps. We could track these callbacks statefully
@@ -112,6 +112,9 @@ def get_auth_callbacks(steps: StepList, workflow: Workflow) -> tuple[Authorizer
     - RESUME callback is explicit RESUME callback, else previous START/RESUME callback
     - RETRY callback is explicit RETRY, else explicit RESUME, else previous RETRY
     """
+    if not workflow:
+        return None, None
+
     # Default to workflow start callbacks
     auth_resume = workflow.authorize_callback
     # auth_retry defaults to the workflow start callback if not otherwise specified.

orchestrator/graphql/schemas/process.py

@@ -86,11 +86,11 @@ async def user_permissions(self, info: OrchestratorInfo) -> FormUserPermissionsT
         oidc_user = await info.context.get_current_user
         workflow = get_workflow(self.workflow_name)
         process = load_process(db.session.get(ProcessTable, self.process_id))  # type: ignore[arg-type]
-        auth_resume, auth_retry = get_auth_callbacks(get_steps_to_evaluate_for_rbac(process), workflow)  # type: ignore[arg-type]
+        auth_resume, auth_retry = get_auth_callbacks(get_steps_to_evaluate_for_rbac(process), workflow)
 
         return FormUserPermissionsType(
-            retryAllowed=auth_retry and auth_retry(oidc_user),  # type: ignore[arg-type]
-            resumeAllowed=auth_resume and auth_resume(oidc_user),  # type: ignore[arg-type]
+            retryAllowed=bool(auth_retry and auth_retry(oidc_user)),
+            resumeAllowed=bool(auth_resume and auth_resume(oidc_user)),
         )
 
     @authenticated_field(description="Returns list of subscriptions of the process")  # type: ignore

test/unit_tests/conftest.py

@@ -42,7 +42,12 @@
 from orchestrator.utils.json import json_dumps
 from orchestrator.utils.redis_client import create_redis_client
 from pydantic_forms.core import FormPage
-from test.unit_tests.fixtures.processes import mocked_processes, mocked_processes_resumeall, test_workflow  # noqa: F401
+from test.unit_tests.fixtures.processes import (  # noqa: F401
+    mocked_processes,
+    mocked_processes_resumeall,
+    test_workflow,
+    test_workflow_soft_deleted,
+)
 from test.unit_tests.fixtures.products.product_blocks.product_block_list_nested import (  # noqa: F401
     test_product_block_list_nested,
     test_product_block_list_nested_db_in_use_by_block,

test/unit_tests/fixtures/processes.py

@@ -5,10 +5,11 @@
 
 import pytest
 import pytz
-from sqlalchemy import delete
+from sqlalchemy import delete, select
 
 from orchestrator.config.assignee import Assignee
 from orchestrator.db import ProcessStepTable, ProcessSubscriptionTable, ProcessTable, db
+from orchestrator.db.models import WorkflowTable
 from orchestrator.workflow import done, init, inputstep, step, workflow
 from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator, UUIDstr
@@ -53,6 +54,18 @@ def workflow_for_testing_processes_py():
         yield wf
 
 
+@pytest.fixture
+def test_workflow_soft_deleted(test_workflow) -> Generator:
+    stmt = select(WorkflowTable).where(WorkflowTable.workflow_id == test_workflow.workflow_id)
+    workflow = db.session.scalar(stmt)
+    assert workflow
+    workflow.deleted_at = datetime.now().tzinfo
+    db.session.add(workflow)
+    db.session.commit()
+
+    yield test_workflow
+
+
 @pytest.fixture
 def mocked_processes(test_workflow, generic_subscription_1, generic_subscription_2):
     first_datetime = datetime(2020, 1, 14, 9, 30, tzinfo=pytz.utc)

test/unit_tests/graphql/test_process.py

@@ -12,13 +12,18 @@
 # limitations under the License.
 import json
 from http import HTTPStatus
+from unittest import mock
 
 
 def build_simple_query(process_id):
     q = """
         query ProcessQuery($processId: UUID!) {
             process(processId: $processId) {
                 processId
+                userPermissions {
+                    retryAllowed
+                    resumeAllowed
+                }
             }
         }
         """
@@ -39,4 +44,28 @@ def test_process(test_client, mocked_processes):
 
     response = test_client.post("/api/graphql", content=test_query, headers={"Content-Type": "application/json"})
     assert response.status_code == HTTPStatus.OK
-    assert response.json() == {"data": {"process": {"processId": str(process_id)}}}
+    assert response.json() == {
+        "data": {
+            "process": {"processId": str(process_id), "userPermissions": {"resumeAllowed": True, "retryAllowed": True}}
+        }
+    }
+
+
+@mock.patch("orchestrator.graphql.schemas.process.get_workflow")
+def test_process_is_allowed_with_historic_workflow_only_left_in_db(
+    mock_get_workflow, test_client, mocked_processes, test_workflow_soft_deleted
+):
+    mock_get_workflow.return_value = None
+    process_id = mocked_processes[0]
+    test_query = build_simple_query(process_id)
+
+    response = test_client.post("/api/graphql", content=test_query, headers={"Content-Type": "application/json"})
+    assert response.status_code == HTTPStatus.OK
+    assert response.json() == {
+        "data": {
+            "process": {
+                "processId": str(process_id),
+                "userPermissions": {"resumeAllowed": False, "retryAllowed": False},
+            }
+        }
+    }