remove api usages from script tests and pyl w/ burnettk

This commit is contained in:
jasquat 2023-05-22 14:09:17 -04:00
parent b575ef09de
commit 1c0d62d366
No known key found for this signature in database
25 changed files with 128 additions and 342 deletions

View File

@ -4,19 +4,11 @@ import shutil
import pytest import pytest
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.bpmn_process import BpmnProcessModel from spiffworkflow_backend.models.bpmn_process import BpmnProcessModel
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.user import UserModel from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.process_model_service import ProcessModelService
@ -62,5 +54,5 @@ def with_db_and_bpmn_file_cleanup() -> None:
@pytest.fixture() @pytest.fixture()
def with_super_admin_user() -> UserModel: def with_super_admin_user() -> UserModel:
raise Exception("HEY") """With_super_admin_user."""
# return BaseTest.create_user_with_permission("super_admin") return BaseTest.create_user_with_permission("super_admin")

View File

@ -1,43 +0,0 @@
"""Save process instance metadata."""
from typing import Any
from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.process_instance_metadata import (
ProcessInstanceMetadataModel,
)
from spiffworkflow_backend.models.script_attributes_context import (
ScriptAttributesContext,
)
from spiffworkflow_backend.scripts.script import Script
class SaveProcessInstanceMetadata(Script):
"""SaveProcessInstanceMetadata."""
def get_description(self) -> str:
"""Get_description."""
return """Save a given dict as process instance metadata (useful for creating reports)."""
def run(
self,
script_attributes_context: ScriptAttributesContext,
*args: Any,
**kwargs: Any,
) -> Any:
"""Run."""
metadata_dict = args[0]
if script_attributes_context.process_instance_id is None:
raise self.get_proces_instance_id_is_missing_error("save_process_instance_metadata")
for key, value in metadata_dict.items():
pim = ProcessInstanceMetadataModel.query.filter_by(
process_instance_id=script_attributes_context.process_instance_id,
key=key,
).first()
if pim is None:
pim = ProcessInstanceMetadataModel(
process_instance_id=script_attributes_context.process_instance_id,
key=key,
)
pim.value = value
db.session.add(pim)
db.session.commit()

View File

@ -25,7 +25,6 @@ from uuid import UUID
import dateparser import dateparser
import pytz import pytz
from flask import current_app from flask import current_app
from flask import g
from lxml import etree # type: ignore from lxml import etree # type: ignore
from lxml.etree import XMLSyntaxError # type: ignore from lxml.etree import XMLSyntaxError # type: ignore
from RestrictedPython import safe_globals # type: ignore from RestrictedPython import safe_globals # type: ignore

View File

@ -1,52 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_96f6665" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="3.0.0-dev">
<bpmn:process id="test_save_process_instance_metadata" isExecutable="true">
<bpmn:startEvent id="Event_0r6oru6">
<bpmn:outgoing>Flow_1j4jzft</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:sequenceFlow id="Flow_1j4jzft" sourceRef="Event_0r6oru6" targetRef="save_key1" />
<bpmn:endEvent id="Event_1s123jg">
<bpmn:incoming>Flow_01xr2ac</bpmn:incoming>
</bpmn:endEvent>
<bpmn:scriptTask id="save_key1">
<bpmn:incoming>Flow_1j4jzft</bpmn:incoming>
<bpmn:outgoing>Flow_10xyk22</bpmn:outgoing>
<bpmn:script>save_process_instance_metadata({"key1": "value1"})</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_10xyk22" sourceRef="save_key1" targetRef="save_key2" />
<bpmn:scriptTask id="save_key2">
<bpmn:incoming>Flow_10xyk22</bpmn:incoming>
<bpmn:outgoing>Flow_01xr2ac</bpmn:outgoing>
<bpmn:script>save_process_instance_metadata({"key2": "value2", "key3": "value3"})</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_01xr2ac" sourceRef="save_key2" targetRef="Event_1s123jg" />
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="test_save_process_instance_metadata">
<bpmndi:BPMNShape id="Event_0r6oru6_di" bpmnElement="Event_0r6oru6">
<dc:Bounds x="162" y="162" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0zfzev2_di" bpmnElement="save_key1">
<dc:Bounds x="250" y="140" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0d1q8x4_di" bpmnElement="save_key2">
<dc:Bounds x="410" y="140" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_1s123jg_di" bpmnElement="Event_1s123jg">
<dc:Bounds x="582" y="162" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNEdge id="Flow_1j4jzft_di" bpmnElement="Flow_1j4jzft">
<di:waypoint x="198" y="180" />
<di:waypoint x="250" y="180" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_10xyk22_di" bpmnElement="Flow_10xyk22">
<di:waypoint x="350" y="180" />
<di:waypoint x="410" y="180" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_01xr2ac_di" bpmnElement="Flow_01xr2ac">
<di:waypoint x="510" y="180" />
<di:waypoint x="582" y="180" />
</bpmndi:BPMNEdge>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

