9235638887

Committed 25 May 2024 12:58PM UTC coverage: 97.61% (-1.1%) from 98.716%

Build # 9235638887

Build Type

Pull #1742

github

Committed by

web-flow

Commit Message

Merge ef9c8656c into abad4f068

Pull Request Pull Request #1742: Replace forum options with account settings

Run Details

166 of 211 new or added lines in 20 files covered. (78.67%)

655 existing lines in 146 files now uncovered.

51625 of 52889 relevant lines covered (97.61%)

0.98 hits per line

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

98.5

/misago/threads/api/threadendpoints/patch.py

from django.contrib.auth import get_user_model
from django.core.exceptions import PermissionDenied, ValidationError
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils.translation import npgettext, pgettext
from rest_framework import serializers
from rest_framework.response import Response

from ....acl import useracl
from ....acl.objectacl import add_acl_to_obj
from ....categories.models import Category
from ....categories.permissions import allow_browse_category, allow_see_category
from ....categories.serializers import CategorySerializer
from ....conf import settings
from ....core.apipatch import ApiPatch
from ....core.shortcuts import get_int_or_404
from ....users.utils import slugify_username
from ...moderation import threads as moderation
from ...participants import (
    add_participant,
    change_owner,
    make_participants_aware,
    remove_participant,
)
from ...permissions import (
    allow_add_participant,
    allow_add_participants,
    allow_approve_thread,
    allow_change_best_answer,
    allow_change_owner,
    allow_edit_thread,
    allow_hide_thread,
    allow_mark_as_best_answer,
    allow_mark_best_answer,
    allow_move_thread,
    allow_pin_thread,
    allow_remove_participant,
    allow_see_post,
    allow_start_thread,
    allow_unhide_thread,
    allow_unmark_best_answer,
)
from ...serializers import ThreadParticipantSerializer
from ...validators import validate_thread_title

User = get_user_model()

thread_patch_dispatcher = ApiPatch()


def patch_acl(request, thread, value):
    """useful little op that updates thread acl to current state"""
    if value:
        add_acl_to_obj(request.user_acl, thread)
        return {"acl": thread.acl}
    return {"acl": None}


thread_patch_dispatcher.add("acl", patch_acl)


def patch_title(request, thread, value):
    try:
        value_cleaned = str(value).strip()
    except (TypeError, ValueError):
        raise PermissionDenied(pgettext("threads api", "Not a valid string."))

    try:
        validate_thread_title(request.settings, value_cleaned)
    except ValidationError as e:
        raise PermissionDenied(e.args[0])

    allow_edit_thread(request.user_acl, thread)

    moderation.change_thread_title(request, thread, value_cleaned)
    return {"title": thread.title}


thread_patch_dispatcher.replace("title", patch_title)


def patch_weight(request, thread, value):
    allow_pin_thread(request.user_acl, thread)

    if not thread.acl.get("can_pin_globally") and thread.weight == 2:
        raise PermissionDenied(
            pgettext(
                "threads api",
                "You can't change globally pinned threads weights in this category.",
            )
        )

    if value == 2:
        if thread.acl.get("can_pin_globally"):
            moderation.pin_thread_globally(request, thread)
        else:
            raise PermissionDenied(
                pgettext(
                    "threads api", "You can't pin threads globally in this category."
                )
            )
    elif value == 1:
        moderation.pin_thread_locally(request, thread)
    elif value == 0:
        moderation.unpin_thread(request, thread)

    return {"weight": thread.weight}


thread_patch_dispatcher.replace("weight", patch_weight)


def patch_move(request, thread, value):
    allow_move_thread(request.user_acl, thread)

    category_pk = get_int_or_404(value)
    new_category = get_object_or_404(
        Category.objects.all_categories().select_related("parent"), pk=category_pk
    )

    add_acl_to_obj(request.user_acl, new_category)
    allow_see_category(request.user_acl, new_category)
    allow_browse_category(request.user_acl, new_category)
    allow_start_thread(request.user_acl, new_category)

    if new_category == thread.category:
        raise PermissionDenied(
            pgettext(
                "threads api", "You can't move thread to the category it's already in."
            )
        )

    moderation.move_thread(request, thread, new_category)

    return {
        "category": CategorySerializer(
            new_category,
            context={
                "settings": request.settings,
            },
        ).data,
    }


