diff --git a/lacommunaute/notification/tests/tests_utils.py b/lacommunaute/notification/tests/tests_utils.py index 66c7d0e3d..9202a3fac 100644 --- a/lacommunaute/notification/tests/tests_utils.py +++ b/lacommunaute/notification/tests/tests_utils.py @@ -1,11 +1,8 @@ -from django.conf import settings -from django.contrib.contenttypes.models import ContentType from django.test import TestCase from faker import Faker -from lacommunaute.forum_conversation.factories import AnonymousPostFactory, PostFactory, TopicFactory +from lacommunaute.forum_conversation.factories import PostFactory, TopicFactory from lacommunaute.forum_conversation.models import Post -from lacommunaute.forum_upvote.models import UpVote from lacommunaute.notification.enums import EmailSentTrackKind from lacommunaute.notification.factories import BouncedEmailFactory, EmailSentTrackFactory from lacommunaute.notification.models import EmailSentTrack @@ -13,8 +10,6 @@ collect_first_replies, collect_following_replies, collect_new_users_for_onboarding, - get_emails_to_be_notified, - get_topics_with_following_replies, last_notification, should_not_be_approved, ) @@ -33,67 +28,13 @@ def test_email_sent_track_with_kind(self): self.assertEqual(last_notification(kind="other"), EmailSentTrack.objects.last().created) -class GetEmailsToBeNotifiedTestCase(TestCase): - @classmethod - def setUp(cls): - cls.topic = TopicFactory(with_post=True) - cls.content_type = ContentType.objects.get_for_model(Post) - - def test_authenticated_poster(self): - PostFactory(topic=self.topic) - self.assertIn(self.topic.first_post.poster.email, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_anonymous_poster(self): - anonymous_post = AnonymousPostFactory(topic=self.topic) - - PostFactory(topic=self.topic) - self.assertIn(anonymous_post.username, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_liker(self): - liker = UserFactory() - self.topic.likers.add(liker) - self.topic.save() - - PostFactory(topic=self.topic) - self.assertIn(liker.email, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_upvoter(self): - upvoter = UserFactory() - UpVote.objects.create(content_object=self.topic.first_post, voter=upvoter) - - PostFactory(topic=self.topic) - self.assertIn(upvoter.email, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_excluded_email_of_the_last_poster_when_authenticated(self): - post = PostFactory(topic=self.topic) - self.assertNotIn(post.poster.email, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_excluded_username_of_the_last_poster_when_anonymous(self): - anonymous_post = AnonymousPostFactory(topic=self.topic) - self.assertNotIn(anonymous_post.username, get_emails_to_be_notified(self.topic, self.content_type)) - - def test_deduplication(self): - UpVote.objects.create(content_object=self.topic.first_post, voter=self.topic.poster) - self.topic.likers.add(self.topic.poster) - self.topic.save() - AnonymousPostFactory(topic=self.topic, username=self.topic.poster.email) - - # Email of the last poster is not significant because he is not notified - PostFactory(topic=self.topic) - - self.assertEqual( - get_emails_to_be_notified(self.topic, self.content_type), - [self.topic.poster.email], - ) - - class CollectFirstRepliesTestCase(TestCase): @classmethod - def setUp(cls): + def setUpTestData(cls): cls.topic = TopicFactory(with_post=True) - def test_no_reply_ever(self): - self.assertEqual(len(collect_first_replies()), 0) + def test_no_reply_to_be_notified(self): + self.assertEqual(len(list(collect_first_replies())), 0) def test_first_reply(self): post = PostFactory(topic=self.topic) @@ -101,137 +42,44 @@ def test_first_reply(self): collect_first_replies(), [ ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{post.topic.get_absolute_url()}", - post.topic.subject, - [post.topic.poster_email], - post.poster_display_name, - ) - ], - ) - - def test_first_reply_with_topic_liker(self): - liker = UserFactory() - self.topic.likers.add(liker) - self.topic.save() - - post = PostFactory(topic=self.topic) - self.assertEqual( - collect_first_replies(), - [ - ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{post.topic.get_absolute_url()}", - post.topic.subject, - sorted(list(set([self.topic.poster_email, liker.email]))), - post.poster_display_name, - ) - ], - ) - - def test_no_reply_since_last_notification(self): - PostFactory(topic=self.topic) - EmailSentTrackFactory(kind="first_reply") - - self.assertEqual(len(collect_first_replies()), 0) - - def test_replies_since_last_notification(self): - EmailSentTrackFactory(kind="first_reply") - post = PostFactory(topic=self.topic) - - self.assertEqual( - collect_first_replies(), - [ - ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{post.topic.get_absolute_url()}", - post.topic.subject, - [post.topic.poster_email], + self.topic.get_absolute_url(with_fqdn=True), + self.topic.subject, + [self.topic.poster_email], post.poster_display_name, ) ], ) - def test_unapproved_reply_in_previous_hour(self): - PostFactory(topic=self.topic, approved=False) - self.assertEqual(len(collect_first_replies()), 0) - - def test_last_emailsenttrack_with_kind(self): + def test_first_reply_since_last_notification(self): PostFactory(topic=self.topic) - EmailSentTrackFactory(kind="other") + EmailSentTrackFactory(kind="dummy") self.assertEqual(len(collect_first_replies()), 1) - EmailSentTrackFactory(kind="first_reply") - + EmailSentTrackFactory(kind=EmailSentTrackKind.FIRST_REPLY) self.assertEqual(len(collect_first_replies()), 0) -class GetTopicsWithFollowingRepliesTestCase(TestCase): - @classmethod - def setUp(cls): - cls.topic = TopicFactory(with_post=True) - cls.previous_notification = last_notification(kind=EmailSentTrackKind.FOLLOWING_REPLIES) - - def test_no_reply_ever(self): - self.assertEqual(len(list(get_topics_with_following_replies(self.previous_notification))), 0) - - def test_first_reply(self): - PostFactory(topic=self.topic) - self.assertEqual(len(list(get_topics_with_following_replies(self.previous_notification))), 0) - - def test_following_replies(self): - for i in range(2, 10): - with self.subTest(i=i): - PostFactory.create_batch(i, topic=self.topic) - - self.assertEqual( - list(get_topics_with_following_replies(self.previous_notification)), - [self.topic], - ) - - def test_no_reply_since_last_notification(self): - PostFactory.create_batch(2, topic=self.topic) - EmailSentTrackFactory(kind=EmailSentTrackKind.FOLLOWING_REPLIES) - - self.assertEqual( - len(list(get_topics_with_following_replies(last_notification(kind=EmailSentTrackKind.FOLLOWING_REPLIES)))), - 0, - ) - - def test_replies_since_last_notification(self): - PostFactory(topic=self.topic) - EmailSentTrackFactory(kind=EmailSentTrackKind.FOLLOWING_REPLIES) - PostFactory(topic=self.topic) - - self.assertEqual( - list(get_topics_with_following_replies(last_notification(kind=EmailSentTrackKind.FOLLOWING_REPLIES))), - [self.topic], - ) - - def test_unapproved_replies(self): - PostFactory(topic=self.topic) - - for i in range(2, 10): - with self.subTest(i=i): - PostFactory.create_batch(i, topic=self.topic, approved=False) - self.assertEqual(len(list(get_topics_with_following_replies(last_notification()))), 0) - - class CollectFollowingRepliesTestCase(TestCase): @classmethod - def setUp(cls): + def setUpTestData(cls): cls.topic = TopicFactory(with_post=True) def test_no_reply_to_be_notified(self): self.assertEqual(len(list(collect_following_replies())), 0) def test_one_reply_to_be_notified(self): + # first reply PostFactory(topic=self.topic, poster__email=self.topic.poster_email) EmailSentTrackFactory(kind=EmailSentTrackKind.FOLLOWING_REPLIES) + + # following reply PostFactory(topic=self.topic) self.assertEqual( list(collect_following_replies()), [ ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{self.topic.get_absolute_url()}", + self.topic.get_absolute_url(with_fqdn=True), self.topic.subject, [self.topic.poster_email], "1 nouvelle réponse", @@ -250,7 +98,7 @@ def test_multiple_replies_to_be_notified(self): list(collect_following_replies()), [ ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{self.topic.get_absolute_url()}", + self.topic.get_absolute_url(with_fqdn=True), self.topic.subject, sorted( list( @@ -266,14 +114,13 @@ def test_multiple_replies_to_be_notified(self): ], ) - def test_last_emailsenttrack_with_kind(self): + def test_following_replies_since_last_notification(self): PostFactory.create_batch(2, topic=self.topic) - EmailSentTrackFactory(kind="other") + EmailSentTrackFactory(kind="other") self.assertEqual(len(list(collect_following_replies())), 1) EmailSentTrackFactory(kind=EmailSentTrackKind.FOLLOWING_REPLIES) - self.assertEqual(len(list(collect_following_replies())), 0) diff --git a/lacommunaute/notification/utils.py b/lacommunaute/notification/utils.py index 415c88c25..65d981742 100644 --- a/lacommunaute/notification/utils.py +++ b/lacommunaute/notification/utils.py @@ -1,12 +1,8 @@ from datetime import datetime -from typing import Any, Generator, List -from django.conf import settings -from django.contrib.contenttypes.models import ContentType -from django.db.models import Count, F, Q from django.utils.timezone import now, timedelta -from lacommunaute.forum_conversation.models import Post, Topic +from lacommunaute.forum_conversation.models import Topic from lacommunaute.notification.enums import EmailSentTrackKind from lacommunaute.notification.models import BouncedEmail, EmailSentTrack from lacommunaute.users.models import User @@ -16,68 +12,30 @@ def last_notification(kind=None) -> datetime: return getattr(EmailSentTrack.objects.filter(kind=kind).last(), "created", now() - timedelta(days=5)) -def get_emails_to_be_notified(topic: Topic, content_type: ContentType) -> List[str]: - authenticated_qs = ( - User.objects.filter( - Q(topic_likes=topic) - | Q(upvotes__content_type=content_type, upvotes__object_id__in=topic.posts.values("id")) - | Q(posts__topic=topic) - ) - .distinct() - .values_list("email", flat=True) - ) - anonymous_qs = ( - Post.objects.filter(topic=topic).exclude(username__isnull=True).values_list("username", flat=True) - ) # .values(email=F("username")) - excluded = topic.last_post.username or topic.last_post.poster.email - - return sorted(email for email in authenticated_qs.union(anonymous_qs) if email != excluded) - - def collect_first_replies(): - content_type = ContentType.objects.get_for_model(Post) return [ ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{post.topic.get_absolute_url()}", - post.topic.subject, - get_emails_to_be_notified(post.topic, content_type), - post.poster_display_name, - ) - for post in Post.objects.filter( - created__gte=last_notification(kind=EmailSentTrackKind.FIRST_REPLY), - approved=True, + topic.get_absolute_url(with_fqdn=True), + topic.subject, + topic.mails_to_notify(), + topic.last_post.poster_display_name, ) - if post.position == 2 + for topic in Topic.objects.with_first_reply(last_notification(kind=EmailSentTrackKind.FIRST_REPLY)) ] -def get_topics_with_following_replies(previous_notification: datetime) -> Generator[Topic, Any, None]: - # vincentporte :disapproved post are not counted in posts_count. Filter them out in unuseful - for topic in ( - Topic.objects.filter(updated__gte=previous_notification, posts_count__gte=3) - .annotate( - new_replies=Count( - "posts", - filter=Q(posts__created__gte=previous_notification) & ~Q(posts__id=F("first_post_id")), - ) - ) - .exclude(new_replies=0) - ): - yield topic - - -def collect_following_replies() -> Generator[tuple[str, str, List[str], str], Any, None]: - content_type = ContentType.objects.get_for_model(Post) - - for topic in get_topics_with_following_replies(last_notification(kind=EmailSentTrackKind.FOLLOWING_REPLIES)): - yield ( - f"{settings.COMMU_PROTOCOL}://{settings.COMMU_FQDN}{topic.get_absolute_url()}", +def collect_following_replies(): + return [ + ( + topic.get_absolute_url(with_fqdn=True), topic.subject, - get_emails_to_be_notified(topic, content_type), + topic.mails_to_notify(), f"{topic.new_replies} nouvelle réponse" if topic.new_replies == 1 else f"{topic.new_replies} nouvelles réponses", ) + for topic in Topic.objects.with_following_replies(last_notification(kind=EmailSentTrackKind.FOLLOWING_REPLIES)) + ] def collect_new_users_for_onboarding():