Merge pull request #321 from codeforboston/ryan/epic-3/delete-incidents

Delete Incidents Route

backend/database/core.py

@@ -20,32 +20,35 @@
 
 from ..config import TestingConfig
 from ..utils import dev_only
+from typing import TypeVar, Type
 
 db = SQLAlchemy()
 
+T = TypeVar("T")
+
 
 class CrudMixin:
     """Mix me into a database model whose CRUD operations you want to expose in
     a convenient manner.
     """
 
-    def create(self, refresh: bool = True):
+    def create(self: T, refresh: bool = True) -> T:
         db.session.add(self)
         db.session.commit()
         if refresh:
             db.session.refresh(self)
         return self
 
-    def delete(self):
+    def delete(self) -> None:
         db.session.delete(self)
         db.session.commit()
 
     @classmethod
-    def get(cls, id: Any, abort_if_null: bool = True):
+    def get(cls: Type[T], id: Any, abort_if_null: bool = True) -> Optional[T]:
         obj = db.session.query(cls).get(id)
         if obj is None and abort_if_null:
             abort(404)
-        return obj
+        return obj  # type: ignore
 
 
 QUERIES_DIR = os.path.abspath(

backend/routes/partners.py

@@ -274,11 +274,13 @@ def get_incidents(partner_id: int):
     """
 
     # Check if the partner exists
-    partner: Partner = Partner.get(partner_id)  # type: ignore
+    partner = Partner.get(partner_id, False)
+    if partner is None:
+        abort(404)
 
     # Check if the user has permission to view incidents for this partner
-    jwt_decoded = get_jwt()  # type: ignore
-    user_id = jwt_decoded["sub"]  # type: ignore
+    jwt_decoded: dict[str, str] = get_jwt()
+    user_id = jwt_decoded["sub"]
 
     # Get the page number from the query parameters (default to 1)
     page = request.args.get("page", 1, type=int)
@@ -310,3 +312,48 @@ def get_incidents(partner_id: int):
         "totalPages": pagination.pages,
         "totalResults": pagination.total,
     }
+
+
+@bp.route("/<int:partner_id>/incidents/<int:incident_id>", methods=["DELETE"])
+@jwt_required()  # type: ignore
+@min_role_required(UserRole.CONTRIBUTOR)
+@validate()  # type: ignore
+def delete_incident(partner_id: int, incident_id: int):
+    """
+    Delete an incident associated with a partner.
+
+    Args:
+        partner_id (int): The ID of the partner.
+        incident_id (int): The ID of the incident.
+
+    Returns:
+        dict: A dictionary with a message indicating the
+        success of the deletion.
+
+    Raises:
+        403: If the user does not have permission to delete the incident.
+        404: If the partner or incident is not found.
+    """
+    jwt_decoded: dict[str, str] = get_jwt()
+    user_id = jwt_decoded["sub"]
+
+    # Check permissions first for security
+    permission = PartnerMember.query.filter(  # type: ignore
+        PartnerMember.user_id == user_id,
+        PartnerMember.partner_id == partner_id,
+        PartnerMember.role.in_((MemberRole.PUBLISHER, MemberRole.ADMIN)),
+    ).first()
+    if permission is None:
+        abort(403)
+
+    partner = Partner.get(partner_id, False)
+    if partner is None:
+        abort(404)
+
+    incident = Incident.get(incident_id, False)
+    if incident is None:
+        abort(404)
+
+    incident.delete()
+
+    return {"message": "Incident deleted successfully"}, 204

backend/tests/conftest.py

@@ -270,7 +270,7 @@ def partner_admin(db_session, example_partner):
 
 
 @pytest.fixture
-def partner_publisher(db_session, example_partner):
+def partner_publisher(db_session: Any, example_partner: PartnerMember):
     user = User(
         email=contributor_email,
         password=user_manager.hash_password(example_password),

backend/tests/test_partners.py

@@ -419,7 +419,6 @@ def test_get_incidents_not_admin(
     client: Any,
     access_token: str,
     example_partner: Partner,
-    example_user: User,
     example_incidents: List[Incident],
 ):
     # Make a request to get the incidents
@@ -432,3 +431,99 @@ def test_get_incidents_not_admin(
 
     # Verify that only public incidents are returned
     assert len(data["results"]) == len(example_incidents)
+
+
+def test_delete_incident(
+    client: Any,
+    partner_publisher: User,
+    example_partner: Partner,
+    example_incidents: List[Incident],
+):
+    """
+    Test that a partner member can delete an incident.
+    """
+    access_token = res = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": partner_publisher.email,
+            "password": example_password,
+        },
+    ).json["access_token"]
+
+    # Make a request to delete the incident
+    res = client.delete(
+        f"/api/v1/partners/{example_partner.id}"
+        + f"/incidents/{example_incidents[0].id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert res.status_code == 204
+
+    # Verify that the incident is deleted
+    deleted_incident = Incident.query.get(example_incidents[0].id)
+    assert deleted_incident is None
+
+
+def test_delete_incident_no_user_role(
+    client: Any,
+    access_token: str,
+):
+    """
+    Test that a user without atlest CONTRIBUTOR role
+    can't delete an incident.
+    """
+    # Make a request to delete the incident
+    res = client.delete(
+        f"/api/v1/partners/{999}" + f"/incidents/{999}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert res.status_code == 403
+
+
+def test_delete_incident_bad_partner_id(
+    client: Any,
+    partner_publisher: User,
+    example_incidents: List[Incident],
+):
+    """
+    Test that a partner member can't delete an incident
+    with a invalid partner id.
+    """
+    access_token = res = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": partner_publisher.email,
+            "password": example_password,
+        },
+    ).json["access_token"]
+
+    # Make a request to delete the incident
+    res = client.delete(
+        f"/api/v1/partners/{999}" + f"/incidents/{example_incidents[0].id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert res.status_code == 403
+
+
+def test_delete_incident_bad_incident_id(
+    client: Any,
+    partner_publisher: User,
+    example_partner: Partner,
+):
+    """
+    Test that a partner member can't delete an incident
+    with a invalid incident id.
+    """
+    access_token = res = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": partner_publisher.email,
+            "password": example_password,
+        },
+    ).json["access_token"]
+
+    # Make a request to delete the incident
+    res = client.delete(
+        f"/api/v1/partners/{example_partner.id}" + f"/incidents/{999}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert res.status_code == 404