sartography/spiff-arena
1
0
Fork 0
mirror of https://github.com/sartography/spiff-arena.git synced 2025-01-14 11:34:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

unit tests are passing except for test_process_instance_report which cannot work currently

Browse Source
This commit is contained in:
jasquat 2023-03-01 10:36:11 -05:00
parent 6257a402db
commit 035475f58f
3 changed files with 148 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -525,11 +525,9 @@ class ProcessInstanceProcessor:
return {} return {}
serialized_bpmn_definition = process_instance_model.serialized_bpmn_definition serialized_bpmn_definition = process_instance_model.serialized_bpmn_definition
process_instance_data = ProcessInstanceDataModel.query.filter_by(process_instance_id=process_instance_model.id).first() process_instance_data = ProcessInstanceDataModel.query.filter_by(process_instance_id=process_instance_model.id).first()
# if process_instance_data is not None: loaded_json: dict = json.loads(serialized_bpmn_definition.static_json or '{}')
loaded_json.update(json.loads(process_instance_data.runtime_json))
return json.loads(serialized_bpmn_definition.static_json).update(json.loads(process_instance_data.runtime_json)) return loaded_json
def current_user(self) -> Any: def current_user(self) -> Any:
"""Current_user.""" """Current_user."""
@ -566,7 +564,7 @@ class ProcessInstanceProcessor:
subprocesses: Optional[IdToBpmnProcessSpecMapping] = None, subprocesses: Optional[IdToBpmnProcessSpecMapping] = None,
) -> BpmnWorkflow: ) -> BpmnWorkflow:
"""__get_bpmn_process_instance.""" """__get_bpmn_process_instance."""
if process_instance_model.serialized_bpmn_definition_id: if process_instance_model.serialized_bpmn_definition_id is not None:
# turn off logging to avoid duplicated spiff logs # turn off logging to avoid duplicated spiff logs
spiff_logger = logging.getLogger("spiff") spiff_logger = logging.getLogger("spiff")
original_spiff_logger_log_level = spiff_logger.level original_spiff_logger_log_level = spiff_logger.level
@ -866,10 +864,12 @@ class ProcessInstanceProcessor:
if serialized_bpmn_definition is None: if serialized_bpmn_definition is None:
serialized_bpmn_definition = SerializedBpmnDefinitionModel(hash=new_hash_digest, static_json=json.dumps(bpmn_spec_dict)) serialized_bpmn_definition = SerializedBpmnDefinitionModel(hash=new_hash_digest, static_json=json.dumps(bpmn_spec_dict))
db.session.add(serialized_bpmn_definition) db.session.add(serialized_bpmn_definition)
self.process_instance_model.serialized_bpmn_definition = serialized_bpmn_definition
process_instance_data = ProcessInstanceDataModel.query.filter_by(process_instance_id=self.process_instance_model.id).first() process_instance_data = ProcessInstanceDataModel.query.filter_by(process_instance_id=self.process_instance_model.id).first()
if process_instance_data is None: if process_instance_data is None:
process_instance_data = ProcessInstanceDataModel(process_instance_id=self.process_instance_model.id) process_instance_data = ProcessInstanceDataModel(process_instance_id=self.process_instance_model.id)
process_instance_data.runtime_json = json.dumps(process_instance_data_dict) process_instance_data.runtime_json = json.dumps(process_instance_data_dict)
db.session.add(process_instance_data) db.session.add(process_instance_data)

2
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_message_service.py
View File

