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

added remaining filter tests w/ burnettk

This commit is contained in:
jasquat 2022-12-20 14:39:04 -05:00
parent 0d3506944c
commit be47b73525
6 changed files with 334 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
migrations/env.py
View File

@ -1,5 +1,3 @@
from __future__ import with_statement
import logging
from logging.config import fileConfig

1
src/spiffworkflow_backend/models/human_task.py
View File

@ -8,7 +8,6 @@ from flask_bpmn.models.db import db
from flask_bpmn.models.db import SpiffworkflowBaseDBModel
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.orm import RelationshipProperty
from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel

3
src/spiffworkflow_backend/models/spiff_step_details.py
View File

@ -1,14 +1,11 @@
"""Spiff_step_details."""
from dataclasses import dataclass
from spiffworkflow_backend.models.human_task import HumanTaskModel
from typing import Optional
from flask_bpmn.models.db import db
from flask_bpmn.models.db import SpiffworkflowBaseDBModel
from sqlalchemy import ForeignKey
from sqlalchemy.orm import deferred
from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel

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

@ -2,7 +2,6 @@
import json
import os
import random
import re
import string
import uuid
from typing import Any
@ -32,10 +31,7 @@ from SpiffWorkflow.task import TaskState
from sqlalchemy import and_
from sqlalchemy import asc
from sqlalchemy import desc
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import aliased
from sqlalchemy.orm import selectinload
from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
ProcessEntityNotFoundError,
@ -79,7 +75,6 @@ from spiffworkflow_backend.models.spec_reference import SpecReferenceSchema
from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel
from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.models.user_group_assignment import UserGroupAssignmentModel
from spiffworkflow_backend.routes.user import verify_token
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService
@ -889,10 +884,7 @@ def process_instance_list(
end_from: Optional[int] = None,
end_to: Optional[int] = None,
process_status: Optional[str] = None,
# initiated_by_me: Optional[bool] = None,
# with_tasks_completed_by_me: Optional[bool] = None,
# with_tasks_completed_by_my_group: Optional[bool] = None,
# with_relation_to_me: Optional[bool] = None,
with_relation_to_me: Optional[bool] = None,
user_filter: Optional[bool] = False,
report_identifier: Optional[str] = None,
report_id: Optional[int] = None,
@ -911,10 +903,7 @@ def process_instance_list(
start_to=start_to,
end_from=end_from,
end_to=end_to,
# initiated_by_me=initiated_by_me,
# with_tasks_completed_by_me=with_tasks_completed_by_me,
# with_tasks_completed_by_my_group=with_tasks_completed_by_my_group,
# with_relation_to_me=with_relation_to_me,
with_relation_to_me=with_relation_to_me,
process_status=process_status.split(",") if process_status else None,
)
else:
@ -928,10 +917,7 @@ def process_instance_list(
end_from=end_from,
end_to=end_to,
process_status=process_status,
# initiated_by_me=initiated_by_me,
# with_tasks_completed_by_me=with_tasks_completed_by_me,
# with_tasks_completed_by_my_group=with_tasks_completed_by_my_group,
# with_relation_to_me=with_relation_to_me,
with_relation_to_me=with_relation_to_me,
)
)
@ -940,7 +926,7 @@ def process_instance_list(
process_instance_report=process_instance_report,
page=page,
per_page=per_page,
user=g.user
user=g.user,
)
return make_response(jsonify(response_json), 200)

57
src/spiffworkflow_backend/services/process_instance_report_service.py
View File

