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

some initial changes to refactor report filters w/ burnettk

This commit is contained in:
jasquat 2022-12-20 11:16:06 -05:00
parent 9bb3d98594
commit 96e14817fa
8 changed files with 270 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
spiffworkflow-backend/migrations/env.py
View File

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

13
spiffworkflow-backend/migrations/versions/b86f7cc3a74b_.py → spiffworkflow-backend/migrations/versions/b99a4cb94b5b_.py
View File

@ -1,8 +1,8 @@
"""empty message
Revision ID: b86f7cc3a74b
Revision ID: b99a4cb94b5b
Revises:
Create Date: 2022-12-19 16:20:27.715487
Create Date: 2022-12-20 10:45:08.295317
"""
from alembic import op
@ -10,7 +10,7 @@ import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'b86f7cc3a74b'
revision = 'b99a4cb94b5b'
down_revision = None
branch_labels = None
depends_on = None
@ -179,8 +179,9 @@ def upgrade():
op.create_table('human_task',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('process_instance_id', sa.Integer(), nullable=False),
sa.Column('actual_owner_id', sa.Integer(), nullable=True),
sa.Column('lane_assignment_id', sa.Integer(), nullable=True),
sa.Column('completed_by_user_id', sa.Integer(), nullable=True),
sa.Column('actual_owner_id', sa.Integer(), nullable=True),
sa.Column('form_file_name', sa.String(length=50), nullable=True),
sa.Column('ui_form_file_name', sa.String(length=50), nullable=True),
sa.Column('updated_at_in_seconds', sa.Integer(), nullable=True),
@ -193,6 +194,7 @@ def upgrade():
sa.Column('process_model_display_name', sa.String(length=255), nullable=True),
sa.Column('completed', sa.Boolean(), nullable=False),
sa.ForeignKeyConstraint(['actual_owner_id'], ['user.id'], ),
sa.ForeignKeyConstraint(['completed_by_user_id'], ['user.id'], ),
sa.ForeignKeyConstraint(['lane_assignment_id'], ['group.id'], ),
sa.ForeignKeyConstraint(['process_instance_id'], ['process_instance.id'], ),
sa.PrimaryKeyConstraint('id'),
@ -259,9 +261,6 @@ def upgrade():
sa.Column('spiff_step', sa.Integer(), nullable=False),
sa.Column('task_json', sa.JSON(), nullable=False),
sa.Column('timestamp', sa.DECIMAL(precision=17, scale=6), nullable=False),
sa.Column('completed_by_user_id', sa.Integer(), nullable=True),
sa.Column('lane_assignment_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['lane_assignment_id'], ['group.id'], ),
sa.ForeignKeyConstraint(['process_instance_id'], ['process_instance.id'], ),
sa.PrimaryKeyConstraint('id')
)

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

@ -31,13 +31,16 @@ class HumanTaskModel(SpiffworkflowBaseDBModel):
db.UniqueConstraint("task_id", "process_instance_id", name="human_task_unique"),
)
actual_owner: RelationshipProperty[UserModel] = relationship(UserModel)
id: int = db.Column(db.Integer, primary_key=True)
process_instance_id: int = db.Column(
ForeignKey(ProcessInstanceModel.id), nullable=False # type: ignore
)
actual_owner_id: int = db.Column(ForeignKey(UserModel.id))
lane_assignment_id: int | None = db.Column(ForeignKey(GroupModel.id))
completed_by_user_id: int = db.Column(ForeignKey(UserModel.id), nullable=True)
actual_owner_id: int = db.Column(ForeignKey(UserModel.id))
# actual_owner: RelationshipProperty[UserModel] = relationship(UserModel)
form_file_name: str | None = db.Column(db.String(50))
ui_form_file_name: str | None = db.Column(db.String(50))

12
spiffworkflow-backend/src/spiffworkflow_backend/models/spiff_step_details.py
View File

