sartography
/
spiffworkflow-backend
mirror of https://github.com/sartography/spiffworkflow-backend.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

pyl w/ burnettk

This commit is contained in:
jasquat 2022-12-21 11:42:45 -05:00
parent dc2a976b9c
commit 7756165b08
5 changed files with 30 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/spiffworkflow_backend/routes/user.py
View File

@ -105,7 +105,9 @@ def verify_token(
) from e
if (
user_info is not None and "error" not in user_info and 'iss' in user_info
user_info is not None
and "error" not in user_info
and "iss" in user_info
): # not sure what to test yet
user_model = (
UserModel.query.filter(UserModel.service == user_info["iss"])

15
src/spiffworkflow_backend/scripts/script.py
View File

@ -1,7 +1,5 @@
"""Script."""
from __future__ import annotations
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel, ProcessInstanceNotFoundError
from spiffworkflow_backend.services.authorization_service import AuthorizationService
import importlib
import os
@ -12,9 +10,12 @@ from typing import Callable
from flask_bpmn.api.api_error import ApiError
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceNotFoundError
from spiffworkflow_backend.models.script_attributes_context import (
ScriptAttributesContext,
)
from spiffworkflow_backend.services.authorization_service import AuthorizationService
# Generally speaking, having some global in a flask app is TERRIBLE.
# This is here, because after loading the application this will never change under
@ -23,7 +24,7 @@ SCRIPT_SUB_CLASSES = None
class ScriptUnauthorizedForUserError(Exception):
pass
"""ScriptUnauthorizedForUserError."""
class Script:
@ -84,10 +85,13 @@ class Script:
instance = subclass()
def check_script_permission() -> None:
"""Check_script_permission."""
if subclass.requires_privileged_permissions():
script_function_name = get_script_function_name(subclass)
uri = f"/v1.0/can-run-privileged-script/{script_function_name}"
process_instance = ProcessInstanceModel.query.filter_by(id=script_attributes_context.process_instance_id).first()
process_instance = ProcessInstanceModel.query.filter_by(
id=script_attributes_context.process_instance_id
).first()
if process_instance is None:
raise ProcessInstanceNotFoundError(
f"Could not find a process instance with id '{script_attributes_context.process_instance_id}' "
@ -103,6 +107,7 @@ class Script:
)
def run_script_if_allowed(*ar: Any, **kw: Any) -> Any:
"""Run_script_if_allowed."""
check_script_permission()
return subclass.run(
instance,
@ -110,9 +115,11 @@ class Script:
*ar,
**kw,
)
return run_script_if_allowed
def get_script_function_name(subclass: type[Script]) -> str:
"""Get_script_function_name."""
return subclass.__module__.split(".")[-1]
execlist = {}

11
src/spiffworkflow_backend/services/process_instance_report_service.py
View File

@ -414,13 +414,16 @@ class ProcessInstanceReportService:
)
if report_filter.with_tasks_assigned_to_my_group is True:
group_model_join_conditions = [GroupModel.id == HumanTaskModel.lane_assignment_id]
group_model_join_conditions = [
GroupModel.id == HumanTaskModel.lane_assignment_id
]
if report_filter.user_group_identifier:
group_model_join_conditions.append(GroupModel.identifier == report_filter.user_group_identifier)
group_model_join_conditions.append(
GroupModel.identifier == report_filter.user_group_identifier
)
process_instance_query = process_instance_query.join(HumanTaskModel)
process_instance_query = process_instance_query.join(
GroupModel,
and_(*group_model_join_conditions)
GroupModel, and_(*group_model_join_conditions)
)
process_instance_query = process_instance_query.join(
UserGroupAssignmentModel,

14
tests/spiffworkflow_backend/scripts/test_add_permission.py
View File

@ -1,12 +1,10 @@
"""Test_get_localtime."""
from flask.app import Flask
from flask_bpmn.api.api_error import ApiError
import pytest
from spiffworkflow_backend.scripts.script import ScriptUnauthorizedForUserError
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.app import Flask
from flask.testing import FlaskClient
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from flask_bpmn.api.api_error import ApiError
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.permission_assignment import PermissionAssignmentModel
@ -16,6 +14,9 @@ from spiffworkflow_backend.models.script_attributes_context import (
)
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.scripts.add_permission import AddPermission
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
class TestAddPermission(BaseTest):
@ -70,11 +71,12 @@ class TestAddPermission(BaseTest):
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_add_permission_script_through_bpmn."""
basic_user = self.find_or_create_user("basic_user")
privileged_user = self.find_or_create_user("privileged_user")
self.add_permissions_to_user(
privileged_user,
target_uri="/v1.0/can-run-privileged-script/*",
target_uri="/v1.0/can-run-privileged-script/add_permission",
permission_names=["create"],
)
process_model = load_test_spec(

3
tests/spiffworkflow_backend/scripts/test_save_process_instance_metadata.py
View File

@ -24,7 +24,6 @@ class TestSaveProcessInstanceMetadata(BaseTest):
with_super_admin_user: UserModel,
) -> None:
"""Test_can_save_process_instance_metadata."""
initiator_user = self.find_or_create_user("initiator_user")
self.create_process_group(
client, with_super_admin_user, "test_group", "test_group"
)
@ -34,7 +33,7 @@ class TestSaveProcessInstanceMetadata(BaseTest):
process_model_source_directory="save_process_instance_metadata",
)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=initiator_user
process_model=process_model, user=with_super_admin_user
)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True)