thread_patch_dispatcher.replace("category", patch_move)


def patch_flatten_categories(request, thread, value):
    try:
        return {"category": thread.category_id}
    except AttributeError:
        return {"category": thread.category_id}


thread_patch_dispatcher.replace("flatten-categories", patch_flatten_categories)


def patch_is_unapproved(request, thread, value):
    allow_approve_thread(request.user_acl, thread)

    if value:
        raise PermissionDenied(
            pgettext("threads api", "Content approval can't be reversed.")
        )

    moderation.approve_thread(request, thread)

    return {
        "is_unapproved": thread.is_unapproved,
        "has_unapproved_posts": thread.has_unapproved_posts,
    }


thread_patch_dispatcher.replace("is-unapproved", patch_is_unapproved)


def patch_is_closed(request, thread, value):
    if thread.acl.get("can_close"):
        if value:
            moderation.close_thread(request, thread)
        else:
            moderation.open_thread(request, thread)

        return {"is_closed": thread.is_closed}
    else:
        if value:
            raise PermissionDenied(
                pgettext(
                    "threads api", "You don't have permission to close this thread."
                )
            )
        else:
            raise PermissionDenied(
                pgettext(
                    "threads api", "You don't have permission to open this thread."
                )
            )


thread_patch_dispatcher.replace("is-closed", patch_is_closed)


def patch_is_hidden(request, thread, value):
    if value:
        allow_hide_thread(request.user_acl, thread)
        moderation.hide_thread(request, thread)
    else:
        allow_unhide_thread(request.user_acl, thread)
        moderation.unhide_thread(request, thread)

    return {"is_hidden": thread.is_hidden}


thread_patch_dispatcher.replace("is-hidden", patch_is_hidden)


def patch_best_answer(request, thread, value):
    try:
        post_id = int(value)
    except (TypeError, ValueError):
        raise PermissionDenied(pgettext("threads api", "A valid integer is required."))

    allow_mark_best_answer(request.user_acl, thread)

    post = get_object_or_404(thread.post_set, id=post_id)
    post.category = thread.category
    post.thread = thread

    allow_see_post(request.user_acl, post)
    allow_mark_as_best_answer(request.user_acl, post)

    if post.is_best_answer:
        raise PermissionDenied(
            pgettext(
                "threads api", "This post is already marked as thread's best answer."
            )
        )

    if thread.has_best_answer:
        allow_change_best_answer(request.user_acl, thread)

    thread.set_best_answer(request.user, post)
    thread.save()

    return {
        "best_answer": thread.best_answer_id,
        "best_answer_is_protected": thread.best_answer_is_protected,
        "best_answer_marked_on": thread.best_answer_marked_on,
        "best_answer_marked_by": thread.best_answer_marked_by_id,
        "best_answer_marked_by_name": thread.best_answer_marked_by_name,
        "best_answer_marked_by_slug": thread.best_answer_marked_by_slug,
    }


thread_patch_dispatcher.replace("best-answer", patch_best_answer)


def patch_unmark_best_answer(request, thread, value):
    try:
        post_id = int(value)
    except (TypeError, ValueError):
        raise PermissionDenied(pgettext("threads api", "A valid integer is required."))

    post = get_object_or_404(thread.post_set, id=post_id)
    post.category = thread.category
    post.thread = thread

    if not post.is_best_answer:
        raise PermissionDenied(
            pgettext(
                "threads api",
                "This post can't be unmarked because it's not currently marked as best answer.",
            )
        )

    allow_unmark_best_answer(request.user_acl, thread)
    thread.clear_best_answer()
    thread.save()

    return {
        "best_answer": None,
        "best_answer_is_protected": False,
        "best_answer_marked_on": None,
        "best_answer_marked_by": None,
        "best_answer_marked_by_name": None,
        "best_answer_marked_by_slug": None,
    }


thread_patch_dispatcher.remove("best-answer", patch_unmark_best_answer)


