15760443583

Committed 19 Jun 2025 02:37PM UTC coverage: 97.274% (+0.06%) from 97.214%

Build # 15760443583

Build Type

push

github

Committed by

web-flow

Commit Message

Replace thread events with updates (#1944)

Run Details

2828 of 2914 new or added lines in 89 files covered. (97.05%)

27 existing lines in 5 files now uncovered.

74115 of 76192 relevant lines covered (97.27%)

0.97 hits per line

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

95.35

/misago/threads/postsfeed.py

from html import escape
from typing import Iterable

from django.http import HttpRequest
from django.urls import reverse

from ..permissions.checkutils import check_permissions
from ..permissions.privatethreads import (
    check_edit_private_thread_post_permission,
)
from ..permissions.threads import (
    check_edit_thread_post_permission,
)
from ..threadupdates.models import ThreadUpdate
from ..threadupdates.actions import thread_updates_renderer
from .hooks import (
    set_posts_feed_related_objects_hook,
)
from .models import Post, Thread
from .prefetch import prefetch_posts_feed_related_objects


class PostsFeed:
    template_name: str = "misago/posts_feed/index.html"
    post_template_name: str = "misago/posts_feed/post.html"
    thread_update_template_name: str = "misago/posts_feed/thread_update.html"

    request: HttpRequest
    thread: Thread
    posts: list[Post]
    updates: list[ThreadUpdate]

    animate_posts: set[int]
    animate_thread_updates: set[int]

    unread_posts: set[int]

    allow_edit_thread: bool
    is_moderator: bool
    counter_start: int

    def __init__(
        self,
        request: HttpRequest,
        thread: Thread,
        posts: list[Post] | None = None,
        thread_updates: list[ThreadUpdate] | None = None,
    ):
        self.request = request
        self.thread = thread
        self.posts = posts or []
        self.thread_updates = thread_updates or []

        self.animate_posts = set()
        self.animate_thread_updates = set()

        self.unread_posts = set()

        self.allow_edit_thread = False
        self.is_moderator = self.get_moderator_status()
        self.counter_start = 0

    def set_animated_posts(self, ids: Iterable[int]):
        self.animate_posts = set(ids)

    def set_animated_thread_updates(self, ids: Iterable[int]):
        self.animate_thread_updates = set(ids)

    def set_unread_posts(self, ids: Iterable[int]):
        self.unread_posts = set(ids)

    def set_counter_start(self, counter_start: int):
        self.counter_start = counter_start

    def get_moderator_status(self) -> bool:
        return False

    def set_allow_edit_thread(self, allow_edit_thread: bool):
        self.allow_edit_thread = allow_edit_thread

    def get_context_data(self, context: dict | None = None) -> dict:
        context_data = {
            "template_name": self.template_name,
            "items": self.get_feed_data(),
        }

        if context:
            context_data.update(context)

        return context_data

    def get_feed_data(self) -> list[dict]:
        feed: list[dict] = []
        for i, post in enumerate(self.posts):
            feed.append(self.get_post_data(post, i + self.counter_start + 1))
        for update in self.thread_updates:
            feed.append(self.get_thread_update_data(update))

        feed.sort(key=lambda item: item["ordering"])

        previous_item = None
        for item in feed:
            item["previous_item"] = previous_item
            if item["type"] == "post":
                previous_item = f"post-{item['post'].id}"
            elif item["type"] == "thread_update":
                previous_item = f"update-{item['thread_update'].id}"

        related_objects = prefetch_posts_feed_related_objects(
            self.request.settings,
            self.request.user_permissions,
            self.posts,
            categories=[self.thread.category],
            threads=[self.thread],
            thread_updates=self.thread_updates,
        )

        set_posts_feed_related_objects_hook(
            self.set_feed_related_objects, feed, related_objects
        )

        return feed

    def get_post_data(self, post: Post, counter: int = 1) -> dict:
        edit_url: str | None = None

        if self.allow_edit_post(post):
            if post.id == self.thread.first_post_id and self.allow_edit_thread:
                edit_url = self.get_edit_thread_post_url()
            else:
                edit_url = self.get_edit_post_url(post)

        return {
            "template_name": self.post_template_name,
            "animate": post.id in self.animate_posts,
            "type": "post",
            "ordering": post.posted_on,
            "post": post,
            "counter": counter,
            "poster": None,
            "poster_name": post.poster_name,
            "unread": post.id in self.unread_posts,
            "rich_text_data": None,
            "attachments": [],
            "edit_url": edit_url,
            "moderation": self.is_moderator,
        }

    def allow_edit_post(self, post: Post) -> bool:
        return False

    def get_edit_thread_post_url(self) -> str | None:
        return None

    def get_edit_post_url(self, post: Post) -> str | None:
        return None

    def get_thread_update_data(self, thread_update: ThreadUpdate) -> dict:
        hide_url: str | None = None
        unhide_url: str | None = None
        delete_url: str | None = None

        if self.is_moderator:
            if thread_update.is_hidden:
                unhide_url = self.get_unhide_thread_update_url(thread_update)
            else:
                hide_url = self.get_hide_thread_update_url(thread_update)

            delete_url = self.get_delete_thread_update_url(thread_update)

        return {
            "template_name": self.thread_update_template_name,
            "animate": thread_update.id in self.animate_thread_updates,
            "ordering": thread_update.created_at,
            "type": "thread_update",
            "thread_update": thread_update,
            "icon": "",
            "description": "",
            "actor": None,
            "actor_name": thread_update.actor_name,
            "context_object": None,
            "hide_url": hide_url,
            "unhide_url": unhide_url,
            "delete_url": delete_url,
            "moderation": self.is_moderator,
        }

    def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return None

    def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return None

    def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return None

    def set_feed_related_objects(self, feed: list[dict], related_objects: dict) -> None:
        for item in feed:
            if item["type"] == "post":
                self.set_post_related_objects(item, item["post"], related_objects)
            if item["type"] == "thread_update":
                self.set_thread_update_related_objects(
                    item, item["thread_update"], related_objects
                )

    def set_post_related_objects(
        self, item: dict, post: Post, related_objects: dict
    ) -> None:
        item["rich_text_data"] = related_objects

        if post.poster_id:
            item["poster"] = related_objects["users"].get(post.poster_id)

        embedded_attachments = post.metadata.get("attachments", [])
        for attachment in related_objects["attachments"].values():
            if (
                attachment.post_id == post.id
                and attachment.id not in embedded_attachments
            ):
                item["attachments"].append(attachment)

        if item["attachments"]:
            item["attachments"].sort(reverse=True, key=lambda a: a.id)

    def set_thread_update_related_objects(
        self, item: dict, thread_update: ThreadUpdate, related_objects: dict
    ) -> None:
        if thread_update.actor_id:
            item["actor"] = related_objects["users"].get(thread_update.actor_id)

        if thread_update.context_type and thread_update.context_id:
            relation_name = None
            if thread_update.context_type == "misago_attachments.attachment":
                relation_name = "attachment"
            if thread_update.context_type == "misago_categories.category":
                relation_name = "categories"
            if thread_update.context_type == "misago_threads.thread":
                relation_name = "threads"
            if thread_update.context_type == "misago_threads.post":
                relation_name = "posts"
            if thread_update.context_type == "misago_users.user":
                relation_name = "users"

            if relation_name:
                item["context_object"] = related_objects[relation_name].get(
                    thread_update.context_id
                )

        if thread_update_data := thread_updates_renderer.render_thread_update(
            thread_update, related_objects
        ):
            item.update(thread_update_data)
        else:
            item.update(
                {"icon": "broken_image", "description": escape(thread_update.action)}
            )


class ThreadPostsFeed(PostsFeed):
    def get_moderator_status(self) -> bool:
        return self.request.user_permissions.is_category_moderator(
            self.thread.category_id
        )

    def allow_edit_post(self, post: Post) -> bool:
        if self.request.user.is_anonymous:
            return False

        with check_permissions() as can_edit_post:
            check_edit_thread_post_permission(
                self.request.user_permissions, self.thread.category, self.thread, post
            )

        return can_edit_post

    def get_edit_thread_post_url(self) -> str | None:
        return reverse(
            "misago:edit-thread",
            kwargs={"id": self.thread.id, "slug": self.thread.slug},
        )

    def get_edit_post_url(self, post: Post) -> str | None:
        return reverse(
            "misago:edit-thread",
            kwargs={"id": self.thread.id, "slug": self.thread.slug, "post": post.id},
        )

    def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:hide-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )

    def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:unhide-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )

    def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:delete-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )


