remaining tests are now passing w/ burnettk

This commit is contained in:
jasquat 2022-10-13 14:59:26 -04:00
parent 50529d04c3
commit 06a7c6485c
2 changed files with 217 additions and 111 deletions

View File

@ -1,15 +1,20 @@
"""Test_logging_service."""
from flask.app import Flask
from flask.testing import FlaskClient
from spiffworkflow_backend.models.user import UserModel
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.user import UserModel
class TestLoggingService(BaseTest):
"""Test logging service."""
def test_logging_service_spiff_logger(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None, with_super_admin_user: UserModel,
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_process_instance_run."""
process_group_id = "test_logging_spiff_logger"

View File

@ -50,6 +50,7 @@ class SecretServiceTestHelpers(BaseTest):
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
@ -77,134 +78,172 @@ class SecretServiceTestHelpers(BaseTest):
class TestSecretService(SecretServiceTestHelpers):
"""TestSecretService."""
def test_add_secret(self, app: Flask, with_db_and_bpmn_file_cleanup: None) -> None:
def test_add_secret(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_add_secret."""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
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.creator_user_id == user.id
assert test_secret.creator_user_id == with_super_admin_user.id
def test_add_secret_duplicate_key_fails(
self, app: Flask, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_add_secret_duplicate_key_fails."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
self.add_test_secret(user)
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) -> None:
def test_get_secret(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_get_secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
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
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_get_secret_bad_service."""
user = self.find_or_create_user()
self.add_test_secret(user)
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
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test update secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
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", user.id)
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_user_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_update_secret_bad_user."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService.update_secret(
self.test_key, "new_secret_value", user.id + 1
self.test_key, "new_secret_value", with_super_admin_user.id + 1
) # noqa: S105
assert (
ae.value.message
== f"User: {user.id+1} cannot update the secret with key : test_key"
== f"User: {with_super_admin_user.id+1} cannot update the secret with key : test_key"
)
def test_update_secret_bad_secret_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_update_secret_bad_secret_fails."""
user = self.find_or_create_user()
secret = self.add_test_secret(user)
secret = self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService.update_secret(secret.key + "x", "some_new_value", user.id)
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
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test delete secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
secrets = SecretModel.query.all()
assert len(secrets) == 1
assert secrets[0].creator_user_id == user.id
SecretService.delete_secret(self.test_key, user.id)
assert secrets[0].creator_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_user_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_delete_secret_bad_user."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService.delete_secret(self.test_key, user.id + 1)
SecretService.delete_secret(self.test_key, with_super_admin_user.id + 1)
assert (
f"User: {user.id+1} cannot delete the secret with key" in ae.value.message
f"User: {with_super_admin_user.id+1} cannot delete the secret with key"
in ae.value.message
)
def test_delete_secret_bad_secret_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_delete_secret_bad_secret_fails."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService.delete_secret(self.test_key + "x", user.id)
SecretService.delete_secret(self.test_key + "x", with_super_admin_user.id)
assert "Resource does not exist" in ae.value.message
def test_secret_add_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_add_allowed_process."""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
process_model_info = self.add_test_process(client, user)
test_secret = self.add_test_secret(with_super_admin_user)
process_model_info = self.add_test_process(client, with_super_admin_user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
allowed_process_model = SecretService().add_allowed_process(
secret_id=test_secret.id,
user_id=user.id,
user_id=with_super_admin_user.id,
allowed_relative_path=process_model_relative_path,
)
@ -219,23 +258,26 @@ class TestSecretService(SecretServiceTestHelpers):
assert test_secret.allowed_processes[0] == allowed_process_model
def test_secret_add_allowed_process_same_process_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Do not allow duplicate entries for secret_id/allowed_relative_path pairs.
We actually take care of this in the db model with a unique constraint
on the 2 columns.
"""
user = self.find_or_create_user()
test_secret = self.add_test_secret(user)
process_model_info = self.add_test_process(client, user)
test_secret = self.add_test_secret(with_super_admin_user)
process_model_info = self.add_test_process(client, with_super_admin_user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
SecretService().add_allowed_process(
secret_id=test_secret.id,
user_id=user.id,
user_id=with_super_admin_user.id,
allowed_relative_path=process_model_relative_path,
)
allowed_processes = SecretAllowedProcessPathModel.query.all()
@ -244,91 +286,114 @@ class TestSecretService(SecretServiceTestHelpers):
with pytest.raises(ApiError) as ae:
SecretService().add_allowed_process(
secret_id=test_secret.id,
user_id=user.id,
user_id=with_super_admin_user.id,
allowed_relative_path=process_model_relative_path,
)
assert "Resource already exists" in ae.value.message
def test_secret_add_allowed_process_bad_user_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_add_allowed_process_bad_user."""
user = self.find_or_create_user()
process_model_info = self.add_test_process(client, user)
process_model_info = self.add_test_process(client, with_super_admin_user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
test_secret = self.add_test_secret(user)
test_secret = self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService().add_allowed_process(
secret_id=test_secret.id,
user_id=user.id + 1,
user_id=with_super_admin_user.id + 1,
allowed_relative_path=process_model_relative_path,
)
assert (
ae.value.message
== f"User: {user.id+1} cannot modify the secret with key : {self.test_key}"
== f"User: {with_super_admin_user.id+1} cannot modify the secret with key : {self.test_key}"
)
def test_secret_add_allowed_process_bad_secret_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_add_allowed_process_bad_secret_fails."""
user = self.find_or_create_user()
process_model_info = self.add_test_process(client, user)
process_model_info = self.add_test_process(client, with_super_admin_user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
test_secret = self.add_test_secret(user)
test_secret = self.add_test_secret(with_super_admin_user)
with pytest.raises(ApiError) as ae:
SecretService().add_allowed_process(
secret_id=test_secret.id + 1,
user_id=user.id,
user_id=with_super_admin_user.id,
allowed_relative_path=process_model_relative_path,
)
assert "Resource does not exist" in ae.value.message
def test_secret_delete_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_delete_allowed_process."""
user = self.find_or_create_user()
allowed_process_model = self.add_test_secret_allowed_process(client, user)
allowed_process_model = self.add_test_secret_allowed_process(
client, with_super_admin_user
)
allowed_processes = SecretAllowedProcessPathModel.query.all()
assert len(allowed_processes) == 1
SecretService().delete_allowed_process(allowed_process_model.id, user.id)
SecretService().delete_allowed_process(
allowed_process_model.id, with_super_admin_user.id
)
allowed_processes = SecretAllowedProcessPathModel.query.all()
assert len(allowed_processes) == 0
def test_secret_delete_allowed_process_bad_user_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_delete_allowed_process_bad_user_fails."""
user = self.find_or_create_user()
allowed_process_model = self.add_test_secret_allowed_process(client, user)
allowed_process_model = self.add_test_secret_allowed_process(
client, with_super_admin_user
)
with pytest.raises(ApiError) as ae:
SecretService().delete_allowed_process(
allowed_process_model.id, user.id + 1
allowed_process_model.id, with_super_admin_user.id + 1
)
message = ae.value.message
assert (
f"User: {user.id+1} cannot delete the allowed_process with id : {allowed_process_model.id}"
f"User: {with_super_admin_user.id+1} cannot delete the allowed_process with id : {allowed_process_model.id}"
in message
)
def test_secret_delete_allowed_process_bad_allowed_process_fails(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_secret_delete_allowed_process_bad_allowed_process_fails."""
user = self.find_or_create_user()
allowed_process_model = self.add_test_secret_allowed_process(client, user)
allowed_process_model = self.add_test_secret_allowed_process(
client, with_super_admin_user
)
with pytest.raises(ApiError) as ae:
SecretService().delete_allowed_process(
allowed_process_model.id + 1, user.id
allowed_process_model.id + 1, with_super_admin_user.id
)
assert "Resource does not exist" in ae.value.message
@ -337,19 +402,22 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
"""TestSecretServiceApi."""
def test_add_secret(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_add_secret."""
user = self.find_or_create_user()
secret_model = SecretModel(
key=self.test_key,
value=self.test_value,
creator_user_id=user.id,
creator_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(user),
headers=self.logged_in_headers(with_super_admin_user),
content_type="application/json",
data=data,
)
@ -359,17 +427,20 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert key in secret.keys()
assert secret["key"] == self.test_key
assert secret["value"] == self.test_value
assert secret["creator_user_id"] == user.id
assert secret["creator_user_id"] == with_super_admin_user.id
def test_get_secret(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test get secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
self.add_test_secret(with_super_admin_user)
secret_response = client.get(
f"/v1.0/secrets/{self.test_key}",
headers=self.logged_in_headers(user),
headers=self.logged_in_headers(with_super_admin_user),
)
assert secret_response
assert secret_response.status_code == 200
@ -377,20 +448,25 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert secret_response.json["value"] == self.test_value
def test_update_secret(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_update_secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
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", creator_user_id=user.id
key=self.test_key,
value="new_secret_value",
creator_user_id=with_super_admin_user.id,
)
response = client.put(
f"/v1.0/secrets/{self.test_key}",
headers=self.logged_in_headers(user),
headers=self.logged_in_headers(with_super_admin_user),
content_type="application/json",
data=json.dumps(SecretModelSchema().dump(secret_model)),
)
@ -402,53 +478,75 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert secret_model.value == "new_secret_value"
def test_delete_secret(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test delete secret."""
user = self.find_or_create_user()
self.add_test_secret(user)
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(user),
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_user(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test_delete_secret_bad_user."""
user_1 = self.find_or_create_user()
user_2 = self.find_or_create_user("test_user_2")
self.add_test_secret(user_1)
# ensure user has permissions to delete the given secret
self.add_permissions_to_user(
user_2,
target_uri=f"/v1.0/secrets/{self.test_key}",
permission_names=["delete"],
)
secret_response = client.delete(
f"/v1.0/secrets/{self.test_key}",
headers=self.logged_in_headers(user_2),
)
assert secret_response.status_code == 401
assert secret_response.json
assert secret_response.json["error_code"] == "delete_secret_error"
def test_delete_secret_bad_key(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""Test delete secret."""
user = self.find_or_create_user()
secret_response = client.delete(
"/v1.0/secrets/bad_secret_key",
headers=self.logged_in_headers(user),
headers=self.logged_in_headers(with_super_admin_user),
)
assert secret_response.status_code == 404
def test_add_secret_allowed_process(
self, app: Flask, client: FlaskClient, with_db_and_bpmn_file_cleanup: None
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> 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)
test_secret = self.add_test_secret(with_super_admin_user)
process_model_info = self.add_test_process(client, with_super_admin_user)
process_model_relative_path = FileSystemService.process_model_relative_path(
process_model_info
)
@ -458,7 +556,7 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
}
response: TestResponse = client.post(
"/v1.0/secrets/allowed_process_paths",
headers=self.logged_in_headers(user),
headers=self.logged_in_headers(with_super_admin_user),
content_type="application/json",
data=json.dumps(data),
)
@ -469,17 +567,20 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
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
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> 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)
test_secret = self.add_test_secret(with_super_admin_user)
process_model_info = self.add_test_process(client, with_super_admin_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
test_secret.id, with_super_admin_user.id, process_model_relative_path
)
allowed_processes = SecretAllowedProcessPathModel.query.all()
assert len(allowed_processes) == 1
@ -487,7 +588,7 @@ class TestSecretServiceApi(SecretServiceTestHelpers):
assert allowed_processes[0].allowed_relative_path == process_model_relative_path
response = client.delete(
f"/v1.0/secrets/allowed_process_paths/{allowed_process.id}",
headers=self.logged_in_headers(user),
headers=self.logged_in_headers(with_super_admin_user),
)
assert response.status_code == 200
allowed_processes = SecretAllowedProcessPathModel.query.all()