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

fixed tests and added test to ensure we are saving error events to the db on error w/ burnettk

This commit is contained in:
jasquat 2023-04-20 13:32:45 -04:00
parent f082f0966c
commit 92c21f2c11
5 changed files with 34 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py
View File

@ -2019,7 +2019,7 @@ class TestProcessApi(BaseTest):
assert response.status_code == 400
api_error = json.loads(response.get_data(as_text=True))
assert api_error["error_code"] == "task_error"
assert api_error["error_code"] == "unexpected_workflow_exception"
assert 'TypeError:can only concatenate str (not "int") to str' in api_error["message"]
process = db.session.query(ProcessInstanceModel).filter(ProcessInstanceModel.id == process_instance_id).first()
@ -2099,7 +2099,7 @@ class TestProcessApi(BaseTest):
processor = ProcessInstanceProcessor(process_instance)
spiff_task = processor.get_task_by_bpmn_identifier("script_task_two", processor.bpmn_process_instance)
assert spiff_task is not None
assert spiff_task.state == TaskState.WAITING
assert spiff_task.state == TaskState.ERROR
assert spiff_task.data == {"my_var": "THE VAR"}
def test_process_model_file_create(

7
spiffworkflow-backend/tests/spiffworkflow_backend/scripts/test_refresh_permissions.py
View File

@ -5,22 +5,19 @@ from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.exceptions.api_error import ApiError
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError
class TestRefreshPermissions(BaseTest):
"""TestRefreshPermissions."""
def test_refresh_permissions_requires_elevated_permission(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_refresh_permissions_requires_elevated_permission."""
basic_user = self.find_or_create_user("basic_user")
privileged_user = self.find_or_create_user("privileged_user")
self.add_permissions_to_user(
@ -38,7 +35,7 @@ class TestRefreshPermissions(BaseTest):
processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError) as exception:
with pytest.raises(WorkflowExecutionServiceError) as exception:
processor.do_engine_steps(save=True)
assert "ScriptUnauthorizedForUserError" in str(exception)

4
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_error_handling_service.py
View File

@ -5,7 +5,6 @@ from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend import db
from spiffworkflow_backend.exceptions.api_error import ApiError
from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
@ -19,6 +18,7 @@ from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError
class TestErrorHandlingService(BaseTest):
@ -34,7 +34,7 @@ class TestErrorHandlingService(BaseTest):
process_model.id, user
)
pip = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError) as e:
with pytest.raises(WorkflowExecutionServiceError) as e:
pip.do_engine_steps(save=True)
ErrorHandlingService().handle_error(process_instance, e.value)
return process_instance

26
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_processor.py
View File

@ -10,12 +10,12 @@ from SpiffWorkflow.task import TaskState
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.exceptions.api_error import ApiError
from spiffworkflow_backend.models.bpmn_process import BpmnProcessModel
from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
from spiffworkflow_backend.models.process_instance_event import ProcessInstanceEventType
from spiffworkflow_backend.models.task import TaskModel # noqa: F401
from spiffworkflow_backend.models.task_definition import TaskDefinitionModel
from spiffworkflow_backend.models.user import UserModel
@ -29,6 +29,7 @@ from spiffworkflow_backend.services.process_instance_processor import (
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError
class TestProcessInstanceProcessor(BaseTest):
@ -713,7 +714,7 @@ class TestProcessInstanceProcessor(BaseTest):
spiff_task = processor.get_task_by_guid(human_task_three.task_id)
ProcessInstanceService.complete_form_task(processor, spiff_task, {}, initiator_user, human_task_three)
def test_task_data_is_set_even_if_process_instance_errors(
def test_task_data_is_set_even_if_process_instance_errors_and_creates_task_failed_event(
self,
app: Flask,
client: FlaskClient,
@ -731,7 +732,7 @@ class TestProcessInstanceProcessor(BaseTest):
)
processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError):
with pytest.raises(WorkflowExecutionServiceError):
processor.do_engine_steps(save=True)
process_instance_final = ProcessInstanceModel.query.filter_by(id=process_instance.id).first()
@ -741,5 +742,22 @@ class TestProcessInstanceProcessor(BaseTest):
"script_task_two", processor_final.bpmn_process_instance
)
assert spiff_task is not None
assert spiff_task.state == TaskState.WAITING
assert spiff_task.state == TaskState.ERROR
assert spiff_task.data == {"my_var": "THE VAR"}
process_instance_events = process_instance.process_instance_events
assert len(process_instance_events) == 4
error_events = [
e for e in process_instance_events if e.event_type == ProcessInstanceEventType.task_failed.value
]
assert len(error_events) == 1
error_event = error_events[0]
assert error_event.task_guid is not None
process_instance_error_details = error_event.error_details
assert len(process_instance_error_details) == 1
error_detail = process_instance_error_details[0]
assert error_detail.message == "NameError:name 'hey' is not defined. Did you mean 'my_var'?"
assert error_detail.task_offset is None
assert error_detail.task_line_number == 1
assert error_detail.task_line_contents == "hey"
assert error_detail.task_trace is not None

20
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_restricted_script_engine.py
View File

@ -5,24 +5,21 @@ from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.exceptions.api_error import ApiError
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError
class TestOpenFile(BaseTest):
"""TestVariousBpmnConstructs."""
def test_dot_notation(
class TestRestrictedScriptEngine(BaseTest):
def test_dot_notation_with_open_file(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_form_data_conversion_to_dot_dict."""
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
process_model = load_test_spec(
"test_group/dangerous",
@ -34,22 +31,17 @@ class TestOpenFile(BaseTest):
process_instance = self.create_process_instance_from_process_model(process_model)
processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError) as exception:
with pytest.raises(WorkflowExecutionServiceError) as exception:
processor.do_engine_steps(save=True)
assert "name 'open' is not defined" in str(exception.value)
class TestImportModule(BaseTest):
"""TestVariousBpmnConstructs."""
def test_dot_notation(
def test_dot_notation_with_import_module(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_form_data_conversion_to_dot_dict."""
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
process_model = load_test_spec(
"test_group/dangerous",
@ -61,6 +53,6 @@ class TestImportModule(BaseTest):
process_instance = self.create_process_instance_from_process_model(process_model)
processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError) as exception:
with pytest.raises(WorkflowExecutionServiceError) as exception:
processor.do_engine_steps(save=True)
assert "Import not allowed: os" in str(exception.value)