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

Merge process_instance_list query filters with report filters (#37)

This commit is contained in:
jbirddog 2022-11-15 15:05:37 -05:00 committed by GitHub
parent b62955deaa
commit 3205b85f42
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 796 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -72,6 +72,9 @@ from spiffworkflow_backend.services.message_service import MessageService
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.process_instance_report_service import (
ProcessInstanceReportService,
)
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
@ -725,11 +728,22 @@ def process_instance_list(
process_status: Optional[str] = None,
) -> flask.wrappers.Response:
"""Process_instance_list."""
process_instance_report = ProcessInstanceReportModel.default_report(g.user)
report_filter = ProcessInstanceReportService.filter_from_metadata_with_overrides(
process_instance_report,
process_model_identifier,
start_from,
start_to,
end_from,
end_to,
process_status,
)
# process_model_identifier = un_modify_modified_process_model_id(modified_process_model_identifier)
process_instance_query = ProcessInstanceModel.query
if process_model_identifier is not None:
if report_filter.process_model_identifier is not None:
process_model = get_process_model(
f"{process_model_identifier}",
f"{report_filter.process_model_identifier}",
)
process_instance_query = process_instance_query.filter_by(
@ -749,36 +763,31 @@ def process_instance_list(
)
)
if start_from is not None:
if report_filter.start_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds >= start_from
ProcessInstanceModel.start_in_seconds >= report_filter.start_from
)
if start_to is not None:
if report_filter.start_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.start_in_seconds <= start_to
ProcessInstanceModel.start_in_seconds <= report_filter.start_to
)
if end_from is not None:
if report_filter.end_from is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds >= end_from
ProcessInstanceModel.end_in_seconds >= report_filter.end_from
)
if end_to is not None:
if report_filter.end_to is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.end_in_seconds <= end_to
ProcessInstanceModel.end_in_seconds <= report_filter.end_to
)
if process_status is not None:
process_status_array = process_status.split(",")
if report_filter.process_status is not None:
process_instance_query = process_instance_query.filter(
ProcessInstanceModel.status.in_(process_status_array) # type: ignore
ProcessInstanceModel.status.in_(report_filter.process_status) # type: ignore
)
process_instances = process_instance_query.order_by(
ProcessInstanceModel.start_in_seconds.desc(), ProcessInstanceModel.id.desc() # type: ignore
).paginate(page=page, per_page=per_page, error_out=False)
process_instance_report = ProcessInstanceReportModel.default_report(g.user)
# TODO need to look into this more - how the filter here interacts with the
# one defined in the report.
# TODO need to look into test failures when the results from result_dict is
# used instead of the process instances

99
spiffworkflow-backend/src/spiffworkflow_backend/services/process_instance_report_service.py Normal file
View File

@ -0,0 +1,99 @@
"""Process_instance_report_service."""
from dataclasses import dataclass
from typing import Optional
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
@dataclass
class ProcessInstanceReportFilter:
"""ProcessInstanceReportFilter."""
process_model_identifier: Optional[str] = None
start_from: Optional[int] = None
start_to: Optional[int] = None
end_from: Optional[int] = None
end_to: Optional[int] = None
process_status: Optional[list[str]] = None
class ProcessInstanceReportService:
"""ProcessInstanceReportService."""
@classmethod
def filter_by_to_dict(
cls, process_instance_report: ProcessInstanceReportModel
) -> dict[str, str]:
"""Filter_by_to_dict."""
metadata = process_instance_report.report_metadata
filter_by = metadata.get("filter_by", [])
filters = {
d["field_name"]: d["field_value"]
for d in filter_by
if "field_name" in d and "field_value" in d
}
return filters
@classmethod
def filter_from_metadata(
cls, process_instance_report: ProcessInstanceReportModel
) -> ProcessInstanceReportFilter:
"""Filter_from_metadata."""
filters = cls.filter_by_to_dict(process_instance_report)
def int_value(key: str) -> Optional[int]:
"""Int_value."""
return int(filters[key]) if key in filters else None
def list_value(key: str) -> Optional[list[str]]:
"""List_value."""
return filters[key].split(",") if key in filters else None
process_model_identifier = filters.get("process_model_identifier")
start_from = int_value("start_from")
start_to = int_value("start_to")
end_from = int_value("end_from")
end_to = int_value("end_to")
process_status = list_value("process_status")
report_filter = ProcessInstanceReportFilter(
process_model_identifier,
start_from,
start_to,
end_from,
end_to,
process_status,
)
return report_filter
@classmethod
def filter_from_metadata_with_overrides(
cls,
process_instance_report: ProcessInstanceReportModel,
process_model_identifier: Optional[str] = None,
start_from: Optional[int] = None,
start_to: Optional[int] = None,
end_from: Optional[int] = None,
end_to: Optional[int] = None,
process_status: Optional[str] = None,
) -> ProcessInstanceReportFilter:
"""Filter_from_metadata_with_overrides."""
report_filter = cls.filter_from_metadata(process_instance_report)
if process_model_identifier is not None:
report_filter.process_model_identifier = process_model_identifier
if start_from is not None:
report_filter.start_from = start_from
if start_to is not None:
report_filter.start_to = start_to
if end_from is not None:
report_filter.end_from = end_from
if end_to is not None:
report_filter.end_to = end_to
if process_status is not None:
report_filter.process_status = process_status.split(",")
return report_filter

671
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_report_service.py Normal file
View File