View File

@ -9,14 +9,12 @@ from typing import Optional
from flask import current_app from flask import current_app
from flask.testing import FlaskClient from flask.testing import FlaskClient
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from spiffworkflow_backend.services.process_instance_service import ProcessInstanceService
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from werkzeug.test import TestResponse # type: ignore from werkzeug.test import TestResponse # type: ignore
from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.exceptions.api_error import ApiError from spiffworkflow_backend.exceptions.api_error import ApiError
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.models.permission_assignment import Permission from spiffworkflow_backend.models.permission_assignment import Permission
from spiffworkflow_backend.models.permission_target import PermissionTargetModel from spiffworkflow_backend.models.permission_target import PermissionTargetModel
from spiffworkflow_backend.models.process_group import ProcessGroup from spiffworkflow_backend.models.process_group import ProcessGroup
@ -29,9 +27,11 @@ from spiffworkflow_backend.models.process_model import ProcessModelInfoSchema
from spiffworkflow_backend.models.user import UserModel from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.file_system_service import FileSystemService from spiffworkflow_backend.services.file_system_service import FileSystemService
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from spiffworkflow_backend.services.process_instance_queue_service import ( from spiffworkflow_backend.services.process_instance_queue_service import (
ProcessInstanceQueueService, ProcessInstanceQueueService,
) )
from spiffworkflow_backend.services.process_instance_service import ProcessInstanceService
from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.user_service import UserService from spiffworkflow_backend.services.user_service import UserService
@ -319,7 +319,7 @@ class BaseTest:
) -> UserModel: ) -> UserModel:
# user = BaseTest.find_or_create_user(username=username) # user = BaseTest.find_or_create_user(username=username)
# return cls.add_permissions_to_user(user, target_uri=target_uri, permission_names=permission_names) # return cls.add_permissions_to_user(user, target_uri=target_uri, permission_names=permission_names)
user = BaseTest.find_or_create_user(username='testadmin1') user = BaseTest.find_or_create_user(username="testadmin1")
AuthorizationService.import_permissions_from_yaml_file(user) AuthorizationService.import_permissions_from_yaml_file(user)
return user return user
@ -418,9 +418,7 @@ class BaseTest:
bpmn_file_name="message_sender.bpmn", # Slightly misnamed, it sends and receives bpmn_file_name="message_sender.bpmn", # Slightly misnamed, it sends and receives
) )
process_instance = self.create_process_instance_from_process_model( process_instance = self.create_process_instance_from_process_model(process_model)
process_model
)
processor_send_receive = ProcessInstanceProcessor(process_instance) processor_send_receive = ProcessInstanceProcessor(process_instance)
processor_send_receive.do_engine_steps(save=True) processor_send_receive.do_engine_steps(save=True)
task = processor_send_receive.get_all_user_tasks()[0] task = processor_send_receive.get_all_user_tasks()[0]

View File

