sartography/spiffworkflow-backend
1
0
Fork 0
mirror of https://github.com/sartography/spiffworkflow-backend.git synced 2025-02-24 05:18:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pre commit

Browse Source
This commit is contained in:
mike cullerton 2022-09-22 16:40:47 -04:00
parent 8a34e9e89d
commit de512fb373
7 changed files with 89 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
src/spiffworkflow_backend/api.yml
View File

@ -85,47 +85,47 @@ paths:
"200":
description: Logout Authenticated User
# /login_api:
# parameters:
# - name: redirect_url
# in: query
# required: false
# schema:
# type: string
# get:
# security: []
# operationId: spiffworkflow_backend.routes.user.login_api
# summary: Authenticate user for API access
# tags:
# - Authentication
# responses:
# "304":
# description: Redirection to the hosted frontend with an auth_token header.
# /login_api_return:
# parameters:
# - name: code
# in: query
# required: true
# schema:
# type: string
# - name: state
# in: query
# required: true
# schema:
# type: string
# - name: session_state
# in: query
# required: false
# schema:
# type: string
# get:
# security: []
# operationId: spiffworkflow_backend.routes.user.login_api_return
# tags:
# - Authentication
# responses:
# "200":
# description: Test Return Response
# /login_api:
# parameters:
# - name: redirect_url
# in: query
# required: false
# schema:
# type: string
# get:
# security: []
# operationId: spiffworkflow_backend.routes.user.login_api
# summary: Authenticate user for API access
# tags:
# - Authentication
# responses:
# "304":
# description: Redirection to the hosted frontend with an auth_token header.
# /login_api_return:
# parameters:
# - name: code
# in: query
# required: true
# schema:
# type: string
# - name: state
# in: query
# required: true
# schema:
# type: string
# - name: session_state
# in: query
# required: false
# schema:
# type: string
# get:
# security: []
# operationId: spiffworkflow_backend.routes.user.login_api_return
# tags:
# - Authentication
# responses:
# "200":
# description: Test Return Response
/status:
get:

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

@ -49,6 +49,7 @@ class SecretAllowedProcessPathModel(SpiffworkflowBaseDBModel):
class SecretAllowedProcessSchema(Schema):
"""SecretAllowedProcessSchema."""
class Meta:
"""Meta."""

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

@ -1108,7 +1108,7 @@ def add_secret(body: Dict) -> Response:
def update_secret(key: str, body: dict) -> None:
"""Update secret."""
SecretService().update_secret(key, body['value'], body['creator_user_id'])
SecretService().update_secret(key, body["value"], body["creator_user_id"])
def delete_secret(key: str) -> None:
@ -1120,10 +1120,13 @@ def delete_secret(key: str) -> None:
def add_allowed_process_path(body: dict) -> Any:
"""Get allowed process paths."""
allowed_process_path = SecretService.add_allowed_process(
body['secret_id'], g.user.id, body["allowed_relative_path"]
body["secret_id"], g.user.id, body["allowed_relative_path"]
)
return Response(
json.dumps(SecretAllowedProcessSchema().dump(allowed_process_path)),
status=201,
mimetype="application/json",
)
return Response(json.dumps(SecretAllowedProcessSchema().dump(allowed_process_path)),
status=201, mimetype="application/json")
def delete_allowed_process_path(allowed_process_path_id: int) -> Any:

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

