mirror of
https://github.com/status-im/spiff-arena.git
synced 2025-02-06 23:24:08 +00:00
1f25521506
1092ed11 Merge branch 'main' of github.com:sartography/spiffworkflow-backend into main 97b96fa9 Merge commit '999e0f4d2b7a3c3665feb806fd4f00dc50f2de8e' 4d2ebbe2 Refresh token (#6) 2c47d862 Revert "Assure that the Active Task Users table is cleared out before deleting the Active Task Record." 6c1660a3 Merge commit 'eff49e1ddb2e70d62e45866b429a1350443107d0' into main f7949ada Merge commit 'ac929cbae7717affed5fa357a56cf1a0256289d5' into main b70804ea Merge commit '65283df3cbfe95c6cd0d69e1cc0204d91c5d535b' 5b9ad7a7 Merge commit 'a59108db95274bef304cbd9246537206d0b449cf' 038c8e68 Merge commit '835160e5a82c7e0518fa25ed7fa1168c9c435739' aae7966b Merge commit 'a1cce807be1c4ac5f04c42d780df06c216a0de9c' 8527c6be Merge commit '7026fe1f779051f6eb99b872bcd45b14c0cbe88f' 1aa9a8da Merge commit '70480e9f91a1053531ea5db32a13a791c39fd89f' 40383711 Merge commit '9a1e33696a744527a18dbec667140edb4038ac94' 2c7b2e28 Merge commit 'd643de93e5ceaae28a6e65044cc85dd039530d2c' e5e70161 Merge commit '2b702661f3bd9b79de887e82e5a5925d07341eb6' b0b92474 updateing poetry lock file. e5fd3df1 Merge commit 'f0b608789b6cdc3ef4303efac053746c98571a48' dfe4e3e1 Merge commit 'c5a2f8b16c9a614b944a6e4610594eba1dac7b9f' into main 78dcaec3 pointless change git-subtree-dir: spiffworkflow-backend git-subtree-split: 1092ed11187d1b825fa1c5e18f0d96592f0b98e7
290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""Test_secret_service."""
|
|
import json
|
|
from typing import Optional
|
|
|
|
import pytest
|
|
from flask.app import Flask
|
|
from flask.testing import FlaskClient
|
|
from flask_bpmn.api.api_error import ApiError
|
|
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
|
|
from werkzeug.test import TestResponse # type: ignore
|
|
|
|
from spiffworkflow_backend.models.process_model import ProcessModelInfo
|
|
from spiffworkflow_backend.models.secret_model import SecretModel
|
|
from spiffworkflow_backend.models.secret_model import SecretModelSchema
|
|
from spiffworkflow_backend.models.user import UserModel
|
|
from spiffworkflow_backend.services.process_model_service import ProcessModelService
|
|
from spiffworkflow_backend.services.secret_service import SecretService
|
|
|
|
|
|
class SecretServiceTestHelpers(BaseTest):
|
|
"""SecretServiceTestHelpers."""
|
|
|
|
test_key = "test_key"
|
|
test_value = "test_value"
|
|
test_process_group_id = "test"
|
|
test_process_group_display_name = "My Test Process Group"
|
|
test_process_model_id = "make_cookies"
|
|
test_process_model_display_name = "Cooooookies"
|
|
test_process_model_description = "Om nom nom delicious cookies"
|
|
|
|
def add_test_secret(self, user: UserModel) -> SecretModel:
|
|
"""Add_test_secret."""
|
|
return SecretService().add_secret(self.test_key, self.test_value, user.id)
|
|
|
|
def add_test_process(
|
|
self, client: FlaskClient, user: UserModel
|
|
) -> ProcessModelInfo:
|
|
"""Add_test_process."""
|
|
self.create_process_group(
|
|
client,
|
|
user,
|
|
self.test_process_group_id,
|
|
display_name=self.test_process_group_display_name,
|
|
)
|
|
self.create_process_model_with_api(
|
|
client,
|
|
process_group_id=self.test_process_group_id,
|
|
process_model_id=self.test_process_model_id,
|
|
process_model_display_name=self.test_process_model_display_name,
|
|
process_model_description=self.test_process_model_description,
|
|
user=user,
|
|
)
|
|
process_model_info = ProcessModelService().get_process_model(
|
|
self.test_process_model_id, self.test_process_group_id
|
|
)
|
|
return process_model_info
|
|
|
|
|
|
class TestSecretService(SecretServiceTestHelpers):
|
|
"""TestSecretService."""
|
|
|
|
def test_add_secret(
|
|
self,
|
|
app: Flask,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_add_secret."""
|
|
test_secret = self.add_test_secret(with_super_admin_user)
|
|
|
|
assert test_secret is not None
|
|
assert test_secret.key == self.test_key
|
|
assert test_secret.value == self.test_value
|
|
assert test_secret.user_id == with_super_admin_user.id
|
|
|
|
def test_add_secret_duplicate_key_fails(
|
|
self,
|
|
app: Flask,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_add_secret_duplicate_key_fails."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
with pytest.raises(ApiError) as ae:
|
|
self.add_test_secret(with_super_admin_user)
|
|
assert ae.value.error_code == "create_secret_error"
|
|
|
|
def test_get_secret(
|
|
self,
|
|
app: Flask,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_get_secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
|
|
secret = SecretService().get_secret(self.test_key)
|
|
assert secret is not None
|
|
assert secret.value == self.test_value
|
|
|
|
def test_get_secret_bad_key_fails(
|
|
self,
|
|
app: Flask,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_get_secret_bad_service."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
|
|
with pytest.raises(ApiError):
|
|
SecretService().get_secret("bad_key")
|
|
|
|
def test_update_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test update secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
secret = SecretService.get_secret(self.test_key)
|
|
assert secret
|
|
assert secret.value == self.test_value
|
|
SecretService.update_secret(
|
|
self.test_key, "new_secret_value", with_super_admin_user.id
|
|
)
|
|
new_secret = SecretService.get_secret(self.test_key)
|
|
assert new_secret
|
|
assert new_secret.value == "new_secret_value" # noqa: S105
|
|
|
|
def test_update_secret_bad_secret_fails(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_update_secret_bad_secret_fails."""
|
|
secret = self.add_test_secret(with_super_admin_user)
|
|
with pytest.raises(ApiError) as ae:
|
|
SecretService.update_secret(
|
|
secret.key + "x", "some_new_value", with_super_admin_user.id
|
|
)
|
|
assert "Resource does not exist" in ae.value.message
|
|
assert ae.value.error_code == "update_secret_error"
|
|
|
|
def test_delete_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test delete secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
secrets = SecretModel.query.all()
|
|
assert len(secrets) == 1
|
|
assert secrets[0].user_id == with_super_admin_user.id
|
|
SecretService.delete_secret(self.test_key, with_super_admin_user.id)
|
|
secrets = SecretModel.query.all()
|
|
assert len(secrets) == 0
|
|
|
|
def test_delete_secret_bad_secret_fails(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_delete_secret_bad_secret_fails."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
with pytest.raises(ApiError) as ae:
|
|
SecretService.delete_secret(self.test_key + "x", with_super_admin_user.id)
|
|
assert "Resource does not exist" in ae.value.message
|
|
|
|
|
|
class TestSecretServiceApi(SecretServiceTestHelpers):
|
|
"""TestSecretServiceApi."""
|
|
|
|
def test_add_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_add_secret."""
|
|
secret_model = SecretModel(
|
|
key=self.test_key,
|
|
value=self.test_value,
|
|
user_id=with_super_admin_user.id,
|
|
)
|
|
data = json.dumps(SecretModelSchema().dump(secret_model))
|
|
response: TestResponse = client.post(
|
|
"/v1.0/secrets",
|
|
headers=self.logged_in_headers(with_super_admin_user),
|
|
content_type="application/json",
|
|
data=data,
|
|
)
|
|
assert response.json
|
|
secret: dict = response.json
|
|
for key in ["key", "value", "user_id"]:
|
|
assert key in secret.keys()
|
|
assert secret["key"] == self.test_key
|
|
assert secret["value"] == self.test_value
|
|
assert secret["user_id"] == with_super_admin_user.id
|
|
|
|
def test_get_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test get secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
secret_response = client.get(
|
|
f"/v1.0/secrets/{self.test_key}",
|
|
headers=self.logged_in_headers(with_super_admin_user),
|
|
)
|
|
assert secret_response
|
|
assert secret_response.status_code == 200
|
|
assert secret_response.json
|
|
assert secret_response.json["value"] == self.test_value
|
|
|
|
def test_update_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test_update_secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
secret: Optional[SecretModel] = SecretService.get_secret(self.test_key)
|
|
assert secret
|
|
assert secret.value == self.test_value
|
|
secret_model = SecretModel(
|
|
key=self.test_key,
|
|
value="new_secret_value",
|
|
user_id=with_super_admin_user.id,
|
|
)
|
|
response = client.put(
|
|
f"/v1.0/secrets/{self.test_key}",
|
|
headers=self.logged_in_headers(with_super_admin_user),
|
|
content_type="application/json",
|
|
data=json.dumps(SecretModelSchema().dump(secret_model)),
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
secret_model = SecretModel.query.filter(
|
|
SecretModel.key == self.test_key
|
|
).first()
|
|
assert secret_model.value == "new_secret_value"
|
|
|
|
def test_delete_secret(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test delete secret."""
|
|
self.add_test_secret(with_super_admin_user)
|
|
secret = SecretService.get_secret(self.test_key)
|
|
assert secret
|
|
assert secret.value == self.test_value
|
|
secret_response = client.delete(
|
|
f"/v1.0/secrets/{self.test_key}",
|
|
headers=self.logged_in_headers(with_super_admin_user),
|
|
)
|
|
assert secret_response.status_code == 200
|
|
with pytest.raises(ApiError):
|
|
secret = SecretService.get_secret(self.test_key)
|
|
|
|
def test_delete_secret_bad_key(
|
|
self,
|
|
app: Flask,
|
|
client: FlaskClient,
|
|
with_db_and_bpmn_file_cleanup: None,
|
|
with_super_admin_user: UserModel,
|
|
) -> None:
|
|
"""Test delete secret."""
|
|
secret_response = client.delete(
|
|
"/v1.0/secrets/bad_secret_key",
|
|
headers=self.logged_in_headers(with_super_admin_user),
|
|
)
|
|
assert secret_response.status_code == 404
|