@ -1,30 +1,29 @@
"""Process_instance_report_service."""
from dataclasses import dataclass
from sqlalchemy.orm import relationship
from sqlalchemy import func
from sqlalchemy.orm import aliased
from sqlalchemy import or_
from sqlalchemy import and_
import re
from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
from spiffworkflow_backend.models.human_task import HumanTaskModel
from spiffworkflow_backend.models.process_instance_metadata import ProcessInstanceMetadataModel
from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel
from spiffworkflow_backend.models.user_group_assignment import UserGroupAssignmentModel
from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel
from spiffworkflow_backend.models.group import GroupModel
from flask_bpmn.api.api_error import ApiError
from sqlalchemy.orm import selectinload
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from dataclasses import dataclass
from typing import Optional
import sqlalchemy
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from sqlalchemy import and_
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import aliased
from sqlalchemy.orm import selectinload
from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.human_task import HumanTaskModel
from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance_metadata import (
ProcessInstanceMetadataModel,
)
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.models.user_group_assignment import UserGroupAssignmentModel
from spiffworkflow_backend.services.process_model_service import ProcessModelService
@ -126,14 +125,17 @@ class ProcessInstanceReportService:
{"Header": "end_in_seconds", "accessor": "end_in_seconds"},
{"Header": "status", "accessor": "status"},
],
"filter_by": [{"field_name": "initiated_by_me", "field_value": True}, {"field_name": "has_terminal_status", "field_value": True}],
"filter_by": [
{"field_name": "initiated_by_me", "field_value": True},
{"field_name": "has_terminal_status", "field_value": True},
],
"order_by": ["-start_in_seconds", "-id"],
},
"system_report_completed_instances_with_tasks_completed_by_me": {
"columns": cls.builtin_column_options(),
"filter_by": [
{"field_name": "with_tasks_completed_by_me", "field_value": True},
{"field_name": "has_terminal_status", "field_value": True}
{"field_name": "has_terminal_status", "field_value": True},
],
"order_by": ["-start_in_seconds", "-id"],
},
@ -144,7 +146,7 @@ class ProcessInstanceReportService:
"field_name": "with_tasks_assigned_to_my_group",
"field_value": True,
},
{"field_name": "has_terminal_status", "field_value": True}
{"field_name": "has_terminal_status", "field_value": True},
],
"order_by": ["-start_in_seconds", "-id"],
},
@ -200,9 +202,7 @@ class ProcessInstanceReportService:
initiated_by_me = bool_value("initiated_by_me")
has_terminal_status = bool_value("has_terminal_status")
with_tasks_completed_by_me = bool_value("with_tasks_completed_by_me")
with_tasks_assigned_to_my_group = bool_value(
"with_tasks_assigned_to_my_group"
)
with_tasks_assigned_to_my_group = bool_value("with_tasks_assigned_to_my_group")
with_relation_to_me = bool_value("with_relation_to_me")
report_filter = ProcessInstanceReportFilter(
@ -320,6 +320,7 @@ class ProcessInstanceReportService:
page: int = 1,
per_page: int = 100,
) -> dict:
"""Run_process_instance_report."""
process_instance_query = ProcessInstanceModel.query
# Always join that hot user table for good performance at serialization time.
process_instance_query = process_instance_query.options(
@ -404,8 +405,8 @@ class ProcessInstanceReportService:
HumanTaskModel,
and_(
HumanTaskModel.process_instance_id == ProcessInstanceModel.id,
HumanTaskModel.completed_by_user_id == user.id
)
HumanTaskModel.completed_by_user_id == user.id,
),
)
if report_filter.with_tasks_assigned_to_my_group is True:
@ -415,9 +416,7 @@ class ProcessInstanceReportService:
GroupModel.identifier == report_filter.user_group_identifier,
)
else:
process_instance_query = process_instance_query.join(
HumanTaskModel
)
process_instance_query = process_instance_query.join(HumanTaskModel)
process_instance_query = process_instance_query.join(
GroupModel,
GroupModel.id == HumanTaskModel.lane_assignment_id,
@ -462,7 +461,9 @@ class ProcessInstanceReportService:
)
process_instance_query = process_instance_query.join(
instance_metadata_alias, and_(*conditions), isouter=isouter
).add_columns(func.max(instance_metadata_alias.value).label(column["accessor"]))
).add_columns(
func.max(instance_metadata_alias.value).label(column["accessor"])
)
order_by_query_array = []
order_by_array = process_instance_report.report_metadata["order_by"]

336
tests/spiffworkflow_backend/unit/test_process_instance_report_service.py
View File

