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

pre commit

This commit is contained in:
mike cullerton 2022-09-13 09:08:24 -04:00
parent 516af8a39b
commit b5da2d35c5
6 changed files with 122 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/spiffworkflow_backend/api.yml
View File

@ -974,7 +974,6 @@ paths:
schema:
$ref: "#/components/schemas/ProcessInstanceLog"
/secrets:
post:
operationId: spiffworkflow_backend.routes.process_api_blueprint.add_secret

8
src/spiffworkflow_backend/load_database_models.py
View File

@ -17,9 +17,6 @@ from spiffworkflow_backend.models.message_instance import (
MessageInstanceModel,
) # noqa: F401
from spiffworkflow_backend.models.message_model import MessageModel # noqa: F401
from spiffworkflow_backend.models.secret_model import SecretModel
from spiffworkflow_backend.models.secret_model import SecretAllowedProcessPathModel
from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel
from spiffworkflow_backend.models.message_triggerable_process_model import (
MessageTriggerableProcessModel,
) # noqa: F401
@ -30,6 +27,11 @@ from spiffworkflow_backend.models.process_instance import (
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
) # noqa: F401
from spiffworkflow_backend.models.secret_model import (
SecretAllowedProcessPathModel,
) # noqa: F401
from spiffworkflow_backend.models.secret_model import SecretModel # noqa: F401
from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel # noqa: F401
from spiffworkflow_backend.models.task_event import TaskEventModel # noqa: F401
from spiffworkflow_backend.models.user import UserModel # noqa: F401
from spiffworkflow_backend.models.user_group_assignment import (

15
src/spiffworkflow_backend/models/secret_model.py
View File

@ -1,14 +1,14 @@
"""Secret_model."""
from flask_bpmn.models.db import db
from flask_bpmn.models.db import SpiffworkflowBaseDBModel
from spiffworkflow_backend.models.user import UserModel
from marshmallow import Schema
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.orm import RelationshipProperty
from spiffworkflow_backend.models.user import UserModel
class SecretModel(SpiffworkflowBaseDBModel):
"""SecretModel."""
__tablename__ = "secret"
id: int = db.Column(db.Integer, primary_key=True)
@ -25,12 +25,11 @@ class SecretModel(SpiffworkflowBaseDBModel):
class SecretAllowedProcessPathModel(SpiffworkflowBaseDBModel):
"""Allowed processes can be Process Groups or Process Models.
We store the path in either case.
"""
We store the path in either case.
"""
__tablename__ = "secret_allowed_process"
id: int = db.Column(db.Integer, primary_key=True)
secret_id: int = db.Column(ForeignKey(SecretModel.id), nullable=False)
allowed_relative_path: str = db.Column(db.String(500), nullable=False, index=True)

25
src/spiffworkflow_backend/routes/process_api_blueprint.py
View File

@ -20,7 +20,6 @@ from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from SpiffWorkflow import Task as SpiffTask # type: ignore
from SpiffWorkflow import TaskState
from spiffworkflow_backend.services.secret_service import SecretService
from sqlalchemy import desc
from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
@ -28,7 +27,6 @@ from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
)
from spiffworkflow_backend.models.active_task import ActiveTaskModel
from spiffworkflow_backend.models.file import FileSchema
from spiffworkflow_backend.models.file import FileType
from spiffworkflow_backend.models.message_instance import MessageInstanceModel
from spiffworkflow_backend.models.message_model import MessageModel
from spiffworkflow_backend.models.message_triggerable_process_model import (
@ -55,6 +53,7 @@ from spiffworkflow_backend.services.process_instance_service import (
ProcessInstanceService,
)
from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.secret_service import SecretService
from spiffworkflow_backend.services.spec_file_service import SpecFileService
process_api_blueprint = Blueprint("process_api", __name__)
@ -280,7 +279,6 @@ def process_model_file_delete(
def add_file(process_group_id: str, process_model_id: str) -> flask.wrappers.Response:
"""Add_file."""
process_model_service = ProcessModelService()
process_model = get_process_model(process_model_id, process_group_id)
request_file = get_file_from_request()
if not request_file.filename:
@ -1031,22 +1029,39 @@ def get_spiff_task_from_process_instance(
# Methods for secrets CRUD - maybe move somewhere else:
#
def get_secret(service: str, client: str) -> str:
"""Get_secret."""
secret = SecretService.get_secret(service, client)
if secret:
return secret
def add_secret(service: str, client: str, secret: str, creator_user_id: int = None, allowed_process: str = None):
def add_secret(
service: str,
client: str,
secret: str,
creator_user_id: int = None,
allowed_process: str = None,
):
"""Add secret."""
...
def update_secret(service: str, client: str, secret: str = None, creator_user_id: int = None, allowed_process: str = None):
def update_secret(
service: str,
client: str,
secret: str = None,
creator_user_id: int = None,
allowed_process: str = None,
):
"""Update secret."""
...
def delete_secret(service: str, client: str):
"""Delete secret."""
...
def get_allowed_process_paths(service: str, client: str):
"""Get allowed process paths."""
...

67
src/spiffworkflow_backend/services/secret_service.py
View File

@ -1,46 +1,77 @@
"""Secret_service."""
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from spiffworkflow_backend.models.secret_model import SecretModel, SecretAllowedProcessPathModel
from spiffworkflow_backend.models.secret_model import SecretAllowedProcessPathModel
from spiffworkflow_backend.models.secret_model import SecretModel
class SecretService:
"""SecretService."""
def add_secret(self, service: str, client: str, key: str, creator_user_id: int = None, allowed_process: str = None):
secret_model = SecretModel(service=service, client=client, key=key, creator_user_id=creator_user_id)
@staticmethod
def add_secret(
service: str,
client: str,
key: str,
creator_user_id: int = None,
allowed_process: str = None,
):
"""Add_secret."""
secret_model = SecretModel(
service=service, client=client, key=key, creator_user_id=creator_user_id
)
db.session.add(secret_model)
try:
db.session.commit()
except Exception as e:
raise ApiError(code='create_secret_failed',
message=f"Cannot create secret for service: {service} and client: {client}. Original error is {e}")
raise ApiError(
code="create_secret_failed",
message=f"Cannot create secret for service: {service} and client: {client}. Original error is {e}",
) from e
return secret_model
@staticmethod
def get_secret(service: str, client: str) -> str:
secret = db.session.query(SecretModel.key).\
filter(SecretModel.service == service).\
filter(SecretModel.client == client).\
scalar()
"""Get_secret."""
secret = (
db.session.query(SecretModel.key)
.filter(SecretModel.service == service)
.filter(SecretModel.client == client)
.scalar()
)
if secret:
return secret
def add_allowed_process(self, secret_id: int, allowed_relative_path: str):
secret_process_model = SecretAllowedProcessPathModel(secret_id=secret_id,
allowed_relative_path=allowed_relative_path)
@staticmethod
def add_allowed_process(secret_id: int, allowed_relative_path: str):
"""Add_allowed_process."""
secret_process_model = SecretAllowedProcessPathModel(
secret_id=secret_id, allowed_relative_path=allowed_relative_path
)
db.session.add(secret_process_model)
try:
db.session.commit()
except Exception as e:
raise ApiError(code='create_allowed_process_failure',
message=f"Count not create an allowed process for for secret: {secret_id} "
f"with path: {allowed_relative_path}. "
f"Original error is {e}")
raise ApiError(
code="create_allowed_process_failure",
message=f"Count not create an allowed process for for secret: {secret_id} "
f"with path: {allowed_relative_path}. "
f"Original error is {e}",
) from e
return secret_process_model
def update_secret(self, service: str, client: str, secret: str = None, creator_user_id: int = None, allowed_process: str = None):
def update_secret(
self,
service: str,
client: str,
secret: str = None,
creator_user_id: int = None,
allowed_process: str = None,
):
"""Does this pass pre commit?"""
...
def delete_secret(self, service: str, client: str):
"""Delete secret."""
...

57
tests/spiffworkflow_backend/integration/test_secret_service.py
View File

@ -1,23 +1,29 @@
"""Test_secret_service."""
from flask.app import Flask
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.secret_model import SecretModel, SecretAllowedProcessPathModel
from spiffworkflow_backend.models.secret_model import SecretAllowedProcessPathModel
from spiffworkflow_backend.services.file_system_service import FileSystemService
from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.secret_service import SecretService
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
class TestSecretService(BaseTest):
"""TestSecretService."""
test_service = 'test_service'
test_client = 'test_client'
test_key = '1234567890'
test_service = "test_service"
test_client = "test_client"
test_key = "1234567890"
def add_test_secret(self, user):
return SecretService().add_secret(self.test_service, self.test_client, self.test_key, user.id)
"""Add_test_secret."""
return SecretService().add_secret(
self.test_service, self.test_client, self.test_key, user.id
)
def test_add_secret(self, app: Flask, with_db_and_bpmn_file_cleanup: None):
"""Test_add_secret."""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
@ -28,6 +34,7 @@ class TestSecretService(BaseTest):
assert test_secret.creator_user_id == user.id
def test_get_secret(self, app: Flask, with_db_and_bpmn_file_cleanup: None):
"""Test_get_secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
@ -35,21 +42,30 @@ class TestSecretService(BaseTest):
assert secret is not None
assert secret == self.test_key
def test_get_secret_bad_service(self, app: Flask, with_db_and_bpmn_file_cleanup: None):
def test_get_secret_bad_service(
self, app: Flask, with_db_and_bpmn_file_cleanup: None
):
"""Test_get_secret_bad_service."""
user = self.find_or_create_user()
self.add_test_secret(user)
bad_secret = SecretService().get_secret('bad_service', self.test_client)
bad_secret = SecretService().get_secret("bad_service", self.test_client)
assert bad_secret is None
def test_get_secret_bad_client(self, app: Flask, with_db_and_bpmn_file_cleanup: None):
def test_get_secret_bad_client(
self, app: Flask, with_db_and_bpmn_file_cleanup: None
):
"""Test_get_secret_bad_client."""
user = self.find_or_create_user()
self.add_test_secret(user)
bad_secret = SecretService().get_secret(self.test_service, 'bad_client')
bad_secret = SecretService().get_secret(self.test_service, "bad_client")
assert bad_secret is None
def test_secret_add_allowed_process(self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None):
def test_secret_add_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
):
"""Test_secret_add_allowed_process."""
process_group_id = "test"
process_group_display_name = "My Test Process Group"
@ -69,22 +85,31 @@ class TestSecretService(BaseTest):
process_model_description=process_model_description,
)
process_model_info = ProcessModelService().get_process_model(process_model_id, process_group_id)
process_model_relative_path = FileSystemService.process_model_relative_path(process_model_info)
process_model_info = ProcessModelService().get_process_model(
process_model_id, process_group_id
)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
test_secret = self.add_test_secret(user)
allowed_process_model = SecretService().add_allowed_process(secret_id=test_secret.id,
allowed_relative_path=process_model_relative_path)
allowed_process_model = SecretService().add_allowed_process(
secret_id=test_secret.id, allowed_relative_path=process_model_relative_path
)
assert allowed_process_model is not None
assert isinstance(allowed_process_model, SecretAllowedProcessPathModel)
assert allowed_process_model.secret_id == test_secret.id
assert allowed_process_model.allowed_relative_path == process_model_relative_path
assert (
allowed_process_model.allowed_relative_path == process_model_relative_path
)
assert len(test_secret.allowed_processes) == 1
assert test_secret.allowed_processes[0] == allowed_process_model
def test_update_secret(self):
"""Test update secret."""
...
def test_delete_secret(self):
"""Test delete secret."""
...