#614972427

Committed 12 Dec 2023 09:36PM UTC coverage: 93.876% (+8.7%) from 85.196%

Build # #614972427

Build Type

Pull #153

travis-ci

Pull Request Pull Request #153: Merge to Master 12.12.23. Before Final Presentation

Run Details

246 of 264 new or added lines in 11 files covered. (93.18%)

1 existing line in 1 file now uncovered.

1211 of 1290 relevant lines covered (93.88%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.77

/api/views.py

from rest_framework import status
from django.utils import timezone
from django.template import loader
from django.http import HttpRequest
from rest_framework import generics
from haversine import haversine, Unit
from rest_framework.views import APIView
from rest_framework.response import Response
from django.contrib.auth import get_user_model
from django.shortcuts import get_object_or_404
from django.core.mail import EmailMultiAlternatives
from rest_framework.permissions import IsAuthenticated
from django.contrib.sites.shortcuts import get_current_site
from rest_framework.authentication import SessionAuthentication

from map.models import ParkingSpace, OccupancyHistory
from users.models import Post, Comment, UserWatchedParkingSpace
from .serializers import (
    ParkingSpaceSerializer,
    PostSerializer,
    UserWatchedParkingSpaceSerializer,
)

from better_profanity import profanity

profanity.load_censor_words()
# Custom swear words can be added to this array
custom_badwords = ["bullshittery", "bitchy"]
User = get_user_model()


class ParkingSpaceNearCenterAPIView(generics.ListAPIView):
    """API endpoint
    /api/spots/?lat=LATITUDE&lon=LONGITUDE
    """

    queryset = ParkingSpace.objects.all()
    serializer_class = ParkingSpaceSerializer

    def get(self, request: HttpRequest) -> Response:
        """handles get requests to API endpoint above

        Args:
            request (HttpRequest): http request object

        Returns:
            Response: JSON Object with either message requesting
            lat and lon as query parameters OR on success
            the parking spots within distance
            (specified in __is_within_dist method)
        """
        lat = request.GET.get("lat")
        lon = request.GET.get("lon")

        if not (lat and lon):
            response_data = {"message": "Bad Request: Missing lat and lon parameters"}
            return Response(response_data, status=status.HTTP_400_BAD_REQUEST)

        center_point = (float(lat), float(lon))
        filtered_spots = [
            spot
            for spot in self.queryset.all()
            if self.__is_within_dist(
                center_point, (float(spot.latitude), float(spot.longitude))
            )
        ]

        for spot in filtered_spots:
            if spot.occupancy_percent:
                spot.occupancy_percent = round(spot.occupancy_percent / 10) * 10

        serializer = self.serializer_class(filtered_spots, many=True)

        return Response(serializer.data, status=status.HTTP_200_OK)

    def __is_within_dist(self, p1, p2):
        """method to create a normal user

        Returns:
            Boolean: p1 is within max_dist of p2
        """
        max_dist = 1
        return haversine(p1, p2, unit=Unit.MILES) < max_dist


class ParkingSpaceChangeOccupancyAPIView(APIView):
    """API endpoint
    /api/spot/occupancy/?percent=PERCENT&id=PARKING_SPACE_ID

    Changes the Occupancy Percent of Parking Space (ID)
    """

    permission_classes = [IsAuthenticated]
    authentication_classes = [SessionAuthentication]

    extra_email_context = {"site_name": "Parkrowd"}
    email_template_name = "api/parkingspace_status_update_email_template.html"
    subject_template_name = "api/parkingspace_status_update_email_subject_template.txt"

    def post(self, request: HttpRequest) -> Response:
        """handles post requests to API endpoint above

        Args:
            request (HttpRequest): http request object

        Returns:
            Response: JSON Object with either parking_spot_id
            sent on success OR
            fail message to update front end
        """
        occupancy_percent = request.data.get("percent")
        parking_spot_id = request.data.get("id")
        available_vehicle_spaces = request.data.get("available_vehicle_spaces")

        if not ((occupancy_percent is not None) and parking_spot_id):
            response_data = {"message": "Bad Request: Missing percent or id parameters"}
            return Response(response_data, status=status.HTTP_400_BAD_REQUEST)

        if isinstance(occupancy_percent, str) and not occupancy_percent.isdigit():
            response_data = {"message": "Bad Request: Percent can only have digits"}
            return Response(response_data, status=status.HTTP_400_BAD_REQUEST)

        occupancy_percent = int(occupancy_percent)
        if occupancy_percent > 100:
            response_data = {"message": "Bad Request: Percent is >100"}
            return Response(response_data, status=status.HTTP_400_BAD_REQUEST)
        if occupancy_percent % 10 != 0:
            response_data = {"message": "Bad Request: Percent is not a multiple of 10"}
            return Response(response_data, status=status.HTTP_400_BAD_REQUEST)

        try:
            # * Retrieve the ParkingSpace instance
            parking_space = ParkingSpace.objects.get(parking_spot_id=parking_spot_id)

            # Update the occupancy_percent field
            parking_space.occupancy_percent = occupancy_percent

            # Update the available_vehicle_spaces
            parking_space.available_vehicle_spaces = available_vehicle_spaces

            # Create New Occupancy History
            history = OccupancyHistory()
            history.user = request.user
            history.parking_space = get_object_or_404(
                ParkingSpace, parking_spot_id=parking_spot_id
            )
            history.updated_at = timezone.now()
            history.occupancy_percent = occupancy_percent

            parking_space.save()
            history.save()

            # send emails to users that put a watch on this spot
            current_site = get_current_site(request)
            email_field_name = User.get_email_field_name()
            user_watches = UserWatchedParkingSpace.objects.filter(
                parking_space=parking_space, threshold__gte=occupancy_percent
            )
            domain = current_site.domain

            for record in user_watches:
                user_email = getattr(record.user, email_field_name)
                context = {
                    "user": record.user,
                    "email": user_email,
                    "domain": domain,
                    "protocol": "https" if self.request.is_secure() else "http",
                    **(self.extra_email_context or {}),
                }
                self.send_mail(
                    self.subject_template_name,
                    self.email_template_name,
                    context,
                    None,
                    user_email,
                )
                record.delete()

            return Response(
                {"message": "Occupancy percent updated successfully."},
                status=status.HTTP_200_OK,
            )

        except ParkingSpace.DoesNotExist:
            return Response(
                {"message": "ParkingSpace with the specified id does not exist."},
                status=status.HTTP_400_BAD_REQUEST,
            )

        except Exception as e:
            return Response(
                {"message": f"An error occurred: {str(e)}"},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )

    def send_mail(
        self,
        subject_template_name,
        email_template_name,
        context,
        from_email,
        to_email,
        html_email_template_name=None,
    ):
        """
        Send a django.core.mail.EmailMultiAlternatives to `to_email`.
        """
        subject = loader.render_to_string(subject_template_name, context)
        # Email subject *must not* contain newlines
        subject = "".join(subject.splitlines())
        body = loader.render_to_string(email_template_name, context)

        email_message = EmailMultiAlternatives(subject, body, from_email, [to_email])
        if html_email_template_name is not None:
            html_email = loader.render_to_string(html_email_template_name, context)
            email_message.attach_alternative(html_email, "text/html")

        email_message.send()


class ParkingSpacePostsAPIView(generics.ListAPIView):
    """GET API endpoint
    /api/spot/posts/<str:spotId>

    get a list of all the posts associated with a spot
    """

    serializer_class = PostSerializer

    def get_queryset(self):
        parking_space_id = self.kwargs["spotId"]
        return Post.objects.filter(parking_space__parking_spot_id=parking_space_id)


class ParkingSpaceAddCommentAPIView(APIView):
    """POST API endpoint
    /api/spot/posts/add-comment/<int:postId>

    get a list of all the posts associated with a spot
    """

    permission_classes = [IsAuthenticated]
    authentication_classes = [SessionAuthentication]

    def post(self, request: HttpRequest, postId: int) -> Response:
        profanity.add_censor_words(custom_badwords)
        comment_content = profanity.censor(request.data["commentContent"])
        if not comment_content:
            return Response("Error: empty comment content", 400)
        try:
            post = Post.objects.get(id=postId)
        except Post.DoesNotExist:
            return Response(f"No post with post id {postId}", 400)

        new_comment = Comment(
            content=comment_content,
            author=request.user,
            post=post,
            created_at=timezone.now(),
        )
        new_comment.save()

        return Response("Comment created!", 200)


class AddWatchOnParkingSpaceAPIView(APIView):
    model = UserWatchedParkingSpace
    permission_classes = [IsAuthenticated]
    authentication_classes = [SessionAuthentication]

    def post(self, request: HttpRequest) -> Response:
        user = request.user
        try:
            parking_space = ParkingSpace.objects.get(
                parking_spot_id=request.data["parking_spot_id"]
            )
        except ParkingSpace.DoesNotExist:
            return Response("Invalid parking spot id", 400)

        threshold = request.data.get("threshold", 80)
        newUserWatchedParkingSpace = UserWatchedParkingSpace(
            user=user, parking_space=parking_space, threshold=threshold
        )
        newUserWatchedParkingSpace.save()

        return Response(f"Watch on spot {parking_space.parking_spot_id} added", 200)


class RemoveWatchOnParkingSpaceAPIView(APIView):
    model = UserWatchedParkingSpace
    permission_classes = [IsAuthenticated]
    authentication_classes = [SessionAuthentication]

    def post(self, request: HttpRequest) -> Response:
        try:
            parkingSpace = ParkingSpace.objects.get(
                parking_spot_id=request.data["parking_spot_id"]
            )
        except ParkingSpace.DoesNotExist:
            return Response("Invalid parking spot id", 400)
        try:
            watchRecord = self.model.objects.get(
                user=request.user, parking_space=parkingSpace
            )
        except self.model.DoesNotExist:
            return Response("No watch found on this parking space", 400)
        watchRecord.delete()

        return Response(f"Watch on spot {parkingSpace.parking_spot_id} removed", 200)


class WatchOnParkingSpaceAPIView(APIView):
    model = UserWatchedParkingSpace
    permission_classes = [IsAuthenticated]
    authentication_classes = [SessionAuthentication]
    serializer_class = UserWatchedParkingSpaceSerializer

    def get(self, request: HttpRequest) -> Response:
        return Response(
            self.serializer_class(
                self.model.objects.filter(user=request.user), many=True
            ).data,
            status.HTTP_200_OK,
        )

1	from rest_framework import status	1✔
2	from django.utils import timezone	1✔
3	from django.template import loader	1✔
4	from django.http import HttpRequest	1✔
5	from rest_framework import generics	1✔
6	from haversine import haversine, Unit	1✔
7	from rest_framework.views import APIView	1✔
8	from rest_framework.response import Response	1✔
9	from django.contrib.auth import get_user_model	1✔
10	from django.shortcuts import get_object_or_404	1✔
11	from django.core.mail import EmailMultiAlternatives	1✔
12	from rest_framework.permissions import IsAuthenticated	1✔
13	from django.contrib.sites.shortcuts import get_current_site	1✔
14	from rest_framework.authentication import SessionAuthentication	1✔
15
16	from map.models import ParkingSpace, OccupancyHistory	1✔
17	from users.models import Post, Comment, UserWatchedParkingSpace	1✔
18	from .serializers import (	1✔
19	ParkingSpaceSerializer,
20	PostSerializer,
21	UserWatchedParkingSpaceSerializer,
22	)
23
24	from better_profanity import profanity	1✔
25
26	profanity.load_censor_words()	1✔
27	# Custom swear words can be added to this array
28	custom_badwords = ["bullshittery", "bitchy"]	1✔
29	User = get_user_model()	1✔
30
31
32	class ParkingSpaceNearCenterAPIView(generics.ListAPIView):	1✔
33	"""API endpoint
34	/api/spots/?lat=LATITUDE&lon=LONGITUDE
35	"""
36
37	queryset = ParkingSpace.objects.all()	1✔
38	serializer_class = ParkingSpaceSerializer	1✔
39
40	def get(self, request: HttpRequest) -> Response:	1✔
41	"""handles get requests to API endpoint above
42
43	Args:
44	request (HttpRequest): http request object
45
46	Returns:
47	Response: JSON Object with either message requesting
48	lat and lon as query parameters OR on success
49	the parking spots within distance
50	(specified in __is_within_dist method)
51	"""
52	lat = request.GET.get("lat")	1✔
53	lon = request.GET.get("lon")	1✔
54
55	if not (lat and lon):	1✔
56	response_data = {"message": "Bad Request: Missing lat and lon parameters"}	1✔
57	return Response(response_data, status=status.HTTP_400_BAD_REQUEST)	1✔
58
59	center_point = (float(lat), float(lon))	1✔
60	filtered_spots = [	1✔
61	spot
62	for spot in self.queryset.all()
63	if self.__is_within_dist(
64	center_point, (float(spot.latitude), float(spot.longitude))
65	)
66	]
67
68	for spot in filtered_spots:	1✔
69	if spot.occupancy_percent:	1✔
70	spot.occupancy_percent = round(spot.occupancy_percent / 10) * 10	×
71
72	serializer = self.serializer_class(filtered_spots, many=True)	1✔
73
74	return Response(serializer.data, status=status.HTTP_200_OK)	1✔
75
76	def __is_within_dist(self, p1, p2):	1✔
77	"""method to create a normal user
78
79	Returns:
80	Boolean: p1 is within max_dist of p2
81	"""
82	max_dist = 1	1✔
83	return haversine(p1, p2, unit=Unit.MILES) < max_dist	1✔
84
85
86	class ParkingSpaceChangeOccupancyAPIView(APIView):	1✔
87	"""API endpoint
88	/api/spot/occupancy/?percent=PERCENT&id=PARKING_SPACE_ID
89
90	Changes the Occupancy Percent of Parking Space (ID)
91	"""
92
93	permission_classes = [IsAuthenticated]	1✔
94	authentication_classes = [SessionAuthentication]	1✔
95
96	extra_email_context = {"site_name": "Parkrowd"}	1✔
97	email_template_name = "api/parkingspace_status_update_email_template.html"	1✔
98	subject_template_name = "api/parkingspace_status_update_email_subject_template.txt"	1✔
99
100	def post(self, request: HttpRequest) -> Response:	1✔
101	"""handles post requests to API endpoint above
102
103	Args:
104	request (HttpRequest): http request object
105
106	Returns:
107	Response: JSON Object with either parking_spot_id
108	sent on success OR
109	fail message to update front end
110	"""
111	occupancy_percent = request.data.get("percent")	1✔
112	parking_spot_id = request.data.get("id")	1✔
113	available_vehicle_spaces = request.data.get("available_vehicle_spaces")	1✔
114
115	if not ((occupancy_percent is not None) and parking_spot_id):	1✔
116	response_data = {"message": "Bad Request: Missing percent or id parameters"}	1✔
117	return Response(response_data, status=status.HTTP_400_BAD_REQUEST)	1✔
118
119	if isinstance(occupancy_percent, str) and not occupancy_percent.isdigit():	1✔
120	response_data = {"message": "Bad Request: Percent can only have digits"}	1✔
121	return Response(response_data, status=status.HTTP_400_BAD_REQUEST)	1✔
122
123	occupancy_percent = int(occupancy_percent)	1✔
124	if occupancy_percent > 100:	1✔
125	response_data = {"message": "Bad Request: Percent is >100"}	1✔
126	return Response(response_data, status=status.HTTP_400_BAD_REQUEST)	1✔
127	if occupancy_percent % 10 != 0:	1✔
128	response_data = {"message": "Bad Request: Percent is not a multiple of 10"}	1✔
129	return Response(response_data, status=status.HTTP_400_BAD_REQUEST)	1✔
130
131	try:	1✔
132	# * Retrieve the ParkingSpace instance
133	parking_space = ParkingSpace.objects.get(parking_spot_id=parking_spot_id)	1✔
134
135	# Update the occupancy_percent field
136	parking_space.occupancy_percent = occupancy_percent	1✔
137
138	# Update the available_vehicle_spaces
139	parking_space.available_vehicle_spaces = available_vehicle_spaces	1✔
140
141	# Create New Occupancy History
142	history = OccupancyHistory()	1✔
143	history.user = request.user	1✔
144	history.parking_space = get_object_or_404(	1✔
145	ParkingSpace, parking_spot_id=parking_spot_id
146	)
147	history.updated_at = timezone.now()	1✔
148	history.occupancy_percent = occupancy_percent	1✔
149
150	parking_space.save()	1✔
151	history.save()	1✔
152
153	# send emails to users that put a watch on this spot
154	current_site = get_current_site(request)	1✔
155	email_field_name = User.get_email_field_name()	1✔
156	user_watches = UserWatchedParkingSpace.objects.filter(	1✔
157	parking_space=parking_space, threshold__gte=occupancy_percent
158	)
159	domain = current_site.domain	1✔
160
161	for record in user_watches:	1✔
162	user_email = getattr(record.user, email_field_name)	1✔
163	context = {	1✔
164	"user": record.user,
165	"email": user_email,
166	"domain": domain,
167	"protocol": "https" if self.request.is_secure() else "http",
168	**(self.extra_email_context or {}),
169	}
170	self.send_mail(	1✔
171	self.subject_template_name,
172	self.email_template_name,
173	context,
174	None,
175	user_email,
176	)
177	record.delete()	1✔
178
179	return Response(	1✔
180	{"message": "Occupancy percent updated successfully."},
181	status=status.HTTP_200_OK,
182	)
183
184	except ParkingSpace.DoesNotExist:	1✔
185	return Response(	1✔
186	{"message": "ParkingSpace with the specified id does not exist."},
187	status=status.HTTP_400_BAD_REQUEST,
188	)
189
190	except Exception as e:	×
UNCOV 191	return Response(	×
192	{"message": f"An error occurred: {str(e)}"},
193	status=status.HTTP_500_INTERNAL_SERVER_ERROR,
194	)
195
196	def send_mail(	1✔
197	self,
198	subject_template_name,
199	email_template_name,
200	context,
201	from_email,
202	to_email,
203	html_email_template_name=None,
204	):
205	"""
206	Send a django.core.mail.EmailMultiAlternatives to `to_email`.
207	"""
208	subject = loader.render_to_string(subject_template_name, context)	1✔
209	# Email subject must not contain newlines
210	subject = "".join(subject.splitlines())	1✔
211	body = loader.render_to_string(email_template_name, context)	1✔
212
213	email_message = EmailMultiAlternatives(subject, body, from_email, [to_email])	1✔
214	if html_email_template_name is not None:	1✔
NEW 215	html_email = loader.render_to_string(html_email_template_name, context)	×
NEW 216	email_message.attach_alternative(html_email, "text/html")	×
217
218	email_message.send()	1✔
219
220
221	class ParkingSpacePostsAPIView(generics.ListAPIView):	1✔
222	"""GET API endpoint
223	/api/spot/posts/<str:spotId>
224
225	get a list of all the posts associated with a spot
226	"""
227
228	serializer_class = PostSerializer	1✔
229
230	def get_queryset(self):	1✔
231	parking_space_id = self.kwargs["spotId"]	1✔
232	return Post.objects.filter(parking_space__parking_spot_id=parking_space_id)	1✔
233
234
235	class ParkingSpaceAddCommentAPIView(APIView):	1✔
236	"""POST API endpoint
237	/api/spot/posts/add-comment/<int:postId>
238
239	get a list of all the posts associated with a spot
240	"""
241
242	permission_classes = [IsAuthenticated]	1✔
243	authentication_classes = [SessionAuthentication]	1✔
244
245	def post(self, request: HttpRequest, postId: int) -> Response:	1✔
246	profanity.add_censor_words(custom_badwords)	1✔
247	comment_content = profanity.censor(request.data["commentContent"])	1✔
248	if not comment_content:	1✔
249	return Response("Error: empty comment content", 400)	1✔
250	try:	1✔
251	post = Post.objects.get(id=postId)	1✔
252	except Post.DoesNotExist:	1✔
253	return Response(f"No post with post id {postId}", 400)	1✔
254
255	new_comment = Comment(	1✔
256	content=comment_content,
257	author=request.user,
258	post=post,
259	created_at=timezone.now(),
260	)
261	new_comment.save()	1✔
262
263	return Response("Comment created!", 200)	1✔
264
265
266	class AddWatchOnParkingSpaceAPIView(APIView):	1✔
267	model = UserWatchedParkingSpace	1✔
268	permission_classes = [IsAuthenticated]	1✔
269	authentication_classes = [SessionAuthentication]	1✔
270
271	def post(self, request: HttpRequest) -> Response:	1✔
272	user = request.user	1✔
273	try:	1✔
274	parking_space = ParkingSpace.objects.get(	1✔
275	parking_spot_id=request.data["parking_spot_id"]
276	)
277	except ParkingSpace.DoesNotExist:	1✔
278	return Response("Invalid parking spot id", 400)	1✔
279
280	threshold = request.data.get("threshold", 80)	1✔
281	newUserWatchedParkingSpace = UserWatchedParkingSpace(	1✔
282	user=user, parking_space=parking_space, threshold=threshold
283	)
284	newUserWatchedParkingSpace.save()	1✔
285
286	return Response(f"Watch on spot {parking_space.parking_spot_id} added", 200)	1✔
287
288
289	class RemoveWatchOnParkingSpaceAPIView(APIView):	1✔
290	model = UserWatchedParkingSpace	1✔
291	permission_classes = [IsAuthenticated]	1✔
292	authentication_classes = [SessionAuthentication]	1✔
293
294	def post(self, request: HttpRequest) -> Response:	1✔
295	try:	1✔
296	parkingSpace = ParkingSpace.objects.get(	1✔
297	parking_spot_id=request.data["parking_spot_id"]
298	)
299	except ParkingSpace.DoesNotExist:	1✔
300	return Response("Invalid parking spot id", 400)	1✔
301	try:	1✔
302	watchRecord = self.model.objects.get(	1✔
303	user=request.user, parking_space=parkingSpace
304	)
NEW 305	except self.model.DoesNotExist:	×
NEW 306	return Response("No watch found on this parking space", 400)	×
307	watchRecord.delete()	1✔
308
309	return Response(f"Watch on spot {parkingSpace.parking_spot_id} removed", 200)	1✔
310
311
312	class WatchOnParkingSpaceAPIView(APIView):	1✔
313	model = UserWatchedParkingSpace	1✔
314	permission_classes = [IsAuthenticated]	1✔
315	authentication_classes = [SessionAuthentication]	1✔
316	serializer_class = UserWatchedParkingSpaceSerializer	1✔
317
318	def get(self, request: HttpRequest) -> Response:	1✔
NEW 319	return Response(	×
320	self.serializer_class(
321	self.model.objects.filter(user=request.user), many=True
322	).data,
323	status.HTTP_200_OK,
324	)

gcivil-nyu-org / Wednesday-Fall2023-Team-2 / #614972427

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous