diff --git a/antarest/core/exceptions.py b/antarest/core/exceptions.py index 2d9ea03f58..aca126e68e 100644 --- a/antarest/core/exceptions.py +++ b/antarest/core/exceptions.py @@ -4,8 +4,6 @@ from fastapi.exceptions import HTTPException -MAX_BINDING_CONSTRAINTS_TO_DISPLAY = 10 - class ShouldNotHappenException(Exception): pass @@ -327,43 +325,33 @@ def __init__(self, is_variant: bool) -> None: super().__init__(HTTPStatus.EXPECTATION_FAILED, "Upgrade not supported for parent of variants") -class AreaDeletionNotAllowed(HTTPException): - def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None: - first_bcs_ids = "" - for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]): - first_bcs_ids += f"{i+1}- {bc}\n" - message = "Area is referenced in the following binding constraints:\n" + first_bcs_ids - msg = f"Area {uuid} is not allowed to be deleted\n{message}" - super().__init__( - HTTPStatus.FORBIDDEN, - msg, - ) - +class BindingConstraintDeletionNotAllowed(HTTPException): + """ + Exception raised when a binding constraint is not allowed to be deleted because it references + other objects: areas, links or thermal clusters. + """ -class LinkDeletionNotAllowed(HTTPException): - def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None: - first_bcs_ids = "" - for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]): - first_bcs_ids += f"{i+1}- {bc}\n" - message = "Link is referenced in the following binding constraints:\n" + first_bcs_ids - msg = f"Link {uuid} is not allowed to be deleted\n{message}" - super().__init__( - HTTPStatus.FORBIDDEN, - msg, + def __init__(self, object_id: str, binding_ids: t.Sequence[str], *, object_type: str) -> None: + """ + Initialize the exception. + + Args: + object_id: ID of the object that is not allowed to be deleted. + binding_ids: Binding constraints IDs that reference the object. + object_type: Type of the object that is not allowed to be deleted: area, link or thermal cluster. + """ + max_count = 10 + first_bcs_ids = ",\n".join(f"{i}- '{bc}'" for i, bc in enumerate(binding_ids[:max_count], 1)) + and_more = f",\nand {len(binding_ids) - max_count} more..." if len(binding_ids) > max_count else "." + message = ( + f"{object_type} '{object_id}' is not allowed to be deleted, because it is referenced" + f" in the following binding constraints:\n{first_bcs_ids}{and_more}" ) + super().__init__(HTTPStatus.FORBIDDEN, message) - -class ClusterDeletionNotAllowed(HTTPException): - def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None: - first_bcs_ids = "" - for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]): - first_bcs_ids += f"{i+1}- {bc}\n" - message = "Cluster is referenced in the following binding constraints:\n" + first_bcs_ids - msg = f"Cluster {uuid} is not allowed to be deleted\n{message}" - super().__init__( - HTTPStatus.FORBIDDEN, - msg, - ) + def __str__(self) -> str: + """Return a string representation of the exception.""" + return self.detail class UnsupportedStudyVersion(HTTPException): diff --git a/antarest/study/business/binding_constraint_management.py b/antarest/study/business/binding_constraint_management.py index 28881ef874..7f42bb7f59 100644 --- a/antarest/study/business/binding_constraint_management.py +++ b/antarest/study/business/binding_constraint_management.py @@ -501,13 +501,12 @@ def terms_to_coeffs(terms: t.Sequence[ConstraintTerm]) -> t.Dict[str, t.List[flo :return: A dictionary of term IDs mapped to a list of their coefficients. """ coeffs = {} - if terms is not None: - for term in terms: - if term.id and term.weight is not None: - coeffs[term.id] = [term.weight] - if term.offset: - coeffs[term.id].append(term.offset) - return coeffs + for term in terms: + if term.id and term.weight is not None: + coeffs[term.id] = [term.weight] + if term.offset: + coeffs[term.id].append(term.offset) + return coeffs def get_binding_constraint(self, study: Study, bc_id: str) -> ConstraintOutput: """ diff --git a/antarest/study/service.py b/antarest/study/service.py index 4ab20a4723..c06703c8b3 100644 --- a/antarest/study/service.py +++ b/antarest/study/service.py @@ -20,12 +20,10 @@ from antarest.core.config import Config from antarest.core.exceptions import ( - AreaDeletionNotAllowed, BadEditInstructionException, - ClusterDeletionNotAllowed, + BindingConstraintDeletionNotAllowed, CommandApplicationError, IncorrectPathError, - LinkDeletionNotAllowed, NotAManagedStudyException, StudyDeletionNotAllowed, StudyNotFoundError, @@ -689,7 +687,7 @@ def create_study( id=sid, name=study_name, workspace=DEFAULT_WORKSPACE_NAME, - path=study_path, + path=str(study_path), created_at=datetime.utcnow(), updated_at=datetime.utcnow(), version=version or NEW_DEFAULT_STUDY_VERSION, @@ -1268,17 +1266,18 @@ def export_task(notifier: TaskUpdateNotifier) -> TaskResult: stopwatch.log_elapsed( lambda x: logger.info(f"Study {study_id} filtered output {output_id} exported in {x}s") ) - return FileResponse( - tmp_export_file, - headers=( - {"Content-Disposition": "inline"} - if filetype == ExportFormat.JSON - else { - "Content-Disposition": f'attachment; filename="output-{output_id}.{"tar.gz" if filetype == ExportFormat.TAR_GZ else "zip"}' - } - ), - media_type=filetype, - ) + + if filetype == ExportFormat.JSON: + headers = {"Content-Disposition": "inline"} + elif filetype == ExportFormat.TAR_GZ: + headers = {"Content-Disposition": f'attachment; filename="output-{output_id}.tar.gz'} + elif filetype == ExportFormat.ZIP: + headers = {"Content-Disposition": f'attachment; filename="output-{output_id}.zip'} + else: # pragma: no cover + raise NotImplementedError(f"Export format {filetype} is not supported") + + return FileResponse(tmp_export_file, headers=headers, media_type=filetype) + else: json_response = json.dumps( matrix.dict(), @@ -1317,26 +1316,20 @@ def set_sim_reference( params: RequestParameters, ) -> None: """ - Set simulation as the reference output + Set simulation as the reference output. + Args: - study_id: study Id - output_id: output id - status: state of the reference status + study_id: study ID. + output_id: The ID of the output to set as reference. + status: state of the reference status. params: request parameters - - Returns: None - """ study = self.get_study(study_id) assert_permission(params.user, study, StudyPermissionType.WRITE) self._assert_study_unarchived(study) logger.info( - "output %s set by user %s as reference (%b) for study %s", - output_id, - params.get_user_id(), - status, - study_id, + f"output {output_id} set by user {params.get_user_id()} as reference ({status}) for study {study_id}" ) self.storage_service.get_storage(study).set_reference_output(study, output_id, status) @@ -1858,6 +1851,18 @@ def update_thermal_cluster_metadata( return self.areas.update_thermal_cluster_metadata(study, area_id, clusters_metadata) def delete_area(self, uuid: str, area_id: str, params: RequestParameters) -> None: + """ + Delete area from study if it is not referenced by a binding constraint, + otherwise raise an HTTP 403 Forbidden error. + + Args: + uuid: The study ID. + area_id: The area ID to delete. + params: The request parameters used to check user permissions. + + Raises: + BindingConstraintDeletionNotAllowed: If the area is referenced by a binding constraint. + """ study = self.get_study(uuid) assert_permission(params.user, study, StudyPermissionType.WRITE) self._assert_study_unarchived(study) @@ -1865,7 +1870,8 @@ def delete_area(self, uuid: str, area_id: str, params: RequestParameters) -> Non study, ConstraintFilters(area_name=area_id) ) if referencing_binding_constraints: - raise AreaDeletionNotAllowed(area_id, [bc.id for bc in referencing_binding_constraints]) + binding_ids = [bc.id for bc in referencing_binding_constraints] + raise BindingConstraintDeletionNotAllowed(area_id, binding_ids, object_type="Area") self.areas.delete_area(study, area_id) self.event_bus.push( Event( @@ -1882,6 +1888,19 @@ def delete_link( area_to: str, params: RequestParameters, ) -> None: + """ + Delete link from study if it is not referenced by a binding constraint, + otherwise raise an HTTP 403 Forbidden error. + + Args: + uuid: The study ID. + area_from: The area from which the link starts. + area_to: The area to which the link ends. + params: The request parameters used to check user permissions. + + Raises: + BindingConstraintDeletionNotAllowed: If the link is referenced by a binding constraint. + """ study = self.get_study(uuid) assert_permission(params.user, study, StudyPermissionType.WRITE) self._assert_study_unarchived(study) @@ -1890,7 +1909,8 @@ def delete_link( study, ConstraintFilters(link_id=link_id) ) if referencing_binding_constraints: - raise LinkDeletionNotAllowed(link_id, [bc.id for bc in referencing_binding_constraints]) + binding_ids = [bc.id for bc in referencing_binding_constraints] + raise BindingConstraintDeletionNotAllowed(link_id, binding_ids, object_type="Link") self.links.delete_link(study, area_from, area_to) self.event_bus.push( Event( @@ -2537,22 +2557,21 @@ def asserts_no_thermal_in_binding_constraints( self, study: Study, area_id: str, cluster_ids: t.Sequence[str] ) -> None: """ - Check that no cluster is referenced in a binding constraint otherwise raise an ClusterDeletionNotAllowed - Exception + Check that no cluster is referenced in a binding constraint, otherwise raise an HTTP 403 Forbidden error. Args: study: input study for which an update is to be committed area_id: area ID to be checked cluster_ids: IDs of the thermal clusters to be checked - Returns: None - - Raises: ClusterDeletionNotAllowed if a cluster is referenced in a binding constraint + Raises: + BindingConstraintDeletionNotAllowed: if a cluster is referenced in a binding constraint """ for cluster_id in cluster_ids: - referencing_binding_constraints = self.binding_constraint_manager.get_binding_constraints( + ref_bcs = self.binding_constraint_manager.get_binding_constraints( study, ConstraintFilters(cluster_id=f"{area_id}.{cluster_id}") ) - if referencing_binding_constraints: - raise ClusterDeletionNotAllowed(cluster_id, [bc.id for bc in referencing_binding_constraints]) + if ref_bcs: + binding_ids = [bc.id for bc in ref_bcs] + raise BindingConstraintDeletionNotAllowed(cluster_id, binding_ids, object_type="Cluster") diff --git a/tests/core/test_exceptions.py b/tests/core/test_exceptions.py new file mode 100644 index 0000000000..28d26ec3f5 --- /dev/null +++ b/tests/core/test_exceptions.py @@ -0,0 +1,25 @@ +from antarest.core.exceptions import BindingConstraintDeletionNotAllowed + + +class TestBindingConstraintDeletionNotAllowed: + def test_few_binding_constraints(self) -> None: + object_id = "france" + binding_ids = ["bc1", "bc2"] + object_type = "Area" + exception = BindingConstraintDeletionNotAllowed(object_id, binding_ids, object_type=object_type) + message = str(exception) + assert f"{object_type} '{object_id}'" in message + assert "bc1" in message + assert "bc2" in message + assert "more..." not in message + + def test_many_binding_constraints(self) -> None: + object_id = "france" + binding_ids = [f"bc{i}" for i in range(1, 50)] + object_type = "Area" + exception = BindingConstraintDeletionNotAllowed(object_id, binding_ids, object_type=object_type) + message = str(exception) + assert f"{object_type} '{object_id}'" in message + assert "bc1" in message + assert "bc2" in message + assert "more..." in message diff --git a/tests/integration/study_data_blueprint/test_thermal.py b/tests/integration/study_data_blueprint/test_thermal.py index 921e1ebab0..77de2566ab 100644 --- a/tests/integration/study_data_blueprint/test_thermal.py +++ b/tests/integration/study_data_blueprint/test_thermal.py @@ -656,7 +656,7 @@ def test_lifecycle( assert res.status_code == 403, res.json() description = res.json()["description"] assert all([elm in description for elm in [fr_gas_conventional, "binding constraint"]]) - assert res.json()["exception"] == "ClusterDeletionNotAllowed" + assert res.json()["exception"] == "BindingConstraintDeletionNotAllowed" # delete the binding constraint res = client.delete( diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py index d7e2f69864..096feec505 100644 --- a/tests/integration/test_integration.py +++ b/tests/integration/test_integration.py @@ -1434,7 +1434,7 @@ def test_area_management(client: TestClient, admin_access_token: str) -> None: description = result.json()["description"] assert all([elm in description for elm in ["area 1", "binding constraint 1"]]) # check the exception - assert result.json()["exception"] == "AreaDeletionNotAllowed" + assert result.json()["exception"] == "BindingConstraintDeletionNotAllowed" # delete binding constraint 1 client.delete(f"/v1/studies/{study_id}/bindingconstraints/binding%20constraint%201") @@ -1715,9 +1715,107 @@ def test_copy(client: TestClient, admin_access_token: str, study_id: str) -> Non assert res["public_mode"] == "READ" -def test_links_deletion(client: TestClient, user_access_token: str, study_id: str) -> None: +def test_areas_deletion_with_binding_constraints(client: TestClient, user_access_token: str, study_id: str) -> None: """ - Test the deletion of links between areas. + Test the deletion of areas that are referenced in binding constraints. + """ + + # set client headers to user access token + client.headers = {"Authorization": f"Bearer {user_access_token}"} + + area1_id = "france" + area2_id = "germany" + cluster_id = "nuclear power plant" + + constraint_terms = [ + { + # Link between two areas + "data": {"area1": area1_id, "area2": area2_id}, + "id": f"{area1_id}%{area2_id}", + "offset": 2, + "weight": 1.0, + }, + { + # Cluster in an area + "data": {"area": area1_id, "cluster": cluster_id.lower()}, + "id": f"{area1_id}.{cluster_id.lower()}", + "offset": 2, + "weight": 1.0, + }, + ] + + for constraint_term in constraint_terms: + # Create an area "area_1" in the study + res = client.post( + f"/v1/studies/{study_id}/areas", + json={"name": area1_id.title(), "type": "AREA", "metadata": {"country": "FR"}}, + ) + res.raise_for_status() + + if set(constraint_term["data"]) == {"area1", "area2"}: + # Create a second area and a link between the two areas + res = client.post( + f"/v1/studies/{study_id}/areas", + json={"name": area2_id.title(), "type": "AREA", "metadata": {"country": "DE"}}, + ) + res.raise_for_status() + res = client.post( + f"/v1/studies/{study_id}/links", + json={"area1": area1_id, "area2": area2_id}, + ) + res.raise_for_status() + + elif set(constraint_term["data"]) == {"area", "cluster"}: + # Create a cluster in the first area + res = client.post( + f"/v1/studies/{study_id}/areas/{area1_id}/clusters/thermal", + json={"name": cluster_id.title(), "group": "Nuclear"}, + ) + res.raise_for_status() + + else: + raise NotImplementedError(f"Unsupported constraint term: {constraint_term}") + + # create a binding constraint that references the link + bc_id = "bc_1" + bc_obj = { + "name": bc_id, + "enabled": True, + "time_step": "daily", + "operator": "less", + "terms": [constraint_term], + } + res = client.post(f"/v1/studies/{study_id}/bindingconstraints", json=bc_obj) + res.raise_for_status() + + if set(constraint_term["data"]) == {"area1", "area2"}: + areas_to_delete = [area1_id, area2_id] + elif set(constraint_term["data"]) == {"area", "cluster"}: + areas_to_delete = [area1_id] + else: + raise NotImplementedError(f"Unsupported constraint term: {constraint_term}") + + for area_id in areas_to_delete: + # try to delete the areas + res = client.delete(f"/v1/studies/{study_id}/areas/{area_id}") + assert res.status_code == 403, res.json() + description = res.json()["description"] + assert all([elm in description for elm in [area_id, bc_id]]) + assert res.json()["exception"] == "BindingConstraintDeletionNotAllowed" + + # delete the binding constraint + res = client.delete(f"/v1/studies/{study_id}/bindingconstraints/{bc_id}") + assert res.status_code == 200, res.json() + + for area_id in areas_to_delete: + # delete the area + res = client.delete(f"/v1/studies/{study_id}/areas/{area_id}") + assert res.status_code == 200, res.json() + + +def test_links_deletion_with_binding_constraints(client: TestClient, user_access_token: str, study_id: str) -> None: + """ + Test the deletion of links that are referenced in binding constraints. """ # set client headers to user access token @@ -1748,10 +1846,7 @@ def test_links_deletion(client: TestClient, user_access_token: str, study_id: st # create a link between the two areas res = client.post( f"/v1/studies/{study_id}/links", - json={ - "area1": "area_1", - "area2": "area_2", - }, + json={"area1": "area_1", "area2": "area_2"}, ) assert res.status_code == 200, res.json() @@ -1769,10 +1864,7 @@ def test_links_deletion(client: TestClient, user_access_token: str, study_id: st } ], } - res = client.post( - f"/v1/studies/{study_id}/bindingconstraints", - json=bc_obj, - ) + res = client.post(f"/v1/studies/{study_id}/bindingconstraints", json=bc_obj) assert res.status_code == 200, res.json() # try to delete the link before deleting the binding constraint @@ -1780,7 +1872,7 @@ def test_links_deletion(client: TestClient, user_access_token: str, study_id: st assert res.status_code == 403, res.json() description = res.json()["description"] assert all([elm in description for elm in ["area_1%area_2", "bc_1"]]) - assert res.json()["exception"] == "LinkDeletionNotAllowed" + assert res.json()["exception"] == "BindingConstraintDeletionNotAllowed" # delete the binding constraint res = client.delete(f"/v1/studies/{study_id}/bindingconstraints/bc_1")