sartography
/
spiff-arena
mirror of https://github.com/sartography/spiff-arena.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

get tests passing

This commit is contained in:
burnettk 2022-12-29 17:13:44 -05:00
parent 6f1510c4f3
commit 2ff73ae0e3
6 changed files with 115 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
spiffworkflow-backend/src/spiffworkflow_backend/api.yml
View File

@ -733,7 +733,7 @@ paths:
schema:
type: string
post:
operationId: spiffworkflow_backend.routes.process_api_blueprint.script_unit_test_create
operationId: spiffworkflow_backend.routes.script_unit_tests_controller.script_unit_test_create
summary: Create script unit test based on given criteria
tags:
- Script Unit Test
@ -754,7 +754,7 @@ paths:
schema:
type: string
post:
operationId: spiffworkflow_backend.routes.process_api_blueprint.script_unit_test_run
operationId: spiffworkflow_backend.routes.script_unit_tests_controller.script_unit_test_run
summary: Run a given script unit test.
tags:
- Script Unit Test

3
spiffworkflow-backend/src/spiffworkflow_backend/routes/messages_controller.py
View File

@ -19,6 +19,9 @@ from spiffworkflow_backend.models.message_triggerable_process_model import (
)
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModelSchema
from spiffworkflow_backend.routes.process_api_blueprint import (
_find_process_instance_by_id_or_raise,
)
from spiffworkflow_backend.services.message_service import MessageService

101
spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py
View File

@ -1,11 +1,8 @@
"""APIs for dealing with process groups, process models, and process instances."""
import json
import uuid
from typing import Any
from typing import Dict
from typing import Optional
from typing import TypedDict
from typing import Union
import connexion # type: ignore
import flask.wrappers
@ -20,9 +17,6 @@ from flask import request
from flask.wrappers import Response
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from lxml import etree # type: ignore
from lxml.builder import ElementMaker # type: ignore
from SpiffWorkflow.task import Task as SpiffTask # type: ignore
from sqlalchemy import and_
from sqlalchemy import or_
@ -54,21 +48,6 @@ from spiffworkflow_backend.services.service_task_service import ServiceTaskServi
from spiffworkflow_backend.services.spec_file_service import SpecFileService
class TaskDataSelectOption(TypedDict):
"""TaskDataSelectOption."""
value: str
label: str
class ReactJsonSchemaSelectOption(TypedDict):
"""ReactJsonSchemaSelectOption."""
type: str
title: str
enum: list[str]
process_api_blueprint = Blueprint("process_api", __name__)
@ -285,64 +264,6 @@ def _get_required_parameter_or_raise(parameter: str, post_body: dict[str, Any])
return return_value
# originally from: https://bitcoden.com/answers/python-nested-dictionary-update-value-where-any-nested-key-matches
def _update_form_schema_with_task_data_as_needed(
in_dict: dict, task_data: dict
) -> None:
"""Update_nested."""
for k, value in in_dict.items():
if "anyOf" == k:
# value will look like the array on the right of "anyOf": ["options_from_task_data_var:awesome_options"]
if isinstance(value, list):
if len(value) == 1:
first_element_in_value_list = value[0]
if isinstance(first_element_in_value_list, str):
if first_element_in_value_list.startswith(
"options_from_task_data_var:"
):
task_data_var = first_element_in_value_list.replace(
"options_from_task_data_var:", ""
)
if task_data_var not in task_data:
raise (
ApiError(
error_code="missing_task_data_var",
message=f"Task data is missing variable: {task_data_var}",
status_code=500,
)
)
select_options_from_task_data = task_data.get(task_data_var)
if isinstance(select_options_from_task_data, list):
if all(
"value" in d and "label" in d
for d in select_options_from_task_data
):
def map_function(
task_data_select_option: TaskDataSelectOption,
) -> ReactJsonSchemaSelectOption:
"""Map_function."""
return {
"type": "string",
"enum": [task_data_select_option["value"]],
"title": task_data_select_option["label"],
}
options_for_react_json_schema_form = list(
map(map_function, select_options_from_task_data)
)
in_dict[k] = options_for_react_json_schema_form
elif isinstance(value, dict):
_update_form_schema_with_task_data_as_needed(value, task_data)
elif isinstance(value, list):
for o in value:
if isinstance(o, dict):
_update_form_schema_with_task_data_as_needed(o, task_data)
def _commit_and_push_to_git(message: str) -> None:
"""Commit_and_push_to_git."""
if current_app.config["GIT_COMMIT_ON_SAVE"]:
@ -499,25 +420,3 @@ def _get_process_model(process_model_id: str) -> ProcessModelInfo:
) from exception
return process_model
def _get_spiff_task_from_process_instance(
task_id: str,
process_instance: ProcessInstanceModel,
processor: Union[ProcessInstanceProcessor, None] = None,
) -> SpiffTask:
"""Get_spiff_task_from_process_instance."""
if processor is None:
processor = ProcessInstanceProcessor(process_instance)
task_uuid = uuid.UUID(task_id)
spiff_task = processor.bpmn_process_instance.get_task(task_uuid)
if spiff_task is None:
raise (
ApiError(
error_code="empty_task",
message="Processor failed to obtain task.",
status_code=500,
)
)
return spiff_task

2
spiffworkflow-backend/src/spiffworkflow_backend/routes/process_instances_controller.py
View File