def patch_add_participant(request, thread, value: str):
    allow_add_participants(request.user_acl, thread)

    try:
        user_slug = slugify_username(value)
        if not user_slug:
            raise PermissionDenied(
                pgettext("threads api", "You have to enter new participant's username.")
            )
        participant = User.objects.get(slug=user_slug)
    except User.DoesNotExist:
        raise PermissionDenied(
            pgettext("threads api", "No user with this name exists.")
        )

    if participant in [p.user for p in thread.participants_list]:
        raise PermissionDenied(
            pgettext("threads api", "This user is already thread participant.")
        )

    participant_acl = useracl.get_user_acl(participant, request.cache_versions)
    allow_add_participant(request.user_acl, participant, participant_acl)
    add_participant(request, thread, participant)

    make_participants_aware(request.user, thread)
    participants = ThreadParticipantSerializer(thread.participants_list, many=True)

    return {"participants": participants.data}


thread_patch_dispatcher.add("participants", patch_add_participant)


def patch_remove_participant(request, thread, value):
    # pylint: disable=undefined-loop-variable
    try:
        user_id = int(value)
    except (ValueError, TypeError):
        raise PermissionDenied(pgettext("threads api", "A valid integer is required."))

    for participant in thread.participants_list:
        if participant.user_id == user_id:
            break
    else:
        raise PermissionDenied(pgettext("threads api", "Participant doesn't exist."))

    allow_remove_participant(request.user_acl, thread, participant.user)
    remove_participant(request, thread, participant.user)

    if len(thread.participants_list) == 1:
        return {"deleted": True}

    make_participants_aware(request.user, thread)
    participants = ThreadParticipantSerializer(thread.participants_list, many=True)

    return {"deleted": False, "participants": participants.data}


thread_patch_dispatcher.remove("participants", patch_remove_participant)


def patch_replace_owner(request, thread, value):
    # pylint: disable=undefined-loop-variable
    try:
        user_id = int(value)
    except (ValueError, TypeError):
        raise PermissionDenied(pgettext("threads api", "A valid integer is required."))

    for participant in thread.participants_list:
        if participant.user_id == user_id:
            if participant.is_owner:
                raise PermissionDenied(
                    pgettext("threads api", "This user already is thread owner.")
                )
            else:
                break
    else:
        raise PermissionDenied(pgettext("threads api", "Participant doesn't exist."))

    allow_change_owner(request.user_acl, thread)
    change_owner(request, thread, participant.user)

    make_participants_aware(request.user, thread)
    participants = ThreadParticipantSerializer(thread.participants_list, many=True)
    return {"participants": participants.data}


thread_patch_dispatcher.replace("owner", patch_replace_owner)


def thread_patch_endpoint(request, thread):
    old_title = thread.title
    old_is_hidden = thread.is_hidden
    old_is_unapproved = thread.is_unapproved
    old_category = thread.category

    response = thread_patch_dispatcher.dispatch(request, thread)

    # diff thread's state against pre-patch and resync category if necessary
    hidden_changed = old_is_hidden != thread.is_hidden
    unapproved_changed = old_is_unapproved != thread.is_unapproved
    category_changed = old_category != thread.category

    title_changed = old_title != thread.title
    if thread.category.last_thread_id != thread.pk:
        title_changed = False  # don't trigger resync on simple title change

    if hidden_changed or unapproved_changed or category_changed:
        thread.category.synchronize()
        thread.category.save()

        if category_changed:
            old_category.synchronize()
            old_category.save()
    elif title_changed:
        thread.category.last_thread_title = thread.title
        thread.category.last_thread_slug = thread.slug
        thread.category.save(update_fields=["last_thread_title", "last_thread_slug"])

    return response


