added api to allow creating script unit tests w/ burnettk

src/spiffworkflow_backend/api.yml

@@ -474,6 +474,33 @@ paths:
                 items:
                   $ref: "#/components/schemas/Workflow"
 
+  /process-models/{process_group_id}/{process_model_id}/script-unit-tests:
+    parameters:
+      - name: process_group_id
+        in: path
+        required: true
+        description: The unique id of an existing process group
+        schema:
+          type: string
+      - name: process_model_id
+        in: path
+        required: true
+        description: The unique id of an existing process model.
+        schema:
+          type: string
+    post:
+      operationId: spiffworkflow_backend.routes.process_api_blueprint.script_unit_test_create
+      summary: Create script unit test based on given criteria
+      tags:
+        - Script Unit Test
+      responses:
+        "200":
+          description: Script Unit Test Result
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Workflow"
+
   /process-models/{process_group_id}/{process_model_id}/script-unit-tests/run:
     parameters:
       - name: process_group_id

src/spiffworkflow_backend/models/task.py

@@ -1,6 +1,6 @@
 """Task."""
 import enum
-from typing import Any
+from typing import Any, Optional
 from typing import Union
 
 import marshmallow
@@ -113,6 +113,7 @@ class Task:
         process_model_display_name: Union[str, None] = None,
         form_schema: Union[str, None] = None,
         form_ui_schema: Union[str, None] = None,
+        parent: Optional[str] = None,
     ):
         """__init__."""
         self.id = id
@@ -123,6 +124,7 @@ class Task:
         self.form = form
         self.documentation = documentation
         self.lane = lane
+        self.parent = parent
 
         self.data = data
         if self.data is None:
@@ -174,6 +176,7 @@ class Task:
             "process_model_display_name": self.process_model_display_name,
             "form_schema": self.form_schema,
             "form_ui_schema": self.form_ui_schema,
+            "parent": self.parent,
         }
 
     @classmethod

src/spiffworkflow_backend/routes/process_api_blueprint.py

@@ -64,6 +64,11 @@ from spiffworkflow_backend.services.service_task_service import ServiceTaskServi
 from spiffworkflow_backend.services.spec_file_service import SpecFileService
 from spiffworkflow_backend.services.user_service import UserService
 
+from lxml import etree
+from lxml.builder import ElementMaker
+import random
+import string
+
 process_api_blueprint = Blueprint("process_api", __name__)
 
 
@@ -909,7 +914,6 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response
                 task.properties["instructionsForEndUser"] = render_jinja_template(
                     task.properties["instructionsForEndUser"], task.data
                 )
-
     return make_response(jsonify(task), 200)
 
 
@@ -975,6 +979,82 @@ def task_submit(
     return Response(json.dumps({"ok": True}), status=202, mimetype="application/json")
 
 
+def script_unit_test_create(
+    process_group_id: str, process_model_id: str, body: Dict[str, Union[str, bool, int]]
+) -> flask.wrappers.Response:
+    """Script_unit_test_run."""
+    bpmn_task_identifier = get_required_parameter_or_raise("bpmn_task_identifier", body)
+    input_json = get_required_parameter_or_raise("input_json", body)
+    expected_output_json = get_required_parameter_or_raise("expected_output_json", body)
+
+    process_model = get_process_model(process_model_id, process_group_id)
+    file = SpecFileService.get_files(process_model, process_model.primary_file_name)[0]
+    if file is None:
+        raise ApiError(
+            code="cannot_find_file",
+            message=f"Could not find the primary bpmn file for process_model: {process_model.id}",
+            status_code=404,
+        )
+
+    # TODO: move this to an xml service or something
+    file_contents = SpecFileService.get_data(process_model, file.name)
+    bpmn_etree_element = SpecFileService.get_etree_element_from_binary_data(
+        file_contents, file.name
+    )
+
+    nsmap = bpmn_etree_element.nsmap
+    spiffElementMaker = ElementMaker(namespace="http://spiffworkflow.org/bpmn/schema/1.0/core", nsmap=nsmap)
+
+    script_task_elements = bpmn_etree_element.xpath(
+        f"//bpmn:scriptTask[@id='{bpmn_task_identifier}']",
+        namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"},
+    )
+    if len(script_task_elements) == 0:
+        raise ApiError(
+            code="missing_script_task",
+            message=f"Cannot find a script task with id: {bpmn_process_identifier}",
+            status_code=404,
+        )
+    script_task_element = script_task_elements[0]
+
+    extension_elements = None
+    extension_elements_array = script_task_element.xpath(
+        f"//bpmn:extensionElements",
+        namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"},
+    )
+    if len(extension_elements_array) == 0:
+        bpmnElementMaker = ElementMaker(namespace="http://www.omg.org/spec/BPMN/20100524/MODEL", nsmap=nsmap)
+        extension_elements = bpmnElementMaker('extensionElements')
+        script_task_element.append(extension_elements)
+    else:
+        extension_elements = extension_elements_array[0]
+
+    unit_test_elements = None
+    unit_test_elements_array = extension_elements.xpath(
+        f"//spiffworkflow:unitTests",
+        namespaces={"spiffworkflow": "http://spiffworkflow.org/bpmn/schema/1.0/core"},
+    )
+    if len(unit_test_elements_array) == 0:
+        unit_test_elements = spiffElementMaker('unitTests')
+        extension_elements.append(unit_test_elements)
+    else:
+        unit_test_elements = unit_test_elements_array[0]
+
+    fuzz = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(7))
+    unit_test_id = f"unit_test_{fuzz}"
+
+    input_json_element = spiffElementMaker('inputJson', json.dumps(input_json))
+    expected_output_json_element = spiffElementMaker('expectedOutputJson', json.dumps(expected_output_json))
+    unit_test_element = spiffElementMaker('unitTest', id=unit_test_id)
+    unit_test_element.append(input_json_element)
+    unit_test_element.append(expected_output_json_element)
+    unit_test_elements.append(unit_test_element)
+    SpecFileService.update_file(process_model, file.name, etree.tostring(bpmn_etree_element))
+
+    return Response(json.dumps({"ok": True}), status=202, mimetype="application/json")
+
+
+
 def script_unit_test_run(
     process_group_id: str, process_model_id: str, body: Dict[str, Union[str, bool, int]]
 ) -> flask.wrappers.Response:

src/spiffworkflow_backend/services/process_instance_service.py

@@ -406,6 +406,10 @@ class ProcessInstanceService:
         else:
             lane = None
 
+        parent_id = None
+        if spiff_task.parent:
+            parent_id = spiff_task.parent.id
+
         task = Task(
             spiff_task.id,
             spiff_task.task_spec.name,
@@ -418,6 +422,7 @@ class ProcessInstanceService:
             multi_instance_index=info["mi_index"],
             process_name=spiff_task.task_spec._wf_spec.description,
             properties=props,
+            parent=parent_id,
         )
 
         return task