Merge process_instance_list query filters with report filters (#37)
This commit is contained in:
parent
92a37b61de
commit
ab4bd7654e
|
@ -72,6 +72,9 @@ from spiffworkflow_backend.services.message_service import MessageService
|
||||||
from spiffworkflow_backend.services.process_instance_processor import (
|
from spiffworkflow_backend.services.process_instance_processor import (
|
||||||
ProcessInstanceProcessor,
|
ProcessInstanceProcessor,
|
||||||
)
|
)
|
||||||
|
from spiffworkflow_backend.services.process_instance_report_service import (
|
||||||
|
ProcessInstanceReportService,
|
||||||
|
)
|
||||||
from spiffworkflow_backend.services.process_instance_service import (
|
from spiffworkflow_backend.services.process_instance_service import (
|
||||||
ProcessInstanceService,
|
ProcessInstanceService,
|
||||||
)
|
)
|
||||||
|
@ -725,11 +728,22 @@ def process_instance_list(
|
||||||
process_status: Optional[str] = None,
|
process_status: Optional[str] = None,
|
||||||
) -> flask.wrappers.Response:
|
) -> flask.wrappers.Response:
|
||||||
"""Process_instance_list."""
|
"""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_model_identifier = un_modify_modified_process_model_id(modified_process_model_identifier)
|
||||||
process_instance_query = ProcessInstanceModel.query
|
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(
|
process_model = get_process_model(
|
||||||
f"{process_model_identifier}",
|
f"{report_filter.process_model_identifier}",
|
||||||
)
|
)
|
||||||
|
|
||||||
process_instance_query = process_instance_query.filter_by(
|
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(
|
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(
|
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(
|
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(
|
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:
|
if report_filter.process_status is not None:
|
||||||
process_status_array = process_status.split(",")
|
|
||||||
process_instance_query = process_instance_query.filter(
|
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(
|
process_instances = process_instance_query.order_by(
|
||||||
ProcessInstanceModel.start_in_seconds.desc(), ProcessInstanceModel.id.desc() # type: ignore
|
ProcessInstanceModel.start_in_seconds.desc(), ProcessInstanceModel.id.desc() # type: ignore
|
||||||
).paginate(page=page, per_page=per_page, error_out=False)
|
).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
|
# TODO need to look into test failures when the results from result_dict is
|
||||||
# used instead of the process instances
|
# used instead of the process instances
|
||||||
|
|
||||||
|
|
|
@ -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
|
|
@ -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"]
|
Loading…
Reference in New Issue