@ -0,0 +1,671 @@
"""Test_process_instance_report_service."""
from typing import Optional
from flask import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_report_service import (
ProcessInstanceReportFilter,
)
from spiffworkflow_backend.services.process_instance_report_service import (
ProcessInstanceReportService,
)
class TestProcessInstanceReportService(BaseTest):
"""TestProcessInstanceReportService."""
def _filter_from_metadata(
self, report_metadata: dict
) -> ProcessInstanceReportFilter:
"""Docstring."""
report = ProcessInstanceReportModel(
identifier="test",
created_by_id=1,
report_metadata=report_metadata,
)
return ProcessInstanceReportService.filter_from_metadata(report)
def _filter_from_metadata_with_overrides(
self,
report_metadata: dict,
process_model_identifier: Optional[str] = None,
start_from: Optional[int] = None,
start_to: Optional[int] = None,
end_from: Optional[int] = None,
end_to: Optional[int] = None,
process_status: Optional[str] = None,
) -> ProcessInstanceReportFilter:
"""Docstring."""
report = ProcessInstanceReportModel(
identifier="test",
created_by_id=1,
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,
)
def _filter_by_dict_from_metadata(self, report_metadata: dict) -> dict[str, str]:
"""Docstring."""
report = ProcessInstanceReportModel(
identifier="test",
created_by_id=1,
report_metadata=report_metadata,
)
return ProcessInstanceReportService.filter_by_to_dict(report)
def test_filter_by_to_dict_no_filter_by(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
filters = self._filter_by_dict_from_metadata(
{
"columns": [],
}
)
assert filters == {}
def test_filter_by_to_dict_empty_filter_by(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
filters = self._filter_by_dict_from_metadata(
{
"columns": [],
"filter_by": [],
}
)
assert filters == {}
def test_filter_by_to_dict_single_filter_by(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
filters = self._filter_by_dict_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "end_to", "field_value": "1234"}],
}
)
assert filters == {"end_to": "1234"}
def test_filter_by_to_dict_mulitple_filter_by(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
filters = self._filter_by_dict_from_metadata(
{
"columns": [],
"filter_by": [
{"field_name": "end_to", "field_value": "1234"},
{"field_name": "end_from", "field_value": "4321"},
],
}
)
assert filters == {"end_to": "1234", "end_from": "4321"}
def test_report_with_no_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_empty_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_unknown_filter_field_name(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "bob", "field_value": "joe"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_unknown_filter_keys(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"_name": "bob", "_value": "joe"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_process_model_identifier_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [
{"field_name": "process_model_identifier", "field_value": "bob"}
],
}
)
assert report_filter.process_model_identifier == "bob"
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_start_from_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "start_from", "field_value": "1234"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from == 1234
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_start_to_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "start_to", "field_value": "1234"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to == 1234
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_end_from_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "end_from", "field_value": "1234"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from == 1234
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_with_end_to_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "end_to", "field_value": "1234"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to == 1234
assert report_filter.process_status is None
def test_report_with_single_startus_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [{"field_name": "process_status", "field_value": "ready"}],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["ready"]
def test_report_with_multiple_startus_filters(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [
{
"field_name": "process_status",
"field_value": "ready,completed,other",
}
],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["ready", "completed", "other"]
def test_report_with_multiple_filters(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata(
{
"columns": [],
"filter_by": [
{"field_name": "start_from", "field_value": "44"},
{"field_name": "end_from", "field_value": "55"},
{"field_name": "process_status", "field_value": "ready"},
],
}
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from == 44
assert report_filter.start_to is None
assert report_filter.end_from == 55
assert report_filter.end_to is None
assert report_filter.process_status == ["ready"]
def test_report_no_override_with_no_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
},
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_override_with_no_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
},
end_to=54321,
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to == 54321
assert report_filter.process_status is None
def test_report_override_process_model_identifier_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [
{"field_name": "process_model_identifier", "field_value": "bob"}
],
},
process_model_identifier="joe",
)
assert report_filter.process_model_identifier == "joe"
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_override_start_from_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [{"field_name": "start_from", "field_value": "123"}],
},
start_from=321,
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from == 321
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_override_start_to_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [{"field_name": "start_to", "field_value": "123"}],
},
start_to=321,
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to == 321
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_override_end_from_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [{"field_name": "end_from", "field_value": "123"}],
},
end_from=321,
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from == 321
assert report_filter.end_to is None
assert report_filter.process_status is None
def test_report_override_end_to_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [{"field_name": "end_to", "field_value": "123"}],
},
end_to=321,
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to == 321
assert report_filter.process_status is None
def test_report_override_process_status_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [
{"field_name": "process_status", "field_value": "joe,bob"}
],
},
process_status="sue",
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["sue"]
def test_report_override_mulitple_process_status_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [{"field_name": "process_status", "field_value": "sue"}],
},
process_status="joe,bob",
)
assert report_filter.process_model_identifier is None
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["joe", "bob"]
def test_report_override_does_not_override_other_filters(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [
{"field_name": "process_model_identifier", "field_value": "sue"},
{"field_name": "process_status", "field_value": "sue"},
],
},
process_status="joe,bob",
)
assert report_filter.process_model_identifier == "sue"
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["joe", "bob"]
def test_report_override_of_none_does_not_override_filter(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Docstring."""
report_filter = self._filter_from_metadata_with_overrides(
{
"columns": [],
"filter_by": [
{"field_name": "process_model_identifier", "field_value": "sue"},
{"field_name": "process_status", "field_value": "sue"},
],
},
process_status=None,
)
assert report_filter.process_model_identifier == "sue"
assert report_filter.start_from is None
assert report_filter.start_to is None
assert report_filter.end_from is None
assert report_filter.end_to is None
assert report_filter.process_status == ["sue"]