create process groups json file at root if it does not exist when creating a data store w/ burnettk (#1056)

Co-authored-by: jasquat <jasquat@users.noreply.github.com>
This commit is contained in:
jasquat 2024-02-16 15:20:16 -05:00 committed by GitHub
parent cdfe4aae68
commit 27ea21b81a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 30 additions and 12 deletions

View File

@ -123,22 +123,18 @@ def _data_store_upsert(body: dict, insert: bool) -> flask.wrappers.Response:
model.schema = schema
model.description = description or ""
_write_specification_to_process_group(data_store_type, model)
db.session.add(model)
db.session.commit()
_write_specification_to_process_group(data_store_type, model)
return make_response(jsonify({"ok": True}), 200)
def _write_specification_to_process_group(data_store_type: str, model: Any) -> None:
# TODO: once the top level is a process group this check should be removed
if model.location == "":
return
process_group = ProcessModelService.get_process_group(model.location, False, False)
if not process_group:
return
process_group = ProcessModelService.get_process_group(
model.location, find_direct_nested_items=False, find_all_nested_items=False, create_if_not_exists=True
)
if data_store_type not in process_group.data_store_specifications:
process_group.data_store_specifications[data_store_type] = {}

View File

@ -456,12 +456,16 @@ class ProcessModelService(FileSystemService):
@classmethod
def get_process_group(
cls, process_group_id: str, find_direct_nested_items: bool = True, find_all_nested_items: bool = True
cls,
process_group_id: str,
find_direct_nested_items: bool = True,
find_all_nested_items: bool = True,
create_if_not_exists: bool = False,
) -> ProcessGroup:
"""Look for a given process_group, and return it."""
if os.path.exists(FileSystemService.root_path()):
process_group_path = FileSystemService.full_path_from_id(process_group_id)
if cls.is_process_group(process_group_path):
if cls.is_process_group(process_group_path) or create_if_not_exists:
return cls.find_or_create_process_group(
process_group_path,
find_direct_nested_items=find_direct_nested_items,

View File

@ -11,6 +11,18 @@ from tests.spiffworkflow_backend.helpers.base_test import BaseTest
@pytest.fixture()
def with_loaded_reference_cache(app: Flask, with_db_and_bpmn_file_cleanup: None) -> Generator[None, None, None]:
reference_objects: dict[str, ReferenceCacheModel] = {}
ReferenceCacheService.add_unique_reference_cache_object(
reference_objects,
ReferenceCacheModel.from_params(
"contacts_datastore_root",
"Contacts Datastore Root",
"data_store",
"contacts_datastore.bpmn",
"",
None,
False,
),
)
ReferenceCacheService.add_unique_reference_cache_object(
reference_objects,
ReferenceCacheModel.from_params(
@ -56,3 +68,9 @@ class TestReferenceCacheService(BaseTest):
def test_does_not_find_data_store_in_non_upsearched_location(self, with_loaded_reference_cache: None) -> None:
location = ReferenceCacheService.upsearch("some/other/place", "contacts_datastore", "data_store")
assert location is None
def test_can_find_data_store_in_upsearched_root_location(self, with_loaded_reference_cache: None) -> None:
location = ReferenceCacheService.upsearch(
"misc/jonjon/generic-data-store-area/test-level-2", "contacts_datastore_root", "data_store"
)
assert location == ""

View File

@ -11,7 +11,7 @@ interface OwnProps {
export default function SpiffTooltip({ title, children }: OwnProps) {
return (
<Tooltip title={title} arrow>
<Tooltip title={title} arrow enterDelay={500}>
{children}
</Tooltip>
);