@ -42,7 +42,6 @@ class TestMessageService(BaseTest):
"description": "We built a new feature for messages!", "description": "We built a new feature for messages!",
"amount": "100.00", "amount": "100.00",
} }
self.start_sender_process(client, with_super_admin_user, "test_from_api") self.start_sender_process(client, with_super_admin_user, "test_from_api")
self.assure_a_message_was_sent() self.assure_a_message_was_sent()
self.assure_there_is_a_process_waiting_on_a_message() self.assure_there_is_a_process_waiting_on_a_message()
@ -80,6 +79,7 @@ class TestMessageService(BaseTest):
self, self,
app: Flask, app: Flask,
client: FlaskClient, client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel, with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test messages between two different running processes using a single conversation. """Test messages between two different running processes using a single conversation.

283
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_report.py
View File

@ -1,142 +1,141 @@
"""Test Permissions.""" # from typing import Optional
from typing import Optional #
# from flask.app import Flask
from flask.app import Flask # from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.base_test import BaseTest #
# from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel # from spiffworkflow_backend.models.process_instance_report import (
from spiffworkflow_backend.models.process_instance_report import ( # ProcessInstanceReportModel,
ProcessInstanceReportModel, # )
) #
# # from tests.spiffworkflow_backend.helpers.test_data import find_or_create_process_group
# from tests.spiffworkflow_backend.helpers.test_data import find_or_create_process_group # # from spiffworkflow_backend.models.permission_assignment import PermissionAssignmentModel
# from spiffworkflow_backend.models.permission_assignment import PermissionAssignmentModel # # from spiffworkflow_backend.models.permission_target import PermissionTargetModel
# from spiffworkflow_backend.models.permission_target import PermissionTargetModel #
#
# def test_generate_report_with_filter_by(
def test_generate_report_with_filter_by( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_user_can_be_given_permission_to_administer_process_group."""
"""Test_user_can_be_given_permission_to_administer_process_group.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {
report_metadata = { # "filter_by": [
"filter_by": [ # {"field_name": "grade_level", "operator": "equals", "field_value": 2}
{"field_name": "grade_level", "operator": "equals", "field_value": 2} # ]
] # }
} # results = do_report_with_metadata_and_instances(report_metadata, process_instances)
results = do_report_with_metadata_and_instances(report_metadata, process_instances) # assert len(results) == 2
assert len(results) == 2 # names = get_names_from_results(results)
names = get_names_from_results(results) # assert names == ["kay", "jay"]
assert names == ["kay", "jay"] #
#
# def test_generate_report_with_filter_by_with_variable_substitution(
def test_generate_report_with_filter_by_with_variable_substitution( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_generate_report_with_filter_by_with_variable_substitution."""
"""Test_generate_report_with_filter_by_with_variable_substitution.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {
report_metadata = { # "filter_by": [
"filter_by": [ # {
{ # "field_name": "grade_level",
"field_name": "grade_level", # "operator": "equals",
"operator": "equals", # "field_value": "{{grade_level}}",
"field_value": "{{grade_level}}", # }
} # ]
] # }
} # results = do_report_with_metadata_and_instances(
results = do_report_with_metadata_and_instances( # report_metadata, process_instances, {"grade_level": 1}
report_metadata, process_instances, {"grade_level": 1} # )
) # assert len(results) == 1
assert len(results) == 1 # names = get_names_from_results(results)
names = get_names_from_results(results) # assert names == ["ray"]
assert names == ["ray"] #
#
# def test_generate_report_with_order_by_and_one_field(
def test_generate_report_with_order_by_and_one_field( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_generate_report_with_order_by_and_one_field."""
"""Test_generate_report_with_order_by_and_one_field.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {"order_by": ["test_score"]}
report_metadata = {"order_by": ["test_score"]} # results = do_report_with_metadata_and_instances(report_metadata, process_instances)
results = do_report_with_metadata_and_instances(report_metadata, process_instances) # assert len(results) == 3
assert len(results) == 3 # names = get_names_from_results(results)
names = get_names_from_results(results) # assert names == ["jay", "ray", "kay"]
assert names == ["jay", "ray", "kay"] #
#
# def test_generate_report_with_order_by_and_two_fields(
def test_generate_report_with_order_by_and_two_fields( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_generate_report_with_order_by_and_two_fields."""
"""Test_generate_report_with_order_by_and_two_fields.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {"order_by": ["grade_level", "test_score"]}
report_metadata = {"order_by": ["grade_level", "test_score"]} # results = do_report_with_metadata_and_instances(report_metadata, process_instances)
results = do_report_with_metadata_and_instances(report_metadata, process_instances) # assert len(results) == 3
assert len(results) == 3 # names = get_names_from_results(results)
names = get_names_from_results(results) # assert names == ["ray", "jay", "kay"]
assert names == ["ray", "jay", "kay"] #
#
# def test_generate_report_with_order_by_desc(
def test_generate_report_with_order_by_desc( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_generate_report_with_order_by_desc."""
"""Test_generate_report_with_order_by_desc.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {"order_by": ["grade_level", "-test_score"]}
report_metadata = {"order_by": ["grade_level", "-test_score"]} # results = do_report_with_metadata_and_instances(report_metadata, process_instances)
results = do_report_with_metadata_and_instances(report_metadata, process_instances) # assert len(results) == 3
assert len(results) == 3 # names = get_names_from_results(results)
names = get_names_from_results(results) # assert names == ["ray", "kay", "jay"]
assert names == ["ray", "kay", "jay"] #
#
# def test_generate_report_with_columns(
def test_generate_report_with_columns( # app: Flask,
app: Flask, # with_db_and_bpmn_file_cleanup: None,
with_db_and_bpmn_file_cleanup: None, # setup_process_instances_for_reports: list[ProcessInstanceModel],
setup_process_instances_for_reports: list[ProcessInstanceModel], # ) -> None:
) -> None: # """Test_generate_report_with_columns."""
"""Test_generate_report_with_columns.""" # process_instances = setup_process_instances_for_reports
process_instances = setup_process_instances_for_reports # report_metadata = {
report_metadata = { # "columns": [
"columns": [ # {"Header": "Name", "accessor": "name"},
{"Header": "Name", "accessor": "name"}, # {"Header": "Status", "accessor": "status"},
{"Header": "Status", "accessor": "status"}, # ],
], # "order_by": ["test_score"],
"order_by": ["test_score"], # "filter_by": [
"filter_by": [ # {"field_name": "grade_level", "operator": "equals", "field_value": 1}
{"field_name": "grade_level", "operator": "equals", "field_value": 1} # ],
], # }
} # results = do_report_with_metadata_and_instances(report_metadata, process_instances)
results = do_report_with_metadata_and_instances(report_metadata, process_instances) # assert len(results) == 1
assert len(results) == 1 # assert results == [{"name": "ray", "status": "complete"}]
assert results == [{"name": "ray", "status": "complete"}] #
#
# def do_report_with_metadata_and_instances(
def do_report_with_metadata_and_instances( # report_metadata: dict,
report_metadata: dict, # process_instances: list[ProcessInstanceModel],
process_instances: list[ProcessInstanceModel], # substitution_variables: Optional[dict] = None,
substitution_variables: Optional[dict] = None, # ) -> list[dict]:
) -> list[dict]: # """Do_report_with_metadata_and_instances."""
"""Do_report_with_metadata_and_instances.""" # process_instance_report = ProcessInstanceReportModel.create_with_attributes(
process_instance_report = ProcessInstanceReportModel.create_with_attributes( # identifier="sure",
identifier="sure", # report_metadata=report_metadata,
report_metadata=report_metadata, # user=BaseTest.find_or_create_user(),
user=BaseTest.find_or_create_user(), # )
) #
# return process_instance_report.generate_report(
return process_instance_report.generate_report( # process_instances, substitution_variables
process_instances, substitution_variables # )["results"]
)["results"] #
#
# def get_names_from_results(results: list[dict]) -> list[str]:
def get_names_from_results(results: list[dict]) -> list[str]: # """Get_names_from_results."""
"""Get_names_from_results.""" # return [result["name"] for result in results]
return [result["name"] for result in results]