sartography/spiffworkflow-backend
1
0
Fork 0
mirror of https://github.com/sartography/spiffworkflow-backend.git synced 2025-02-24 05:18:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

pre commit

Browse Source
This commit is contained in:
mike cullerton 2022-06-21 10:55:51 -04:00
parent 8988d3b362
commit f06b1c2fc2
3 changed files with 37 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/spiffworkflow_backend/routes/process_api_blueprint.py
View File

@ -244,10 +244,17 @@ def process_instance_create(process_group_id, process_model_id):
)
def process_instance_list(process_group_id, process_model_id,
page=1, per_page=100,
start_from=None, start_till=None,
end_from=None, end_till=None, process_status=None):
def process_instance_list(
process_group_id,
process_model_id,
page=1,
per_page=100,
start_from=None,
start_till=None,
end_from=None,
end_till=None,
process_status=None,
):
"""Process_instance_list."""
process_model = ProcessModelService().get_process_model(
process_model_id, group_id=process_group_id
@ -261,7 +268,9 @@ def process_instance_list(process_group_id, process_model_id,
)
)
results = ProcessInstanceModel.query.filter_by(process_model_identifier=process_model.id)
results = ProcessInstanceModel.query.filter_by(
process_model_identifier=process_model.id
)
if start_from is not None:
results = results.filter(ProcessInstanceModel.start_in_seconds >= start_from)
if start_till is not None:
@ -273,13 +282,9 @@ def process_instance_list(process_group_id, process_model_id,
if process_status is not None:
results = results.filter(ProcessInstanceModel.status == process_status)
process_instances = (
results
.order_by(
ProcessInstanceModel.start_in_seconds.desc(), ProcessInstanceModel.id.desc()
)
.paginate(page, per_page, False)
)
process_instances = results.order_by(
ProcessInstanceModel.start_in_seconds.desc(), ProcessInstanceModel.id.desc()
).paginate(page, per_page, False)
serialized_results = []
for process_instance in process_instances.items:

2
src/spiffworkflow_backend/scripts/script.py
View File

@ -49,6 +49,7 @@ class Script:
We may be able to remove the task for each of these calls if we are not using it other than potentially
updating the task data.
"""
def make_closure(subclass, task, workflow_id):
"""Yes - this is black magic.
@ -84,6 +85,7 @@ class Script:
We may be able to remove the task for each of these calls if we are not using it other than potentially
updating the task data.
"""
def make_closure_validate(subclass, task, workflow_id):
"""Make_closure_validate."""
instance = subclass()

36
tests/spiffworkflow_backend/integration/test_process_api.py
View File

@ -3,7 +3,6 @@ import io
import json
import os
import shutil
from datetime import datetime
from typing import Dict
from typing import Iterator
@ -22,7 +21,8 @@ from werkzeug.test import TestResponse
from spiffworkflow_backend.models.file import FileType
from spiffworkflow_backend.models.process_group import ProcessGroup
from spiffworkflow_backend.models.process_group import ProcessGroupSchema
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel, ProcessInstanceStatus
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.process_model import ProcessModelInfoSchema
from spiffworkflow_backend.services.process_model_service import ProcessModelService
@ -654,7 +654,7 @@ def test_process_instance_list_with_paginated_items(
def test_process_instance_list_filter(
app: Flask, client: FlaskClient, with_bpmn_file_cleanup: None
app: Flask, client: FlaskClient, with_bpmn_file_cleanup: None
) -> None:
"""Test_process_instance_list_filter."""
db.session.query(ProcessInstanceModel).delete()
@ -664,7 +664,7 @@ def test_process_instance_list_filter(
test_process_model_id = "sample"
user = find_or_create_user()
load_test_spec(app, test_process_model_id, process_group_id=test_process_group_id)
statuses = ('not_started', 'user_input_required', 'waiting', 'complete', 'erroring')
statuses = ("not_started", "user_input_required", "waiting", "complete", "erroring")
# create 5 instances with different status, and different start_in_seconds/end_in_seconds
for i in range(5):
@ -676,7 +676,7 @@ def test_process_instance_list_filter(
last_updated=datetime.now(),
start_in_seconds=(1000 * i) + 1000,
end_in_seconds=(1000 * i) + 2000,
bpmn_json=json.dumps({'i': i})
bpmn_json=json.dumps({"i": i}),
)
db.session.add(process_instance)
db.session.commit()
@ -686,7 +686,7 @@ def test_process_instance_list_filter(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 5
# filter for each of the status
@ -696,9 +696,9 @@ def test_process_instance_list_filter(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?process_status={ProcessInstanceStatus[statuses[i]].value}",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 1
assert results[0]['status'] == ProcessInstanceStatus[statuses[i]].value
assert results[0]["status"] == ProcessInstanceStatus[statuses[i]].value
# filter by start/end seconds
# start > 1000 - this should eliminate the first
@ -706,40 +706,40 @@ def test_process_instance_list_filter(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=1001",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 4
for i in range(4):
assert json.loads(results[i]['bpmn_json'])['i'] in (1, 2, 3, 4)
assert json.loads(results[i]["bpmn_json"])["i"] in (1, 2, 3, 4)
# start > 2000, end < 5000 - this should eliminate the first 2 and the last
response = client.get(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=2001&end_till=5999",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 2
assert json.loads(results[0]['bpmn_json'])['i'] in (2, 3)
assert json.loads(results[1]['bpmn_json'])['i'] in (2, 3)
assert json.loads(results[0]["bpmn_json"])["i"] in (2, 3)
assert json.loads(results[1]["bpmn_json"])["i"] in (2, 3)
# start > 1000, start < 4000 - this should eliminate the first and the last 2
response = client.get(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=1001&start_till=3999",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 2
assert json.loads(results[0]['bpmn_json'])['i'] in (1, 2)
assert json.loads(results[1]['bpmn_json'])['i'] in (1, 2)
assert json.loads(results[0]["bpmn_json"])["i"] in (1, 2)
assert json.loads(results[1]["bpmn_json"])["i"] in (1, 2)
# end > 2000, end < 6000 - this should eliminate the first and the last
response = client.get(
f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?end_from=2001&end_till=5999",
headers=logged_in_headers(user),
)
results = response.json['results']
results = response.json["results"]
assert len(results) == 3
for i in range(3):
assert json.loads(results[i]['bpmn_json'])['i'] in (1, 2, 3)
assert json.loads(results[i]["bpmn_json"])["i"] in (1, 2, 3)
def test_process_instance_report_with_default_list(