asst syntax cleanup

This commit is contained in:
mike cullerton 2022-11-19 15:45:13 -05:00
parent fd93d558e2
commit dc13c1c1ca
3 changed files with 49 additions and 34 deletions

View File

@ -237,8 +237,13 @@ def process_group_show(
def process_group_move( def process_group_move(
modified_process_group_id: str, new_location: str modified_process_group_id: str, new_location: str
) -> flask.wrappers.Response: ) -> flask.wrappers.Response:
original_process_group_id = un_modify_modified_process_model_id(modified_process_group_id) """process_group_move."""
new_process_group = ProcessModelService().process_group_move(original_process_group_id, new_location) original_process_group_id = un_modify_modified_process_model_id(
modified_process_group_id
)
new_process_group = ProcessModelService().process_group_move(
original_process_group_id, new_location
)
return make_response(jsonify(new_process_group), 201) return make_response(jsonify(new_process_group), 201)
@ -331,8 +336,13 @@ def process_model_show(modified_process_model_identifier: str) -> Any:
def process_model_move( def process_model_move(
modified_process_model_identifier: str, new_location: str modified_process_model_identifier: str, new_location: str
) -> flask.wrappers.Response: ) -> flask.wrappers.Response:
original_process_model_id = un_modify_modified_process_model_id(modified_process_model_identifier) """process_model_move."""
new_process_model = ProcessModelService().process_model_move(original_process_model_id, new_location) original_process_model_id = un_modify_modified_process_model_id(
modified_process_model_identifier
)
new_process_model = ProcessModelService().process_model_move(
original_process_model_id, new_location
)
return make_response(jsonify(new_process_model), 201) return make_response(jsonify(new_process_model), 201)

View File

@ -115,13 +115,18 @@ class ProcessModelService(FileSystemService):
path = f"{FileSystemService.root_path()}/{process_model_id}" path = f"{FileSystemService.root_path()}/{process_model_id}"
shutil.rmtree(path) shutil.rmtree(path)
def process_model_move(self, original_process_model_id: str, new_location: str) -> ProcessModelInfo: def process_model_move(
original_model_path = os.path.abspath(os.path.join(FileSystemService.root_path(), original_process_model_id)) self, original_process_model_id: str, new_location: str
) -> ProcessModelInfo:
"""process_model_move."""
original_model_path = os.path.abspath(
os.path.join(FileSystemService.root_path(), original_process_model_id)
)
_, model_id = os.path.split(original_model_path) _, model_id = os.path.split(original_model_path)
new_relative_path = f"{new_location}/{model_id}" new_relative_path = f"{new_location}/{model_id}"
new_model_path = os.path.abspath(os.path.join( new_model_path = os.path.abspath(
FileSystemService.root_path(), new_relative_path os.path.join(FileSystemService.root_path(), new_relative_path)
)) )
shutil.move(original_model_path, new_model_path) shutil.move(original_model_path, new_model_path)
new_process_model = self.get_process_model(new_relative_path) new_process_model = self.get_process_model(new_relative_path)
return new_process_model return new_process_model
@ -242,7 +247,10 @@ class ProcessModelService(FileSystemService):
self.write_json_file(json_path, serialized_process_group) self.write_json_file(json_path, serialized_process_group)
return process_group return process_group
def process_group_move(self, original_process_group_id: str, new_location: str) -> ProcessGroup: def process_group_move(
self, original_process_group_id: str, new_location: str
) -> ProcessGroup:
"""process_group_move."""
original_group_path = self.process_group_path(original_process_group_id) original_group_path = self.process_group_path(original_process_group_id)
original_root, original_group_id = os.path.split(original_group_path) original_root, original_group_id = os.path.split(original_group_path)
new_root = f"{FileSystemService.root_path()}/{new_location}" new_root = f"{FileSystemService.root_path()}/{new_location}"

View File

@ -2371,18 +2371,14 @@ class TestProcessApi(BaseTest):
print("test_script_unit_test_run") print("test_script_unit_test_run")
def setup_initial_groups_for_move_tests( def setup_initial_groups_for_move_tests(
self, client: FlaskClient, with_super_admin_user: UserModel self, client: FlaskClient, with_super_admin_user: UserModel
) -> None: ) -> None:
groups = [ """setup_initial_groups_for_move_tests."""
'group_a', 'group_b', 'group_b/group_bb' groups = ["group_a", "group_b", "group_b/group_bb"]
]
# setup initial groups # setup initial groups
for group in groups: for group in groups:
self.create_process_group( self.create_process_group(
client, client, with_super_admin_user, group, display_name=group
with_super_admin_user,
group,
display_name=group
) )
# make sure initial groups exist # make sure initial groups exist
for group in groups: for group in groups:
@ -2400,7 +2396,7 @@ class TestProcessApi(BaseTest):
"""test_move_model.""" """test_move_model."""
self.setup_initial_groups_for_move_tests(client, with_super_admin_user) self.setup_initial_groups_for_move_tests(client, with_super_admin_user)
process_model_id = 'test_model' process_model_id = "test_model"
original_location = "group_a" original_location = "group_a"
original_process_model_path = f"{original_location}/{process_model_id}" original_process_model_path = f"{original_location}/{process_model_id}"
@ -2410,31 +2406,35 @@ class TestProcessApi(BaseTest):
original_process_model_path, original_process_model_path,
user=with_super_admin_user, user=with_super_admin_user,
process_model_display_name=process_model_id, process_model_display_name=process_model_id,
process_model_description=process_model_id process_model_description=process_model_id,
) )
persisted = ProcessModelService().get_process_model(original_process_model_path) persisted = ProcessModelService().get_process_model(original_process_model_path)
assert persisted is not None assert persisted is not None
assert persisted.id == original_process_model_path assert persisted.id == original_process_model_path
# move model to `group_b/group_bb` # move model to `group_b/group_bb`
new_location = 'group_b/group_bb' new_location = "group_b/group_bb"
new_process_model_path = f"{new_location}/{process_model_id}" new_process_model_path = f"{new_location}/{process_model_id}"
modified_original_process_model_id = original_process_model_path.replace("/", ":") modified_original_process_model_id = original_process_model_path.replace(
"/", ":"
)
response = client.put( response = client.put(
f"/v1.0/process-models/{modified_original_process_model_id}/move?new_location={new_location}", f"/v1.0/process-models/{modified_original_process_model_id}/move?new_location={new_location}",
headers=self.logged_in_headers(with_super_admin_user), headers=self.logged_in_headers(with_super_admin_user),
) )
assert response.status_code == 201 assert response.status_code == 201
assert response.json['id'] == new_process_model_path assert response.json["id"] == new_process_model_path
# make sure the original model does not exist # make sure the original model does not exist
with pytest.raises(ProcessEntityNotFoundError) as e: with pytest.raises(ProcessEntityNotFoundError) as e:
ProcessModelService().get_process_model(original_process_model_path) ProcessModelService().get_process_model(original_process_model_path)
assert e.value.args[0] == 'process_model_not_found' assert e.value.args[0] == "process_model_not_found"
# make sure the new model does exist # make sure the new model does exist
new_process_model = ProcessModelService().get_process_model(new_process_model_path) new_process_model = ProcessModelService().get_process_model(
new_process_model_path
)
assert new_process_model is not None assert new_process_model is not None
assert new_process_model.id == new_process_model_path assert new_process_model.id == new_process_model_path
@ -2449,14 +2449,11 @@ class TestProcessApi(BaseTest):
self.setup_initial_groups_for_move_tests(client, with_super_admin_user) self.setup_initial_groups_for_move_tests(client, with_super_admin_user)
# add sub group to `group_a` # add sub group to `group_a`
sub_group_id = 'sub_group' sub_group_id = "sub_group"
original_location = 'group_a' original_location = "group_a"
original_sub_path = f"{original_location}/{sub_group_id}" original_sub_path = f"{original_location}/{sub_group_id}"
self.create_process_group( self.create_process_group(
client, client, with_super_admin_user, original_sub_path, display_name=sub_group_id
with_super_admin_user,
original_sub_path,
display_name=sub_group_id
) )
# make sure original subgroup exists # make sure original subgroup exists
persisted = ProcessModelService().get_process_group(original_sub_path) persisted = ProcessModelService().get_process_group(original_sub_path)
@ -2464,7 +2461,7 @@ class TestProcessApi(BaseTest):
assert persisted.id == original_sub_path assert persisted.id == original_sub_path
# move sub_group to `group_b/group_bb` # move sub_group to `group_b/group_bb`
new_location = 'group_b/group_bb' new_location = "group_b/group_bb"
new_sub_path = f"{new_location}/{sub_group_id}" new_sub_path = f"{new_location}/{sub_group_id}"
modified_original_process_group_id = original_sub_path.replace("/", ":") modified_original_process_group_id = original_sub_path.replace("/", ":")
response = client.put( response = client.put(
@ -2472,13 +2469,13 @@ class TestProcessApi(BaseTest):
headers=self.logged_in_headers(with_super_admin_user), headers=self.logged_in_headers(with_super_admin_user),
) )
assert response.status_code == 201 assert response.status_code == 201
assert response.json['id'] == new_sub_path assert response.json["id"] == new_sub_path
# make sure the original subgroup does not exist # make sure the original subgroup does not exist
with pytest.raises(ProcessEntityNotFoundError) as e: with pytest.raises(ProcessEntityNotFoundError) as e:
ProcessModelService().get_process_group(original_sub_path) ProcessModelService().get_process_group(original_sub_path)
assert e.value.args[0] == 'process_group_not_found' assert e.value.args[0] == "process_group_not_found"
assert e.value.args[1] == f"Process Group Id: {original_sub_path}" assert e.value.args[1] == f"Process Group Id: {original_sub_path}"
# make sure the new subgroup does exist # make sure the new subgroup does exist