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

script to get last user completing a task is working w/ burnettk

This commit is contained in:
jasquat 2023-02-27 12:08:07 -05:00
parent e07a122674
commit 9f84564457
11 changed files with 468 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
spiffworkflow-backend/migrations/versions/b91143f4e414_.py → spiffworkflow-backend/migrations/versions/d6e5b3af0908_.py
View File

@ -1,8 +1,8 @@
"""empty message
Revision ID: b91143f4e414
Revision ID: d6e5b3af0908
Revises: 63fc8d693b9f
Create Date: 2023-02-25 22:46:03.533624
Create Date: 2023-02-27 11:10:28.058014
"""
from alembic import op
@ -10,7 +10,7 @@ import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'b91143f4e414'
revision = 'd6e5b3af0908'
down_revision = '63fc8d693b9f'
branch_labels = None
depends_on = None
@ -18,11 +18,11 @@ depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('human_task', sa.Column('process_model_identifier', sa.String(length=255), nullable=True))
op.add_column('human_task', sa.Column('bpmn_process_identifier', sa.String(length=255), nullable=True))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('human_task', 'process_model_identifier')
op.drop_column('human_task', 'bpmn_process_identifier')
# ### end Alembic commands ###

8
spiffworkflow-backend/src/spiffworkflow_backend/models/human_task.py
View File

@ -34,6 +34,8 @@ class HumanTaskModel(SpiffworkflowBaseDBModel):
lane_assignment_id: int | None = db.Column(ForeignKey(GroupModel.id))
completed_by_user_id: int = db.Column(ForeignKey(UserModel.id), nullable=True) # type: ignore
completed_by_user = relationship("UserModel", foreign_keys=[completed_by_user_id])
actual_owner_id: int = db.Column(ForeignKey(UserModel.id)) # type: ignore
# actual_owner: RelationshipProperty[UserModel] = relationship(UserModel)
@ -49,7 +51,7 @@ class HumanTaskModel(SpiffworkflowBaseDBModel):
task_type: str = db.Column(db.String(50))
task_status: str = db.Column(db.String(50))
process_model_display_name: str = db.Column(db.String(255))
process_model_identifier: str = db.Column(db.String(255))
bpmn_process_identifier: str = db.Column(db.String(255))
completed: bool = db.Column(db.Boolean, default=False, nullable=False, index=True)
human_task_users = relationship("HumanTaskUserModel", cascade="delete")
@ -75,8 +77,8 @@ class HumanTaskModel(SpiffworkflowBaseDBModel):
new_task.process_model_display_name = task.process_model_display_name
if hasattr(task, "process_group_identifier"):
new_task.process_group_identifier = task.process_group_identifier
if hasattr(task, "process_model_identifier"):
new_task.process_model_identifier = task.process_model_identifier
if hasattr(task, "bpmn_process_identifier"):
new_task.bpmn_process_identifier = task.bpmn_process_identifier
# human tasks only have status when getting the list on the home page
# and it comes from the process_instance. it should not be confused with task_status.

3
spiffworkflow-backend/src/spiffworkflow_backend/models/task.py
View File

@ -45,6 +45,7 @@ class Task:
process_model_display_name: Union[str, None] = None,
process_group_identifier: Union[str, None] = None,
process_model_identifier: Union[str, None] = None,
bpmn_process_identifier: Union[str, None] = None,
form_schema: Union[dict, None] = None,
form_ui_schema: Union[dict, None] = None,
parent: Optional[str] = None,
@ -76,6 +77,7 @@ class Task:
self.process_instance_status = process_instance_status
self.process_group_identifier = process_group_identifier
self.process_model_identifier = process_model_identifier
self.bpmn_process_identifier = bpmn_process_identifier
self.process_model_display_name = process_model_display_name
self.form_schema = form_schema
self.form_ui_schema = form_ui_schema
@ -122,6 +124,7 @@ class Task:
"process_model_display_name": self.process_model_display_name,
"process_group_identifier": self.process_group_identifier,
"process_model_identifier": self.process_model_identifier,
"bpmn_process_identifier": self.bpmn_process_identifier,
"form_schema": self.form_schema,
"form_ui_schema": self.form_ui_schema,
"parent": self.parent,

37
spiffworkflow-backend/src/spiffworkflow_backend/scripts/get_last_user_completing_task.py
View File

