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

do not raise errors if a process cannot be found in the spec reference cache when trying to get a task trace for an error

This commit is contained in:
jasquat 2023-05-04 09:45:01 -04:00
parent d769143c83
commit f65b301635
6 changed files with 53 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
spiffworkflow-backend/src/spiffworkflow_backend/config/__init__.py
View File

@ -18,13 +18,13 @@ def setup_database_uri(app: Flask) -> None:
if app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_URI") is None:
database_name = f"spiffworkflow_backend_{app.config['ENV_IDENTIFIER']}"
if app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_TYPE") == "sqlite":
app.config[
"SQLALCHEMY_DATABASE_URI"
] = f"sqlite:///{app.instance_path}/db_{app.config['ENV_IDENTIFIER']}.sqlite3"
app.config["SQLALCHEMY_DATABASE_URI"] = (
f"sqlite:///{app.instance_path}/db_{app.config['ENV_IDENTIFIER']}.sqlite3"
)
elif app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_TYPE") == "postgres":
app.config[
"SQLALCHEMY_DATABASE_URI"
] = f"postgresql://spiffworkflow_backend:spiffworkflow_backend@localhost:5432/{database_name}"
app.config["SQLALCHEMY_DATABASE_URI"] = (
f"postgresql://spiffworkflow_backend:spiffworkflow_backend@localhost:5432/{database_name}"
)
else:
# use pswd to trick flake8 with hardcoded passwords
db_pswd = app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_PASSWORD")

4
spiffworkflow-backend/src/spiffworkflow_backend/exceptions/api_error.py
View File

@ -127,8 +127,8 @@ class ApiError(Exception):
instance.task_trace = TaskModelError.get_task_trace(task_model)
try:
spec_reference = TaskService.get_spec_reference_from_bpmn_process(task_model.bpmn_process)
instance.file_name = spec_reference.file_name
spec_reference_filename = TaskService.get_spec_filename_from_bpmn_process(task_model.bpmn_process)
instance.file_name = spec_reference_filename
except Exception as exception:
current_app.logger.error(exception)

6
spiffworkflow-backend/src/spiffworkflow_backend/models/process_instance.py
View File

@ -129,9 +129,9 @@ class ProcessInstanceModel(SpiffworkflowBaseDBModel):
def serialized_with_metadata(self) -> dict[str, Any]:
process_instance_attributes = self.serialized
process_instance_attributes["process_metadata"] = self.process_metadata
process_instance_attributes[
"process_model_with_diagram_identifier"
] = self.process_model_with_diagram_identifier
process_instance_attributes["process_model_with_diagram_identifier"] = (
self.process_model_with_diagram_identifier
)
return process_instance_attributes
@property

12
spiffworkflow-backend/src/spiffworkflow_backend/services/process_instance_processor.py
View File