def bulk_patch_endpoint(
    request, viewmodel
):  # pylint: disable=too-many-branches, too-many-locals
    serializer = BulkPatchSerializer(
        data=request.data, context={"settings": request.settings}
    )
    if not serializer.is_valid():
        return Response(serializer.errors, status=400)

    threads = clean_threads_for_patch(request, viewmodel, serializer.data["ids"])

    old_titles = [t.title for t in threads]
    old_is_hidden = [t.is_hidden for t in threads]
    old_is_unapproved = [t.is_unapproved for t in threads]
    old_category = [t.category_id for t in threads]

    response = thread_patch_dispatcher.dispatch_bulk(request, threads)

    new_titles = [t.title for t in threads]
    new_is_hidden = [t.is_hidden for t in threads]
    new_is_unapproved = [t.is_unapproved for t in threads]
    new_category = [t.category_id for t in threads]

    # sync titles
    if new_titles != old_titles:
        for i, t in enumerate(threads):
            if t.title != old_titles[i] and t.category.last_thread_id == t.pk:
                t.category.last_thread_title = t.title
                t.category.last_thread_slug = t.slug
                t.category.save(update_fields=["last_thread_title", "last_thread_slug"])

    # sync categories
    sync_categories = []

    if new_is_hidden != old_is_hidden:
        for i, t in enumerate(threads):
            if t.is_hidden != old_is_hidden[i] and t.category_id not in sync_categories:
                sync_categories.append(t.category_id)

    if new_is_unapproved != old_is_unapproved:
        for i, t in enumerate(threads):
            if (
                t.is_unapproved != old_is_unapproved[i]
                and t.category_id not in sync_categories
            ):
                sync_categories.append(t.category_id)

    if new_category != old_category:
        for i, t in enumerate(threads):
            if t.category_id != old_category[i]:
                if t.category_id not in sync_categories:
                    sync_categories.append(t.category_id)
                if old_category[i] not in sync_categories:
                    sync_categories.append(old_category[i])

    if sync_categories:
        for category in Category.objects.filter(id__in=sync_categories):
            category.synchronize()
            category.save()

    return response


def clean_threads_for_patch(request, viewmodel, threads_ids):
    threads = []
    for thread_id in sorted(set(threads_ids), reverse=True):
        try:
            threads.append(viewmodel(request, thread_id).unwrap())
        except (Http404, PermissionDenied):
            raise PermissionDenied(
                pgettext(
                    "threads api", "One or more threads to update could not be found."
                )
            )
    return threads


class BulkPatchSerializer(serializers.Serializer):
    ids = serializers.ListField(
        child=serializers.IntegerField(min_value=1), min_length=1
    )
    ops = serializers.ListField(
        child=serializers.DictField(), min_length=1, max_length=10
    )

    def validate_ids(self, data):
        limit = self.context["settings"].threads_per_page
        if len(data) > limit:
            message = npgettext(
                "threads api",
                "No more than %(limit)s thread can be updated at a single time.",
                "No more than %(limit)s threads can be updated at a single time.",
                limit,
            )
            raise ValidationError(message % {"limit": limit})
        return data