@ -1,31 +1,21 @@
"""Test_get_localtime."""
from operator import itemgetter from operator import itemgetter
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.script_attributes_context import ( from spiffworkflow_backend.models.script_attributes_context import (
ScriptAttributesContext, ScriptAttributesContext,
) )
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.scripts.get_all_permissions import GetAllPermissions from spiffworkflow_backend.scripts.get_all_permissions import GetAllPermissions
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
class TestGetAllPermissions(BaseTest): class TestGetAllPermissions(BaseTest):
"""TestGetAllPermissions."""
def test_can_get_all_permissions( def test_can_get_all_permissions(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test_can_get_all_permissions."""
self.find_or_create_user("test_user")
# now that we have everything, try to clear it out... # now that we have everything, try to clear it out...
script_attributes_context = ScriptAttributesContext( script_attributes_context = ScriptAttributesContext(
task=None, task=None,

View File

@ -3,14 +3,12 @@ import json
from flask import g from flask import g
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.script_attributes_context import ( from spiffworkflow_backend.models.script_attributes_context import (
ScriptAttributesContext, ScriptAttributesContext,
) )
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.scripts.get_current_user import GetCurrentUser from spiffworkflow_backend.scripts.get_current_user import GetCurrentUser
@ -18,11 +16,8 @@ class TestGetCurrentUser(BaseTest):
def test_get_current_user( def test_get_current_user(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test_can_get_members_of_a_group."""
testuser1 = self.find_or_create_user("testuser1") testuser1 = self.find_or_create_user("testuser1")
testuser1.tenant_specific_field_1 = "456" testuser1.tenant_specific_field_1 = "456"
db.session.add(testuser1) db.session.add(testuser1)

View File

@ -1,12 +1,9 @@
"""Test_get_localtime."""
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.group import GroupModel from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )
@ -14,16 +11,11 @@ from spiffworkflow_backend.services.user_service import UserService
class TestGetGroupMembers(BaseTest): class TestGetGroupMembers(BaseTest):
"""TestGetGroupMembers."""
def test_can_get_members_of_a_group( def test_can_get_members_of_a_group(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test_can_get_members_of_a_group."""
initiator_user = self.find_or_create_user("initiator_user") initiator_user = self.find_or_create_user("initiator_user")
testuser1 = self.find_or_create_user("testuser1") testuser1 = self.find_or_create_user("testuser1")
testuser2 = self.find_or_create_user("testuser2") testuser2 = self.find_or_create_user("testuser2")
@ -38,7 +30,6 @@ class TestGetGroupMembers(BaseTest):
UserService.add_user_to_group(testuser2, group_a) UserService.add_user_to_group(testuser2, group_a)
UserService.add_user_to_group(testuser3, group_b) UserService.add_user_to_group(testuser3, group_b)
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
process_model = load_test_spec( process_model = load_test_spec(
process_model_id="test_group/get_group_members", process_model_id="test_group/get_group_members",
bpmn_file_name="get_group_members.bpmn", bpmn_file_name="get_group_members.bpmn",

View File

@ -1,10 +1,7 @@
"""Test_get_localtime."""
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
@ -18,12 +15,8 @@ class TestGetLastUserCompletingTask(BaseTest):
def test_get_last_user_completing_task_script_works( def test_get_last_user_completing_task_script_works(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test_sets_permission_correctly_on_human_task."""
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
initiator_user = self.find_or_create_user("initiator_user") initiator_user = self.find_or_create_user("initiator_user")
assert initiator_user.principal is not None assert initiator_user.principal is not None
AuthorizationService.import_permissions_from_yaml_file() AuthorizationService.import_permissions_from_yaml_file()

View File

@ -1,9 +1,7 @@
"""Test_get_localtime."""
import datetime import datetime
import pytz import pytz
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
@ -20,10 +18,7 @@ from spiffworkflow_backend.services.process_instance_service import (
class TestGetLocaltime(BaseTest): class TestGetLocaltime(BaseTest):
"""TestProcessAPi."""
def test_get_localtime_script_directly(self) -> None: def test_get_localtime_script_directly(self) -> None:
"""Test_get_localtime_script_directly."""
current_time = datetime.datetime.now() current_time = datetime.datetime.now()
timezone = "US/Pacific" timezone = "US/Pacific"
process_model_identifier = "test_process_model" process_model_identifier = "test_process_model"
@ -44,17 +39,14 @@ class TestGetLocaltime(BaseTest):
def test_get_localtime_script_through_bpmn( def test_get_localtime_script_through_bpmn(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
"""Test_process_instance_run."""
initiator_user = self.find_or_create_user("initiator_user") initiator_user = self.find_or_create_user("initiator_user")
self.add_permissions_to_user( self.add_permissions_to_user(
initiator_user, initiator_user,
target_uri="/v1.0/process-groups", target_uri="/v1.0/process-groups",
permission_names=["read", "create"], permission_names=["read", "create"],
) )
self.create_process_group_with_api(client=client, user=initiator_user, process_group_id="test_group")
process_model = load_test_spec( process_model = load_test_spec(
process_model_id="test_group/get_localtime", process_model_id="test_group/get_localtime",
bpmn_file_name="get_localtime.bpmn", bpmn_file_name="get_localtime.bpmn",

View File

@ -1,10 +1,8 @@
"""Test_get_localtime.""" """Test_get_localtime."""
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
@ -18,19 +16,15 @@ class TestGetProcessInitiatorUser(BaseTest):
def test_get_process_initiator_user( def test_get_process_initiator_user(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None: ) -> None:
"""Test_sets_permission_correctly_on_human_task.""" """Test_sets_permission_correctly_on_human_task."""
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
initiator_user = self.find_or_create_user("initiator_user") initiator_user = self.find_or_create_user("initiator_user")
assert initiator_user.principal is not None assert initiator_user.principal is not None
AuthorizationService.import_permissions_from_yaml_file() AuthorizationService.import_permissions_from_yaml_file()
process_model = load_test_spec( process_model = load_test_spec(
process_model_id="misc/category_number_one/simple_form", process_model_id="misc/category_number_one/simple_form",
# bpmn_file_name="simp.bpmn",
process_model_source_directory="simple_form", process_model_source_directory="simple_form",
) )
process_instance = self.create_process_instance_from_process_model( process_instance = self.create_process_instance_from_process_model(

View File

@ -1,7 +1,6 @@
"""Test_get_localtime.""" """Test_get_localtime."""
import pytest import pytest
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
@ -15,7 +14,6 @@ class TestRefreshPermissions(BaseTest):
def test_refresh_permissions_requires_elevated_permission( def test_refresh_permissions_requires_elevated_permission(
self, self,
app: Flask, app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
basic_user = self.find_or_create_user("basic_user") basic_user = self.find_or_create_user("basic_user")

View File

@ -1,42 +0,0 @@
"""Test_get_localtime."""
from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.process_instance_metadata import (
ProcessInstanceMetadataModel,
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
class TestSaveProcessInstanceMetadata(BaseTest):
"""TestSaveProcessInstanceMetadata."""
def test_can_save_process_instance_metadata(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_can_save_process_instance_metadata."""
self.create_process_group_with_api(client, with_super_admin_user, "test_group", "test_group")
process_model = load_test_spec(
process_model_id="save_process_instance_metadata/save_process_instance_metadata",
bpmn_file_name="save_process_instance_metadata.bpmn",
process_model_source_directory="save_process_instance_metadata",
)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=with_super_admin_user
)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True)
process_instance_metadata = ProcessInstanceMetadataModel.query.filter_by(
process_instance_id=process_instance.id
).all()
assert len(process_instance_metadata) == 3

View File

@ -1,12 +1,11 @@
"""Test_message_service.""" """Test_message_service."""
import pytest import pytest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask import Flask from flask import Flask
from flask.testing import FlaskClient from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.group import GroupModel from spiffworkflow_backend.models.group import GroupModel
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.authorization_service import GroupPermissionsDict from spiffworkflow_backend.services.authorization_service import GroupPermissionsDict
from spiffworkflow_backend.services.authorization_service import InvalidPermissionError from spiffworkflow_backend.services.authorization_service import InvalidPermissionError
@ -17,7 +16,6 @@ from spiffworkflow_backend.services.process_instance_processor import (
from spiffworkflow_backend.services.process_instance_service import ( from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService, ProcessInstanceService,
) )
from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.user_service import UserService from spiffworkflow_backend.services.user_service import UserService
@ -137,7 +135,7 @@ class TestAuthorizationService(BaseTest):
"delete", "delete",
), ),
("/process-instances/some-process-group:some-process-model:*", "read"), ("/process-instances/some-process-group:some-process-model:*", "read"),
('/process-model-natural-language/some-process-group:some-process-model:*', "create"), ("/process-model-natural-language/some-process-group:some-process-model:*", "create"),
("/process-model-publish/some-process-group:some-process-model:*", "create"), ("/process-model-publish/some-process-group:some-process-model:*", "create"),
("/process-models/some-process-group:some-process-model:*", "create"), ("/process-models/some-process-group:some-process-model:*", "create"),
("/process-models/some-process-group:some-process-model:*", "delete"), ("/process-models/some-process-group:some-process-model:*", "delete"),
@ -160,26 +158,28 @@ class TestAuthorizationService(BaseTest):
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
"""Test_explode_permissions_start_on_process_group.""" """Test_explode_permissions_start_on_process_group."""
expected_permissions = sorted([ expected_permissions = sorted(
("/event-error-details/some-process-group:some-process-model:*", "read"), [
( ("/event-error-details/some-process-group:some-process-model:*", "read"),
"/logs/some-process-group:some-process-model:*", (
"read", "/logs/some-process-group:some-process-model:*",
), "read",
( ),
"/logs/typeahead-filter-values/some-process-group:some-process-model:*", (
"read", "/logs/typeahead-filter-values/some-process-group:some-process-model:*",
), "read",
( ),
"/process-data-file-download/some-process-group:some-process-model:*", (
"read", "/process-data-file-download/some-process-group:some-process-model:*",
), "read",
( ),
"/process-instances/for-me/some-process-group:some-process-model:*", (
"read", "/process-instances/for-me/some-process-group:some-process-model:*",
), "read",
("/process-instances/some-process-group:some-process-model:*", "create"), ),
]) ("/process-instances/some-process-group:some-process-model:*", "create"),
]
)
permissions_to_assign = AuthorizationService.explode_permissions( permissions_to_assign = AuthorizationService.explode_permissions(
"start", "PG:/some-process-group/some-process-model" "start", "PG:/some-process-group/some-process-model"
) )
@ -219,7 +219,7 @@ class TestAuthorizationService(BaseTest):
"delete", "delete",
), ),
("/process-instances/some-process-group:some-process-model/*", "read"), ("/process-instances/some-process-group:some-process-model/*", "read"),
('/process-model-natural-language/some-process-group:some-process-model/*', "create"), ("/process-model-natural-language/some-process-group:some-process-model/*", "create"),
("/process-model-publish/some-process-group:some-process-model/*", "create"), ("/process-model-publish/some-process-group:some-process-model/*", "create"),
("/process-models/some-process-group:some-process-model/*", "create"), ("/process-models/some-process-group:some-process-model/*", "create"),
("/process-models/some-process-group:some-process-model/*", "delete"), ("/process-models/some-process-group:some-process-model/*", "delete"),
@ -242,26 +242,28 @@ class TestAuthorizationService(BaseTest):
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
"""Test_explode_permissions_start_on_process_model.""" """Test_explode_permissions_start_on_process_model."""
expected_permissions = sorted([ expected_permissions = sorted(
( [
"/event-error-details/some-process-group:some-process-model/*", (
"read", "/event-error-details/some-process-group:some-process-model/*",
), "read",
( ),
"/logs/some-process-group:some-process-model/*", (
"read", "/logs/some-process-group:some-process-model/*",
), "read",
("/logs/typeahead-filter-values/some-process-group:some-process-model/*", "read"), ),
( ("/logs/typeahead-filter-values/some-process-group:some-process-model/*", "read"),
"/process-data-file-download/some-process-group:some-process-model/*", (
"read", "/process-data-file-download/some-process-group:some-process-model/*",
), "read",
( ),
"/process-instances/for-me/some-process-group:some-process-model/*", (
"read", "/process-instances/for-me/some-process-group:some-process-model/*",
), "read",
("/process-instances/some-process-group:some-process-model/*", "create"), ),
]) ("/process-instances/some-process-group:some-process-model/*", "create"),
]
)
permissions_to_assign = AuthorizationService.explode_permissions( permissions_to_assign = AuthorizationService.explode_permissions(
"start", "PM:/some-process-group/some-process-model" "start", "PM:/some-process-group/some-process-model"
) )
@ -274,30 +276,32 @@ class TestAuthorizationService(BaseTest):
client: FlaskClient, client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
expected_permissions = sorted([ expected_permissions = sorted(
("/active-users/*", "create"), [
("/connector-proxy/typeahead/*", "read"), ("/active-users/*", "create"),
("/debug/version-info", "read"), ("/connector-proxy/typeahead/*", "read"),
("/process-groups", "read"), ("/debug/version-info", "read"),
("/process-instances/find-by-id/*", "read"), ("/process-groups", "read"),
("/process-instances/for-me", "create"), ("/process-instances/find-by-id/*", "read"),
("/process-instances/report-metadata", "read"), ("/process-instances/for-me", "create"),
("/process-instances/reports/*", "create"), ("/process-instances/report-metadata", "read"),
("/process-instances/reports/*", "delete"), ("/process-instances/reports/*", "create"),
("/process-instances/reports/*", "read"), ("/process-instances/reports/*", "delete"),
("/process-instances/reports/*", "update"), ("/process-instances/reports/*", "read"),
("/process-models", "read"), ("/process-instances/reports/*", "update"),
("/processes", "read"), ("/process-models", "read"),
("/processes/callers", "read"), ("/processes", "read"),
("/service-tasks", "read"), ("/processes/callers", "read"),
("/tasks/*", "create"), ("/service-tasks", "read"),
("/tasks/*", "delete"), ("/tasks/*", "create"),
("/tasks/*", "read"), ("/tasks/*", "delete"),
("/tasks/*", "update"), ("/tasks/*", "read"),
("/user-groups/for-current-user", "read"), ("/tasks/*", "update"),
("/users/exists/by-username", "create"), ("/user-groups/for-current-user", "read"),
("/users/search", "read"), ("/users/exists/by-username", "create"),
]) ("/users/search", "read"),
]
)
permissions_to_assign = AuthorizationService.explode_permissions("all", "BASIC") permissions_to_assign = AuthorizationService.explode_permissions("all", "BASIC")
permissions_to_assign_tuples = sorted([(p.target_uri, p.permission) for p in permissions_to_assign]) permissions_to_assign_tuples = sorted([(p.target_uri, p.permission) for p in permissions_to_assign])
assert permissions_to_assign_tuples == expected_permissions assert permissions_to_assign_tuples == expected_permissions
@ -308,28 +312,30 @@ class TestAuthorizationService(BaseTest):
client: FlaskClient, client: FlaskClient,
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
expected_permissions = sorted([ expected_permissions = sorted(
("/authentications", "read"), [
("/can-run-privileged-script/*", "create"), ("/authentications", "read"),
("/debug/*", "create"), ("/can-run-privileged-script/*", "create"),
("/messages", "read"), ("/debug/*", "create"),
("/messages/*", "create"), ("/messages", "read"),
("/process-instance-reset/*", "create"), ("/messages/*", "create"),
("/process-instance-resume/*", "create"), ("/process-instance-reset/*", "create"),
("/process-instance-suspend/*", "create"), ("/process-instance-resume/*", "create"),
("/process-instance-terminate/*", "create"), ("/process-instance-suspend/*", "create"),
("/process-instances/*", "create"), ("/process-instance-terminate/*", "create"),
("/process-instances/*", "delete"), ("/process-instances/*", "create"),
("/process-instances/*", "read"), ("/process-instances/*", "delete"),
("/process-instances/*", "update"), ("/process-instances/*", "read"),
("/secrets/*", "create"), ("/process-instances/*", "update"),
("/secrets/*", "delete"), ("/secrets/*", "create"),
("/secrets/*", "read"), ("/secrets/*", "delete"),
("/secrets/*", "update"), ("/secrets/*", "read"),
("/send-event/*", "create"), ("/secrets/*", "update"),
("/task-complete/*", "create"), ("/send-event/*", "create"),
("/task-data/*", "update"), ("/task-complete/*", "create"),
]) ("/task-data/*", "update"),
]
)
permissions_to_assign = AuthorizationService.explode_permissions("all", "ELEVATED") permissions_to_assign = AuthorizationService.explode_permissions("all", "ELEVATED")
permissions_to_assign_tuples = sorted([(p.target_uri, p.permission) for p in permissions_to_assign]) permissions_to_assign_tuples = sorted([(p.target_uri, p.permission) for p in permissions_to_assign])
assert permissions_to_assign_tuples == expected_permissions assert permissions_to_assign_tuples == expected_permissions

View File

@ -1,10 +1,9 @@
"""Test_various_bpmn_constructs.""" """Test_various_bpmn_constructs."""
from flask.app import Flask from flask.app import Flask
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.testing import FlaskClient from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )
@ -42,7 +41,9 @@ class TestDotNotation(BaseTest):
"invoice.invoiceAmount": "1000.00", "invoice.invoiceAmount": "1000.00",
"invoice.dueDate": "09/30/2022", "invoice.dueDate": "09/30/2022",
} }
ProcessInstanceService.complete_form_task(processor, user_task, form_data, process_instance.process_initiator, human_task) ProcessInstanceService.complete_form_task(
processor, user_task, form_data, process_instance.process_initiator, human_task
)
expected = { expected = {
"contibutorName": "Elizabeth", "contibutorName": "Elizabeth",

View File

@ -9,14 +9,10 @@ from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
from spiffworkflow_backend.models.process_model import ProcessModelInfo from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )
from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError from spiffworkflow_backend.services.workflow_execution_service import WorkflowExecutionServiceError
@ -27,12 +23,8 @@ class TestErrorHandlingService(BaseTest):
Like it can fire off BPMN messages in case a BPMN Task is waiting for that message. Like it can fire off BPMN messages in case a BPMN Task is waiting for that message.
""" """
def run_process_model_and_handle_error( def run_process_model_and_handle_error(self, process_model: ProcessModelInfo) -> ProcessInstanceModel:
self, process_model: ProcessModelInfo process_instance = self.create_process_instance_from_process_model(process_model)
) -> ProcessInstanceModel:
process_instance = self.create_process_instance_from_process_model(
process_model
)
pip = ProcessInstanceProcessor(process_instance) pip = ProcessInstanceProcessor(process_instance)
with pytest.raises(WorkflowExecutionServiceError) as e: with pytest.raises(WorkflowExecutionServiceError) as e:
pip.do_engine_steps(save=True) pip.do_engine_steps(save=True)

View File

@ -1,15 +1,13 @@
"""Test_message_instance.""" """Test_message_instance."""
import pytest import pytest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask import Flask from flask import Flask
from flask.testing import FlaskClient from flask.testing import FlaskClient
from spiffworkflow_backend.models.process_model import ProcessModelInfo
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.message_instance import MessageInstanceModel from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.models.user import UserModel from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.services.process_model_service import ProcessModelService
class TestMessageInstance(BaseTest): class TestMessageInstance(BaseTest):

View File

@ -93,7 +93,6 @@ class TestMessageService(BaseTest):
with_db_and_bpmn_file_cleanup: None, with_db_and_bpmn_file_cleanup: None,
) -> None: ) -> None:
"""Test_can_send_message_to_multiple_process_models.""" """Test_can_send_message_to_multiple_process_models."""
process_group_id = "test_group_multi"
# self.create_process_group_with_api(client, with_super_admin_user, process_group_id, process_group_id) # self.create_process_group_with_api(client, with_super_admin_user, process_group_id, process_group_id)
process_model_sender = load_test_spec( process_model_sender = load_test_spec(

View File

@ -8,7 +8,6 @@ from spiffworkflow_backend.models.group import GroupModel
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
from spiffworkflow_backend.models.principal import PrincipalModel from spiffworkflow_backend.models.principal import PrincipalModel
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.user_service import UserService from spiffworkflow_backend.services.user_service import UserService

View File

@ -18,7 +18,6 @@ from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
from spiffworkflow_backend.models.process_instance_event import ProcessInstanceEventType from spiffworkflow_backend.models.process_instance_event import ProcessInstanceEventType
from spiffworkflow_backend.models.task import TaskModel # noqa: F401 from spiffworkflow_backend.models.task import TaskModel # noqa: F401
from spiffworkflow_backend.models.task_definition import TaskDefinitionModel from spiffworkflow_backend.models.task_definition import TaskDefinitionModel
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.authorization_service import ( from spiffworkflow_backend.services.authorization_service import (
UserDoesNotHaveAccessToTaskError, UserDoesNotHaveAccessToTaskError,
@ -440,9 +439,7 @@ class TestProcessInstanceProcessor(BaseTest):
process_model_id="test_group/boundary_event_reset", process_model_id="test_group/boundary_event_reset",
process_model_source_directory="boundary_event_reset", process_model_source_directory="boundary_event_reset",
) )
process_instance = self.create_process_instance_from_process_model( process_instance = self.create_process_instance_from_process_model(process_model=process_model)
process_model=process_model
)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True) processor.do_engine_steps(save=True)
assert len(process_instance.active_human_tasks) == 1 assert len(process_instance.active_human_tasks) == 1
@ -467,7 +464,9 @@ class TestProcessInstanceProcessor(BaseTest):
human_task_one = process_instance.active_human_tasks[0] human_task_one = process_instance.active_human_tasks[0]
assert human_task_one.task_title == "Manual Task #1" assert human_task_one.task_title == "Manual Task #1"
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
processor.manual_complete_task(str(human_task_one.task_id), execute=True, user=process_instance.process_initiator) processor.manual_complete_task(
str(human_task_one.task_id), execute=True, user=process_instance.process_initiator
)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
processor.resume() processor.resume()
processor.do_engine_steps(save=True) processor.do_engine_steps(save=True)
@ -490,15 +489,15 @@ class TestProcessInstanceProcessor(BaseTest):
process_model_id="test_group/step_through_gateway", process_model_id="test_group/step_through_gateway",
process_model_source_directory="step_through_gateway", process_model_source_directory="step_through_gateway",
) )
process_instance = self.create_process_instance_from_process_model( process_instance = self.create_process_instance_from_process_model(process_model=process_model)
process_model=process_model
)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True) processor.do_engine_steps(save=True)
assert len(process_instance.active_human_tasks) == 1 assert len(process_instance.active_human_tasks) == 1
human_task_one = process_instance.active_human_tasks[0] human_task_one = process_instance.active_human_tasks[0]
processor.bpmn_process_instance.get_task_from_id(UUID(human_task_one.task_id)) processor.bpmn_process_instance.get_task_from_id(UUID(human_task_one.task_id))
processor.manual_complete_task(str(human_task_one.task_id), execute=True, user=process_instance.process_initiator) processor.manual_complete_task(
str(human_task_one.task_id), execute=True, user=process_instance.process_initiator
)
processor.save() processor.save()
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
step1_task = processor.get_task_by_bpmn_identifier("step_1", processor.bpmn_process_instance) step1_task = processor.get_task_by_bpmn_identifier("step_1", processor.bpmn_process_instance)
@ -866,9 +865,7 @@ class TestProcessInstanceProcessor(BaseTest):
bpmn_file_name="script_error_with_task_data.bpmn", bpmn_file_name="script_error_with_task_data.bpmn",
process_model_source_directory="error", process_model_source_directory="error",
) )
process_instance = self.create_process_instance_from_process_model( process_instance = self.create_process_instance_from_process_model(process_model=process_model)
process_model=process_model
)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(WorkflowExecutionServiceError): with pytest.raises(WorkflowExecutionServiceError):

View File

@ -9,7 +9,6 @@ from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.process_instance_file_data import ( from spiffworkflow_backend.models.process_instance_file_data import (
ProcessInstanceFileDataModel, ProcessInstanceFileDataModel,
) )
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_service import ( from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService, ProcessInstanceService,
) )

View File

@ -1,11 +1,9 @@
"""Test_various_bpmn_constructs."""
import pytest import pytest
from flask.app import Flask from flask.app import Flask
from flask.testing import FlaskClient from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )
@ -24,8 +22,6 @@ class TestRestrictedScriptEngine(BaseTest):
bpmn_file_name="read_etc_passwd.bpmn", bpmn_file_name="read_etc_passwd.bpmn",
process_model_source_directory="dangerous-scripts", process_model_source_directory="dangerous-scripts",
) )
self.find_or_create_user()
process_instance = self.create_process_instance_from_process_model(process_model) process_instance = self.create_process_instance_from_process_model(process_model)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)
@ -44,8 +40,6 @@ class TestRestrictedScriptEngine(BaseTest):
bpmn_file_name="read_env.bpmn", bpmn_file_name="read_env.bpmn",
process_model_source_directory="dangerous-scripts", process_model_source_directory="dangerous-scripts",
) )
self.find_or_create_user()
process_instance = self.create_process_instance_from_process_model(process_model) process_instance = self.create_process_instance_from_process_model(process_model)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)

View File

@ -3,7 +3,6 @@ from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )

View File

@ -11,7 +11,6 @@ from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.db import db from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.spec_reference import SpecReferenceCache from spiffworkflow_backend.models.spec_reference import SpecReferenceCache
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.spec_file_service import ( from spiffworkflow_backend.services.spec_file_service import (
ProcessModelFileInvalidError, ProcessModelFileInvalidError,
@ -39,7 +38,7 @@ class TestSpecFileService(BaseTest):
load_test_spec( load_test_spec(
process_model_id=self.process_model_id, process_model_id=self.process_model_id,
bpmn_file_name=self.bpmn_file_name, bpmn_file_name=self.bpmn_file_name,
process_model_source_directory='call_activity_nested', process_model_source_directory="call_activity_nested",
) )
bpmn_process_id_lookups = SpecReferenceCache.query.all() bpmn_process_id_lookups = SpecReferenceCache.query.all()
assert len(bpmn_process_id_lookups) == 1 assert len(bpmn_process_id_lookups) == 1
@ -57,7 +56,7 @@ class TestSpecFileService(BaseTest):
load_test_spec( load_test_spec(
process_model_id=self.process_model_id, process_model_id=self.process_model_id,
bpmn_file_name=self.bpmn_file_name, bpmn_file_name=self.bpmn_file_name,
process_model_source_directory='call_activity_nested', process_model_source_directory="call_activity_nested",
) )
bpmn_process_id_lookups = SpecReferenceCache.query.all() bpmn_process_id_lookups = SpecReferenceCache.query.all()
assert len(bpmn_process_id_lookups) == 1 assert len(bpmn_process_id_lookups) == 1
@ -94,7 +93,7 @@ class TestSpecFileService(BaseTest):
load_test_spec( load_test_spec(
process_model_id=self.process_model_id, process_model_id=self.process_model_id,
bpmn_file_name=self.bpmn_file_name, bpmn_file_name=self.bpmn_file_name,
process_model_source_directory='call_activity_nested', process_model_source_directory="call_activity_nested",
) )
bpmn_process_id_lookups = SpecReferenceCache.query.all() bpmn_process_id_lookups = SpecReferenceCache.query.all()
@ -144,7 +143,7 @@ class TestSpecFileService(BaseTest):
load_test_spec( load_test_spec(
process_model_id=self.process_model_id, process_model_id=self.process_model_id,
bpmn_file_name=self.bpmn_file_name, bpmn_file_name=self.bpmn_file_name,
process_model_source_directory='call_activity_nested', process_model_source_directory="call_activity_nested",
) )
bpmn_process_id_lookups = SpecReferenceCache.query.all() bpmn_process_id_lookups = SpecReferenceCache.query.all()
@ -173,7 +172,7 @@ class TestSpecFileService(BaseTest):
process_model_id = "test_group/call_activity_nested" process_model_id = "test_group/call_activity_nested"
process_model = load_test_spec( process_model = load_test_spec(
process_model_id=process_model_id, process_model_id=process_model_id,
process_model_source_directory='call_activity_nested', process_model_source_directory="call_activity_nested",
) )
files = SpecFileService.get_files(process_model) files = SpecFileService.get_files(process_model)

View File

@ -1,14 +1,12 @@
"""Test_various_bpmn_constructs.""" """Test_various_bpmn_constructs."""
from flask.app import Flask from flask.app import Flask
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.testing import FlaskClient from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, ProcessInstanceProcessor,
) )
from spiffworkflow_backend.services.process_model_service import ProcessModelService
class TestVariousBpmnConstructs(BaseTest): class TestVariousBpmnConstructs(BaseTest):
@ -20,7 +18,7 @@ class TestVariousBpmnConstructs(BaseTest):
) -> None: ) -> None:
process_model = load_test_spec( process_model = load_test_spec(
process_model_id="test_group/timer_intermediate_catch_event", process_model_id="test_group/timer_intermediate_catch_event",
process_model_source_directory='timer_intermediate_catch_event', process_model_source_directory="timer_intermediate_catch_event",
) )
process_instance = self.create_process_instance_from_process_model(process_model) process_instance = self.create_process_instance_from_process_model(process_model)
processor = ProcessInstanceProcessor(process_instance) processor = ProcessInstanceProcessor(process_instance)