class PrivateThreadPostsFeed(PostsFeed):
    def get_moderator_status(self) -> bool:
        return self.request.user_permissions.is_private_threads_moderator

    def allow_edit_post(self, post: Post) -> bool:
        with check_permissions() as can_edit_post:
            check_edit_private_thread_post_permission(
                self.request.user_permissions, self.thread, post
            )

        return can_edit_post

    def get_edit_thread_post_url(self) -> str | None:
        return reverse(
            "misago:edit-private-thread",
            kwargs={"id": self.thread.id, "slug": self.thread.slug},
        )

    def get_edit_post_url(self, post: Post) -> str | None:
        return reverse(
            "misago:edit-private-thread",
            kwargs={"id": self.thread.id, "slug": self.thread.slug, "post": post.id},
        )

    def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:hide-private-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )

    def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:unhide-private-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )

    def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str | None:
        return reverse(
            "misago:delete-private-thread-update",
            kwargs={
                "id": self.thread.id,
                "slug": self.thread.slug,
                "thread_update": thread_update.id,
            },
        )

1	from html import escape	1✔
2	from typing import Iterable	1✔
3
4	from django.http import HttpRequest	1✔
5	from django.urls import reverse	1✔
6
7	from ..permissions.checkutils import check_permissions	1✔
8	from ..permissions.privatethreads import (	1✔
9	check_edit_private_thread_post_permission,
10	)
11	from ..permissions.threads import (	1✔
12	check_edit_thread_post_permission,
13	)
14	from ..threadupdates.models import ThreadUpdate	1✔
15	from ..threadupdates.actions import thread_updates_renderer	1✔
16	from .hooks import (	1✔
17	set_posts_feed_related_objects_hook,
18	)
19	from .models import Post, Thread	1✔
20	from .prefetch import prefetch_posts_feed_related_objects	1✔
21
22
23	class PostsFeed:	1✔
24	template_name: str = "misago/posts_feed/index.html"	1✔
25	post_template_name: str = "misago/posts_feed/post.html"	1✔
26	thread_update_template_name: str = "misago/posts_feed/thread_update.html"	1✔
27
28	request: HttpRequest	1✔
29	thread: Thread	1✔
30	posts: list[Post]	1✔
31	updates: list[ThreadUpdate]	1✔
32
33	animate_posts: set[int]	1✔
34	animate_thread_updates: set[int]	1✔
35
36	unread_posts: set[int]	1✔
37
38	allow_edit_thread: bool	1✔
39	is_moderator: bool	1✔
40	counter_start: int	1✔
41
42	def __init__(	1✔
43	self,
44	request: HttpRequest,
45	thread: Thread,
46	posts: list[Post] \| None = None,
47	thread_updates: list[ThreadUpdate] \| None = None,
48	):
49	self.request = request	1✔
50	self.thread = thread	1✔
51	self.posts = posts or []	1✔
52	self.thread_updates = thread_updates or []	1✔
53
54	self.animate_posts = set()	1✔
55	self.animate_thread_updates = set()	1✔
56
57	self.unread_posts = set()	1✔
58
59	self.allow_edit_thread = False	1✔
60	self.is_moderator = self.get_moderator_status()	1✔
61	self.counter_start = 0	1✔
62
63	def set_animated_posts(self, ids: Iterable[int]):	1✔
64	self.animate_posts = set(ids)	1✔
65
66	def set_animated_thread_updates(self, ids: Iterable[int]):	1✔
67	self.animate_thread_updates = set(ids)	1✔
68
69	def set_unread_posts(self, ids: Iterable[int]):	1✔
70	self.unread_posts = set(ids)	1✔
71
72	def set_counter_start(self, counter_start: int):	1✔
73	self.counter_start = counter_start	1✔
74
75	def get_moderator_status(self) -> bool:	1✔
76	return False	1✔
77
78	def set_allow_edit_thread(self, allow_edit_thread: bool):	1✔
79	self.allow_edit_thread = allow_edit_thread	1✔
80
81	def get_context_data(self, context: dict \| None = None) -> dict:	1✔
82	context_data = {	1✔
83	"template_name": self.template_name,
84	"items": self.get_feed_data(),
85	}
86
87	if context:	1✔
88	context_data.update(context)	1✔
89
90	return context_data	1✔
91
92	def get_feed_data(self) -> list[dict]:	1✔
93	feed: list[dict] = []	1✔
94	for i, post in enumerate(self.posts):	1✔
95	feed.append(self.get_post_data(post, i + self.counter_start + 1))	1✔
96	for update in self.thread_updates:	1✔
97	feed.append(self.get_thread_update_data(update))	1✔
98
99	feed.sort(key=lambda item: item["ordering"])	1✔
100
101	previous_item = None	1✔
102	for item in feed:	1✔
103	item["previous_item"] = previous_item	1✔
104	if item["type"] == "post":	1✔
105	previous_item = f"post-{item['post'].id}"	1✔
106	elif item["type"] == "thread_update":	1✔
107	previous_item = f"update-{item['thread_update'].id}"	1✔
108
109	related_objects = prefetch_posts_feed_related_objects(	1✔
110	self.request.settings,
111	self.request.user_permissions,
112	self.posts,
113	categories=[self.thread.category],
114	threads=[self.thread],
115	thread_updates=self.thread_updates,
116	)
117
118	set_posts_feed_related_objects_hook(	1✔
119	self.set_feed_related_objects, feed, related_objects
120	)
121
122	return feed	1✔
123
124	def get_post_data(self, post: Post, counter: int = 1) -> dict:	1✔
125	edit_url: str \| None = None	1✔
126
127	if self.allow_edit_post(post):	1✔
128	if post.id == self.thread.first_post_id and self.allow_edit_thread:	1✔
129	edit_url = self.get_edit_thread_post_url()	1✔
130	else:
131	edit_url = self.get_edit_post_url(post)	1✔
132
133	return {	1✔
134	"template_name": self.post_template_name,
135	"animate": post.id in self.animate_posts,
136	"type": "post",
137	"ordering": post.posted_on,
138	"post": post,
139	"counter": counter,
140	"poster": None,
141	"poster_name": post.poster_name,
142	"unread": post.id in self.unread_posts,
143	"rich_text_data": None,
144	"attachments": [],
145	"edit_url": edit_url,
146	"moderation": self.is_moderator,
147	}
148
149	def allow_edit_post(self, post: Post) -> bool:	1✔
150	return False	1✔
151
152	def get_edit_thread_post_url(self) -> str \| None:	1✔
153	return None	×
154
155	def get_edit_post_url(self, post: Post) -> str \| None:	1✔
156	return None	×
157
158	def get_thread_update_data(self, thread_update: ThreadUpdate) -> dict:	1✔
159	hide_url: str \| None = None	1✔
160	unhide_url: str \| None = None	1✔
161	delete_url: str \| None = None	1✔
162
163	if self.is_moderator:	1✔
164	if thread_update.is_hidden:	1✔
165	unhide_url = self.get_unhide_thread_update_url(thread_update)	1✔
166	else:
167	hide_url = self.get_hide_thread_update_url(thread_update)	1✔
168
169	delete_url = self.get_delete_thread_update_url(thread_update)	1✔
170
171	return {	1✔
172	"template_name": self.thread_update_template_name,
173	"animate": thread_update.id in self.animate_thread_updates,
174	"ordering": thread_update.created_at,
175	"type": "thread_update",
176	"thread_update": thread_update,
177	"icon": "",
178	"description": "",
179	"actor": None,
180	"actor_name": thread_update.actor_name,
181	"context_object": None,
182	"hide_url": hide_url,
183	"unhide_url": unhide_url,
184	"delete_url": delete_url,
185	"moderation": self.is_moderator,
186	}
187
188	def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
NEW 189	return None	×
190
191	def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
NEW 192	return None	×
193
194	def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
NEW 195	return None	×
196
197	def set_feed_related_objects(self, feed: list[dict], related_objects: dict) -> None:	1✔
198	for item in feed:	1✔
199	if item["type"] == "post":	1✔
200	self.set_post_related_objects(item, item["post"], related_objects)	1✔
201	if item["type"] == "thread_update":	1✔
202	self.set_thread_update_related_objects(	1✔
203	item, item["thread_update"], related_objects
204	)
205
206	def set_post_related_objects(	1✔
207	self, item: dict, post: Post, related_objects: dict
208	) -> None:
209	item["rich_text_data"] = related_objects	1✔
210
211	if post.poster_id:	1✔
212	item["poster"] = related_objects["users"].get(post.poster_id)	1✔
213
214	embedded_attachments = post.metadata.get("attachments", [])	1✔
215	for attachment in related_objects["attachments"].values():	1✔
216	if (	1✔
217	attachment.post_id == post.id
218	and attachment.id not in embedded_attachments
219	):
220	item["attachments"].append(attachment)	1✔
221
222	if item["attachments"]:	1✔
223	item["attachments"].sort(reverse=True, key=lambda a: a.id)	1✔
224
225	def set_thread_update_related_objects(	1✔
226	self, item: dict, thread_update: ThreadUpdate, related_objects: dict
227	) -> None:
228	if thread_update.actor_id:	1✔
229	item["actor"] = related_objects["users"].get(thread_update.actor_id)	1✔
230
231	if thread_update.context_type and thread_update.context_id:	1✔
232	relation_name = None	1✔
233	if thread_update.context_type == "misago_attachments.attachment":	1✔
NEW 234	relation_name = "attachment"	×
235	if thread_update.context_type == "misago_categories.category":	1✔
236	relation_name = "categories"	1✔
237	if thread_update.context_type == "misago_threads.thread":	1✔
238	relation_name = "threads"	1✔
239	if thread_update.context_type == "misago_threads.post":	1✔
NEW 240	relation_name = "posts"	×
241	if thread_update.context_type == "misago_users.user":	1✔
242	relation_name = "users"	1✔
243
244	if relation_name:	1✔
245	item["context_object"] = related_objects[relation_name].get(	1✔
246	thread_update.context_id
247	)
248
249	if thread_update_data := thread_updates_renderer.render_thread_update(	1✔
250	thread_update, related_objects
251	):
252	item.update(thread_update_data)	1✔
253	else:
NEW 254	item.update(	×
255	{"icon": "broken_image", "description": escape(thread_update.action)}
256	)
257
258
259	class ThreadPostsFeed(PostsFeed):	1✔
260	def get_moderator_status(self) -> bool:	1✔
261	return self.request.user_permissions.is_category_moderator(	1✔
262	self.thread.category_id
263	)
264
265	def allow_edit_post(self, post: Post) -> bool:	1✔
266	if self.request.user.is_anonymous:	1✔
267	return False	1✔
268
269	with check_permissions() as can_edit_post:	1✔
270	check_edit_thread_post_permission(	1✔
271	self.request.user_permissions, self.thread.category, self.thread, post
272	)
273
274	return can_edit_post	1✔
275
276	def get_edit_thread_post_url(self) -> str \| None:	1✔
277	return reverse(	1✔
278	"misago:edit-thread",
279	kwargs={"id": self.thread.id, "slug": self.thread.slug},
280	)
281
282	def get_edit_post_url(self, post: Post) -> str \| None:	1✔
283	return reverse(	1✔
284	"misago:edit-thread",
285	kwargs={"id": self.thread.id, "slug": self.thread.slug, "post": post.id},
286	)
287
288	def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
289	return reverse(	1✔
290	"misago:hide-thread-update",
291	kwargs={
292	"id": self.thread.id,
293	"slug": self.thread.slug,
294	"thread_update": thread_update.id,
295	},
296	)
297
298	def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
299	return reverse(	1✔
300	"misago:unhide-thread-update",
301	kwargs={
302	"id": self.thread.id,
303	"slug": self.thread.slug,
304	"thread_update": thread_update.id,
305	},
306	)
307
308	def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
309	return reverse(	1✔
310	"misago:delete-thread-update",
311	kwargs={
312	"id": self.thread.id,
313	"slug": self.thread.slug,
314	"thread_update": thread_update.id,
315	},
316	)
317
318
319	class PrivateThreadPostsFeed(PostsFeed):	1✔
320	def get_moderator_status(self) -> bool:	1✔
321	return self.request.user_permissions.is_private_threads_moderator	1✔
322
323	def allow_edit_post(self, post: Post) -> bool:	1✔
324	with check_permissions() as can_edit_post:	1✔
325	check_edit_private_thread_post_permission(	1✔
326	self.request.user_permissions, self.thread, post
327	)
328
329	return can_edit_post	1✔
330
331	def get_edit_thread_post_url(self) -> str \| None:	1✔
332	return reverse(	1✔
333	"misago:edit-private-thread",
334	kwargs={"id": self.thread.id, "slug": self.thread.slug},
335	)
336
337	def get_edit_post_url(self, post: Post) -> str \| None:	1✔
338	return reverse(	1✔
339	"misago:edit-private-thread",
340	kwargs={"id": self.thread.id, "slug": self.thread.slug, "post": post.id},
341	)
342
343	def get_hide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
344	return reverse(	1✔
345	"misago:hide-private-thread-update",
346	kwargs={
347	"id": self.thread.id,
348	"slug": self.thread.slug,
349	"thread_update": thread_update.id,
350	},
351	)
352
353	def get_unhide_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
354	return reverse(	1✔
355	"misago:unhide-private-thread-update",
356	kwargs={
357	"id": self.thread.id,
358	"slug": self.thread.slug,
359	"thread_update": thread_update.id,
360	},
361	)
362
363	def get_delete_thread_update_url(self, thread_update: ThreadUpdate) -> str \| None:	1✔
364	return reverse(	1✔
365	"misago:delete-private-thread-update",
366	kwargs={
367	"id": self.thread.id,
368	"slug": self.thread.slug,
369	"thread_update": thread_update.id,
370	},
371	)

rafalp / Misago / 15760443583

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