Bugfix issue 144 (#148)

* updated format for passing workflow prompt variables

* code formatting

* updated documentation

* added latest bugfix

* test prompts passed to run_workflow_definition

* fixed issue with pre-epoch dates on Windows

Author: jlwalke2

CHANGELOG.md

@@ -3,6 +3,11 @@ Unreleased
 **Improvements**
  - `folders.get_folder()` can now handle folder paths and delegates (e.g. @public).
 
+**Bugfixes**
+ - Fixed an issue with `model_management.execute_model_workflow_definition()` where input values for
+   workflow prompts were not correctly submitted.  Note that the `input=` parameter was renamed to
+   `prompts=` to avoid conflicting with the built-in `input()`.
+
 v1.8.1 (2023-01-19)
 ----------
 **Changes**

src/sasctl/_services/model_management.py

@@ -311,15 +311,13 @@ def list_model_workflow_definition(cls):
 
         Returns
         -------
-        RestObj
+        list of RestObj
             The list of workflows
 
         """
         from .workflow import Workflow
 
-        wf = Workflow()
-
-        return wf.list_enabled_definitions()
+        return Workflow.list_enabled_definitions()
 
     @classmethod
     @experimental
@@ -339,9 +337,7 @@ def list_model_workflow_prompt(cls, workflowName):
         """
         from .workflow import Workflow
 
-        wf = Workflow()
-
-        return wf.list_workflow_prompt(workflowName)
+        return Workflow.list_workflow_prompt(workflowName)
 
     @classmethod
     @experimental
@@ -373,7 +369,9 @@ def list_model_workflow_executed(cls, projectName):
 
     @classmethod
     @experimental
-    def execute_model_workflow_definition(cls, project_name, workflow_name, input=None):
+    def execute_model_workflow_definition(
+        cls, project_name, workflow_name, prompts=None
+    ):
         """Runs specific Workflow Processes Definitions.
 
         Parameters
@@ -382,29 +380,31 @@ def execute_model_workflow_definition(cls, project_name, workflow_name, input=No
             Name of the Project that will execute workflow
         workflow_name : str
             Name or ID of an enabled workflow to execute
-        input : dict, optional
-            Input values for the workflow for initial workflow prompt
+        prompts : dict, optional
+            Input values to provide for the initial workflow prompts.  Should be
+            specified as name:value pairs.
 
         Returns
         -------
         RestObj
             The executing workflow
 
+        .. versionchanged:: 1.8.2
+            Renamed the `input` parameter to `prompts`.
+
         """
         from .model_repository import ModelRepository
         from .workflow import Workflow
 
         mr = ModelRepository()
-        wf = Workflow()
 
         project = mr.get_project(project_name)
 
-        workflow = wf.run_workflow_definition(workflow_name, input=input)
-
-        # Associations running workflow to model project, note workflow has to be running
-        # THINK ABOUT: do we do a check on status of the workflow to determine if it is still running before associating?
+        workflow = Workflow.run_workflow_definition(workflow_name, prompts=prompts)
 
-        input = {
+        # Associate running workflow to model project.
+        # NOTE: workflow has to be running
+        data = {
             "processName": workflow["name"],
             "processId": workflow["id"],
             "objectType": "MM_Project",
@@ -414,14 +414,15 @@ def execute_model_workflow_definition(cls, project_name, workflow_name, input=No
             "solutionObjectMediaType": "application/vnd.sas.models.project+json",
         }
 
-        # Note, you can get a HTTP Error 404: {"errorCode":74052,"message":"The workflow process for id <> cannot be found.
-        #                                    Associations can only be made to running processes.","details":["correlator:
-        #                                    e62c5562-2b11-45db-bcb7-933200cb0f0a","traceId: 3118c0fb1eb9702d","path:
-        #                                    /modelManagement/workflowAssociations"],"links":[],"version":2,"httpStatusCode":404}
+        # Note: you can get a HTTP Error 404:
+        # {"errorCode":74052,"message":"The workflow process for id <> cannot be found.
+        # Associations can only be made to running processes.","details":["correlator:
+        # e62c5562-2b11-45db-bcb7-933200cb0f0a","traceId: 3118c0fb1eb9702d","path:
+        # /modelManagement/workflowAssociations"],"links":[],"version":2,"httpStatusCode":404}
         # Which is fine and expected like the Visual Experience.
         return cls.post(
             "/workflowAssociations",
-            json=input,
+            json=data,
             headers={
                 "Content-Type": "application/vnd.sas.workflow.object.association+json"
             },

src/sasctl/_services/workflow.py

@@ -3,18 +3,30 @@
 #
 # Copyright © 2019, SAS Institute Inc., Cary, NC, USA.  All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+import datetime
 
+from ..core import PagedItemIterator
 from .service import Service
 
 
 class Workflow(Service):
     """The Workflow API provides basic resources for list, prompt details,
     and running workflow processes.
+
+    Warnings
+    --------
+    Note that this service is intentionally not included in the sasctl documentation.
+    As of 2022.09 this service is not publicly documented on developer.sas.com and is
+    only partially implemented here in order to provide functionality for the Model
+    Manager service.  All methods included in this service should be treated as
+    experimental and are subject to change without notice.
+
     """
 
     _SERVICE_ROOT = "/workflow"
 
-    def list_definitions(self, include_enabled=True, include_disabled=False):
+    @classmethod
+    def list_definitions(cls, include_enabled=True, include_disabled=False):
         """List workflow definitions.
 
         Parameters
@@ -40,9 +52,16 @@ def list_definitions(self, include_enabled=True, include_disabled=False):
             return []
 
         # Header required to prevent 400 ERROR Bad Request
-        return self.get(url, headers={"Accept-Language": "en-US"})
+        results = cls.get(url, headers={"Accept-Language": "en-US"})
 
-    def list_enabled_definitions(self):
+        if results is None:
+            return []
+        if isinstance(results, (list, PagedItemIterator)):
+            return results
+        return [results]
+
+    @classmethod
+    def list_enabled_definitions(cls):
         """List process definitions that are currently enabled.
 
         Returns
@@ -51,9 +70,10 @@ def list_enabled_definitions(self):
             The list of definitions.
 
         """
-        return self.list_definitions(include_enabled=True, include_disabled=False)
+        return cls.list_definitions(include_enabled=True, include_disabled=False)
 
-    def list_workflow_prompt(self, name):
+    @classmethod
+    def list_workflow_prompt(cls, name):
         """List prompt Workflow Processes Definitions.
 
         Parameters
@@ -68,7 +88,7 @@ def list_workflow_prompt(self, name):
 
         """
 
-        ret = self._find_specific_workflow(name)
+        ret = cls._find_specific_workflow(name)
         if ret is None:
             raise ValueError("No Workflow enabled for %s name or id." % name)
 
@@ -78,44 +98,86 @@ def list_workflow_prompt(self, name):
             # No prompt inputs on workflow
             return None
 
-    def run_workflow_definition(self, name, input=None):
+    @classmethod
+    def run_workflow_definition(cls, name, prompts=None):
         """Runs specific Workflow Processes Definitions.
 
         Parameters
         ----------
         name : str
             Name or ID of an enabled workflow to execute
-        input : dict, optional
-            Input values for the workflow for initial workflow prompt
+        prompts : dict, optional
+            Input values to provide for the initial workflow prompts.  Should be
+            specified as name:value pairs.
 
         Returns
         -------
         RestObj
             The executing workflow
 
         """
-
-        workflow = self._find_specific_workflow(name)
+        workflow = cls._find_specific_workflow(name)
         if workflow is None:
             raise ValueError("No Workflow enabled for %s name or id." % name)
 
-        if input is None:
-            return self.post(
+        if prompts is None:
+            return cls.post(
                 "/processes?definitionId=" + workflow["id"],
                 headers={"Content-Type": "application/vnd.sas.workflow.variables+json"},
             )
-        if isinstance(input, dict):
-            return self.post(
+        if isinstance(prompts, dict):
+
+            variables = []
+
+            # For each prompt defined in the workflow, check if a value was provided.
+            for prompt in workflow.prompts:
+                if prompt["variableName"] in prompts:
+                    name = prompt["variableName"]
+                    value = prompts[name]
+
+                    if type(value) == datetime.datetime:
+                        # NOTE: do not use isinstance() to compare types as
+                        # datetime.date will also evaluate as True.
+
+                        # Explicitly convert to local time zone if not set.
+                        if value.tzinfo is None:
+                            try:
+                                value = value.astimezone()
+                            except OSError:
+                                # On Windows pre-1970 dates will cause an issue.
+                                # See https://bugs.python.org/issue36759
+                                pass
+
+                        if value.tzinfo is None:
+                            # Failed to convert to local time.  Have to just assume it's UTC.
+                            # Example: 2023-01-25T13:49:40.726162Z
+                            value = value.isoformat() + "Z"
+                        else:
+                            # Example: 2023-01-25T13:49:40.726162-05:00
+                            value = value.isoformat()
+
+                    variables.append(
+                        {
+                            "name": name,
+                            "value": value,
+                            "scope": "local",
+                            "type": prompt["variableType"],
+                            "version": prompt["version"],
+                        }
+                    )
+
+            return cls.post(
                 "/processes?definitionId=" + workflow["id"],
-                json=input,
+                json={"variables": variables},
                 headers={"Content-Type": "application/vnd.sas.workflow.variables+json"},
             )
 
-    def _find_specific_workflow(self, name):
+    @classmethod
+    def _find_specific_workflow(cls, name):
         # Internal helper method
         # Finds a workflow with the name (can be a name or id)
         # Returns a dict objects of the workflow
-        listendef = self.list_enabled_definitions()
+        listendef = cls.list_enabled_definitions()
         for tmp in listendef:
             if tmp["name"] == name:
                 return tmp

tests/unit/test_workflow.py

@@ -4,10 +4,12 @@
 # Copyright © 2019, SAS Institute Inc., Cary, NC, USA.  All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import datetime
 from unittest import mock
 
 import pytest
 
+from sasctl.core import RestObj
 from sasctl._services import workflow
 
 
@@ -92,3 +94,76 @@ def test_list_workflow_prompt_workflowprompt():
         testresult = wf.list_workflow_prompt("98765")
         print(testresult)
         assert testresult is not None
+
+
+@mock.patch("sasctl._services.workflow.Workflow.post")
+@mock.patch("sasctl._services.workflow.Workflow._find_specific_workflow")
+def test_run_workflow_definition_no_prompts(get_workflow, post):
+    """Verify correct REST call to run a workflow with no inputs."""
+
+    DEFINITION_ID = "abc-123"
+
+    get_workflow.return_value = RestObj(name="Inquisition", id=DEFINITION_ID)
+    post.return_value.status_code = 200
+    post.return_value.json.return_value = {}
+
+    workflow.Workflow.run_workflow_definition("inquisition")
+
+    assert post.call_count == 1
+    url, params = post.call_args
+    assert url[0].startswith("/processes")
+    assert DEFINITION_ID in url[0]
+    assert "json" not in params  # should not have passed any prompt info
+
+
+@mock.patch("sasctl._services.workflow.Workflow.post")
+@mock.patch("sasctl._services.workflow.Workflow._find_specific_workflow")
+def test_run_workflow_definition_with_prompts(get_workflow, post):
+    """Verify correct REST call and prompt formatting."""
+
+    # Mock response from looking up workflow information.
+    # NOTE: needs to include prompt information to match with prompt input values.
+    WORKFLOW = RestObj(
+        {
+            "name": "Inquisition",
+            "id": "abc-123",
+            "prompts": [
+                {"variableName": "moto", "variableType": "string", "version": 3},
+                {"variableName": "date", "variableType": "dateTime", "version": 1},
+            ],
+        }
+    )
+
+    PROMPTS = {
+        "moto": "No one expects the Spanish Inquisition!",
+        "date": datetime.datetime(1912, 1, 1, 17, 30),
+    }
+
+    # Return mock workflow when asked
+    get_workflow.return_value = WORKFLOW
+
+    # Return a valid HTTP response for POSTs
+    post.return_value.status_code = 200
+    post.return_value.json.return_value = {}
+
+    workflow.Workflow.run_workflow_definition("inquisition", prompts=PROMPTS)
+
+    assert post.call_count == 1
+    url, params = post.call_args
+
+    assert url[0].startswith("/processes")
+    assert WORKFLOW["id"] in url[0]
+
+    # Check each prompt value that was passed and ensure it was correctly
+    # matched to the prompts defined by the workflow.
+    for name, value in PROMPTS.items():
+
+        # Find the matching variable entry in the POST data
+        variable = next(v for v in params["json"]["variables"] if v["name"] == name)
+
+        # String and datetime prompts should be passed as string values
+        if isinstance(value, (str, datetime.datetime)):
+            assert isinstance(variable["value"], str)
+
+        if isinstance(value, datetime.datetime):
+            assert variable["type"] == "dateTime"