spiffworkflow-backend/conftest.py

128 lines
4.0 KiB
Python
Raw Normal View History

2022-05-17 20:47:55 +00:00
"""Conftest."""
2022-05-19 18:38:47 +00:00
import os
2022-06-21 19:13:02 +00:00
import shutil
2022-05-19 18:38:47 +00:00
import pytest
from flask.app import Flask
from flask.testing import FlaskClient
2022-11-18 21:45:44 +00:00
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.db import SpiffworkflowBaseDBModel
from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
2022-07-10 05:24:12 +00:00
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.user import UserModel
2022-07-10 05:24:12 +00:00
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
2022-06-21 19:13:02 +00:00
from spiffworkflow_backend.services.process_model_service import ProcessModelService
2022-05-17 20:47:55 +00:00
2022-11-09 18:45:16 +00:00
# from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
2022-05-20 15:46:44 +00:00
# We need to call this before importing spiffworkflow_backend
2022-05-20 15:46:44 +00:00
# otherwise typeguard cannot work. hence the noqa: E402
if os.environ.get("RUN_TYPEGUARD") == "true":
2022-05-20 15:55:09 +00:00
from typeguard.importhook import install_import_hook
install_import_hook(packages="spiffworkflow_backend")
2022-05-20 15:46:44 +00:00
from spiffworkflow_backend import create_app # noqa: E402
2022-05-17 20:47:55 +00:00
@pytest.fixture(scope="session")
def app() -> Flask:
2022-05-17 20:47:55 +00:00
"""App."""
os.environ["SPIFFWORKFLOW_BACKEND_ENV"] = "unit_testing"
os.environ["FLASK_SESSION_SECRET_KEY"] = "super_secret_key"
app = create_app()
2022-05-19 20:08:38 +00:00
return app
2022-06-21 19:13:02 +00:00
@pytest.fixture()
2022-07-10 05:24:12 +00:00
def with_db_and_bpmn_file_cleanup() -> None:
2022-06-21 19:13:02 +00:00
"""Process_group_resource."""
db.session.query(HumanTaskUserModel).delete()
for model in SpiffworkflowBaseDBModel._all_subclasses():
db.session.query(model).delete()
db.session.commit()
2022-06-21 19:13:02 +00:00
try:
yield
finally:
process_model_service = ProcessModelService()
if os.path.exists(process_model_service.root_path()):
shutil.rmtree(process_model_service.root_path())
2022-07-10 05:24:12 +00:00
@pytest.fixture()
def with_super_admin_user() -> UserModel:
"""With_super_admin_user."""
return BaseTest.create_user_with_permission("super_admin")
2022-07-10 05:24:12 +00:00
@pytest.fixture()
2022-11-09 18:45:16 +00:00
def setup_process_instances_for_reports(
client: FlaskClient, with_super_admin_user: UserModel
) -> list[ProcessInstanceModel]:
2022-07-10 05:24:12 +00:00
"""Setup_process_instances_for_reports."""
user = with_super_admin_user
2022-07-10 05:24:12 +00:00
process_group_id = "runs_without_input"
process_model_id = "sample"
2022-11-09 18:45:16 +00:00
# bpmn_file_name = "sample.bpmn"
bpmn_file_location = "sample"
process_model_identifier = BaseTest().create_group_and_model_with_bpmn(
client,
with_super_admin_user,
process_group_id=process_group_id,
process_model_id=process_model_id,
# bpmn_file_name=bpmn_file_name,
2022-11-09 18:45:16 +00:00
bpmn_file_location=bpmn_file_location,
)
# BaseTest().create_process_group(
# client=client, user=user, process_group_id=process_group_id, display_name=process_group_id
# )
# process_model_id = "runs_without_input/sample"
# load_test_spec(
# process_model_id=f"{process_group_id}/{process_model_id}",
# process_model_source_directory="sample"
# )
2022-07-10 05:24:12 +00:00
process_instances = []
for data in [kay(), ray(), jay()]:
2022-11-24 17:00:26 +00:00
process_instance = ProcessInstanceService.create_process_instance_from_process_model_identifier(
# process_group_identifier=process_group_id,
process_model_identifier=process_model_identifier,
2022-07-10 05:24:12 +00:00
user=user,
)
processor = ProcessInstanceProcessor(process_instance)
processor.slam_in_data(data)
process_instance.status = "complete"
db.session.add(process_instance)
db.session.commit()
2022-07-10 05:24:12 +00:00
process_instances.append(process_instance)
return process_instances
def kay() -> dict:
"""Kay."""
return {"name": "kay", "grade_level": 2, "test_score": 10}
def ray() -> dict:
"""Ray."""
return {"name": "ray", "grade_level": 1, "test_score": 9}
def jay() -> dict:
"""Jay."""
return {"name": "jay", "grade_level": 2, "test_score": 8}