@ -184,8 +184,8 @@ def encode_auth_token(sub: str, token_type: Optional[str] = None) -> str:
"""
payload = {"sub": sub}
if token_type is None:
token_type = 'internal'
payload['token_type'] = token_type
token_type = "internal" # noqa: S105
payload["token_type"] = token_type
if "SECRET_KEY" in current_app.config:
secret_key = current_app.config.get("SECRET_KEY")
else:
@ -326,10 +326,12 @@ def get_user_from_decoded_internal_token(decoded_token: dict) -> Optional[UserMo
# user: UserModel = UserModel.query.filter()
if user:
return user
user = UserModel(username=service_id,
uid=service_id,
service=service,
service_id=service_id,
name="API User")
user = UserModel(
username=service_id,
uid=service_id,
service=service,
service_id=service_id,
name="API User",
)
return user

4
src/spiffworkflow_backend/services/logging_service.py
View File

@ -132,7 +132,9 @@ def setup_logger(app: Flask) -> None:
spiff_logger_filehandler = None
if app.config["SPIFFWORKFLOW_BACKEND_LOG_TO_FILE"]:
spiff_logger_filehandler = logging.FileHandler(f"{app.root_path}/../../log/{app.env}.log")
spiff_logger_filehandler = logging.FileHandler(
f"{app.root_path}/../../log/{app.env}.log"
)
spiff_logger_filehandler.setLevel(spiff_log_level)
spiff_logger_filehandler.setFormatter(log_formatter)

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

@ -1,8 +1,6 @@
"""Secret_service."""
import logging
from typing import Optional
from flask import current_app
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from sqlalchemy.exc import IntegrityError
@ -181,7 +179,7 @@ class SecretService:
secret = SecretModel.query.filter(
SecretModel.id == allowed_process.secret_id
).first()
assert secret
assert secret # noqa: S101
if secret.creator_user_id == user_id:
db.session.delete(allowed_process)
try:

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

@ -374,14 +374,17 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert secret_response.status_code == 200
assert secret_response.json == self.test_value
def test_update_secret(self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None) -> None:
def test_update_secret(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
) -> None:
"""Test_update_secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
secret = SecretService.get_secret(self.test_key)
assert secret == self.test_value
secret_model = SecretModel(key=self.test_key,
value="new_secret_value",
creator_user_id=user.id)
secret_model = SecretModel(
key=self.test_key, value="new_secret_value", creator_user_id=user.id
)
response = client.put(
f"/v1.0/secrets/{self.test_key}",
headers=self.logged_in_headers(user),
@ -390,7 +393,9 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
)
assert response.status_code == 204
secret_model = SecretModel.query.filter(SecretModel.key == self.test_key).first()
secret_model = SecretModel.query.filter(
SecretModel.key == self.test_key
).first()
assert secret_model.value == "new_secret_value"
def test_delete_secret(
@ -435,22 +440,25 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert secret_response.status_code == 404
print("test_delete_secret_bad_key")
def test_add_secret_allowed_process(self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None) -> None:
"""Test add secret allowed process"""
def test_add_secret_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
) -> None:
"""Test add secret allowed process."""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
process_model_info = self.add_test_process(client, user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
data = {"secret_id": test_secret.id,
"allowed_relative_path": process_model_relative_path
}
data = {
"secret_id": test_secret.id,
"allowed_relative_path": process_model_relative_path,
}
response: TestResponse = client.post(
"/v1.0/secrets/allowed_process_paths",
headers=self.logged_in_headers(user),
content_type='application/json',
data=json.dumps(data)
content_type="application/json",
data=json.dumps(data),
)
assert response.status_code == 201
allowed_processes = SecretAllowedProcessPathModel.query.all()
@ -458,15 +466,19 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert allowed_processes[0].allowed_relative_path == process_model_relative_path
assert allowed_processes[0].secret_id == test_secret.id
def test_delete_secret_allowed_process(self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None) -> None:
"""Test delete secret allowed process"""
def test_delete_secret_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
) -> None:
"""Test delete secret allowed process."""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
process_model_info = self.add_test_process(client, user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
allowed_process = SecretService.add_allowed_process(test_secret.id, user.id, process_model_relative_path)
allowed_process = SecretService.add_allowed_process(
test_secret.id, user.id, process_model_relative_path
)
allowed_processes = SecretAllowedProcessPathModel.query.all()
assert len(allowed_processes) == 1
assert allowed_processes[0].secret_id == test_secret.id