@ -1,5 +1,6 @@
"""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
@ -20,10 +21,13 @@ class SpiffStepDetailsModel(SpiffworkflowBaseDBModel):
process_instance_id: int = db.Column(
ForeignKey(ProcessInstanceModel.id), nullable=False # type: ignore
)
# human_task_id: int = db.Column(
# ForeignKey(HumanTaskModel.id) # type: ignore
# )
spiff_step: int = db.Column(db.Integer, nullable=False)
task_json: dict = deferred(db.Column(db.JSON, nullable=False)) # type: ignore
timestamp: float = db.Column(db.DECIMAL(17, 6), nullable=False)
completed_by_user_id: int = db.Column(db.Integer, nullable=True)
lane_assignment_id: Optional[int] = db.Column(
ForeignKey(GroupModel.id), nullable=True
)
# completed_by_user_id: int = db.Column(db.Integer, nullable=True)
# lane_assignment_id: Optional[int] = db.Column(
# ForeignKey(GroupModel.id), nullable=True
# )

203
spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py
View File

@ -860,7 +860,7 @@ def process_instance_list_for_me(
user_filter: Optional[bool] = False,
report_identifier: Optional[str] = None,
report_id: Optional[int] = None,
group_identifier: Optional[str] = None,
user_group_identifier: Optional[str] = None,
) -> flask.wrappers.Response:
"""Process_instance_list_for_me."""
return process_instance_list(
@ -875,7 +875,7 @@ def process_instance_list_for_me(
user_filter=user_filter,
report_identifier=report_identifier,
report_id=report_id,
group_identifier=group_identifier,
user_group_identifier=user_group_identifier,
with_relation_to_me=True,
)
@ -896,183 +896,46 @@ def process_instance_list(
user_filter: Optional[bool] = False,
report_identifier: Optional[str] = None,
report_id: Optional[int] = None,
group_identifier: Optional[str] = None,
user_group_identifier: Optional[str] = None,
) -> flask.wrappers.Response:
"""Process_instance_list."""
process_instance_report = ProcessInstanceReportService.report_with_identifier(
g.user, report_id, report_identifier
)
print(f"with_relation_to_me: {with_relation_to_me}")
if user_filter:
report_filter = ProcessInstanceReportFilter(
process_model_identifier,
start_from,
start_to,
end_from,
end_to,
process_status.split(",") if process_status else None,
initiated_by_me,
with_tasks_completed_by_me,
with_tasks_completed_by_my_group,
with_relation_to_me,
process_model_identifier=process_model_identifier,
user_group_identifier=user_group_identifier,
start_from=start_from,
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,
process_status=process_status.split(",") if process_status else None,
)
else:
report_filter = (
ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report,
process_model_identifier,
start_from,
start_to,
end_from,
end_to,
process_status,
initiated_by_me,
with_tasks_completed_by_me,
with_tasks_completed_by_my_group,
with_relation_to_me,
process_instance_report=process_instance_report,
process_model_identifier=process_model_identifier,
user_group_identifier=user_group_identifier,
start_from=start_from,
start_to=start_to,
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,
)
)
process_instance_query = ProcessInstanceModel.query
# Always join that hot user table for good performance at serialization time.
process_instance_query = process_instance_query.options(
selectinload(ProcessInstanceModel.process_initiator)
)
if report_filter.process_model_identifier is not None:
process_model = get_process_model(
f"{report_filter.process_model_identifier}",
)
process_instance_query = process_instance_query.filter_by(
process_model_identifier=process_model.id
)
# this can never happen. obviously the class has the columns it defines. this is just to appease mypy.
if (
ProcessInstanceModel.start_in_seconds is None
or ProcessInstanceModel.end_in_seconds is None
):
raise (
ApiError(
error_code="unexpected_condition",
message="Something went very wrong",
status_code=500,
)
)
if report_filter.start_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds >= report_filter.start_from
)
if report_filter.start_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds <= report_filter.start_to
)
if report_filter.end_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds >= report_filter.end_from
)
if report_filter.end_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds <= report_filter.end_to
)
if report_filter.process_status is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(report_filter.process_status) # type: ignore
)
if report_filter.initiated_by_me is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
process_instance_query = process_instance_query.filter_by(
process_initiator=g.user
)
if report_filter.with_relation_to_me is True:
process_instance_query = process_instance_query.outerjoin(
HumanTaskModel
).outerjoin(
HumanTaskUserModel,
and_(
HumanTaskModel.id == HumanTaskUserModel.human_task_id,
HumanTaskUserModel.user_id == g.user.id,
),
)
process_instance_query = process_instance_query.filter(
or_(
HumanTaskUserModel.id.is_not(None),
ProcessInstanceModel.process_initiator_id == g.user.id,
)
)
# TODO: not sure if this is exactly what is wanted
if report_filter.with_tasks_completed_by_me is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
# process_instance_query = process_instance_query.join(UserModel, UserModel.id == ProcessInstanceModel.process_initiator_id)
# process_instance_query = process_instance_query.add_columns(UserModel.username)
# search for process_instance.UserModel.username in this file for more details about why adding columns is annoying.
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.process_initiator_id != g.user.id
)
process_instance_query = process_instance_query.join(
SpiffStepDetailsModel,
ProcessInstanceModel.id == SpiffStepDetailsModel.process_instance_id,
)
process_instance_query = process_instance_query.join(
SpiffLoggingModel,
ProcessInstanceModel.id == SpiffLoggingModel.process_instance_id,
)
process_instance_query = process_instance_query.filter(
SpiffLoggingModel.message.contains("COMPLETED") # type: ignore
)
process_instance_query = process_instance_query.filter(
SpiffLoggingModel.spiff_step == SpiffStepDetailsModel.spiff_step
)
process_instance_query = process_instance_query.filter(
SpiffStepDetailsModel.completed_by_user_id == g.user.id
)
if report_filter.with_tasks_completed_by_my_group is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
process_instance_query = process_instance_query.join(
SpiffStepDetailsModel,
ProcessInstanceModel.id == SpiffStepDetailsModel.process_instance_id,
)
process_instance_query = process_instance_query.join(
SpiffLoggingModel,
ProcessInstanceModel.id == SpiffLoggingModel.process_instance_id,
)
process_instance_query = process_instance_query.filter(
SpiffLoggingModel.message.contains("COMPLETED") # type: ignore
)
process_instance_query = process_instance_query.filter(
SpiffLoggingModel.spiff_step == SpiffStepDetailsModel.spiff_step
)
if group_identifier:
process_instance_query = process_instance_query.join(
GroupModel,
GroupModel.identifier == group_identifier,
)
else:
process_instance_query = process_instance_query.join(
GroupModel,
GroupModel.id == SpiffStepDetailsModel.lane_assignment_id,
)
process_instance_query = process_instance_query.join(
UserGroupAssignmentModel,
UserGroupAssignmentModel.group_id == GroupModel.id,
)
process_instance_query = process_instance_query.filter(
UserGroupAssignmentModel.user_id == g.user.id
)
process_instance_query = ProcessInstanceReportService.run_process_instance_report(report_filter, g.user)
instance_metadata_aliases = {}
stock_columns = ProcessInstanceReportService.get_column_names_for_model(
@ -1470,11 +1333,11 @@ def task_list_for_me(page: int = 1, per_page: int = 100) -> flask.wrappers.Respo
def task_list_for_my_groups(
group_identifier: Optional[str] = None, page: int = 1, per_page: int = 100
user_group_identifier: Optional[str] = None, page: int = 1, per_page: int = 100
) -> flask.wrappers.Response:
"""Task_list_for_my_groups."""
return get_tasks(
group_identifier=group_identifier,
user_group_identifier=user_group_identifier,
processes_started_by_user=False,
page=page,
per_page=per_page,
@ -1494,7 +1357,7 @@ def get_tasks(
has_lane_assignment_id: bool = True,
page: int = 1,
per_page: int = 100,
group_identifier: Optional[str] = None,
user_group_identifier: Optional[str] = None,
) -> flask.wrappers.Response:
"""Get_tasks."""
user_id = g.user.id
@ -1532,9 +1395,9 @@ def get_tasks(
),
)
if has_lane_assignment_id:
if group_identifier:
if user_group_identifier:
human_tasks_query = human_tasks_query.filter(
GroupModel.identifier == group_identifier
GroupModel.identifier == user_group_identifier
)
else:
human_tasks_query = human_tasks_query.filter(
@ -1550,7 +1413,7 @@ def get_tasks(
ProcessInstanceModel.updated_at_in_seconds,
ProcessInstanceModel.created_at_in_seconds,
UserModel.username,
GroupModel.identifier.label("group_identifier"),
GroupModel.identifier.label("user_group_identifier"),
HumanTaskModel.task_name,
HumanTaskModel.task_title,
HumanTaskModel.process_model_display_name,

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

@ -558,7 +558,7 @@ class ProcessInstanceProcessor:
"spiff_step": self.process_instance_model.spiff_step or 1,
"task_json": task_json,
"timestamp": round(time.time()),
"completed_by_user_id": self.current_user().id,
# "completed_by_user_id": self.current_user().id,
}
def spiff_step_details(self) -> SpiffStepDetailsModel:
@ -569,14 +569,14 @@ class ProcessInstanceProcessor:
spiff_step=details_mapping["spiff_step"],
task_json=details_mapping["task_json"],
timestamp=details_mapping["timestamp"],
completed_by_user_id=details_mapping["completed_by_user_id"],
# completed_by_user_id=details_mapping["completed_by_user_id"],
)
return details_model
def save_spiff_step_details(self, human_task: HumanTaskModel) -> None:
"""SaveSpiffStepDetails."""
details_model = self.spiff_step_details()
details_model.lane_assignment_id = human_task.lane_assignment_id
# details_model.lane_assignment_id = human_task.lane_assignment_id
db.session.add(details_model)
db.session.commit()

186
spiffworkflow-backend/src/spiffworkflow_backend/services/process_instance_report_service.py
View File

@ -1,5 +1,16 @@
"""Process_instance_report_service."""
from dataclasses import dataclass
from sqlalchemy import or_
from sqlalchemy import and_
from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
from spiffworkflow_backend.models.human_task import HumanTaskModel
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 typing import Optional
import sqlalchemy
@ -9,6 +20,7 @@ from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_model_service import ProcessModelService
@dataclass
@ -16,6 +28,7 @@ class ProcessInstanceReportFilter:
"""ProcessInstanceReportFilter."""
process_model_identifier: Optional[str] = None
user_group_identifier: Optional[str] = None
start_from: Optional[int] = None
start_to: Optional[int] = None
end_from: Optional[int] = None
@ -32,6 +45,8 @@ class ProcessInstanceReportFilter:
if self.process_model_identifier is not None:
d["process_model_identifier"] = self.process_model_identifier
if self.user_group_identifier is not None:
d["user_group_identifier"] = self.user_group_identifier
if self.start_from is not None:
d["start_from"] = str(self.start_from)
if self.start_to is not None:
@ -167,6 +182,7 @@ class ProcessInstanceReportService:
return filters[key].split(",") if key in filters else None
process_model_identifier = filters.get("process_model_identifier")
user_group_identifier = filters.get("user_group_identifier")
start_from = int_value("start_from")
start_to = int_value("start_to")
end_from = int_value("end_from")
@ -181,6 +197,7 @@ class ProcessInstanceReportService:
report_filter = ProcessInstanceReportFilter(
process_model_identifier,
user_group_identifier,
start_from,
start_to,
end_from,
@ -199,6 +216,7 @@ class ProcessInstanceReportService:
cls,
process_instance_report: ProcessInstanceReportModel,
process_model_identifier: Optional[str] = None,
user_group_identifier: Optional[str] = None,
start_from: Optional[int] = None,
start_to: Optional[int] = None,
end_from: Optional[int] = None,
@ -214,6 +232,8 @@ class ProcessInstanceReportService:
if process_model_identifier is not None:
report_filter.process_model_identifier = process_model_identifier
if user_group_identifier is not None:
report_filter.user_group_identifier = user_group_identifier
if start_from is not None:
report_filter.start_from = start_from
if start_to is not None:
@ -276,3 +296,169 @@ class ProcessInstanceReportService:
{"Header": "Username", "accessor": "username", "filterable": False},
{"Header": "Status", "accessor": "status", "filterable": False},
]
@classmethod
def run_process_instance_report(cls, report_filter: ProcessInstanceReportFilter, user: UserModel) -> None:
process_instance_query = ProcessInstanceModel.query
# Always join that hot user table for good performance at serialization time.
process_instance_query = process_instance_query.options(
selectinload(ProcessInstanceModel.process_initiator)
)
if report_filter.process_model_identifier is not None:
process_model = ProcessModelService.get_process_model(
f"{report_filter.process_model_identifier}",
)
process_instance_query = process_instance_query.filter_by(
process_model_identifier=process_model.id
)
# this can never happen. obviously the class has the columns it defines. this is just to appease mypy.
if (
ProcessInstanceModel.start_in_seconds is None
or ProcessInstanceModel.end_in_seconds is None
):
raise (
ApiError(
error_code="unexpected_condition",
message="Something went very wrong",
status_code=500,
)
)
if report_filter.start_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds >= report_filter.start_from
)
if report_filter.start_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds <= report_filter.start_to
)
if report_filter.end_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds >= report_filter.end_from
)
if report_filter.end_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds <= report_filter.end_to
)
if report_filter.process_status is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(report_filter.process_status) # type: ignore
)
if report_filter.initiated_by_me is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
process_instance_query = process_instance_query.filter_by(
process_initiator=user
)
if report_filter.with_relation_to_me is True:
process_instance_query = process_instance_query.outerjoin(
HumanTaskModel
).outerjoin(
HumanTaskUserModel,
and_(
HumanTaskModel.id == HumanTaskUserModel.human_task_id,
HumanTaskUserModel.user_id == user.id,
),
)
process_instance_query = process_instance_query.filter(
or_(
HumanTaskUserModel.id.is_not(None),
ProcessInstanceModel.process_initiator_id == user.id,
)
)
# TODO: not sure if this is exactly what is wanted
if report_filter.with_tasks_completed_by_me is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
# process_instance_query = process_instance_query.join(UserModel, UserModel.id == ProcessInstanceModel.process_initiator_id)
# process_instance_query = process_instance_query.add_columns(UserModel.username)
# search for process_instance.UserModel.username in this file for more details about why adding columns is annoying.
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.process_initiator_id != user.id
)
process_instance_query = process_instance_query.join(
HumanTaskModel,
and_(
HumanTaskModel.process_instance_id == ProcessInstanceModel.id,
HumanTaskModel.completed_by_user_id == user.id
)
)
# ).join(
# HumanTaskUserModel,
# and_(
# HumanTaskModel.id == HumanTaskUserModel.human_task_id,
# HumanTaskUserModel.user_id == user.id,
# ),
# )
# process_instance_query = process_instance_query.join(
# SpiffStepDetailsModel,
# ProcessInstanceModel.id == SpiffStepDetailsModel.process_instance_id,
# )
# process_instance_query = process_instance_query.join(
# SpiffLoggingModel,
# ProcessInstanceModel.id == SpiffLoggingModel.process_instance_id,
# )
# process_instance_query = process_instance_query.filter(
# SpiffLoggingModel.message.contains("COMPLETED") # type: ignore
# )
# process_instance_query = process_instance_query.filter(
# SpiffLoggingModel.spiff_step == SpiffStepDetailsModel.spiff_step
# )
# process_instance_query = process_instance_query.filter(
# HumanTaskModel.completed_by_user_id == user.id
# )
if report_filter.with_tasks_completed_by_my_group is True:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(ProcessInstanceModel.terminal_statuses()) # type: ignore
)
# process_instance_query = process_instance_query.join(
# SpiffStepDetailsModel,
# ProcessInstanceModel.id == SpiffStepDetailsModel.process_instance_id,
# )
# process_instance_query = process_instance_query.join(
# SpiffLoggingModel,
# ProcessInstanceModel.id == SpiffLoggingModel.process_instance_id,
# )
# process_instance_query = process_instance_query.filter(
# SpiffLoggingModel.message.contains("COMPLETED") # type: ignore
# )
# process_instance_query = process_instance_query.filter(
# SpiffLoggingModel.spiff_step == SpiffStepDetailsModel.spiff_step
# )
if report_filter.user_group_identifier:
process_instance_query = process_instance_query.join(
GroupModel,
GroupModel.identifier == report_filter.user_group_identifier,
)
else:
process_instance_query = process_instance_query.join(
HumanTaskModel
).join(
HumanTaskUserModel,
and_(
HumanTaskModel.id == HumanTaskUserModel.human_task_id,
HumanTaskUserModel.user_id == user.id,
),
)
process_instance_query = process_instance_query.join(
GroupModel,
GroupModel.id == HumanTaskModel.lane_assignment_id,
)
process_instance_query = process_instance_query.join(
UserGroupAssignmentModel,
UserGroupAssignmentModel.group_id == GroupModel.id,
)
process_instance_query = process_instance_query.filter(
UserGroupAssignmentModel.user_id == user.id
)
return process_instance_query

34
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_report_service.py
View File

@ -1,8 +1,11 @@
"""Test_process_instance_report_service."""
from typing import Optional
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 tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.process_instance_report import (
@ -122,13 +125,13 @@ class TestProcessInstanceReportService(BaseTest):
report_metadata=report_metadata,
)
return ProcessInstanceReportService.filter_from_metadata_with_overrides(
report,
process_model_identifier,
start_from,
start_to,
end_from,
end_to,
process_status,
process_instance_report=report,
process_model_identifier=process_model_identifier,
start_from=start_from,
start_to=start_to,
end_from=end_from,
end_to=end_to,
process_status=process_status,
)
def _filter_by_dict_from_metadata(self, report_metadata: dict) -> dict[str, str]:
@ -743,3 +746,20 @@ class TestProcessInstanceReportService(BaseTest):
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["sue"]
def test_can_filter_by_with_relation_to_me(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
process_model_id = "runs_without_input/sample"
bpmn_file_location = "sample"
load_test_spec(
process_model_id,
process_model_source_directory=bpmn_file_location,
)
user_one = self.find_or_create_user(username="user_one")
user_two = self.find_or_create_user(username="user_two")
process_instance_created_by_user_one_one = ProcessInstanceModel()