import datetime import time import urlparse from bs4 import BeautifulSoup from django.test import TestCase from django.contrib.auth.models import User, Group from group_messaging.models import Message from group_messaging.models import MessageMemo from group_messaging.models import SenderList from group_messaging.models import LastVisitTime from group_messaging.models import get_personal_group from group_messaging.models import create_personal_group from group_messaging.views import ThreadsList from mock import Mock MESSAGE_TEXT = 'test message text' def create_user(name): """creates a user and a personal group, returns the created user""" user = User.objects.create_user(name, name + '@example.com') #note that askbot will take care of three lines below automatically try: group = get_personal_group(user) except Group.DoesNotExist: group = create_personal_group(user) group_name = '_personal_%d' % user.id group, created = Group.objects.get_or_create(name=group_name) user.groups.add(group) return user def get_html_message(mail_message): """mail message is an item from the django.core.mail.outbox""" return mail_message.alternatives[0][0] class GroupMessagingTests(TestCase): """base class for the test cases in this app""" def setUp(self): self.sender = create_user('sender') self.recipient = create_user('recipient') def create_thread(self, sender, recipient_groups): return Message.objects.create_thread( sender=sender, recipients=recipient_groups, text=MESSAGE_TEXT ) def create_thread_for_user(self, sender, recipient): group = get_personal_group(recipient) return self.create_thread(sender, [group]) def setup_three_message_thread(self, original_poster=None, responder=None): """talk in this order: sender, recipient, sender""" original_poster = original_poster or self.sender responder = responder or self.recipient root_message = self.create_thread_for_user(original_poster, responder) response = Message.objects.create_response( sender=responder, text='some response', parent=root_message ) response2 = Message.objects.create_response( sender=original_poster, text='some response2', parent=response ) return root_message, response, response2 class ViewsTests(GroupMessagingTests): def get_view_context(self, view_class, data=None, user=None, method='GET'): spec = ['REQUEST', 'user'] assert(method in ('GET', 'POST')) spec.append(method) request = Mock(spec=spec) request.REQUEST = data setattr(request, method, data) request.user = user return view_class().get_context(request) def test_new_response_marks_thread_heading_as_new(self): root = self.create_thread_for_user(self.sender, self.recipient) response = Message.objects.create_response( sender=self.recipient, text='some response', parent=root ) #response must show as "new" to the self.sender context = self.get_view_context( ThreadsList, data={'sender_id': '-1'}, user=self.sender ) self.assertEqual(context['threads_data'][root.id]['status'], 'new') #"visit" the thread: todo - make a method last_visit_time, created = LastVisitTime.objects.get_or_create( user=self.sender, message=root ) last_visit_time.at = datetime.datetime.now() last_visit_time.save() time.sleep(1.5) #response must show as "seen" context = self.get_view_context( ThreadsList, data={'sender_id': '-1'}, user=self.sender ) self.assertEqual(context['threads_data'][root.id]['status'], 'seen') #self.recipient makes another response response = Message.objects.create_response( sender=self.recipient, text='some response', parent=response ) #thread must be "new" again context = self.get_view_context( ThreadsList, data={'sender_id': '-1'}, user=self.sender ) self.assertEqual(context['threads_data'][root.id]['status'], 'new') def test_answer_to_deleted_thread_undeletes_thread(self): #setup: message, reply, responder deletes thread root_message = self.create_thread_for_user(self.sender, self.recipient) response = Message.objects.create_response( sender=self.recipient, text='some response', parent=root_message ) memo1, created = MessageMemo.objects.get_or_create( message=root_message, user=self.recipient, status=MessageMemo.ARCHIVED ) #OP sends reply to reply response2 = Message.objects.create_response( sender=self.sender, text='some response2', parent=response ) context = self.get_view_context( ThreadsList, data={'sender_id': '-1'}, user=self.recipient ) self.assertEqual(len(context['threads']), 1) thread_id = context['threads'][0].id thread_data = context['threads_data'][thread_id] self.assertEqual(thread_data['status'], 'new') def test_emailed_message_url_works_for_post_recipient(self): root = self.create_thread_for_user(self.sender, self.recipient) from django.core.mail import outbox html_message = get_html_message(outbox[0]) link = BeautifulSoup(html_message).find('a', attrs={'class': 'thread-link'}) url = link['href'].replace('&', '&') parsed_url = urlparse.urlparse(url) url_data = urlparse.parse_qsl(parsed_url.query) self.client.login(user_id=self.recipient.id, method='force') response = self.client.get(parsed_url.path, url_data) dom = BeautifulSoup(response.content) threads = dom.find_all('ul', attrs={'class': 'thread'}) self.assertEquals(len(threads), 1) thread_lists = dom.find_all('table', attrs={'class': 'threads-list'}) self.assertEquals(len(thread_lists), 0) def test_sent_thread_is_visited_by_sender(self): root = self.create_thread_for_user(self.sender, self.recipient) context = self.get_view_context( ThreadsList, data={'sender_id': str(self.sender.id)}, user=self.sender ) thread_data = context['threads_data'][root.id] self.assertEqual(thread_data['status'], 'seen') class ModelsTests(GroupMessagingTests): """test cases for the `private_messaging` models""" def test_create_thread_for_user(self): """the basic create thread with one recipient tests that the recipient is there""" message = self.create_thread_for_user(self.sender, self.recipient) #message type is stored self.assertEqual(message.message_type, Message.STORED) #recipient is in the list of recipients recipients = set(message.recipients.all()) recipient_group = get_personal_group(self.recipient) #sender_group = get_personal_group(self.sender) #maybe add this too expected_recipients = set([recipient_group]) self.assertEqual(recipients, expected_recipients) #self.assertRaises( # MessageMemo.DoesNotExist, # MessageMemo.objects.get, # message=message #) #make sure that the original senders memo to the root #message is marke ad seen memos = MessageMemo.objects.filter( message=message, user=self.sender ) self.assertEquals(memos.count(), 1) self.assertEqual(memos[0].status, MessageMemo.SEEN) def test_get_senders_for_user(self): """this time send thread to a real group test that member of the group has updated the sender list""" group = Group.objects.create(name='somegroup') self.recipient.groups.add(group) message = self.create_thread(self.sender, [group]) senders = SenderList.objects.get_senders_for_user(self.recipient) self.assertEqual(set(senders), set([self.sender])) def test_create_thread_response(self): """create a thread with one response, then load thread for the user test that only the root message is retrieved""" root_message = self.create_thread_for_user(self.sender, self.recipient) response = Message.objects.create_response( sender=self.recipient, text='some response', parent=root_message ) self.assertEqual(response.message_type, Message.STORED) #assert that there is only one "seen" memo for the response memos = MessageMemo.objects.filter(message=response) self.assertEqual(memos.count(), 1) self.assertEqual(memos[0].user, self.recipient) self.assertEqual(memos[0].status, MessageMemo.SEEN) #assert that recipients are the two people who are part of #this conversation recipients = set(response.recipients.all()) sender_group = get_personal_group(self.sender) recipient_group = get_personal_group(self.recipient) expected_recipients = set([sender_group, recipient_group]) self.assertEqual(recipients, expected_recipients) def test_get_threads(self): root_message = self.create_thread_for_user(self.sender, self.recipient) threads = set(Message.objects.get_threads(recipient=self.sender)) self.assertEqual(threads, set([])) threads = set(Message.objects.get_threads(recipient=self.recipient)) self.assertEqual(threads, set([root_message])) response = Message.objects.create_response( sender=self.recipient, text='some response', parent=root_message ) threads = set(Message.objects.get_threads(recipient=self.sender)) self.assertEqual(threads, set([root_message])) threads = set(Message.objects.get_threads(recipient=self.recipient)) self.assertEqual(threads, set([root_message])) def test_deleting_thread_is_user_specific(self): """when one user deletes thread, that same thread should not end up deleted by another user """ root, response, response2 = self.setup_three_message_thread() threads = Message.objects.get_threads(recipient=self.sender) self.assertEquals(threads.count(), 1) threads = Message.objects.get_threads(recipient=self.recipient) self.assertEquals(threads.count(), 1) memo1, created = MessageMemo.objects.get_or_create( message=root, user=self.recipient, status=MessageMemo.ARCHIVED ) threads = Message.objects.get_threads(recipient=self.sender) self.assertEquals(threads.count(), 1) threads = Message.objects.get_threads(recipient=self.recipient) self.assertEquals(threads.count(), 0) threads = Message.objects.get_threads( recipient=self.recipient, deleted=True ) self.assertEquals(threads.count(), 1) def test_user_specific_inboxes(self): self.create_thread_for_user(self.sender, self.recipient) threads = Message.objects.get_threads( recipient=self.recipient, sender=self.sender ) self.assertEqual(threads.count(), 1) threads = Message.objects.get_threads( recipient=self.sender, sender=self.recipient ) self.assertEqual(threads.count(), 0) def test_response_updates_thread_headline(self): root = self.create_thread_for_user(self.sender, self.recipient) response = Message.objects.create_response( sender=self.recipient, text='some response', parent=root ) self.assertEqual(root.headline, 'some response') def test_email_alert_sent(self): root = self.create_thread_for_user(self.sender, self.recipient) from django.core.mail import outbox self.assertEqual(len(outbox), 1) self.assertEqual(len(outbox[0].recipients()), 1) self.assertEqual(outbox[0].recipients()[0], self.recipient.email) html_message = get_html_message(outbox[0]) self.assertTrue(root.text in html_message) soup = BeautifulSoup(html_message) links = soup.find_all('a', attrs={'class': 'thread-link'}) self.assertEqual(len(links), 1) parse_result = urlparse.urlparse(links[0]['href']) query = urlparse.parse_qs(parse_result.query.replace('&', '&')) self.assertEqual(query['thread_id'][0], str(root.id)) def test_get_sent_threads(self): root1, re11, re12 = self.setup_three_message_thread() root2, re21, re22 = self.setup_three_message_thread( original_poster=self.recipient, responder=self.sender ) root3, re31, re32 = self.setup_three_message_thread() #mark root2 as seen root2.mark_as_seen(self.sender) #mark root3 as deleted root3.archive(self.sender) threads = Message.objects.get_sent_threads(sender=self.sender) self.assertEqual(threads.count(), 2) self.assertEqual(set(threads), set([root1, root2]))#root3 is deleted def test_recipient_lists_are_in_senders_info(self): thread = self.create_thread_for_user(self.sender, self.recipient) self.assertTrue(self.recipient.username in thread.senders_info) def test_self_response_not_in_senders_inbox(self): root = self.create_thread_for_user(self.sender, self.recipient) response = Message.objects.create_response( sender=self.sender, text='some response', parent=root ) threads = Message.objects.get_threads(recipient=self.sender) self.assertEqual(threads.count(), 0) def test_sent_message_is_seen_by_the_sender(self): root = self.create_thread_for_user(self.sender, self.recipient) time.sleep(1.5) last_visits = LastVisitTime.objects.filter(message=root, user=self.sender) self.assertEqual(last_visits.count(), 1)