@ -1,14 +1,13 @@
"""Get current user."""
from typing import Any
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.models.human_task import HumanTaskModel
from flask import current_app
from flask import g
from spiffworkflow_backend.models.human_task import HumanTaskModel
from spiffworkflow_backend.models.script_attributes_context import (
ScriptAttributesContext,
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.scripts.script import Script
@ -25,22 +24,28 @@ class GetLastUserCompletingTask(Script):
self,
script_attributes_context: ScriptAttributesContext,
*_args: Any,
**kwargs: Any
**kwargs: Any,
) -> Any:
"""Run."""
# dump the user using our json encoder and then load it back up as a dict
# to remove unwanted field types
if len(_args) == 2:
process_model_identifier = _args[0]
task_bpmn_identifier = _args[1]
bpmn_process_identifier = _args[0]
task_name = _args[1]
else:
process_model_identifier = kwargs["process_model_identifier"]
task_bpmn_identifier = kwargs["task_bpmn_identifier"]
process_model_identifier = _args[0] or kwargs["process_model_identifier"]
print(f"process_model_identifier: {process_model_identifier}")
import pdb; pdb.set_trace()
# human_task = HumanTaskModel.query.filter_by(process_model_identifier=process_model_identifier).order_by(HumanTaskModel.id.desc()).first()
human_task = HumanTaskModel.query.filter_by(process_model_identifier=process_model_identifier).order_by(HumanTaskModel.id.desc()).join(UserModel, UserModel.id == HumanTaskModel.completed_by_user_id).first()
return human_task.completed_by_user
# user_as_json_string = current_app.json.dumps(g.user)
# return current_app.json.loads(user_as_json_string)
bpmn_process_identifier = kwargs["bpmn_process_identifier"]
task_name = kwargs["task_bpmn_identifier"]
human_task = (
HumanTaskModel.query.filter_by(
bpmn_process_identifier=bpmn_process_identifier, task_name=task_name
)
.order_by(HumanTaskModel.id.desc()) # type: ignore
.join(UserModel, UserModel.id == HumanTaskModel.completed_by_user_id)
.first()
)
# dump the user using our json encoder and then load it back up as a dict
# to remove unwanted field types
user_as_json_string = current_app.json.dumps(human_task.completed_by_user)
return current_app.json.loads(user_as_json_string)

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