@ -423,9 +423,9 @@ class ProcessInstanceProcessor:
tld.process_instance_id = process_instance_model.id
# we want this to be the fully qualified path to the process model including all group subcomponents
current_app.config[
"THREAD_LOCAL_DATA"
].process_model_identifier = f"{process_instance_model.process_model_identifier}"
current_app.config["THREAD_LOCAL_DATA"].process_model_identifier = (
f"{process_instance_model.process_model_identifier}"
)
self.process_instance_model = process_instance_model
self.process_model_service = ProcessModelService()
@ -585,9 +585,9 @@ class ProcessInstanceProcessor:
bpmn_subprocess_definition.bpmn_identifier
] = bpmn_process_definition_dict
spiff_bpmn_process_dict["subprocess_specs"][bpmn_subprocess_definition.bpmn_identifier]["task_specs"] = {}
bpmn_subprocess_definition_bpmn_identifiers[
bpmn_subprocess_definition.id
] = bpmn_subprocess_definition.bpmn_identifier
bpmn_subprocess_definition_bpmn_identifiers[bpmn_subprocess_definition.id] = (
bpmn_subprocess_definition.bpmn_identifier
)
task_definitions = TaskDefinitionModel.query.filter(
TaskDefinitionModel.bpmn_process_definition_id.in_( # type: ignore

17
spiffworkflow-backend/src/spiffworkflow_backend/services/task_service.py
View File

@ -89,15 +89,15 @@ class TaskModelError(Exception):
task_definition = task_model.task_definition
task_bpmn_name = TaskService.get_name_for_display(task_definition)
bpmn_process = task_model.bpmn_process
spec_reference = TaskService.get_spec_reference_from_bpmn_process(bpmn_process)
spec_reference_filename = TaskService.get_spec_filename_from_bpmn_process(bpmn_process)
task_trace = [f"{task_bpmn_name} ({spec_reference.file_name})"]
task_trace = [f"{task_bpmn_name} ({spec_reference_filename})"]
while bpmn_process.guid is not None:
caller_task_model = TaskModel.query.filter_by(guid=bpmn_process.guid).first()
bpmn_process = BpmnProcessModel.query.filter_by(id=bpmn_process.direct_parent_process_id).first()
spec_reference = TaskService.get_spec_reference_from_bpmn_process(bpmn_process)
spec_reference_filename = TaskService.get_spec_filename_from_bpmn_process(bpmn_process)
task_trace.append(
f"{TaskService.get_name_for_display(caller_task_model.task_definition)} ({spec_reference.file_name})"
f"{TaskService.get_name_for_display(caller_task_model.task_definition)} ({spec_reference_filename})"
)
return task_trace
@ -630,6 +630,15 @@ class TaskService:
result.append({"event": event_definition, "label": extensions["signalButtonLabel"]})
return result
@classmethod
def get_spec_filename_from_bpmn_process(cls, bpmn_process: BpmnProcessModel) -> Optional[str]:
"""Just return the filename if the bpmn process is found in spec reference cache."""
try:
filename: Optional[str] = cls.get_spec_reference_from_bpmn_process(bpmn_process).file_name
return filename
except SpecReferenceNotFoundError:
return None
@classmethod
def get_spec_reference_from_bpmn_process(cls, bpmn_process: BpmnProcessModel) -> SpecReferenceCache:
"""Get the bpmn file for a given task model.

46
spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_for_good_errors.py
View File

@ -15,27 +15,6 @@ from spiffworkflow_backend.routes.tasks_controller import _dequeued_interstitial
class TestForGoodErrors(BaseTest):
"""Assure when certain errors happen when rendering a jinaj2 error that it makes some sense."""
def get_next_user_task(
self,
process_instance_id: int,
client: FlaskClient,
with_super_admin_user: UserModel,
) -> Any:
# Call this to assure all engine-steps are fully processed before we search for human tasks.
_dequeued_interstitial_stream(process_instance_id)
"""Returns the next available user task for a given process instance, if possible."""
human_tasks = (
db.session.query(HumanTaskModel).filter(HumanTaskModel.process_instance_id == process_instance_id).all()
)
assert len(human_tasks) > 0, "No human tasks found for process."
human_task = human_tasks[0]
response = client.get(
f"/v1.0/tasks/{process_instance_id}/{human_task.task_id}",
headers=self.logged_in_headers(with_super_admin_user),
)
return response
def test_invalid_form(
self,
app: Flask,
@ -61,7 +40,7 @@ class TestForGoodErrors(BaseTest):
headers=self.logged_in_headers(with_super_admin_user),
)
assert response.status_code == 200
response = self.get_next_user_task(process_instance_id, client, with_super_admin_user)
response = self._get_next_user_task(process_instance_id, client, with_super_admin_user)
assert response.json is not None
assert response.json["error_type"] == "TemplateSyntaxError"
assert response.json["line_number"] == 3
@ -88,7 +67,7 @@ class TestForGoodErrors(BaseTest):
f"/v1.0/process-instances/{self.modify_process_identifier_for_path_param(process_model.id)}/{process_instance.id}/run",
headers=self.logged_in_headers(with_super_admin_user),
)
response = self.get_next_user_task(process_instance.id, client, with_super_admin_user)
response = self._get_next_user_task(process_instance.id, client, with_super_admin_user)
assert response.status_code == 400
assert response.json is not None
@ -99,3 +78,24 @@ class TestForGoodErrors(BaseTest):
assert "instructions for end user" in response.json["message"]
assert "Jinja2" in response.json["message"]
assert "unexpected '='" in response.json["message"]
def _get_next_user_task(
self,
process_instance_id: int,
client: FlaskClient,
with_super_admin_user: UserModel,
) -> Any:
# Call this to assure all engine-steps are fully processed before we search for human tasks.
_dequeued_interstitial_stream(process_instance_id)
"""Returns the next available user task for a given process instance, if possible."""
human_tasks = (
db.session.query(HumanTaskModel).filter(HumanTaskModel.process_instance_id == process_instance_id).all()
)
assert len(human_tasks) > 0, "No human tasks found for process."
human_task = human_tasks[0]
response = client.get(
f"/v1.0/tasks/{process_instance_id}/{human_task.task_id}",
headers=self.logged_in_headers(with_super_admin_user),
)
return response