@ -1,15 +1,14 @@
"""Test_process_instance_report_service."""
from typing import Optional
from flask_bpmn.models.db import db
from spiffworkflow_backend.models.human_task import HumanTaskModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask import Flask
from flask.testing import FlaskClient
from spiffworkflow_backend.models import process_instance_report
from flask_bpmn.models.db import db
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.human_task import HumanTaskModel
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
@ -20,6 +19,7 @@ from spiffworkflow_backend.services.process_instance_report_service import (
from spiffworkflow_backend.services.process_instance_report_service import (
ProcessInstanceReportService,
)
from spiffworkflow_backend.services.user_service import UserService
class TestProcessInstanceReportFilter(BaseTest):
@ -755,6 +755,7 @@ class TestProcessInstanceReportService(BaseTest):
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_can_filter_by_completed_instances_initiated_by_me."""
process_model_id = "runs_without_input/sample"
bpmn_file_location = "sample"
process_model = load_test_spec(
@ -765,30 +766,43 @@ class TestProcessInstanceReportService(BaseTest):
user_two = self.find_or_create_user(username="user_two")
# Several processes to ensure they do not return in the result
_process_instance_created_by_user_one_one = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_one)
_process_instance_created_by_user_one_two = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_one)
_process_instance_created_by_user_one_three = self.create_process_instance_from_process_model(process_model=process_model, status="waiting", user=user_one)
_process_instance_created_by_user_two_one = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_two)
_process_instance_created_by_user_two_two = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_two)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_one
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
process_instance_report = ProcessInstanceReportService.report_with_identifier(
user=user_one, report_identifier="system_report_completed_instances_initiated_by_me"
user=user_one,
report_identifier="system_report_completed_instances_initiated_by_me",
)
report_filter = ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
report_filter = (
ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
)
)
response_json = ProcessInstanceReportService.run_process_instance_report(
report_filter=report_filter,
process_instance_report=process_instance_report,
user=user_one
user=user_one,
)
assert len(response_json['results']) == 2
assert response_json['results'][0]['process_initiator_id'] == user_one.id
assert response_json['results'][1]['process_initiator_id'] == user_one.id
assert response_json['results'][0]['status'] == 'complete'
assert response_json['results'][1]['status'] == 'complete'
assert len(response_json["results"]) == 2
assert response_json["results"][0]["process_initiator_id"] == user_one.id
assert response_json["results"][1]["process_initiator_id"] == user_one.id
assert response_json["results"][0]["status"] == "complete"
assert response_json["results"][1]["status"] == "complete"
def test_can_filter_by_completed_instances_with_tasks_completed_by_me(
self,
@ -796,6 +810,7 @@ class TestProcessInstanceReportService(BaseTest):
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_can_filter_by_completed_instances_with_tasks_completed_by_me."""
process_model_id = "runs_without_input/sample"
bpmn_file_location = "sample"
process_model = load_test_spec(
@ -806,12 +821,30 @@ class TestProcessInstanceReportService(BaseTest):
user_two = self.find_or_create_user(username="user_two")
# Several processes to ensure they do not return in the result
process_instance_created_by_user_one_one = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_one)
_process_instance_created_by_user_one_two = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_one)
process_instance_created_by_user_one_three = self.create_process_instance_from_process_model(process_model=process_model, status="waiting", user=user_one)
process_instance_created_by_user_two_one = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_two)
_process_instance_created_by_user_two_two = self.create_process_instance_from_process_model(process_model=process_model, status="complete", user=user_two)
process_instance_created_by_user_two_three = self.create_process_instance_from_process_model(process_model=process_model, status="waiting", user=user_two)
process_instance_created_by_user_one_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
process_instance_created_by_user_one_three = (
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_one
)
)
process_instance_created_by_user_two_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_two
)
human_task_for_user_one_one = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_one.id,
@ -846,19 +879,252 @@ class TestProcessInstanceReportService(BaseTest):
db.session.commit()
process_instance_report = ProcessInstanceReportService.report_with_identifier(
user=user_one, report_identifier="system_report_completed_instances_with_tasks_completed_by_me"
user=user_one,
report_identifier="system_report_completed_instances_with_tasks_completed_by_me",
)
report_filter = ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
report_filter = (
ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
)
)
response_json = ProcessInstanceReportService.run_process_instance_report(
report_filter=report_filter,
process_instance_report=process_instance_report,
user=user_one
user=user_one,
)
assert len(response_json['results']) == 1
assert response_json['results'][0]['process_initiator_id'] == user_two.id
assert response_json['results'][0]['id'] == process_instance_created_by_user_two_one.id
assert response_json['results'][0]['status'] == 'complete'
assert len(response_json["results"]) == 1
assert response_json["results"][0]["process_initiator_id"] == user_two.id
assert (
response_json["results"][0]["id"]
== process_instance_created_by_user_two_one.id
)
assert response_json["results"][0]["status"] == "complete"
def test_can_filter_by_completed_instances_with_tasks_completed_by_my_groups(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_can_filter_by_completed_instances_with_tasks_completed_by_my_groups."""
process_model_id = "runs_without_input/sample"
bpmn_file_location = "sample"
process_model = load_test_spec(
process_model_id,
process_model_source_directory=bpmn_file_location,
)
user_group_one = GroupModel(identifier="group_one")
user_group_two = GroupModel(identifier="group_two")
db.session.add(user_group_one)
db.session.add(user_group_two)
db.session.commit()
user_one = self.find_or_create_user(username="user_one")
user_two = self.find_or_create_user(username="user_two")
user_three = self.find_or_create_user(username="user_three")
UserService.add_user_to_group(user_one, user_group_one)
UserService.add_user_to_group(user_two, user_group_one)
UserService.add_user_to_group(user_three, user_group_two)
# Several processes to ensure they do not return in the result
process_instance_created_by_user_one_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
process_instance_created_by_user_one_three = (
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_one
)
)
process_instance_created_by_user_two_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_two
)
human_task_for_user_group_one_one = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_one.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_one_two = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_three.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_one_three = HumanTaskModel(
process_instance_id=process_instance_created_by_user_two_one.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_two_one = HumanTaskModel(
process_instance_id=process_instance_created_by_user_two_one.id,
lane_assignment_id=user_group_two.id,
)
human_task_for_user_group_two_two = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_one.id,
lane_assignment_id=user_group_two.id,
)
db.session.add(human_task_for_user_group_one_one)
db.session.add(human_task_for_user_group_one_two)
db.session.add(human_task_for_user_group_one_three)
db.session.add(human_task_for_user_group_two_one)
db.session.add(human_task_for_user_group_two_two)
db.session.commit()
process_instance_report = ProcessInstanceReportService.report_with_identifier(
user=user_one,
report_identifier="system_report_completed_instances_with_tasks_completed_by_my_groups",
)
report_filter = (
ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
)
)
response_json = ProcessInstanceReportService.run_process_instance_report(
report_filter=report_filter,
process_instance_report=process_instance_report,
user=user_one,
)
assert len(response_json["results"]) == 2
assert response_json["results"][0]["process_initiator_id"] == user_two.id
assert (
response_json["results"][0]["id"]
== process_instance_created_by_user_two_one.id
)
assert response_json["results"][0]["status"] == "complete"
assert response_json["results"][1]["process_initiator_id"] == user_one.id
assert (
response_json["results"][1]["id"]
== process_instance_created_by_user_one_one.id
)
assert response_json["results"][1]["status"] == "complete"
def test_can_filter_by_with_relation_to_me(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_can_filter_by_with_relation_to_me."""
process_model_id = "runs_without_input/sample"
bpmn_file_location = "sample"
process_model = load_test_spec(
process_model_id,
process_model_source_directory=bpmn_file_location,
)
user_group_one = GroupModel(identifier="group_one")
user_group_two = GroupModel(identifier="group_two")
db.session.add(user_group_one)
db.session.add(user_group_two)
db.session.commit()
user_one = self.find_or_create_user(username="user_one")
user_two = self.find_or_create_user(username="user_two")
user_three = self.find_or_create_user(username="user_three")
UserService.add_user_to_group(user_one, user_group_one)
UserService.add_user_to_group(user_two, user_group_one)
UserService.add_user_to_group(user_three, user_group_two)
# Several processes to ensure they do not return in the result
process_instance_created_by_user_one_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
)
process_instance_created_by_user_one_two = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_one
)
)
process_instance_created_by_user_one_three = (
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_one
)
)
process_instance_created_by_user_two_one = (
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
)
self.create_process_instance_from_process_model(
process_model=process_model, status="complete", user=user_two
)
self.create_process_instance_from_process_model(
process_model=process_model, status="waiting", user=user_two
)
human_task_for_user_group_one_one = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_one.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_one_two = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_three.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_one_three = HumanTaskModel(
process_instance_id=process_instance_created_by_user_two_one.id,
lane_assignment_id=user_group_one.id,
)
human_task_for_user_group_two_one = HumanTaskModel(
process_instance_id=process_instance_created_by_user_two_one.id,
lane_assignment_id=user_group_two.id,
)
human_task_for_user_group_two_two = HumanTaskModel(
process_instance_id=process_instance_created_by_user_one_one.id,
lane_assignment_id=user_group_two.id,
)
db.session.add(human_task_for_user_group_one_one)
db.session.add(human_task_for_user_group_one_two)
db.session.add(human_task_for_user_group_one_three)
db.session.add(human_task_for_user_group_two_one)
db.session.add(human_task_for_user_group_two_two)
db.session.commit()
UserService.add_user_to_human_tasks_if_appropriate(user_one)
process_instance_report = ProcessInstanceReportService.report_with_identifier(
user=user_one
)
report_filter = (
ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report=process_instance_report,
process_model_identifier=process_model.id,
with_relation_to_me=True,
)
)
response_json = ProcessInstanceReportService.run_process_instance_report(
report_filter=report_filter,
process_instance_report=process_instance_report,
user=user_one,
)
assert len(response_json["results"]) == 4
process_instance_ids_in_results = [r["id"] for r in response_json["results"]]
assert (
process_instance_created_by_user_one_one.id
in process_instance_ids_in_results
)
assert (
process_instance_created_by_user_one_two.id
in process_instance_ids_in_results
)
assert (
process_instance_created_by_user_one_three.id
in process_instance_ids_in_results
)
assert (
process_instance_created_by_user_two_one.id
in process_instance_ids_in_results
)