@ -13,7 +13,7 @@ from flask import request
from flask.wrappers import Response
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from SpiffWorkflow.task import TaskState
from SpiffWorkflow.task import TaskState # type: ignore
from spiffworkflow_backend.models.process_instance import ProcessInstanceApiSchema
from spiffworkflow_backend.models.process_instance import (

6
spiffworkflow-backend/src/spiffworkflow_backend/routes/script_unit_tests_controller.py
View File

@ -1,7 +1,5 @@
"""APIs for dealing with process groups, process models, and process instances."""
import json
from spiffworkflow_backend.routes.process_api_blueprint import _get_required_parameter_or_raise
from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model
import random
import string
from typing import Dict
@ -18,6 +16,10 @@ from lxml import etree # type: ignore
from lxml.builder import ElementMaker # type: ignore
from spiffworkflow_backend.models.principal import PrincipalModel
from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model
from spiffworkflow_backend.routes.process_api_blueprint import (
_get_required_parameter_or_raise,
)
from spiffworkflow_backend.services.script_unit_test_runner import ScriptUnitTestRunner
from spiffworkflow_backend.services.spec_file_service import SpecFileService

106
spiffworkflow-backend/src/spiffworkflow_backend/routes/tasks_controller.py
View File

@ -1,10 +1,11 @@
"""APIs for dealing with process groups, process models, and process instances."""
import json
from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model, _get_required_parameter_or_raise
import os
import uuid
from typing import Any
from typing import Dict
from typing import Optional
from typing import TypedDict
from typing import Union
import flask.wrappers
@ -14,6 +15,7 @@ from flask import jsonify
from flask import make_response
from flask.wrappers import Response
from flask_bpmn.api.api_error import ApiError
from SpiffWorkflow.task import Task as SpiffTask # type: ignore
from SpiffWorkflow.task import TaskState
from sqlalchemy import and_
from sqlalchemy import asc
@ -26,6 +28,13 @@ from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.routes.process_api_blueprint import (
_find_principal_or_raise,
)
from spiffworkflow_backend.routes.process_api_blueprint import (
_find_process_instance_by_id_or_raise,
)
from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.file_system_service import FileSystemService
from spiffworkflow_backend.services.process_instance_processor import (
@ -38,6 +47,21 @@ from spiffworkflow_backend.services.process_model_service import ProcessModelSer
from spiffworkflow_backend.services.spec_file_service import SpecFileService
class TaskDataSelectOption(TypedDict):
"""TaskDataSelectOption."""
value: str
label: str
class ReactJsonSchemaSelectOption(TypedDict):
"""ReactJsonSchemaSelectOption."""
type: str
title: str
enum: list[str]
# TODO: see comment for before_request
# @process_api_blueprint.route("/v1.0/tasks", methods=["GET"])
def task_list_my_tasks(page: int = 1, per_page: int = 100) -> flask.wrappers.Response:
@ -418,3 +442,83 @@ def _render_jinja_template(unprocessed_template: str, data: dict[str, Any]) -> s
)
template = jinja_environment.from_string(unprocessed_template)
return template.render(**data)
def _get_spiff_task_from_process_instance(
task_id: str,
process_instance: ProcessInstanceModel,
processor: Union[ProcessInstanceProcessor, None] = None,
) -> SpiffTask:
"""Get_spiff_task_from_process_instance."""
if processor is None:
processor = ProcessInstanceProcessor(process_instance)
task_uuid = uuid.UUID(task_id)
spiff_task = processor.bpmn_process_instance.get_task(task_uuid)
if spiff_task is None:
raise (
ApiError(
error_code="empty_task",
message="Processor failed to obtain task.",
status_code=500,
)
)
return spiff_task
# originally from: https://bitcoden.com/answers/python-nested-dictionary-update-value-where-any-nested-key-matches
def _update_form_schema_with_task_data_as_needed(
in_dict: dict, task_data: dict
) -> None:
"""Update_nested."""
for k, value in in_dict.items():
if "anyOf" == k:
# value will look like the array on the right of "anyOf": ["options_from_task_data_var:awesome_options"]
if isinstance(value, list):
if len(value) == 1:
first_element_in_value_list = value[0]
if isinstance(first_element_in_value_list, str):
if first_element_in_value_list.startswith(
"options_from_task_data_var:"
):
task_data_var = first_element_in_value_list.replace(
"options_from_task_data_var:", ""
)
if task_data_var not in task_data:
raise (
ApiError(
error_code="missing_task_data_var",
message=f"Task data is missing variable: {task_data_var}",
status_code=500,
)
)
select_options_from_task_data = task_data.get(task_data_var)
if isinstance(select_options_from_task_data, list):
if all(
"value" in d and "label" in d
for d in select_options_from_task_data
):
def map_function(
task_data_select_option: TaskDataSelectOption,
) -> ReactJsonSchemaSelectOption:
"""Map_function."""
return {
"type": "string",
"enum": [task_data_select_option["value"]],
"title": task_data_select_option["label"],
}
options_for_react_json_schema_form = list(
map(map_function, select_options_from_task_data)
)
in_dict[k] = options_for_react_json_schema_form
elif isinstance(value, dict):
_update_form_schema_with_task_data_as_needed(value, task_data)
elif isinstance(value, list):
for o in value:
if isinstance(o, dict):
_update_form_schema_with_task_data_as_needed(o, task_data)