@ -874,14 +874,13 @@ class ProcessInstanceProcessor:
process_instance_id=self.process_instance_model.id, completed=False
).all()
ready_or_waiting_tasks = self.get_all_ready_or_waiting_tasks()
process_model_display_name = ""
process_model_identifier = ""
process_model_info = self.process_model_service.get_process_model(
self.process_instance_model.process_model_identifier
)
if process_model_info is not None:
process_model_display_name = process_model_info.display_name
process_model_identifier = process_model_info.id
self.extract_metadata(process_model_info)
@ -894,6 +893,10 @@ class ProcessInstanceProcessor:
)
extensions = task_spec.extensions
# in the xml, it's the id attribute. this identifies the process where the activity lives.
# if it's in a subprocess, it's the inner process.
bpmn_process_identifier = ready_or_waiting_task.workflow.name
form_file_name = None
ui_form_file_name = None
if "properties" in extensions:
@ -913,7 +916,7 @@ class ProcessInstanceProcessor:
human_task = HumanTaskModel(
process_instance_id=self.process_instance_model.id,
process_model_display_name=process_model_display_name,
process_model_identifier=process_model_identifier,
bpmn_process_identifier=bpmn_process_identifier,
form_file_name=form_file_name,
ui_form_file_name=ui_form_file_name,
task_id=str(ready_or_waiting_task.id),
@ -1744,7 +1747,6 @@ class ProcessInstanceProcessor:
details_model.end_in_seconds = time.time()
details_model.task_json = self.get_task_json_from_spiff_task(task)
db.session.add(details_model)
# this is the thing that actually commits the db transaction (on behalf of the other updates above as well)
self.save()

353
spiffworkflow-backend/t2.json Normal file
View File

@ -0,0 +1,353 @@
{
"data": {
"validate_only": false,
"spiff__python_env_state": {}
},
"last_task": "d139f273-e735-4c8c-8183-66cb946eb8a6",
"success": true,
"tasks": {
"d278c748-e3b1-4eeb-a64e-e927e61a80ed": {
"id": "d278c748-e3b1-4eeb-a64e-e927e61a80ed",
"parent": null,
"children": [
"42edeac7-afb6-41d9-b02c-53f6889b530b"
],
"last_state_change": 1677513437.654198,
"state": 32,
"task_spec": "Root",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"42edeac7-afb6-41d9-b02c-53f6889b530b": {
"id": "42edeac7-afb6-41d9-b02c-53f6889b530b",
"parent": "d278c748-e3b1-4eeb-a64e-e927e61a80ed",
"children": [
"d139f273-e735-4c8c-8183-66cb946eb8a6"
],
"last_state_change": 1677513437.6548965,
"state": 32,
"task_spec": "Start",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"d139f273-e735-4c8c-8183-66cb946eb8a6": {
"id": "d139f273-e735-4c8c-8183-66cb946eb8a6",
"parent": "42edeac7-afb6-41d9-b02c-53f6889b530b",
"children": [
"83f7924b-01a0-43f7-9a50-f9dca568f0d8"
],
"last_state_change": 1677513437.655577,
"state": 32,
"task_spec": "StartEvent_1",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {
"event_fired": true
},
"data": {}
},
"83f7924b-01a0-43f7-9a50-f9dca568f0d8": {
"id": "83f7924b-01a0-43f7-9a50-f9dca568f0d8",
"parent": "d139f273-e735-4c8c-8183-66cb946eb8a6",
"children": [
"55c92211-cf14-43f8-8f25-950c0691fb30"
],
"last_state_change": 1677513437.6557715,
"state": 16,
"task_spec": "initiator_one",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"55c92211-cf14-43f8-8f25-950c0691fb30": {
"id": "55c92211-cf14-43f8-8f25-950c0691fb30",
"parent": "83f7924b-01a0-43f7-9a50-f9dca568f0d8",
"children": [
"022792f0-6a6c-4bcb-8b7a-67c59a2c8c69"
],
"last_state_change": 1677513437.6543038,
"state": 4,
"task_spec": "finance_approval",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"022792f0-6a6c-4bcb-8b7a-67c59a2c8c69": {
"id": "022792f0-6a6c-4bcb-8b7a-67c59a2c8c69",
"parent": "55c92211-cf14-43f8-8f25-950c0691fb30",
"children": [
"0795b348-5eef-4b67-8cfa-6444d1fb1a49"
],
"last_state_change": 1677513437.6543193,
"state": 4,
"task_spec": "initiator_two",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"0795b348-5eef-4b67-8cfa-6444d1fb1a49": {
"id": "0795b348-5eef-4b67-8cfa-6444d1fb1a49",
"parent": "022792f0-6a6c-4bcb-8b7a-67c59a2c8c69",
"children": [
"35d0aecc-23d6-4d42-a593-571b8de0cbcf"
],
"last_state_change": 1677513437.6543357,
"state": 4,
"task_spec": "Event_06f4e68",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"35d0aecc-23d6-4d42-a593-571b8de0cbcf": {
"id": "35d0aecc-23d6-4d42-a593-571b8de0cbcf",
"parent": "0795b348-5eef-4b67-8cfa-6444d1fb1a49",
"children": [
"7a7af0ac-2f31-4025-b739-27822470f3ec"
],
"last_state_change": 1677513437.654353,
"state": 4,
"task_spec": "Proccess_yhito9d.EndJoin",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
},
"7a7af0ac-2f31-4025-b739-27822470f3ec": {
"id": "7a7af0ac-2f31-4025-b739-27822470f3ec",
"parent": "35d0aecc-23d6-4d42-a593-571b8de0cbcf",
"children": [],
"last_state_change": 1677513437.654371,
"state": 4,
"task_spec": "End",
"triggered": false,
"workflow_name": "Proccess_yhito9d",
"internal_data": {},
"data": {}
}
},
"root": "d278c748-e3b1-4eeb-a64e-e927e61a80ed",
"spec": {
"name": "Proccess_yhito9d",
"description": "Proccess_yhito9d",
"file": "lanes.bpmn",
"task_specs": {
"Start": {
"id": "Proccess_yhito9d_1",
"name": "Start",
"description": "",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [],
"outputs": [
"StartEvent_1"
],
"typename": "StartTask"
},
"Proccess_yhito9d.EndJoin": {
"id": "Proccess_yhito9d_2",
"name": "Proccess_yhito9d.EndJoin",
"description": "",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"Event_06f4e68"
],
"outputs": [
"End"
],
"typename": "_EndJoin"
},
"End": {
"id": "Proccess_yhito9d_3",
"name": "End",
"description": "",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"Proccess_yhito9d.EndJoin"
],
"outputs": [],
"typename": "Simple"
},
"StartEvent_1": {
"id": "Proccess_yhito9d_4",
"name": "StartEvent_1",
"description": null,
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"Start"
],
"outputs": [
"initiator_one"
],
"lane": "Process Initiator",
"documentation": null,
"position": {
"x": 179,
"y": 159
},
"data_input_associations": [],
"data_output_associations": [],
"io_specification": null,
"event_definition": {
"internal": false,
"external": false,
"typename": "NoneEventDefinition"
},
"typename": "StartEvent",
"extensions": {}
},
"initiator_one": {
"id": "Proccess_yhito9d_5",
"name": "initiator_one",
"description": "Initiator One",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"StartEvent_1"
],
"outputs": [
"finance_approval"
],
"lane": "Process Initiator",
"documentation": null,
"position": {
"x": 270,
"y": 137
},
"data_input_associations": [],
"data_output_associations": [],
"io_specification": null,
"prescript": null,
"postscript": "user_completing_task = get_last_user_completing_task(\"misc/category_number_one/lanes\", \"initiator_one\")",
"typename": "ManualTask",
"extensions": {
"instructionsForEndUser": "This is for the initiator user",
"postScript": "user_completing_task = get_last_user_completing_task(\"misc/category_number_one/lanes\", \"initiator_one\")"
}
},
"finance_approval": {
"id": "Proccess_yhito9d_6",
"name": "finance_approval",
"description": "Finance Approval",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"initiator_one"
],
"outputs": [
"initiator_two"
],
"lane": "Finance Team",
"documentation": null,
"position": {
"x": 310,
"y": 320
},
"data_input_associations": [],
"data_output_associations": [],
"io_specification": null,
"prescript": null,
"postscript": null,
"typename": "ManualTask",
"extensions": {
"instructionsForEndUser": "This is for a Finance Team user"
}
},
"initiator_two": {
"id": "Proccess_yhito9d_7",
"name": "initiator_two",
"description": "Initiator Two",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"finance_approval"
],
"outputs": [
"Event_06f4e68"
],
"lane": "Process Initiator",
"documentation": null,
"position": {
"x": 440,
"y": 137
},
"data_input_associations": [],
"data_output_associations": [],
"io_specification": null,
"prescript": null,
"postscript": null,
"typename": "ManualTask",
"extensions": {
"instructionsForEndUser": "This is initiator again",
"postScript": null
}
},
"Event_06f4e68": {
"id": "Proccess_yhito9d_8",
"name": "Event_06f4e68",
"description": null,
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [
"initiator_two"
],
"outputs": [
"Proccess_yhito9d.EndJoin"
],
"lane": "Process Initiator",
"documentation": null,
"position": {
"x": 572,
"y": 159
},
"data_input_associations": [],
"data_output_associations": [],
"io_specification": null,
"event_definition": {
"internal": false,
"external": false,
"typename": "NoneEventDefinition"
},
"typename": "EndEvent",
"extensions": {}
},
"Root": {
"id": "Proccess_yhito9d_9",
"name": "Root",
"description": "",
"manual": false,
"internal": false,
"lookahead": 2,
"inputs": [],
"outputs": [],
"typename": "Simple"
}
},
"io_specification": null,
"data_objects": {},
"correlation_keys": {},
"typename": "BpmnProcessSpec"
},
"subprocess_specs": {},
"subprocesses": {},
"bpmn_messages": [],
"serializer_version": "1.0-spiffworkflow-backend"
}

1
spiffworkflow-backend/test.json Normal file
View File

File diff suppressed because one or more lines are too long

1
spiffworkflow-backend/tests/data/model_with_lanes/lanes.bpmn
View File

@ -23,7 +23,6 @@
<bpmn:manualTask id="initiator_one" name="Initiator One">
<bpmn:extensionElements>
<spiffworkflow:instructionsForEndUser>This is for the initiator user</spiffworkflow:instructionsForEndUser>
<spiffworkflow:postScript>user_completing_task = get_last_user_completing_task("misc/category_number_one/lanes", "initiator_one")</spiffworkflow:postScript>
</bpmn:extensionElements>
<bpmn:incoming>Flow_1tbyols</bpmn:incoming>
<bpmn:outgoing>Flow_16ppta1</bpmn:outgoing>

5
spiffworkflow-backend/tests/data/simple_form/simple_form.bpmn
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:spiffworkflow="http://spiffworkflow.org/bpmn/schema/1.0/core" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_96f6665" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="3.0.0-dev">
<bpmn:process id="Proccess_WithForm" name="Process With Form" isExecutable="true">
<bpmn:process id="Process_WithForm" name="Process With Form" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_0smvjir</bpmn:outgoing>
</bpmn:startEvent>
@ -14,6 +14,7 @@
<spiffworkflow:instructionsForEndUser>Hello {{ name }}
Department: {{ department }}
</spiffworkflow:instructionsForEndUser>
<spiffworkflow:postScript>user_completing_task = get_last_user_completing_task("Process_WithForm", "Activity_SimpleForm")</spiffworkflow:postScript>
</bpmn:extensionElements>
<bpmn:incoming>Flow_1ly1khd</bpmn:incoming>
<bpmn:outgoing>Flow_1boyhcj</bpmn:outgoing>
@ -31,7 +32,7 @@ Department: {{ department }}
</bpmn:userTask>
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Proccess_WithForm">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_WithForm">
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="179" y="159" width="36" height="36" />
</bpmndi:BPMNShape>

71
spiffworkflow-backend/tests/spiffworkflow_backend/scripts/test_get_last_user_completing_task.py Normal file
View File

@ -0,0 +1,71 @@
"""Test_get_localtime."""
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
class TestGetLastUserCompletingTask(BaseTest):
def test_get_last_user_completing_task_script_works(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_sets_permission_correctly_on_human_task."""
self.create_process_group(
client, with_super_admin_user, "test_group", "test_group"
)
initiator_user = self.find_or_create_user("initiator_user")
assert initiator_user.principal is not None
AuthorizationService.import_permissions_from_yaml_file()
process_model = load_test_spec(
process_model_id="misc/category_number_one/simple_form",
# bpmn_file_name="simp.bpmn",
process_model_source_directory="simple_form",
)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=initiator_user
)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True)
assert len(process_instance.active_human_tasks) == 1
human_task = process_instance.active_human_tasks[0]
assert len(human_task.potential_owners) == 1
assert human_task.potential_owners[0] == initiator_user
spiff_task = processor.__class__.get_task_by_bpmn_identifier(
human_task.task_name, processor.bpmn_process_instance
)
ProcessInstanceService.complete_form_task(
processor, spiff_task, {"name": "HEY"}, initiator_user, human_task
)
assert len(process_instance.active_human_tasks) == 1
human_task = process_instance.active_human_tasks[0]
spiff_task = processor.__class__.get_task_by_bpmn_identifier(
human_task.task_name, processor.bpmn_process_instance
)
ProcessInstanceService.complete_form_task(
processor, spiff_task, {}, initiator_user, human_task
)
assert spiff_task is not None
assert (
initiator_user.username
== spiff_task.get_data("user_completing_task")["username"]
)

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

@ -65,54 +65,6 @@ class TestProcessInstanceProcessor(BaseTest):
app.config["THREAD_LOCAL_DATA"].process_model_identifier = None
app.config["THREAD_LOCAL_DATA"].process_instance_id = None
def test_get_last_user_completing_task_script_works(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_sets_permission_correctly_on_human_task."""
self.create_process_group(
client, with_super_admin_user, "test_group", "test_group"
)
initiator_user = self.find_or_create_user("initiator_user")
finance_user = self.find_or_create_user("testuser2")
assert initiator_user.principal is not None
assert finance_user.principal is not None
AuthorizationService.import_permissions_from_yaml_file()
finance_group = GroupModel.query.filter_by(identifier="Finance Team").first()
assert finance_group is not None
process_model = load_test_spec(
process_model_id="misc/category_number_one/lanes",
bpmn_file_name="lanes.bpmn",
process_model_source_directory="model_with_lanes",
)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=initiator_user
)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True)
assert len(process_instance.active_human_tasks) == 1
human_task = process_instance.active_human_tasks[0]
assert human_task.lane_assignment_id is None
assert len(human_task.potential_owners) == 1
assert human_task.potential_owners[0] == initiator_user
spiff_task = processor.__class__.get_task_by_bpmn_identifier(
human_task.task_name, processor.bpmn_process_instance
)
ProcessInstanceService.complete_form_task(
processor, spiff_task, {}, initiator_user, human_task
)
print(f"initiator_user.username: {initiator_user.username}")
print(f"data: {processor.get_data()}")
print(f"task_data: {spiff_task.data}")
assert initiator_user.username == spiff_task.get_data("user_completing_task")["username"]
def test_sets_permission_correctly_on_human_task(
self,
app: Flask,