1	from django.contrib.auth import get_user_model	1✔
2	from django.core.exceptions import PermissionDenied, ValidationError	1✔
3	from django.http import Http404	1✔
4	from django.shortcuts import get_object_or_404	1✔
5	from django.utils.translation import npgettext, pgettext	1✔
6	from rest_framework import serializers	1✔
7	from rest_framework.response import Response	1✔
8
9	from ....acl import useracl	1✔
10	from ....acl.objectacl import add_acl_to_obj	1✔
11	from ....categories.models import Category	1✔
12	from ....categories.permissions import allow_browse_category, allow_see_category	1✔
13	from ....categories.serializers import CategorySerializer	1✔
14	from ....conf import settings	1✔
15	from ....core.apipatch import ApiPatch	1✔
16	from ....core.shortcuts import get_int_or_404	1✔
17	from ....users.utils import slugify_username	1✔
18	from ...moderation import threads as moderation	1✔
19	from ...participants import (	1✔
20	add_participant,
21	change_owner,
22	make_participants_aware,
23	remove_participant,
24	)
25	from ...permissions import (	1✔
26	allow_add_participant,
27	allow_add_participants,
28	allow_approve_thread,
29	allow_change_best_answer,
30	allow_change_owner,
31	allow_edit_thread,
32	allow_hide_thread,
33	allow_mark_as_best_answer,
34	allow_mark_best_answer,
35	allow_move_thread,
36	allow_pin_thread,
37	allow_remove_participant,
38	allow_see_post,
39	allow_start_thread,
40	allow_unhide_thread,
41	allow_unmark_best_answer,
42	)
43	from ...serializers import ThreadParticipantSerializer	1✔
44	from ...validators import validate_thread_title	1✔
45
46	User = get_user_model()	1✔
47
48	thread_patch_dispatcher = ApiPatch()	1✔
49
50
51	def patch_acl(request, thread, value):	1✔
52	"""useful little op that updates thread acl to current state"""
53	if value:	1✔
54	add_acl_to_obj(request.user_acl, thread)	1✔
55	return {"acl": thread.acl}	1✔
56	return {"acl": None}	1✔
57
58
59	thread_patch_dispatcher.add("acl", patch_acl)	1✔
60
61
62	def patch_title(request, thread, value):	1✔
63	try:	1✔
64	value_cleaned = str(value).strip()	1✔
UNCOV 65	except (TypeError, ValueError):	×
UNCOV 66	raise PermissionDenied(pgettext("threads api", "Not a valid string."))	×
67
68	try:	1✔
69	validate_thread_title(request.settings, value_cleaned)	1✔
70	except ValidationError as e:	1✔
71	raise PermissionDenied(e.args[0])	1✔
72
73	allow_edit_thread(request.user_acl, thread)	1✔
74
75	moderation.change_thread_title(request, thread, value_cleaned)	1✔
76	return {"title": thread.title}	1✔
77
78
79	thread_patch_dispatcher.replace("title", patch_title)	1✔
80
81
82	def patch_weight(request, thread, value):	1✔
83	allow_pin_thread(request.user_acl, thread)	1✔
84
85	if not thread.acl.get("can_pin_globally") and thread.weight == 2:	1✔
86	raise PermissionDenied(	1✔
87	pgettext(
88	"threads api",
89	"You can't change globally pinned threads weights in this category.",
90	)
91	)
92
93	if value == 2:	1✔
94	if thread.acl.get("can_pin_globally"):	1✔
95	moderation.pin_thread_globally(request, thread)	1✔
96	else:
97	raise PermissionDenied(	1✔
98	pgettext(
99	"threads api", "You can't pin threads globally in this category."
100	)
101	)
102	elif value == 1:	1✔
103	moderation.pin_thread_locally(request, thread)	1✔
104	elif value == 0:	1✔
105	moderation.unpin_thread(request, thread)	1✔
106
107	return {"weight": thread.weight}	1✔
108
109
110	thread_patch_dispatcher.replace("weight", patch_weight)	1✔
111
112
113	def patch_move(request, thread, value):	1✔
114	allow_move_thread(request.user_acl, thread)	1✔
115
116	category_pk = get_int_or_404(value)	1✔
117	new_category = get_object_or_404(	1✔
118	Category.objects.all_categories().select_related("parent"), pk=category_pk
119	)
120
121	add_acl_to_obj(request.user_acl, new_category)	1✔
122	allow_see_category(request.user_acl, new_category)	1✔
123	allow_browse_category(request.user_acl, new_category)	1✔
124	allow_start_thread(request.user_acl, new_category)	1✔
125
126	if new_category == thread.category:	1✔
127	raise PermissionDenied(	1✔
128	pgettext(
129	"threads api", "You can't move thread to the category it's already in."
130	)
131	)
132
133	moderation.move_thread(request, thread, new_category)	1✔
134
135	return {	1✔
136	"category": CategorySerializer(
137	new_category,
138	context={
139	"settings": request.settings,
140	},
141	).data,
142	}
143
144
145	thread_patch_dispatcher.replace("category", patch_move)	1✔
146
147
148	def patch_flatten_categories(request, thread, value):	1✔
149	try:	1✔
150	return {"category": thread.category_id}	1✔
UNCOV 151	except AttributeError:	×
UNCOV 152	return {"category": thread.category_id}	×
153
154
155	thread_patch_dispatcher.replace("flatten-categories", patch_flatten_categories)	1✔
156
157
158	def patch_is_unapproved(request, thread, value):	1✔
159	allow_approve_thread(request.user_acl, thread)	1✔
160
161	if value:	1✔
162	raise PermissionDenied(	1✔
163	pgettext("threads api", "Content approval can't be reversed.")
164	)
165
166	moderation.approve_thread(request, thread)	1✔
167
168	return {	1✔
169	"is_unapproved": thread.is_unapproved,
170	"has_unapproved_posts": thread.has_unapproved_posts,
171	}
172
173
174	thread_patch_dispatcher.replace("is-unapproved", patch_is_unapproved)	1✔
175
176
177	def patch_is_closed(request, thread, value):	1✔
178	if thread.acl.get("can_close"):	1✔
179	if value:	1✔
180	moderation.close_thread(request, thread)	1✔
181	else:
182	moderation.open_thread(request, thread)	1✔
183
184	return {"is_closed": thread.is_closed}	1✔
185	else:
186	if value:	1✔
187	raise PermissionDenied(	1✔
188	pgettext(
189	"threads api", "You don't have permission to close this thread."
190	)
191	)
192	else:
193	raise PermissionDenied(	1✔
194	pgettext(
195	"threads api", "You don't have permission to open this thread."
196	)
197	)
198
199
200	thread_patch_dispatcher.replace("is-closed", patch_is_closed)	1✔
201
202
203	def patch_is_hidden(request, thread, value):	1✔
204	if value:	1✔
205	allow_hide_thread(request.user_acl, thread)	1✔
206	moderation.hide_thread(request, thread)	1✔
207	else:
208	allow_unhide_thread(request.user_acl, thread)	1✔
209	moderation.unhide_thread(request, thread)	1✔
210
211	return {"is_hidden": thread.is_hidden}	1✔
212
213
214	thread_patch_dispatcher.replace("is-hidden", patch_is_hidden)	1✔
215
216
217	def patch_best_answer(request, thread, value):	1✔
218	try:	1✔
219	post_id = int(value)	1✔
220	except (TypeError, ValueError):	1✔
221	raise PermissionDenied(pgettext("threads api", "A valid integer is required."))	1✔
222
223	allow_mark_best_answer(request.user_acl, thread)	1✔
224
225	post = get_object_or_404(thread.post_set, id=post_id)	1✔
226	post.category = thread.category	1✔
227	post.thread = thread	1✔
228
229	allow_see_post(request.user_acl, post)	1✔
230	allow_mark_as_best_answer(request.user_acl, post)	1✔
231
232	if post.is_best_answer:	1✔
233	raise PermissionDenied(	1✔
234	pgettext(
235	"threads api", "This post is already marked as thread's best answer."
236	)
237	)
238
239	if thread.has_best_answer:	1✔
240	allow_change_best_answer(request.user_acl, thread)	1✔
241
242	thread.set_best_answer(request.user, post)	1✔
243	thread.save()	1✔
244
245	return {	1✔
246	"best_answer": thread.best_answer_id,
247	"best_answer_is_protected": thread.best_answer_is_protected,
248	"best_answer_marked_on": thread.best_answer_marked_on,
249	"best_answer_marked_by": thread.best_answer_marked_by_id,
250	"best_answer_marked_by_name": thread.best_answer_marked_by_name,
251	"best_answer_marked_by_slug": thread.best_answer_marked_by_slug,
252	}
253
254
255	thread_patch_dispatcher.replace("best-answer", patch_best_answer)	1✔
256
257
258	def patch_unmark_best_answer(request, thread, value):	1✔
259	try:	1✔
260	post_id = int(value)	1✔
261	except (TypeError, ValueError):	1✔
262	raise PermissionDenied(pgettext("threads api", "A valid integer is required."))	1✔
263
264	post = get_object_or_404(thread.post_set, id=post_id)	1✔
265	post.category = thread.category	1✔
266	post.thread = thread	1✔
267
268	if not post.is_best_answer:	1✔
269	raise PermissionDenied(	1✔
270	pgettext(
271	"threads api",
272	"This post can't be unmarked because it's not currently marked as best answer.",
273	)
274	)
275
276	allow_unmark_best_answer(request.user_acl, thread)	1✔
277	thread.clear_best_answer()	1✔
278	thread.save()	1✔
279
280	return {	1✔
281	"best_answer": None,
282	"best_answer_is_protected": False,
283	"best_answer_marked_on": None,
284	"best_answer_marked_by": None,
285	"best_answer_marked_by_name": None,
286	"best_answer_marked_by_slug": None,
287	}
288
289
290	thread_patch_dispatcher.remove("best-answer", patch_unmark_best_answer)	1✔
291
292
293	def patch_add_participant(request, thread, value: str):	1✔
294	allow_add_participants(request.user_acl, thread)	1✔
295
296	try:	1✔
297	user_slug = slugify_username(value)	1✔
298	if not user_slug:	1✔
299	raise PermissionDenied(	1✔
300	pgettext("threads api", "You have to enter new participant's username.")
301	)
302	participant = User.objects.get(slug=user_slug)	1✔
303	except User.DoesNotExist:	1✔
304	raise PermissionDenied(	1✔
305	pgettext("threads api", "No user with this name exists.")
306	)
307
308	if participant in [p.user for p in thread.participants_list]:	1✔
309	raise PermissionDenied(	1✔
310	pgettext("threads api", "This user is already thread participant.")
311	)
312
313	participant_acl = useracl.get_user_acl(participant, request.cache_versions)	1✔
314	allow_add_participant(request.user_acl, participant, participant_acl)	1✔
315	add_participant(request, thread, participant)	1✔
316
317	make_participants_aware(request.user, thread)	1✔
318	participants = ThreadParticipantSerializer(thread.participants_list, many=True)	1✔
319
320	return {"participants": participants.data}	1✔
321
322
323	thread_patch_dispatcher.add("participants", patch_add_participant)	1✔
324
325
326	def patch_remove_participant(request, thread, value):	1✔
327	# pylint: disable=undefined-loop-variable
328	try:	1✔
329	user_id = int(value)	1✔
330	except (ValueError, TypeError):	1✔
331	raise PermissionDenied(pgettext("threads api", "A valid integer is required."))	1✔
332
333	for participant in thread.participants_list:	1✔
334	if participant.user_id == user_id:	1✔
335	break	1✔
336	else:
337	raise PermissionDenied(pgettext("threads api", "Participant doesn't exist."))	1✔
338
339	allow_remove_participant(request.user_acl, thread, participant.user)	1✔
340	remove_participant(request, thread, participant.user)	1✔
341
342	if len(thread.participants_list) == 1:	1✔
343	return {"deleted": True}	1✔
344
345	make_participants_aware(request.user, thread)	1✔
346	participants = ThreadParticipantSerializer(thread.participants_list, many=True)	1✔
347
348	return {"deleted": False, "participants": participants.data}	1✔
349
350
351	thread_patch_dispatcher.remove("participants", patch_remove_participant)	1✔
352
353
354	def patch_replace_owner(request, thread, value):	1✔
355	# pylint: disable=undefined-loop-variable
356	try:	1✔
357	user_id = int(value)	1✔
358	except (ValueError, TypeError):	1✔
359	raise PermissionDenied(pgettext("threads api", "A valid integer is required."))	1✔
360
361	for participant in thread.participants_list:	1✔
362	if participant.user_id == user_id:	1✔
363	if participant.is_owner:	1✔
364	raise PermissionDenied(	1✔
365	pgettext("threads api", "This user already is thread owner.")
366	)
367	else:
368	break	1✔
369	else:
370	raise PermissionDenied(pgettext("threads api", "Participant doesn't exist."))	1✔
371
372	allow_change_owner(request.user_acl, thread)	1✔
373	change_owner(request, thread, participant.user)	1✔
374
375	make_participants_aware(request.user, thread)	1✔
376	participants = ThreadParticipantSerializer(thread.participants_list, many=True)	1✔
377	return {"participants": participants.data}	1✔
378
379
380	thread_patch_dispatcher.replace("owner", patch_replace_owner)	1✔
381
382
383	def thread_patch_endpoint(request, thread):	1✔
384	old_title = thread.title	1✔
385	old_is_hidden = thread.is_hidden	1✔
386	old_is_unapproved = thread.is_unapproved	1✔
387	old_category = thread.category	1✔
388
389	response = thread_patch_dispatcher.dispatch(request, thread)	1✔
390
391	# diff thread's state against pre-patch and resync category if necessary
392	hidden_changed = old_is_hidden != thread.is_hidden	1✔
393	unapproved_changed = old_is_unapproved != thread.is_unapproved	1✔
394	category_changed = old_category != thread.category	1✔
395
396	title_changed = old_title != thread.title	1✔
397	if thread.category.last_thread_id != thread.pk:	1✔
398	title_changed = False # don't trigger resync on simple title change	1✔
399
400	if hidden_changed or unapproved_changed or category_changed:	1✔
401	thread.category.synchronize()	1✔
402	thread.category.save()	1✔
403
404	if category_changed:	1✔
405	old_category.synchronize()	1✔
406	old_category.save()	1✔
407	elif title_changed:	1✔
408	thread.category.last_thread_title = thread.title	1✔
409	thread.category.last_thread_slug = thread.slug	1✔
410	thread.category.save(update_fields=["last_thread_title", "last_thread_slug"])	1✔
411
412	return response	1✔
413
414
415	def bulk_patch_endpoint(	1✔
416	request, viewmodel
417	): # pylint: disable=too-many-branches, too-many-locals
418	serializer = BulkPatchSerializer(	1✔
419	data=request.data, context={"settings": request.settings}
420	)
421	if not serializer.is_valid():	1✔
422	return Response(serializer.errors, status=400)	1✔
423
424	threads = clean_threads_for_patch(request, viewmodel, serializer.data["ids"])	1✔
425
426	old_titles = [t.title for t in threads]	1✔
427	old_is_hidden = [t.is_hidden for t in threads]	1✔
428	old_is_unapproved = [t.is_unapproved for t in threads]	1✔
429	old_category = [t.category_id for t in threads]	1✔
430
431	response = thread_patch_dispatcher.dispatch_bulk(request, threads)	1✔
432
433	new_titles = [t.title for t in threads]	1✔
434	new_is_hidden = [t.is_hidden for t in threads]	1✔
435	new_is_unapproved = [t.is_unapproved for t in threads]	1✔
436	new_category = [t.category_id for t in threads]	1✔
437
438	# sync titles
439	if new_titles != old_titles:	1✔
440	for i, t in enumerate(threads):	1✔
441	if t.title != old_titles[i] and t.category.last_thread_id == t.pk:	1✔
442	t.category.last_thread_title = t.title	1✔
443	t.category.last_thread_slug = t.slug	1✔
444	t.category.save(update_fields=["last_thread_title", "last_thread_slug"])	1✔
445
446	# sync categories
447	sync_categories = []	1✔
448
449	if new_is_hidden != old_is_hidden:	1✔
450	for i, t in enumerate(threads):	1✔
451	if t.is_hidden != old_is_hidden[i] and t.category_id not in sync_categories:	1✔
452	sync_categories.append(t.category_id)	1✔
453
454	if new_is_unapproved != old_is_unapproved:	1✔
455	for i, t in enumerate(threads):	1✔
456	if (	1✔
457	t.is_unapproved != old_is_unapproved[i]
458	and t.category_id not in sync_categories
459	):
460	sync_categories.append(t.category_id)	1✔
461
462	if new_category != old_category:	1✔
463	for i, t in enumerate(threads):	1✔
464	if t.category_id != old_category[i]:	1✔
465	if t.category_id not in sync_categories:	1✔
466	sync_categories.append(t.category_id)	1✔
467	if old_category[i] not in sync_categories:	1✔
468	sync_categories.append(old_category[i])	1✔
469
470	if sync_categories:	1✔
471	for category in Category.objects.filter(id__in=sync_categories):	1✔
472	category.synchronize()	1✔
473	category.save()	1✔
474
475	return response	1✔
476
477
478	def clean_threads_for_patch(request, viewmodel, threads_ids):	1✔
479	threads = []	1✔
480	for thread_id in sorted(set(threads_ids), reverse=True):	1✔
481	try:	1✔
482	threads.append(viewmodel(request, thread_id).unwrap())	1✔
483	except (Http404, PermissionDenied):	1✔
484	raise PermissionDenied(	1✔
485	pgettext(
486	"threads api", "One or more threads to update could not be found."
487	)
488	)
489	return threads	1✔
490
491
492	class BulkPatchSerializer(serializers.Serializer):	1✔
493	ids = serializers.ListField(	1✔
494	child=serializers.IntegerField(min_value=1), min_length=1
495	)
496	ops = serializers.ListField(	1✔
497	child=serializers.DictField(), min_length=1, max_length=10
498	)
499
500	def validate_ids(self, data):	1✔
501	limit = self.context["settings"].threads_per_page	1✔
502	if len(data) > limit:	1✔
503	message = npgettext(	1✔
504	"threads api",
505	"No more than %(limit)s thread can be updated at a single time.",
506	"No more than %(limit)s threads can be updated at a single time.",
507	limit,
508	)
509	raise ValidationError(message % {"limit": limit})	1✔
510	return data	1✔

rafalp / Misago / 9235638887

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