Create incidents endpoint and tests

backend/routes/incidents.py

@@ -5,7 +5,7 @@
 from backend.auth.jwt import min_role_required, contributor_has_partner
 from backend.mixpanel.mix import track_to_mp
 from mixpanel import MixpanelException
-from flask import Blueprint, abort, current_app, request
+from flask import Blueprint, abort, request
 from flask_jwt_extended.view_decorators import jwt_required
 from flask_jwt_extended import get_jwt
 from pydantic import BaseModel
@@ -47,14 +47,28 @@ def get_incident(incident_id: int):
 @validate(json=CreateIncidentSchema)
 def create_incident():
     """Create a single incident.
-
-    Cannot be called in production environments
     """
-    if current_app.env == "production":
-        abort(418)
+    body = request.context.json
+    jwt_decoded: dict[str, str] = get_jwt()
+    user_id = jwt_decoded["sub"]
+    permission = PartnerMember.query.filter(
+        PartnerMember.user_id == user_id,
+        PartnerMember.role.in_((MemberRole.PUBLISHER, MemberRole.ADMIN)),
+    ).first()
+    if permission is None:
+        abort(403)
 
+    existing_incident = Incident.query.filter_by(
+        source_id=body.source_id,
+        time_of_incident=body.time_of_incident,
+        longitude=body.longitude,
+        latitude=body.latitude,
+        location=body.location,
+    ).first()
+    if existing_incident:
+        abort(409, "Incident already exists")
     try:
-        incident = incident_to_orm(request.context.json)
+        incident = incident_to_orm(body)
     except Exception:
         abort(400)
 

backend/tests/test_incidents.py

@@ -5,6 +5,8 @@
 from backend.database import Incident, Partner, PrivacyStatus, User
 from typing import Any
 
+member_email = "[email protected]"
+example_password = "my_password"
 mock_partners = {
     "cpdp": {"name": "Citizens Police Data Project"},
     "mpv": {"name": "Mapping Police Violence"},
@@ -53,18 +55,25 @@
 
 
 @pytest.fixture
-def example_incidents(db_session, client, contributor_access_token):
+def example_incidents(db_session, client , partner_admin):
     for id, mock in mock_partners.items():
         db_session.add(Partner(**mock))
         db_session.commit()
 
     created = {}
+    access_token = res = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": partner_admin.email ,
+            "password": "my_password"
+        },
+    ).json["access_token"]
     for name, mock in mock_incidents.items():
         res = client.post(
             "/api/v1/incidents/create",
             json=mock,
             headers={
-                "Authorization": "Bearer {0}".format(contributor_access_token)
+                "Authorization": "Bearer {0}".format(access_token)
             },
         )
         assert res.status_code == 200
@@ -73,9 +82,6 @@ def example_incidents(db_session, client, contributor_access_token):
 
 
 def test_create_incident(db_session, example_incidents):
-    # TODO: test that the User actually has permission to create an
-    # incident for the partner
-    # expected = mock_incidents["domestic"]
     created = example_incidents["domestic"]
 
     incident_obj = (
@@ -88,7 +94,95 @@ def test_create_incident(db_session, example_incidents):
             incident_obj.perpetrators[i].id == created["perpetrators"][i]["id"]
         )
     assert incident_obj.use_of_force[0].id == created["use_of_force"][0]["id"]
-    # assert incident_obj.source == expected["source"]
+    assert incident_obj.location == created["location"]
+    assert incident_obj.description == created["description"]
+
+
+"""
+test for creating a new incident and
+creating same incident
+"""
+
+
+def test_create_incident_exists(
+        client,
+        partner_admin,
+
+):
+    created = {}
+    access_token = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": partner_admin.email ,
+            "password": "my_password"
+        },
+    ).json["access_token"]
+    # creating new incident
+    res = client.post(
+        "/api/v1/incidents/create",
+        headers={"Authorization": f"Bearer {access_token}"},
+        json=mock_incidents["domestic"]
+    )
+    created["domestic"] = res.json
+    domestic_instance = created["domestic"]
+    print(created)
+    assert res.status_code == 200
+    expected = mock_incidents["domestic"]
+    incident_obj = Incident.query.filter_by(
+        time_of_incident=expected["time_of_incident"]
+    ).first()
+    date_format = '%Y-%m-%d %H:%M:%S'
+    date_obj = datetime.strptime(
+        expected["time_of_incident"],
+        date_format)
+    assert incident_obj.time_of_incident == date_obj
+    for i in [0, 1]:
+        assert (
+            incident_obj.perpetrators[i].id ==
+            domestic_instance["perpetrators"][i]["id"]
+        )
+    assert (incident_obj.use_of_force[0].id ==
+            domestic_instance["use_of_force"][0]["id"])
+    assert incident_obj.location == domestic_instance["location"]
+    assert incident_obj.description == domestic_instance["description"]
+
+    # creating the same incident
+    # should not be able to create incident
+    res = client.post(
+        "/api/v1/incidents/create",
+        headers={"Authorization": f"Bearer {access_token}"},
+        json=mock_incidents["domestic"]
+    )
+
+    assert res.status_code == 409
+
+
+"""
+creating incident when user
+does not have permission
+"""
+
+
+def test_create_incident_no_permission(
+        client,
+        example_user
+
+):
+    access_token = client.post(
+        "api/v1/auth/login",
+        json={
+            "email": example_user.email,
+            "password": "my_password"
+        },
+    ).json["access_token"]
+    print(access_token)
+    # creating new incident
+    res = client.post(
+        "/api/v1/incidents/create",
+        headers={"Authorization": f"Bearer {access_token}"},
+        json=mock_incidents["domestic"]
+    )
+    assert res.status_code == 403
 
 
 def test_get_incident(app, client, db_session, access_token):