13355009925

Committed 16 Feb 2025 12:31PM UTC coverage: 97.179% (+0.04%) from 97.137%

Build # 13355009925

Build Type

push

github

Committed by

web-flow

Commit Message

Attachments v2 (#1813)

Run Details

8499 of 8679 new or added lines in 191 files covered. (97.93%)

70 existing lines in 14 files now uncovered.

69483 of 71500 relevant lines covered (97.18%)

0.97 hits per line

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

98.69

/misago/threads/api/threadposts.py

from django.core.exceptions import PermissionDenied
from django.db import transaction
from django.utils.translation import pgettext
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

from ...acl.objectacl import add_acl_to_obj
from ...categories import PRIVATE_THREADS_ROOT_NAME, THREADS_ROOT_NAME
from ...core.shortcuts import get_int_or_404
from ...users.online.utils import make_users_status_aware
from ..models import Post
from ..permissions import allow_edit_post, allow_reply_thread
from ..serializers import AttachmentSerializer, PostSerializer
from ..viewmodels import ForumThread, PrivateThread, ThreadPost, ThreadPosts
from .postendpoints.delete import delete_bulk, delete_post
from .postendpoints.edits import get_edit_endpoint, revert_post_endpoint
from .postendpoints.likes import likes_list_endpoint
from .postendpoints.merge import posts_merge_endpoint
from .postendpoints.move import posts_move_endpoint
from .postendpoints.patch_event import event_patch_endpoint
from .postendpoints.patch_post import bulk_patch_endpoint, post_patch_endpoint
from .postendpoints.split import posts_split_endpoint
from .postingendpoint import PostingEndpoint


class ViewSet(viewsets.ViewSet):
    thread = None
    posts = ThreadPosts
    post_ = ThreadPost

    def get_thread(
        self, request, pk, path_aware=False, read_aware=False, watch_aware=False
    ):
        return self.thread(  # pylint: disable=not-callable
            request,
            get_int_or_404(pk),
            path_aware=path_aware,
            read_aware=read_aware,
            watch_aware=watch_aware,
        )

    def get_posts(self, request, thread, page):
        return self.posts(request, thread, page)

    def get_post(self, request, thread, pk):
        return self.post_(request, thread, get_int_or_404(pk))

    def list(self, request, thread_pk):
        page = get_int_or_404(request.query_params.get("page", 0))
        if page == 1:
            page = 0  # api allows explicit first page

        thread = self.get_thread(
            request,
            thread_pk,
            path_aware=True,
            read_aware=True,
            watch_aware=True,
        )
        posts = self.get_posts(request, thread, page)

        data = thread.get_frontend_context()
        data["post_set"] = posts.get_frontend_context()

        return Response(data)

    @action(detail=False, methods=["post"])
    @transaction.atomic
    def merge(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk).unwrap()
        return posts_merge_endpoint(request, thread)

    @action(detail=False, methods=["post"])
    @transaction.atomic
    def move(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk).unwrap()
        return posts_move_endpoint(request, thread, self.thread)

    @action(detail=False, methods=["post"])
    @transaction.atomic
    def split(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk).unwrap()
        return posts_split_endpoint(request, thread)

    @transaction.atomic
    def create(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk).unwrap()
        allow_reply_thread(request.user_acl, thread)

        post = Post(thread=thread, category=thread.category)

        # Put them through posting pipeline
        posting = PostingEndpoint(
            request,
            PostingEndpoint.REPLY,
            tree_name=self.tree_name,
            thread=thread,
            post=post,
        )

        if not posting.is_valid():
            return Response(posting.errors, status=400)

        user_posts = request.user.posts

        posting.save()

        # setup extra data for serialization
        post.is_read = False
        post.is_new = True
        post.poster.posts = user_posts + 1

        make_users_status_aware(request, [post.poster])

        return Response(PostSerializer(post, context={"user": request.user}).data)

    @transaction.atomic
    def update(self, request, thread_pk, pk=None):
        thread = self.get_thread(request, thread_pk).unwrap()
        post = self.get_post(request, thread, pk).unwrap()

        allow_edit_post(request.user_acl, post)

        posting = PostingEndpoint(
            request,
            PostingEndpoint.EDIT,
            tree_name=self.tree_name,
            thread=thread,
            post=post,
        )

        if not posting.is_valid():
            return Response(posting.errors, status=400)

        post_edits = post.edits

        posting.save()

        post.is_read = True
        post.is_new = False
        post.edits = post_edits + 1

        if post.poster:
            make_users_status_aware(request, [post.poster])

        return Response(PostSerializer(post, context={"user": request.user}).data)

    def patch(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk)
        return bulk_patch_endpoint(request, thread.unwrap())

    @transaction.atomic
    def partial_update(self, request, thread_pk, pk):
        thread = self.get_thread(request, thread_pk)
        post = self.get_post(request, thread, pk).unwrap()

        if post.is_event:
            return event_patch_endpoint(request, post)
        return post_patch_endpoint(request, post)

    @transaction.atomic
    def delete(self, request, thread_pk, pk=None):
        thread = self.get_thread(request, thread_pk)

        if pk:
            post = self.get_post(request, thread, pk).unwrap()
            return delete_post(request, thread.unwrap(), post)

        return delete_bulk(request, thread.unwrap())

    @action(detail=True, methods=["get"], url_name="editor")
    def post_editor(self, request, thread_pk, pk=None):
        thread = self.get_thread(request, thread_pk)
        post = self.get_post(request, thread, pk).unwrap()

        allow_edit_post(request.user_acl, post)

        attachments = []
        for attachment in post.attachment_set.order_by("-id"):
            add_acl_to_obj(request.user_acl, attachment)
            attachments.append(attachment)
        attachments_json = AttachmentSerializer(
            attachments, many=True, context={"user": request.user}
        ).data

        return Response(
            {
                "id": post.pk,
                "api": post.get_api_url(),
                "post": post.original,
                "attachments": attachments_json,
                "can_protect": bool(thread.category.acl["can_protect_posts"]),
                "is_protected": post.is_protected,
                "poster": post.poster_name,
            }
        )

    @action(detail=False, methods=["get"], url_name="editor")
    def reply_editor(self, request, thread_pk):
        thread = self.get_thread(request, thread_pk).unwrap()
        allow_reply_thread(request.user_acl, thread)

        if "reply" not in request.query_params:
            return Response({})

        reply_to = self.get_post(
            request, thread, request.query_params["reply"]
        ).unwrap()

        if reply_to.is_event:
            raise PermissionDenied(pgettext("posts api", "Events can't be replied to."))
        if reply_to.is_hidden and not reply_to.acl["can_see_hidden"]:
            raise PermissionDenied(
                pgettext("posts api", "You can't reply to hidden posts.")
            )

        return Response(
            {
                "id": reply_to.pk,
                "post": reply_to.original,
                "poster": reply_to.poster_name,
            }
        )

    @action(detail=True, methods=["get", "post"])
    def edits(self, request, thread_pk, pk=None):
        if request.method == "GET":
            thread = self.get_thread(request, thread_pk)
            post = self.get_post(request, thread, pk).unwrap()

            return get_edit_endpoint(request, post)

        if request.method == "POST":
            with transaction.atomic():
                thread = self.get_thread(request, thread_pk)
                post = self.get_post(request, thread, pk).unwrap()

                allow_edit_post(request.user_acl, post)

                return revert_post_endpoint(request, post)

    @action(detail=True, methods=["get"])
    def likes(self, request, thread_pk, pk=None):
        thread = self.get_thread(request, thread_pk)
        post = self.get_post(request, thread, pk).unwrap()

        if post.acl["can_see_likes"] < 2:
            raise PermissionDenied(
                pgettext("posts api", "You can't see who liked this post.")
            )

        return likes_list_endpoint(request, post)


class ThreadPostsViewSet(ViewSet):
    tree_name = THREADS_ROOT_NAME
    thread = ForumThread


class PrivateThreadPostsViewSet(ViewSet):
    tree_name = PRIVATE_THREADS_ROOT_NAME
    thread = PrivateThread

1	from django.core.exceptions import PermissionDenied	1✔
2	from django.db import transaction	1✔
3	from django.utils.translation import pgettext	1✔
4	from rest_framework import viewsets	1✔
5	from rest_framework.decorators import action	1✔
6	from rest_framework.response import Response	1✔
7
8	from ...acl.objectacl import add_acl_to_obj	1✔
9	from ...categories import PRIVATE_THREADS_ROOT_NAME, THREADS_ROOT_NAME	1✔
10	from ...core.shortcuts import get_int_or_404	1✔
11	from ...users.online.utils import make_users_status_aware	1✔
12	from ..models import Post	1✔
13	from ..permissions import allow_edit_post, allow_reply_thread	1✔
14	from ..serializers import AttachmentSerializer, PostSerializer	1✔
15	from ..viewmodels import ForumThread, PrivateThread, ThreadPost, ThreadPosts	1✔
16	from .postendpoints.delete import delete_bulk, delete_post	1✔
17	from .postendpoints.edits import get_edit_endpoint, revert_post_endpoint	1✔
18	from .postendpoints.likes import likes_list_endpoint	1✔
19	from .postendpoints.merge import posts_merge_endpoint	1✔
20	from .postendpoints.move import posts_move_endpoint	1✔
21	from .postendpoints.patch_event import event_patch_endpoint	1✔
22	from .postendpoints.patch_post import bulk_patch_endpoint, post_patch_endpoint	1✔
23	from .postendpoints.split import posts_split_endpoint	1✔
24	from .postingendpoint import PostingEndpoint	1✔
25
26
27	class ViewSet(viewsets.ViewSet):	1✔
28	thread = None	1✔
29	posts = ThreadPosts	1✔
30	post_ = ThreadPost	1✔
31
32	def get_thread(	1✔
33	self, request, pk, path_aware=False, read_aware=False, watch_aware=False
34	):
35	return self.thread( # pylint: disable=not-callable	1✔
36	request,
37	get_int_or_404(pk),
38	path_aware=path_aware,
39	read_aware=read_aware,
40	watch_aware=watch_aware,
41	)
42
43	def get_posts(self, request, thread, page):	1✔
44	return self.posts(request, thread, page)	1✔
45
46	def get_post(self, request, thread, pk):	1✔
47	return self.post_(request, thread, get_int_or_404(pk))	1✔
48
49	def list(self, request, thread_pk):	1✔
50	page = get_int_or_404(request.query_params.get("page", 0))	1✔
51	if page == 1:	1✔
52	page = 0 # api allows explicit first page	1✔
53
54	thread = self.get_thread(	1✔
55	request,
56	thread_pk,
57	path_aware=True,
58	read_aware=True,
59	watch_aware=True,
60	)
61	posts = self.get_posts(request, thread, page)	1✔
62
63	data = thread.get_frontend_context()	1✔
64	data["post_set"] = posts.get_frontend_context()	1✔
65
66	return Response(data)	1✔
67
68	@action(detail=False, methods=["post"])	1✔
69	@transaction.atomic	1✔
70	def merge(self, request, thread_pk):	1✔
71	thread = self.get_thread(request, thread_pk).unwrap()	1✔
72	return posts_merge_endpoint(request, thread)	1✔
73
74	@action(detail=False, methods=["post"])	1✔
75	@transaction.atomic	1✔
76	def move(self, request, thread_pk):	1✔
77	thread = self.get_thread(request, thread_pk).unwrap()	1✔
78	return posts_move_endpoint(request, thread, self.thread)	1✔
79
80	@action(detail=False, methods=["post"])	1✔
81	@transaction.atomic	1✔
82	def split(self, request, thread_pk):	1✔
83	thread = self.get_thread(request, thread_pk).unwrap()	1✔
84	return posts_split_endpoint(request, thread)	1✔
85
86	@transaction.atomic	1✔
87	def create(self, request, thread_pk):	1✔
88	thread = self.get_thread(request, thread_pk).unwrap()	1✔
89	allow_reply_thread(request.user_acl, thread)	1✔
90
91	post = Post(thread=thread, category=thread.category)	1✔
92
93	# Put them through posting pipeline
94	posting = PostingEndpoint(	1✔
95	request,
96	PostingEndpoint.REPLY,
97	tree_name=self.tree_name,
98	thread=thread,
99	post=post,
100	)
101
102	if not posting.is_valid():	1✔
103	return Response(posting.errors, status=400)	1✔
104
105	user_posts = request.user.posts	1✔
106
107	posting.save()	1✔
108
109	# setup extra data for serialization
110	post.is_read = False	1✔
111	post.is_new = True	1✔
112	post.poster.posts = user_posts + 1	1✔
113
114	make_users_status_aware(request, [post.poster])	1✔
115
116	return Response(PostSerializer(post, context={"user": request.user}).data)	1✔
117
118	@transaction.atomic	1✔
119	def update(self, request, thread_pk, pk=None):	1✔
120	thread = self.get_thread(request, thread_pk).unwrap()	1✔
121	post = self.get_post(request, thread, pk).unwrap()	1✔
122
123	allow_edit_post(request.user_acl, post)	1✔
124
125	posting = PostingEndpoint(	1✔
126	request,
127	PostingEndpoint.EDIT,
128	tree_name=self.tree_name,
129	thread=thread,
130	post=post,
131	)
132
133	if not posting.is_valid():	1✔
134	return Response(posting.errors, status=400)	1✔
135
136	post_edits = post.edits	1✔
137
138	posting.save()	1✔
139
140	post.is_read = True	1✔
141	post.is_new = False	1✔
142	post.edits = post_edits + 1	1✔
143
144	if post.poster:	1✔
145	make_users_status_aware(request, [post.poster])	1✔
146
147	return Response(PostSerializer(post, context={"user": request.user}).data)	1✔
148
149	def patch(self, request, thread_pk):	1✔
150	thread = self.get_thread(request, thread_pk)	1✔
151	return bulk_patch_endpoint(request, thread.unwrap())	1✔
152
153	@transaction.atomic	1✔
154	def partial_update(self, request, thread_pk, pk):	1✔
155	thread = self.get_thread(request, thread_pk)	1✔
156	post = self.get_post(request, thread, pk).unwrap()	1✔
157
158	if post.is_event:	1✔
159	return event_patch_endpoint(request, post)	1✔
160	return post_patch_endpoint(request, post)	1✔
161
162	@transaction.atomic	1✔
163	def delete(self, request, thread_pk, pk=None):	1✔
164	thread = self.get_thread(request, thread_pk)	1✔
165
166	if pk:	1✔
167	post = self.get_post(request, thread, pk).unwrap()	1✔
168	return delete_post(request, thread.unwrap(), post)	1✔
169
170	return delete_bulk(request, thread.unwrap())	1✔
171
172	@action(detail=True, methods=["get"], url_name="editor")	1✔
173	def post_editor(self, request, thread_pk, pk=None):	1✔
174	thread = self.get_thread(request, thread_pk)	1✔
175	post = self.get_post(request, thread, pk).unwrap()	1✔
176
177	allow_edit_post(request.user_acl, post)	1✔
178
179	attachments = []	1✔
180	for attachment in post.attachment_set.order_by("-id"):	1✔
UNCOV 181	add_acl_to_obj(request.user_acl, attachment)	×
UNCOV 182	attachments.append(attachment)	×
183	attachments_json = AttachmentSerializer(	1✔
184	attachments, many=True, context={"user": request.user}
185	).data
186
187	return Response(	1✔
188	{
189	"id": post.pk,
190	"api": post.get_api_url(),
191	"post": post.original,
192	"attachments": attachments_json,
193	"can_protect": bool(thread.category.acl["can_protect_posts"]),
194	"is_protected": post.is_protected,
195	"poster": post.poster_name,
196	}
197	)
198
199	@action(detail=False, methods=["get"], url_name="editor")	1✔
200	def reply_editor(self, request, thread_pk):	1✔
201	thread = self.get_thread(request, thread_pk).unwrap()	1✔
202	allow_reply_thread(request.user_acl, thread)	1✔
203
204	if "reply" not in request.query_params:	1✔
205	return Response({})	1✔
206
207	reply_to = self.get_post(	1✔
208	request, thread, request.query_params["reply"]
209	).unwrap()
210
211	if reply_to.is_event:	1✔
212	raise PermissionDenied(pgettext("posts api", "Events can't be replied to."))	1✔
213	if reply_to.is_hidden and not reply_to.acl["can_see_hidden"]:	1✔
214	raise PermissionDenied(	1✔
215	pgettext("posts api", "You can't reply to hidden posts.")
216	)
217
218	return Response(	1✔
219	{
220	"id": reply_to.pk,
221	"post": reply_to.original,
222	"poster": reply_to.poster_name,
223	}
224	)
225
226	@action(detail=True, methods=["get", "post"])	1✔
227	def edits(self, request, thread_pk, pk=None):	1✔
228	if request.method == "GET":	1✔
229	thread = self.get_thread(request, thread_pk)	1✔
230	post = self.get_post(request, thread, pk).unwrap()	1✔
231
232	return get_edit_endpoint(request, post)	1✔
233
234	if request.method == "POST":	1✔
235	with transaction.atomic():	1✔
236	thread = self.get_thread(request, thread_pk)	1✔
237	post = self.get_post(request, thread, pk).unwrap()	1✔
238
239	allow_edit_post(request.user_acl, post)	1✔
240
241	return revert_post_endpoint(request, post)	1✔
242
243	@action(detail=True, methods=["get"])	1✔
244	def likes(self, request, thread_pk, pk=None):	1✔
245	thread = self.get_thread(request, thread_pk)	1✔
246	post = self.get_post(request, thread, pk).unwrap()	1✔
247
248	if post.acl["can_see_likes"] < 2:	1✔
249	raise PermissionDenied(	1✔
250	pgettext("posts api", "You can't see who liked this post.")
251	)
252
253	return likes_list_endpoint(request, post)	1✔
254
255
256	class ThreadPostsViewSet(ViewSet):	1✔
257	tree_name = THREADS_ROOT_NAME	1✔
258	thread = ForumThread	1✔
259
260
261	class PrivateThreadPostsViewSet(ViewSet):	1✔
262	tree_name = PRIVATE_THREADS_ROOT_NAME	1✔
263	thread = PrivateThread	1✔

rafalp / Misago / 13355009925

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